fix(webui): bind destructive dataflow controls to help

README.md

@@ -137,6 +137,13 @@ records the effective Policy decision and ancestor limits. Inherited
 definitions remain read-only; lower scopes may narrow, but not broaden,
 execution, reuse, inheritance, or automation permissions.
 
+Derived definitions report when their source has a newer immutable revision;
+the source never mutates the child silently. Adopting an update requires the
+reviewed source revision and hash plus a reason. It appends a new child
+revision, retains the previous graph and all run evidence, records reviewer
+and Policy provenance, and returns the child to draft before the changed graph
+can run or receive automation.
+
 Complete active flows support explicit user/API starts, administrative
 backfills, one-time schedules, interval schedules, and exact-match platform
 events. Trigger deliveries are durable and idempotent. They enqueue the same
@@ -147,11 +154,15 @@ the run before source access or output publication.
 Confidential and restricted events are not accepted through the direct
 ingress; those require Core's transactional event bridge.
 
-Reusable subflow nodes pin a template reference, version, graph snapshot, and
-parameter values. Their single input is bound to an explicitly marked inline
-source inside the snapshot, parameter substitution is data-only, and nesting
-is bounded. This keeps completed run definitions reproducible even when the
-source template changes later.
+Reusable subflow nodes select a Policy-authorized complete flow or template and
+an immutable revision. The server resolves the graph instead of accepting a
+caller-supplied snapshot, records the source hash and Policy decision, and pins
+closed typed input/output contracts. Their single input is bound to an
+explicitly marked typed inline source inside the snapshot, parameter
+substitution is data-only, and cycles across nested references are rejected.
+Incompatible caller schemas fail validation before execution. This keeps
+completed run definitions reproducible even when the source definition changes
+later.
 
 The executable fixtures in `fixtures/golden` cover monthly structured-file
 reconciliation, sanctions screening, a HEICO-style current-status export, and

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-dataflow"
-version = "0.1.19"
+version = "0.1.23"
 description = "Governed graphical and SQL data pipelines for GovOPlaN."
 readme = "README.md"
 requires-python = ">=3.12"

src/govoplan_dataflow/__init__.py

@@ -1,3 +1,3 @@
 from __future__ import annotations
 
-__version__ = "0.1.19"
+__version__ = "0.1.23"

src/govoplan_dataflow/backend/german_documentation.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from typing import Iterable
+
+from govoplan_core.core.modules import DocumentationTopic
+
+
+_TRANSLATIONS = {
+    "dataflow.data-subject-requests": {
+        "title": "Datenschutzanfragen zu Dataflow-Daten",
+        "summary": "Gespeicherte Transformationsdetails minimieren, ohne abgeleitete Datenflüsse als führende Betroffenendaten zu behandeln.",
+        "body": (
+            "Dataflow gleicht exakte mandantenbezogene Pipeline-, Revisions-, Abgleich-, Lauf-, Bereitstellungs-, Trigger- und Zustellkennungen sowie minimierte Konto-, Identitäts- und Mitgliedschaftszuordnungen ab. Ergebnisse enthalten niemals Graphen, SQL, Anfrage- oder Ereignisinhalte, Abgleichkorrekturen, Autorisierungssnapshots, Provenienztexte, Fehler, Quelldetails, Hashwerte, Zugangsdaten oder Ausgabezeilen. Dataflow durchsucht beliebige Transformationsinhalte nicht nach Personen; das führende Eingabemodul muss Betroffenendaten auffinden und berichtigen. Eindeutig bestimmte abgeschlossene Lauf- und Zustelldetails können idempotent minimiert und personenbezogene Automatisierungsbefugnisse deaktiviert und widerrufen werden. Definitionen, Abgleichnachweise, aktive Arbeiten, Bereitstellungen, umfassende Pipeline-Pakete, veröffentlichte Datasource-Ausgaben und institutionelle Zuordnungen benötigen eine autorisierte Prüfung oder Aufbewahrung. Nach der Prüfung sind die Quellen zu berichtigen und Ableitungen in Datasources, Search und Reporting zu aktualisieren."
+        ),
+    },
+    "dataflow.module-boundary": {
+        "title": "Modulgrenze von Dataflow",
+        "summary": "Versionierte tabellarische Transformationen grafisch oder mit eingeschränktem SQL bearbeiten.",
+        "body": (
+            "Dataflow verantwortet kanonische Pipeline-Graphen, unveränderliche Revisionen, Validierung, eingeschränkte SQL-Kompilierung, Vorschau- und Laufdiagnosen sowie Herkunftsreferenzen. Datasources verantwortet den gesteuerten Katalog und Materialisierungen, Connectors den externen Abruf und Zugangsdaten, Reporting die analytische Darstellung und Exporte, Workflow die Orchestrierung und menschliche Übergaben und Risk Compliance die Sanktionsprüfung und Richtliniengrenzen. Benutzer-SQL wird in freigegebene Transformationen übersetzt und nie ungeprüft an eine Datenbank weitergereicht."
+        ),
+    },
+    "dataflow.reference.nodes-and-expressions": {
+        "title": "Dataflow-Knoten und Ausdrücke",
+        "summary": "Typisierte Knoteneingaben, Ausdrücke, Schemafortschreibung und begrenzte Zwischenergebnisse verstehen.",
+        "body": (
+            "Jeder Graphknoten definiert typisierte Eingaben, Konfiguration, Ausgabeschema und Validierungsregeln. Quellknoten binden Inline-Inhalte oder gesteuerte Datasource-Referenzen; Verknüpfungs-, Filter-, Transformations-, Qualitäts-, Abgleich-, Teilfluss- und Ausgabeknoten bleiben im kanonischen Graphen ausdrücklich sichtbar. Wiederverwendbare Teilflüsse wählen eine durch Policy erlaubte unveränderliche Fluss- oder Vorlagenrevision. Der Server löst Graph, Quell-Hash, Policy-Entscheidung und geschlossene Ein-/Ausgabeverträge auf und bindet sie; mitgelieferte Graphkopien werden ignoriert, unvereinbare Eingaben und zyklische Referenzen abgelehnt. Abgleichzeilen führen stabile Schlüssel- und Eingabe-Hashes sowie Vorher-/Nachher-Werte. Prüfentscheidungen werden als unveränderliche, mandanteneigene Entscheidungssätze gespeichert; geänderte Eingaben werden ungültig, ohne Fachdaten still umzuschreiben. Ausdrücke führen weder Host- noch Datenbankcode aus. Knoten-Vorschauen sind begrenzt, für die handelnde Person datenschutzgefiltert und werden nicht als Laufergebnis gespeichert. SQL wird in denselben Graphen kompiliert; nicht unterstützte Anweisungen erscheinen als Diagnose."
+        ),
+    },
+    "dataflow.reference.fields-and-consequences": {
+        "title": "Dataflow-Felder und Lebenszyklusfolgen",
+        "summary": "Bedeutung von Geltungsbereich, Revision, Wiederverwendung, Automatisierung, Ausführung, Veröffentlichung, Promotion und Löschung.",
+        "body": (
+            "Der Geltungsbereich bestimmt Eigentum und Policy-Vererbung. Vorlagen können abgeleitet, aber nicht ausgeführt werden; vollständige Flüsse dürfen bei wirksamer Policy geprüft, versioniert, automatisiert und ausgeführt werden. Speichern fügt eine unveränderliche Revision an. Eine bereichsbezogene Kopie bindet Quellrevision und Inhalts-Hash; eine neuere Quelle ändert sie nicht automatisch. Die geprüfte Übernahme benötigt den exakten Quell-Hash und eine Begründung, fügt eine Kopierevision an, protokolliert Policy-Entscheidung und prüfende Person und setzt die Kopie zur erneuten Aktivierung auf Entwurf. Trigger binden Revision und Autorisierungsnachweis und prüfen ihre Befugnis je Zustellung neu. „Läufe zulassen“ ist nur die Zulassungsgrenze der Definition und gewährt niemandem eine Berechtigung. Einmalige Läufe verwenden lokale Mandantenzeit; versäumte Intervalle werden je Richtlinie zu einem Lauf zusammengefasst oder übersprungen, niemals ungeprüft vollständig nachgeholt. Das Parallelitätslimit begrenzt aktive Zustellungen und erhöht keine Worker-Kapazität. Läufe erzeugen dauerhafte Befehls- und Recovery-Nachweise. Veröffentlichung erstellt eine gesteuerte Datasource-Materialisierung; Promotion wählt eine unveränderliche Revision für Staging oder Produktion aus. Abgleichentscheidungen werden mit optimistischer Nebenläufigkeit als neue Revision gespeichert und verändern die geprüfte Fachzeile nicht. Löschen verhindert künftige Nutzung, während Lauf-, Bereitstellungs-, Herkunfts-, Audit- und Recovery-Nachweise ihrer Aufbewahrung folgen."
+        ),
+    },
+    "dataflow.execution-and-recovery": {
+        "title": "Dataflow ausführen, veröffentlichen und wiederherstellen",
+        "summary": "Gebundene Läufe, Umgebungsfreigaben, Ausgabeveröffentlichung, Abbruch, Abgleich und Nachweise betreiben.",
+        "body": (
+            "Jeder Lauf ist an eine unveränderliche Revision und einen Idempotenzschlüssel gebunden. Die Warteschlange erfasst handelnde Person, Befugnis, Umgebung, Fortschritt, Abbruch, Ausgabe und Recovery-Zustand. Reine Datenbankläufe werden atomar gespeichert. Die Veröffentlichung in eine gesteuerte Datasource nutzt Vorwärts-Recovery: Ein unbekanntes Anbieterergebnis wird vor einer Wiederholung abgeglichen, damit keine Ausgabe doppelt entsteht. Staging- und Produktionsfreigaben sind ausdrücklich und schreiben keine Revision um. Das Einfrieren einer Veröffentlichung versieht exakt die unveränderliche Ausgabe mit einer dauerhaften Bezeichnung; Daten werden weder kopiert noch von Aufbewahrungs-, Hold- und Zugriffsregeln der Datasource getrennt. Artefaktbasierte und Inline-Ausgaben liefern dieselben stabilen Veröffentlichungs-, Datasource- und Materialisierungsreferenzen. Warnungen und prüfpflichtige Zustände bleiben für Workflow sichtbar. Ein Abbruch ist nach Beginn externer Arbeit nur bestmöglich; der Abschlussnachweis unterscheidet gestoppt, abgeschlossen, fehlgeschlagen und abgleichpflichtig. Vor Annahme eines geplanten, ereignisbasierten oder eingereihten Laufs wird die Modulberechtigung des Mandanten geprüft. Eine Deaktivierung stoppt neue Annahmen und überlässt bereits angenommene Läufe einer ausdrücklichen Betriebsentscheidung. Reporting darf nur einen erfolgreichen veröffentlichten Lauf binden; Dataflow prüft Befugnis und Datasource-Zugriff erneut und liest exakt die protokollierte Materialisierung ohne erneute Parametrisierung oder Ausführung."
+        ),
+    },
+}
+
+
+def localize_documentation_topics(
+    topics: Iterable[DocumentationTopic],
+) -> tuple[DocumentationTopic, ...]:
+    localized: list[DocumentationTopic] = []
+    for topic in topics:
+        german = _TRANSLATIONS.get(topic.id)
+        if german is None:
+            localized.append(topic)
+            continue
+        translations = {
+            locale: dict(value) for locale, value in topic.translations.items()
+        }
+        translations["de"] = {**translations.get("de", {}), **german}
+        localized.append(replace(topic, translations=translations))
+    return tuple(localized)

src/govoplan_dataflow/backend/german_structured_documentation.py

@@ -0,0 +1,95 @@
+"""German translations for public structured documentation metadata."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+GERMAN_STRUCTURED_TRANSLATIONS: dict[str, dict[str, Any]] = {'dataflow.execution-and-recovery': {'consequence_classes': {'cancel_run': 'Stornierung von '
+                                                                           'Anfragen; bereits '
+                                                                           'anerkannte externe '
+                                                                           'Effekte können '
+                                                                           'verbleiben.',
+                                                             'promote_revision': 'Macht eine '
+                                                                                 'unveränderliche '
+                                                                                 'Revision in '
+                                                                                 'einer höheren '
+                                                                                 'Ausführungsumgebung '
+                                                                                 'geeignet.',
+                                                             'publish_output': 'Erstellt oder '
+                                                                               'aktualisiert eine '
+                                                                               'geregelte '
+                                                                               'Datenquelle und '
+                                                                               'fügt eine '
+                                                                               'Materialisierung '
+                                                                               'hinzu.',
+                                                             'queue_run': 'Erstellt einen '
+                                                                          'dauerhaften asynchronen '
+                                                                          'Befehls- und '
+                                                                          'Autorisierungsnachweis.'}},
+ 'dataflow.reference.fields-and-consequences': {'consequence_classes': {'configure_trigger': 'Erstellt '
+                                                                                             'oder '
+                                                                                             'ändert '
+                                                                                             'einen '
+                                                                                             'Automatisierungsbefehl '
+                                                                                             'mit '
+                                                                                             'Revisions- '
+                                                                                             'und '
+                                                                                             'Autorisierungsnachweisen.',
+                                                                        'delete_pipeline': 'Verhindert '
+                                                                                           'die '
+                                                                                           'zukünftige '
+                                                                                           'Verwendung, '
+                                                                                           'während '
+                                                                                           'beibehaltene '
+                                                                                           'Nachweise '
+                                                                                           'geregelt '
+                                                                                           'bleiben.',
+                                                                        'derive_copy': 'Erstellt '
+                                                                                       'eine '
+                                                                                       'separat '
+                                                                                       'verwaltete '
+                                                                                       'Kopie, die '
+                                                                                       'an die '
+                                                                                       'Quellrevision '
+                                                                                       'und den '
+                                                                                       'Hash '
+                                                                                       'gebunden '
+                                                                                       'ist.',
+                                                                        'rebase_copy': 'Hängt die '
+                                                                                       'genaue '
+                                                                                       'überprüfte '
+                                                                                       'Quellrevision '
+                                                                                       'an eine '
+                                                                                       'Scope-Kopie '
+                                                                                       'an, '
+                                                                                       'zeichnet '
+                                                                                       'die '
+                                                                                       'Herkunft '
+                                                                                       'des '
+                                                                                       'Reviewers '
+                                                                                       'auf und '
+                                                                                       'gibt sie '
+                                                                                       'in den '
+                                                                                       'Entwurf '
+                                                                                       'zurück.',
+                                                                        'record_decision': 'Fügt '
+                                                                                           'eine '
+                                                                                           'vom '
+                                                                                           'handelnde '
+                                                                                           'Person '
+                                                                                           'zugewiesene '
+                                                                                           'unveränderliche '
+                                                                                           'Entscheidungsrevision '
+                                                                                           'gegen '
+                                                                                           'einen '
+                                                                                           'genauen '
+                                                                                           'Eingabe-Hash '
+                                                                                           'an.',
+                                                                        'save_revision': 'Fügt '
+                                                                                         'eine '
+                                                                                         'unveränderliche '
+                                                                                         'Überarbeitung '
+                                                                                         'der '
+                                                                                         'Pipelinedefinition '
+                                                                                         'an.'}}}

src/govoplan_dataflow/backend/governance.py

@@ -121,6 +121,7 @@ def definition_governance_payload(
     *,
     principal: ApiPrincipal,
     registry: object | None,
+    source_update: Mapping[str, object] | None = None,
 ) -> dict[str, object]:
     actions = {
         action: definition_decision(
@@ -142,6 +143,25 @@ def definition_governance_payload(
         "derived_from_pipeline_id": pipeline.derived_from_pipeline_id,
         "derived_from_revision": pipeline.derived_from_revision,
         "derived_from_hash": pipeline.derived_from_hash,
+        "source_available": bool(
+            source_update and source_update.get("source_available") is True
+        ),
+        "source_name": (
+            source_update.get("source_name") if source_update else None
+        ),
+        "source_current_revision": (
+            source_update.get("source_current_revision")
+            if source_update
+            else None
+        ),
+        "source_current_hash": (
+            source_update.get("source_current_hash")
+            if source_update
+            else None
+        ),
+        "update_available": bool(
+            source_update and source_update.get("update_available") is True
+        ),
         "derivation_provenance": dict(pipeline.derivation_provenance),
         "actions": actions,
     }

src/govoplan_dataflow/backend/manifest.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+from govoplan_core.core.modules import with_documentation_structured_translations
+from govoplan_dataflow.backend.german_structured_documentation import GERMAN_STRUCTURED_TRANSLATIONS
+
 from pathlib import Path
 
 from govoplan_core.core.module_guards import (
@@ -18,6 +21,7 @@ from govoplan_core.core.dataflows import (
 )
 from govoplan_core.core.modules import (
     CapabilityDocumentation,
+    DocumentationCondition,
     DocumentationTopic,
     FrontendModule,
     FrontendRoute,
@@ -52,11 +56,14 @@ from govoplan_dataflow.backend.dsar_provider import (
     DATAFLOW_DSAR_CAPABILITY,
     DataflowDsarProvider,
 )
+from govoplan_dataflow.backend.german_documentation import (
+    localize_documentation_topics,
+)
 
 
 MODULE_ID = "dataflow"
 MODULE_NAME = "Dataflow"
-MODULE_VERSION = "0.1.19"
+MODULE_VERSION = "0.1.23"
 
 READ_SCOPE = "dataflow:pipeline:read"
 WRITE_SCOPE = "dataflow:pipeline:write"
@@ -146,7 +153,7 @@ ROLE_TEMPLATES = (
     ),
 )
 
-DOCUMENTATION = (
+DOCUMENTATION = localize_documentation_topics((
     DocumentationTopic(
         id="dataflow.data-subject-requests",
         title="Dataflow data-subject requests",
@@ -197,6 +204,19 @@ DOCUMENTATION = (
             "audit",
         ),
         metadata={
+            "kind": "workflow",
+            "route": "/dataflow",
+            "screen": "Dataflow",
+            "prerequisites": [
+                "You may read Dataflow pipeline definitions in the active tenant.",
+            ],
+            "steps": [
+                "Open Dataflow and select a pipeline or create an authorized draft.",
+                "Inspect the graph, immutable revision, diagnostics, and declared source references.",
+                "Use preview or run actions only when the effective permissions and Policy allow them.",
+            ],
+            "outcome": "The pipeline remains a governed transformation definition with explicit module boundaries and source authority.",
+            "verification": "Confirm the active revision, validation diagnostics, source references, and permitted actions in the Dataflow workspace.",
             "first_slice": (
                 "Inline and governed datasources, union, join, filter, deduplication, select, "
                 "typed expressions, conversion, quality and reconciliation, reusable subflows, "
@@ -218,6 +238,27 @@ DOCUMENTATION = (
                 "dataflow.state.read-only",
             ],
         },
+        conditions=(
+            DocumentationCondition(
+                required_modules=("dataflow",),
+                required_scopes=(READ_SCOPE,),
+            ),
+        ),
+        structured_translation_version="1",
+        structured_translations={
+            "de": {
+                "prerequisites": [
+                    "Sie dürfen Dataflow-Pipeline-Definitionen im aktiven Mandanten lesen.",
+                ],
+                "steps": [
+                    "Öffnen Sie Dataflow und wählen Sie eine Pipeline oder legen Sie einen autorisierten Entwurf an.",
+                    "Prüfen Sie Graph, unveränderliche Revision, Diagnosen und ausgewiesene Quellreferenzen.",
+                    "Verwenden Sie Vorschau- oder Laufaktionen nur, wenn wirksame Berechtigungen und Policy sie erlauben.",
+                ],
+                "outcome": "Die Pipeline bleibt eine gesteuerte Transformationsdefinition mit ausdrücklichen Modulgrenzen und Quellenautorität.",
+                "verification": "Prüfen Sie aktive Revision, Validierungsdiagnosen, Quellreferenzen und erlaubte Aktionen im Dataflow-Arbeitsbereich.",
+            }
+        },
     ),
     DocumentationTopic(
         id="dataflow.reference.nodes-and-expressions",
@@ -227,6 +268,9 @@ DOCUMENTATION = (
             "Every graph node declares typed inputs, configuration, output schema, and validation rules. "
             "Source nodes pin inline content or governed Datasource references; combine, filter, transform, "
             "quality, reconciliation, reusable-subflow, and output nodes remain explicit in the canonical graph. "
+            "Reusable subflows select a Policy-authorized immutable flow or template revision. The server resolves "
+            "and pins its graph, source hash, Policy decision, and closed typed input/output contracts; caller-supplied "
+            "graph snapshots are ignored, incompatible inputs fail validation, and nested reference cycles are rejected. "
             "Reconciliation rows expose stable key hashes, explicit before/after values, and input hashes. The "
             "review dialog records accept, reject, correct, or defer decisions in tenant-owned immutable decision "
             "sets. Their current projection is a fingerprinted Dataflow source; every superseded revision retains "
@@ -244,11 +288,15 @@ DOCUMENTATION = (
         order=76,
         related_modules=("datasources", "connectors", "policy", "audit"),
         metadata={
+            "kind": "reference",
             "help_contexts": [
                 "dataflow.field.node-name",
                 "dataflow.field.source",
                 "dataflow.field.expression",
                 "dataflow.field.schema",
+                "dataflow.field.reusable-input-binding",
+                "dataflow.field.subflow-reference",
+                "dataflow.field.subflow-revision",
                 "dataflow.action.preview-node",
                 "dataflow.action.review-decisions",
             ],
@@ -261,7 +309,10 @@ DOCUMENTATION = (
         body=(
             "Scope determines ownership and Policy inheritance. Templates can be derived but not run; complete "
             "flows may be previewed, revisioned, automated, and executed when effective Policy allows it. Saving "
-            "appends an immutable revision. A scoped copy pins its source revision and content hash. Triggers pin "
+            "appends an immutable revision. A scoped copy pins its source revision and content hash. A newer "
+            "source revision is reported without changing the copy. Adopting it requires an exact reviewed source "
+            "hash and a reason, appends an immutable copy revision, records the Policy decision and reviewer, and "
+            "returns the copy to draft so runs and automation cannot use the changed graph before activation. Triggers pin "
             "the revision and authorization grant, then re-evaluate authority for every delivery. Allow runs is "
             "the definition-level admission boundary and does not grant a caller permission. A one-time run uses "
             "the configured tenant-local date and time. The missed-run policy either coalesces elapsed interval "
@@ -287,6 +338,7 @@ DOCUMENTATION = (
             "audit",
         ),
         metadata={
+            "kind": "reference",
             "help_contexts": [
                 "dataflow.field.scope",
                 "dataflow.field.definition-kind",
@@ -294,8 +346,10 @@ DOCUMENTATION = (
                 "dataflow.field.trigger-run-at",
                 "dataflow.field.trigger-missed-runs",
                 "dataflow.field.trigger-concurrency",
+                "dataflow.field.rebase-reason",
                 "dataflow.action.save",
                 "dataflow.action.derive",
+                "dataflow.action.rebase",
                 "dataflow.action.trigger",
                 "dataflow.action.record-decision",
                 "dataflow.action.delete",
@@ -303,6 +357,7 @@ DOCUMENTATION = (
             "consequence_classes": {
                 "save_revision": "Appends an immutable pipeline definition revision.",
                 "derive_copy": "Creates a separately governed copy pinned to the source revision and hash.",
+                "rebase_copy": "Appends the exact reviewed source revision to a scoped copy, records reviewer provenance, and returns it to draft.",
                 "configure_trigger": "Creates or changes an automation command with revision and authorization evidence.",
                 "record_decision": "Appends an actor-attributed immutable decision revision against an exact input hash.",
                 "delete_pipeline": "Prevents future use while retained evidence remains governed.",
@@ -351,7 +406,7 @@ DOCUMENTATION = (
             },
         },
     ),
-)
+))
 
 
 def _dataflow_router(context: ModuleContext):
@@ -747,6 +802,11 @@ manifest = ModuleManifest(
 )
 
 
+manifest = with_documentation_structured_translations(
+    manifest, locale="de", translations=GERMAN_STRUCTURED_TRANSLATIONS
+)
+
+
 def get_manifest() -> ModuleManifest:
     return manifest
 

src/govoplan_dataflow/backend/node_library.py

@@ -613,14 +613,13 @@ _NODE_TYPES = (
         type="subflow",
         category="transform",
         label="Reusable subflow",
-        description="Run a pinned parameterized template snapshot as one node.",
+        description="Run a Policy-authorized, server-resolved immutable definition revision as one node.",
         icon="boxes",
         input_ports=(NodePortDefinition(id="input", label="Input"),),
         config_fields=(
             NodeConfigField(id="template_ref", label="Template reference", kind="text", required=True),
             NodeConfigField(id="template_version", label="Template version", kind="text", required=True),
             NodeConfigField(id="parameters", label="Parameters", kind="json", required=True),
-            NodeConfigField(id="graph", label="Pinned graph", kind="json", required=True),
         ),
         default_config={
             "template_ref": "",

src/govoplan_dataflow/backend/router.py

@@ -68,6 +68,7 @@ from govoplan_dataflow.backend.schemas import (
     PipelineRunMetricsResponse,
     PipelineRunResponse,
     PipelinePromotionRequest,
+    PipelineRebaseRequest,
     PipelineResponse,
     PipelineSqlResponse,
     PipelineUpdateRequest,
@@ -109,6 +110,7 @@ from govoplan_dataflow.backend.service import (
     pipeline_run_response,
     preview_pipeline,
     promote_pipeline,
+    rebase_pipeline,
     render_graph_sql,
     start_pipeline_run,
     update_pipeline,
@@ -544,6 +546,8 @@ def api_create_pipeline(
             tenant_id=tenant_id or principal.tenant_id,
             actor_id=_actor_id(principal),
             payload=payload,
+            principal=principal,
+            registry=get_registry(),
         )
     except (PermissionError, ValueError) as exc:
         raise _governance_http_error(exc) from exc
@@ -842,6 +846,8 @@ def api_update_pipeline(
             pipeline_id=pipeline_id,
             actor_id=_actor_id(principal),
             payload=payload,
+            principal=principal,
+            registry=get_registry(),
         )
     except (PermissionError, ValueError) as exc:
         raise _governance_http_error(exc) from exc
@@ -968,6 +974,64 @@ def api_derive_pipeline(
     return response
 
 
+@router.post(
+    "/pipelines/{pipeline_id}/rebase",
+    response_model=PipelineResponse,
+)
+def api_rebase_pipeline(
+    pipeline_id: str,
+    payload: PipelineRebaseRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> PipelineResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        existing = get_pipeline(
+            session,
+            tenant_id=principal.tenant_id,
+            pipeline_id=pipeline_id,
+        )
+        require_definition_action(
+            existing,
+            principal=principal,
+            registry=get_registry(),
+            action="edit",
+        )
+        pipeline = rebase_pipeline(
+            session,
+            tenant_id=principal.tenant_id,
+            pipeline_id=pipeline_id,
+            actor_id=_actor_id(principal),
+            principal=principal,
+            registry=get_registry(),
+            payload=payload,
+        )
+    except PermissionError as exc:
+        raise _governance_http_error(exc) from exc
+    except DataflowError as exc:
+        raise _http_error(exc) from exc
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action="dataflow.pipeline.rebased",
+        object_type="dataflow_pipeline",
+        object_id=pipeline.id,
+        details={
+            "child_revision": pipeline.current_revision,
+            "source_pipeline_id": pipeline.derived_from_pipeline_id,
+            "source_revision": pipeline.derived_from_revision,
+            "source_hash": pipeline.derived_from_hash,
+            "status": pipeline.status,
+            "reason": payload.reason.strip(),
+        },
+    )
+    response = _pipeline_response(session, pipeline, principal)
+    session.commit()
+    return response
+
+
 @router.get(
     "/pipelines/{pipeline_id}/triggers",
     response_model=DataflowTriggerListResponse,
@@ -1503,10 +1567,22 @@ def api_promote_pipeline(
 @router.post("/validate", response_model=PipelineValidationResponse)
 def api_validate_pipeline(
     payload: PipelineDraftRequest,
+    session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> PipelineValidationResponse:
     _require_any_scope(principal, READ_SCOPE, WRITE_SCOPE, RUN_SCOPE, ADMIN_SCOPE)
-    return validate_draft(payload)
+    try:
+        return validate_draft(
+            payload,
+            session=session,
+            tenant_id=principal.tenant_id,
+            principal=principal,
+            registry=get_registry(),
+        )
+    except PermissionError as exc:
+        raise _governance_http_error(exc) from exc
+    except DataflowError as exc:
+        raise _http_error(exc) from exc
 
 
 @router.post("/sql/compile", response_model=PipelineSqlResponse)
@@ -1543,6 +1619,8 @@ def api_preview_pipeline(
             registry=get_registry(),
             payload=payload,
         )
+    except PermissionError as exc:
+        raise _governance_http_error(exc) from exc
     except DataflowError as exc:
         raise _http_error(exc) from exc
     if response.pipeline_id:

src/govoplan_dataflow/backend/schema_validation.py

@@ -129,7 +129,14 @@ def _propagation_context(
 def _inline_source(
     context: SchemaPropagationContext,
 ) -> SchemaPropagationResult:
-    return SchemaPropagationResult(_inline_schema(context.node.config.get("rows")))
+    configured = _configured_schema(
+        context.node.config.get("contract_schema")
+    )
+    return SchemaPropagationResult(
+        configured
+        if configured.columns
+        else _inline_schema(context.node.config.get("rows"))
+    )
 
 
 def _reference_source(
@@ -724,16 +731,57 @@ def _comparison_columns(value: object) -> tuple[list[str], list[str]]:
 
 
 def _subflow(context: SchemaPropagationContext) -> SchemaPropagationResult:
+    input_schema = _configured_schema(
+        context.node.config.get("input_schema")
+    )
     output_schema = _configured_schema(
         context.node.config.get("output_schema")
     )
+    diagnostics: list[DataflowDiagnostic] = []
+    if input_schema.columns and not context.input_state.open:
+        missing = sorted(input_schema.columns - context.input_state.columns)
+        if missing:
+            diagnostics.append(
+                _error(
+                    "subflow.input_contract.missing",
+                    "Subflow input is missing required contract columns: "
+                    + ", ".join(missing),
+                    node_id=context.node.id,
+                    field="input_schema",
+                )
+            )
+        incompatible = sorted(
+            column
+            for column in input_schema.columns & context.input_state.columns
+            if not _compatible_contract_type(
+                context.input_state.type_of(column),
+                input_schema.type_of(column),
+            )
+        )
+        if incompatible:
+            diagnostics.append(
+                _error(
+                    "subflow.input_contract.type",
+                    "Subflow input has incompatible contract types for: "
+                    + ", ".join(incompatible),
+                    node_id=context.node.id,
+                    field="input_schema",
+                )
+            )
     return SchemaPropagationResult(
         output_schema
         if output_schema.columns
-        else unknown_schema()
+        else unknown_schema(),
+        tuple(diagnostics),
     )
 
 
+def _compatible_contract_type(actual: str, expected: str) -> bool:
+    if "unknown" in {actual, expected} or actual == expected:
+        return True
+    return {actual, expected} <= {"integer", "number"}
+
+
 def _identity(context: SchemaPropagationContext) -> SchemaPropagationResult:
     return SchemaPropagationResult(context.input_state)
 

src/govoplan_dataflow/backend/schemas.py

@@ -122,6 +122,11 @@ class PipelineGovernanceResponse(BaseModel):
     derived_from_pipeline_id: str | None
     derived_from_revision: int | None
     derived_from_hash: str | None
+    source_available: bool = False
+    source_name: str | None = None
+    source_current_revision: int | None = None
+    source_current_hash: str | None = None
+    update_available: bool = False
     derivation_provenance: dict[str, Any] = Field(default_factory=dict)
     actions: dict[str, DefinitionActionDecisionResponse]
 
@@ -246,7 +251,23 @@ class PipelineDeriveRequest(BaseModel):
     allow_automation: bool = False
 
 
+class PipelineRebaseRequest(BaseModel):
+    expected_revision: int = Field(ge=1)
+    source_revision: int = Field(ge=1)
+    source_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
+    reason: str = Field(min_length=3, max_length=4_000)
+
+    @field_validator("reason")
+    @classmethod
+    def validate_reason(cls, value: str) -> str:
+        cleaned = value.strip()
+        if len(cleaned) < 3:
+            raise ValueError("A meaningful rebase review reason is required.")
+        return cleaned
+
+
 class PipelineDraftRequest(BaseModel):
+    pipeline_id: str | None = Field(default=None, max_length=36)
     graph: PipelineGraph | None = None
     sql_text: str | None = Field(default=None, max_length=100_000)
     source_nodes: list[GraphNode] = Field(default_factory=list, max_length=20)

src/govoplan_dataflow/backend/service.py

@@ -60,6 +60,7 @@ from govoplan_dataflow.backend.graph import (
     preserve_compatible_graph_layout,
     validate_graph,
 )
+from govoplan_dataflow.backend.ir import graph_to_ir
 from govoplan_dataflow.backend.schemas import (
     DataflowDiagnostic,
     GraphNode,
@@ -72,6 +73,7 @@ from govoplan_dataflow.backend.schemas import (
     PipelinePreviewResponse,
     PipelineDeploymentResponse,
     PipelinePromotionRequest,
+    PipelineRebaseRequest,
     PipelineResponse,
     PipelineRevisionResponse,
     PipelineRunResponse,
@@ -187,15 +189,177 @@ def get_pipeline_revision(
     return item
 
 
+def _resolve_reusable_subflows(
+    session: Session,
+    *,
+    tenant_id: str,
+    graph: PipelineGraph,
+    principal: ApiPrincipal | None,
+    registry: object | None,
+    target_pipeline_id: str | None,
+    ancestry: tuple[str, ...] = (),
+) -> PipelineGraph:
+    if not any(node.type == "subflow" for node in graph.nodes):
+        return graph
+    if principal is None:
+        raise DataflowConflictError(
+            "Reusable subflows require a tenant principal and current Policy "
+            "decision."
+        )
+    resolved_nodes: list[GraphNode] = []
+    for node in graph.nodes:
+        if node.type != "subflow":
+            resolved_nodes.append(node)
+            continue
+        source_id = _pipeline_id_from_ref(node.config.get("template_ref"))
+        if source_id == target_pipeline_id:
+            raise DataflowConflictError(
+                "A pipeline cannot reference itself as a reusable subflow."
+            )
+        source = get_pipeline(
+            session,
+            tenant_id=tenant_id,
+            pipeline_id=source_id,
+        )
+        reuse_decision = require_definition_action(
+            source,
+            principal=principal,
+            registry=registry,
+            action="reuse",
+        )
+        source_revision_number = _subflow_revision(
+            node.config.get("template_version")
+        )
+        source_revision = get_pipeline_revision(
+            session,
+            pipeline=source,
+            revision=source_revision_number,
+        )
+        reference_key = f"{source.id}:{source_revision.revision}"
+        if reference_key in ancestry:
+            raise DataflowConflictError(
+                "Reusable subflow references contain a cycle at "
+                f"pipeline:{source.id} revision {source_revision.revision}."
+            )
+        nested = _resolve_reusable_subflows(
+            session,
+            tenant_id=tenant_id,
+            graph=PipelineGraph.model_validate(source_revision.graph),
+            principal=principal,
+            registry=registry,
+            target_pipeline_id=target_pipeline_id,
+            ancestry=(*ancestry, reference_key),
+        )
+        input_nodes = [
+            item
+            for item in nested.nodes
+            if item.type == "source.inline"
+            and item.config.get("input_binding") is True
+        ]
+        if len(input_nodes) != 1:
+            raise DataflowConflictError(
+                "A referenced reusable definition must declare exactly one "
+                "inline template input binding."
+            )
+        typed = graph_to_ir(nested)
+        typed_by_id = {item.id: item for item in typed.nodes}
+        output_nodes = [item for item in nested.nodes if item.type == "output"]
+        if len(output_nodes) != 1:
+            raise DataflowConflictError(
+                "A referenced reusable definition must have exactly one "
+                "typed output."
+            )
+        input_contract = _typed_contract(
+            typed_by_id[input_nodes[0].id].output_schema,
+            label="input",
+        )
+        output_contract = _typed_contract(
+            typed_by_id[output_nodes[0].id].output_schema,
+            label="output",
+        )
+        config = {
+            **node.config,
+            "template_ref": f"pipeline:{source.id}",
+            "template_version": str(source_revision.revision),
+            "template_hash": source_revision.content_hash,
+            "graph": canonical_graph_payload(nested),
+            "input_schema": input_contract,
+            "output_schema": output_contract,
+            "reference_provenance": {
+                "source_scope": {
+                    "scope_type": source.scope_type,
+                    "scope_id": source.scope_id,
+                },
+                "source_definition_kind": source.definition_kind,
+                "policy_decision": reuse_decision.to_dict(),
+            },
+        }
+        resolved_nodes.append(
+            node.model_copy(update={"config": config}, deep=True)
+        )
+    return graph.model_copy(update={"nodes": resolved_nodes}, deep=True)
+
+
+def _pipeline_id_from_ref(value: object) -> str:
+    text = str(value or "").strip()
+    if not text.startswith("pipeline:") or len(text) <= len("pipeline:"):
+        raise DataflowConflictError(
+            "Reusable subflows require a canonical pipeline reference."
+        )
+    return text.removeprefix("pipeline:")
+
+
+def _subflow_revision(value: object) -> int:
+    try:
+        revision = int(str(value).strip())
+    except (TypeError, ValueError) as exc:
+        raise DataflowConflictError(
+            "Reusable subflows require a valid immutable source revision."
+        ) from exc
+    if revision < 1:
+        raise DataflowConflictError(
+            "Reusable subflow revisions must be positive."
+        )
+    return revision
+
+
+def _typed_contract(schema: object, *, label: str) -> list[dict[str, object]]:
+    fields = tuple(getattr(schema, "fields", ()))
+    if not fields or any(getattr(item, "type", "unknown") == "unknown" for item in fields):
+        raise DataflowConflictError(
+            f"The reusable definition needs a closed typed {label} contract. "
+            "Provide representative typed rows at its template input."
+        )
+    return [
+        {
+            "name": str(item.name),
+            "type": str(item.type),
+            "nullable": bool(item.nullable),
+        }
+        for item in fields
+    ]
+
+
 def create_pipeline(
     session: Session,
     *,
     tenant_id: str,
     actor_id: str | None,
     payload: PipelineCreateRequest,
+    principal: ApiPrincipal | None = None,
+    registry: object | None = None,
 ) -> DataflowPipeline:
-    definition = normalize_definition(
+    pipeline_id = new_uuid()
+    graph = _resolve_reusable_subflows(
+        session,
+        tenant_id=tenant_id,
         graph=payload.graph,
+        principal=principal,
+        registry=registry,
+        target_pipeline_id=pipeline_id,
+    )
+    definition = normalize_definition(
+        graph=graph,
         sql_text=payload.sql_text,
         editor_mode=payload.editor_mode,
     )
@@ -209,6 +373,7 @@ def create_pipeline(
         else payload.scope_id
     )
     pipeline = DataflowPipeline(
+        id=pipeline_id,
         tenant_id=stored_tenant_id,
         scope_type=payload.scope_type,
         scope_id=scope_id,
@@ -248,6 +413,8 @@ def update_pipeline(
     pipeline_id: str,
     actor_id: str | None,
     payload: PipelineUpdateRequest,
+    principal: ApiPrincipal | None = None,
+    registry: object | None = None,
 ) -> DataflowPipeline:
     pipeline = get_pipeline(session, tenant_id=tenant_id, pipeline_id=pipeline_id)
     if payload.expected_revision != pipeline.current_revision:
@@ -270,8 +437,16 @@ def update_pipeline(
         raise DataflowConflictError(
             "Definition kind is immutable; derive a flow or template instead."
         )
-    definition = normalize_definition(
+    graph = _resolve_reusable_subflows(
+        session,
+        tenant_id=tenant_id,
         graph=payload.graph,
+        principal=principal,
+        registry=registry,
+        target_pipeline_id=pipeline.id,
+    )
+    definition = normalize_definition(
+        graph=graph,
         sql_text=payload.sql_text,
         editor_mode=payload.editor_mode,
     )
@@ -417,6 +592,195 @@ def derive_pipeline(
     return pipeline
 
 
+def pipeline_source_update_status(
+    session: Session,
+    *,
+    tenant_id: str,
+    pipeline: DataflowPipeline,
+) -> dict[str, object]:
+    source_id = pipeline.derived_from_pipeline_id
+    if not source_id:
+        return {
+            "source_available": False,
+            "source_name": None,
+            "source_current_revision": None,
+            "source_current_hash": None,
+            "update_available": False,
+        }
+    source = session.scalar(
+        select(DataflowPipeline).where(
+            DataflowPipeline.id == source_id,
+            or_(
+                DataflowPipeline.tenant_id == tenant_id,
+                DataflowPipeline.tenant_id.is_(None),
+            ),
+            DataflowPipeline.deleted_at.is_(None),
+        )
+    )
+    if source is None:
+        return {
+            "source_available": False,
+            "source_name": None,
+            "source_current_revision": None,
+            "source_current_hash": None,
+            "update_available": False,
+        }
+    revision = get_pipeline_revision(session, pipeline=source)
+    return {
+        "source_available": True,
+        "source_name": source.name,
+        "source_current_revision": revision.revision,
+        "source_current_hash": revision.content_hash,
+        "update_available": (
+            revision.revision != pipeline.derived_from_revision
+            or revision.content_hash != pipeline.derived_from_hash
+        ),
+    }
+
+
+def rebase_pipeline(
+    session: Session,
+    *,
+    tenant_id: str,
+    pipeline_id: str,
+    actor_id: str | None,
+    principal: ApiPrincipal,
+    registry: object | None,
+    payload: PipelineRebaseRequest,
+) -> DataflowPipeline:
+    pipeline = get_pipeline(
+        session,
+        tenant_id=tenant_id,
+        pipeline_id=pipeline_id,
+    )
+    if payload.expected_revision != pipeline.current_revision:
+        raise DataflowConflictError(
+            "Derived pipeline changed on the server; expected revision "
+            f"{payload.expected_revision}, current revision is "
+            f"{pipeline.current_revision}."
+        )
+    source_id = pipeline.derived_from_pipeline_id
+    if not source_id:
+        raise DataflowConflictError(
+            "Only a pipeline derived from another definition can adopt a "
+            "source update."
+        )
+    source = get_pipeline(
+        session,
+        tenant_id=tenant_id,
+        pipeline_id=source_id,
+    )
+    reuse_decision = require_definition_action(
+        source,
+        principal=principal,
+        registry=registry,
+        action="derive",
+    )
+    source_revision = get_pipeline_revision(
+        session,
+        pipeline=source,
+        revision=payload.source_revision,
+    )
+    if source_revision.content_hash != payload.source_hash:
+        raise DataflowConflictError(
+            "The reviewed source hash no longer matches the requested "
+            "revision; reload before adopting the update."
+        )
+    if (
+        pipeline.derived_from_revision is not None
+        and source_revision.revision <= pipeline.derived_from_revision
+    ):
+        raise DataflowConflictError(
+            "A source update must use a revision newer than the currently "
+            "pinned revision."
+        )
+
+    previous_child_revision = pipeline.current_revision
+    previous_child = get_pipeline_revision(session, pipeline=pipeline)
+    previous_source_revision = pipeline.derived_from_revision
+    previous_source_hash = pipeline.derived_from_hash
+    source_limits = _effective_governance_limits(
+        source,
+        decision_details=reuse_decision.details,
+    )
+    effective_limits = {
+        "inherit_to_lower_scopes": (
+            pipeline.inherit_to_lower_scopes
+            and source_limits["inherit_to_lower_scopes"]
+        ),
+        "allow_run": pipeline.allow_run and source_limits["allow_run"],
+        "allow_reuse": pipeline.allow_reuse and source_limits["allow_reuse"],
+        "allow_automation": (
+            pipeline.allow_automation and source_limits["allow_automation"]
+        ),
+    }
+    next_child_revision = previous_child_revision + 1
+    rebased_at = utcnow()
+    history_value = pipeline.derivation_provenance.get("rebase_history", [])
+    history = list(history_value) if isinstance(history_value, list) else []
+    history.append(
+        {
+            "child_revision_before": previous_child_revision,
+            "child_hash_before": previous_child.content_hash,
+            "child_revision_after": next_child_revision,
+            "source_revision_before": previous_source_revision,
+            "source_hash_before": previous_source_hash,
+            "source_revision_after": source_revision.revision,
+            "source_hash_after": source_revision.content_hash,
+            "policy_decision": reuse_decision.to_dict(),
+            "reason": payload.reason.strip(),
+            "rebased_by": actor_id,
+            "rebased_at": rebased_at.isoformat(),
+        }
+    )
+    provenance = dict(pipeline.derivation_provenance)
+    provenance.update(
+        {
+            "source_ref": f"pipeline:{source.id}",
+            "source_scope": {
+                "scope_type": source.scope_type,
+                "scope_id": source.scope_id,
+            },
+            "source_definition_kind": source.definition_kind,
+            "source_revision": source_revision.revision,
+            "source_hash": source_revision.content_hash,
+            "source_effective_limits": effective_limits,
+            "policy_decision": reuse_decision.to_dict(),
+            "last_rebased_by": actor_id,
+            "last_rebased_at": rebased_at.isoformat(),
+            "last_rebase_reason": payload.reason.strip(),
+            "rebase_history": history,
+        }
+    )
+
+    pipeline.current_revision = next_child_revision
+    pipeline.status = "draft"
+    pipeline.inherit_to_lower_scopes = effective_limits[
+        "inherit_to_lower_scopes"
+    ]
+    pipeline.allow_run = effective_limits["allow_run"]
+    pipeline.allow_reuse = effective_limits["allow_reuse"]
+    pipeline.allow_automation = effective_limits["allow_automation"]
+    pipeline.derived_from_revision = source_revision.revision
+    pipeline.derived_from_hash = source_revision.content_hash
+    pipeline.derivation_provenance = provenance
+    pipeline.updated_by = actor_id
+    pipeline.revisions.append(
+        DataflowPipelineRevision(
+            tenant_id=pipeline.tenant_id,
+            revision=next_child_revision,
+            schema_version=source_revision.schema_version,
+            graph=json.loads(json.dumps(source_revision.graph)),
+            sql_text=source_revision.sql_text,
+            editor_mode=source_revision.editor_mode,
+            content_hash=source_revision.content_hash,
+            created_by=actor_id,
+        )
+    )
+    session.flush()
+    return pipeline
+
+
 def delete_pipeline(
     session: Session,
     *,
@@ -455,11 +819,36 @@ def pipeline_response(
             pipeline,
             principal=principal,
             registry=registry,
+            source_update=pipeline_source_update_status(
+                session,
+                tenant_id=principal.tenant_id,
+                pipeline=pipeline,
+            ),
         ),
     )
 
 
-def validate_draft(payload: PipelineDraftRequest) -> PipelineValidationResponse:
+def validate_draft(
+    payload: PipelineDraftRequest,
+    *,
+    session: Session | None = None,
+    tenant_id: str | None = None,
+    principal: ApiPrincipal | None = None,
+    registry: object | None = None,
+) -> PipelineValidationResponse:
+    if payload.graph is not None and session is not None and tenant_id is not None:
+        payload = payload.model_copy(
+            update={
+                "graph": _resolve_reusable_subflows(
+                    session,
+                    tenant_id=tenant_id,
+                    graph=payload.graph,
+                    principal=principal,
+                    registry=registry,
+                    target_pipeline_id=payload.pipeline_id,
+                )
+            }
+        )
     if payload.sql_text and payload.sql_text.strip():
         try:
             graph, sql_text, diagnostics = compile_sql(
@@ -598,7 +987,13 @@ def preview_pipeline(
             sql_text=payload.sql_text,
             source_nodes=payload.source_nodes,
         )
-        validated = validate_draft(draft)
+        validated = validate_draft(
+            draft,
+            session=session,
+            tenant_id=tenant_id,
+            principal=principal,
+            registry=registry,
+        )
         if not validated.valid or validated.graph is None:
             return PipelinePreviewResponse(
                 run_id=None,
@@ -2290,11 +2685,13 @@ __all__ = [
     "list_pipelines",
     "normalize_definition",
     "pipeline_response",
+    "pipeline_source_update_status",
     "pipeline_deployment_response",
     "pipeline_run_descriptor",
     "pipeline_run_request",
     "pipeline_run_response",
     "promote_pipeline",
+    "rebase_pipeline",
     "preview_pipeline",
     "render_graph_sql",
     "start_pipeline_run",

tests/test_documentation_contract.py

@@ -0,0 +1,19 @@
+from govoplan_dataflow.backend.manifest import get_manifest
+
+
+def test_static_documentation_has_complete_german_reference_copy() -> None:
+    for topic in get_manifest().documentation:
+        german = topic.translations.get("de", {})
+        assert all(german.get(field, "").strip() for field in ("title", "summary", "body")), topic.id
+
+
+def test_documentation_exposes_conditioned_workflow_and_reference() -> None:
+    topics = {topic.id: topic for topic in get_manifest().documentation}
+    workflow = topics["dataflow.module-boundary"]
+    assert workflow.metadata.get("kind") == "workflow"
+    assert any(condition.required_scopes for condition in workflow.conditions)
+    assert workflow.structured_translations.get("de")
+
+    reference = topics["dataflow.reference.fields-and-consequences"]
+    assert reference.metadata.get("kind") == "reference"
+    assert reference.metadata.get("consequence_classes")

tests/test_triggers.py

@@ -25,12 +25,16 @@ from govoplan_dataflow.backend.schemas import (
     DataflowTriggerSchedule,
     PipelineCreateRequest,
     PipelineDeriveRequest,
+    PipelineRebaseRequest,
     PipelineUpdateRequest,
 )
 from govoplan_dataflow.backend.service import (
     DataflowConflictError,
+    DataflowValidationError,
     create_pipeline,
     derive_pipeline,
+    pipeline_response,
+    rebase_pipeline,
     start_pipeline_run,
     update_pipeline,
 )
@@ -46,10 +50,87 @@ POLICY_CAPABILITY = "policy.definitionGovernance"
 AUTOMATION_CAPABILITY = "auth.automationPrincipalProvider"
 
 
-def sample_graph():
+def sample_graph(*, minimum: int = 10):
     from test_service import sample_graph as build_graph
 
-    return build_graph()
+    return build_graph(minimum=minimum)
+
+
+def reusable_graph(*, minimum: int = 10):
+    graph = sample_graph(minimum=minimum)
+    return graph.model_copy(
+        update={
+            "nodes": [
+                (
+                    node.model_copy(
+                        update={
+                            "config": {
+                                **node.config,
+                                "input_binding": True,
+                            }
+                        },
+                        deep=True,
+                    )
+                    if node.id == "source"
+                    else node
+                )
+                for node in graph.nodes
+            ]
+        },
+        deep=True,
+    )
+
+
+def referencing_graph(
+    source_ref: str,
+    source_revision: int,
+    *,
+    omit_amount: bool = False,
+    input_binding: bool = False,
+):
+    graph = sample_graph()
+    return graph.model_copy(
+        update={
+            "nodes": [
+                (
+                    node.model_copy(
+                        update={
+                            "config": {
+                                **node.config,
+                                "rows": (
+                                    [{"id": 1}]
+                                    if omit_amount
+                                    else node.config["rows"]
+                                ),
+                                "input_binding": input_binding,
+                            }
+                        },
+                        deep=True,
+                    )
+                    if node.id == "source"
+                    else node.model_copy(
+                        update={
+                            "type": "subflow",
+                            "label": "Governed reusable flow",
+                            "config": {
+                                "template_ref": source_ref,
+                                "template_version": str(source_revision),
+                                "parameters": {},
+                                "graph": sample_graph(
+                                    minimum=999
+                                ).model_dump(mode="json"),
+                            },
+                        },
+                        deep=True,
+                    )
+                    if node.id == "filter"
+                    else node
+                )
+                for node in graph.nodes
+            ]
+        },
+        deep=True,
+    )
 
 
 def principal() -> ApiPrincipal:
@@ -482,6 +563,152 @@ class DataflowTriggerTests(unittest.TestCase):
                 ),
             )
 
+    def test_reusable_reference_is_policy_resolved_with_typed_contracts(
+        self,
+    ) -> None:
+        template = create_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="template-author",
+            payload=PipelineCreateRequest(
+                name="Typed reusable filter",
+                graph=reusable_graph(minimum=10),
+                definition_kind="template",
+                allow_reuse=True,
+            ),
+        )
+        consumer = create_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="account-1",
+            principal=self.actor,
+            registry=self.registry,
+            payload=PipelineCreateRequest(
+                name="Resolved consumer",
+                graph=referencing_graph(f"pipeline:{template.id}", 1),
+            ),
+        )
+        self.session.commit()
+
+        stored = consumer.revisions[0].graph
+        subflow = next(
+            node for node in stored["nodes"] if node["id"] == "filter"
+        )
+        self.assertEqual(template.revisions[0].content_hash, subflow["config"]["template_hash"])
+        self.assertEqual(
+            10,
+            next(
+                node
+                for node in subflow["config"]["graph"]["nodes"]
+                if node["id"] == "filter"
+            )["config"]["value"],
+        )
+        self.assertEqual(
+            {"id", "amount"},
+            {
+                field["name"]
+                for field in subflow["config"]["input_schema"]
+            },
+        )
+        self.assertEqual(
+            {"id", "amount"},
+            {
+                field["name"]
+                for field in subflow["config"]["output_schema"]
+            },
+        )
+        self.assertTrue(
+            subflow["config"]["reference_provenance"][
+                "policy_decision"
+            ]["allowed"]
+        )
+
+        with self.assertRaises(DataflowValidationError):
+            create_pipeline(
+                self.session,
+                tenant_id="tenant-1",
+                actor_id="account-1",
+                principal=self.actor,
+                registry=self.registry,
+                payload=PipelineCreateRequest(
+                    name="Incompatible consumer",
+                    graph=referencing_graph(
+                        f"pipeline:{template.id}",
+                        1,
+                        omit_amount=True,
+                    ),
+                ),
+            )
+
+    def test_reusable_reference_cycles_are_rejected_across_revisions(
+        self,
+    ) -> None:
+        left = create_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="account-1",
+            payload=PipelineCreateRequest(
+                name="Left template",
+                graph=reusable_graph(),
+                definition_kind="template",
+                allow_reuse=True,
+            ),
+        )
+        right = create_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="account-1",
+            payload=PipelineCreateRequest(
+                name="Right template",
+                graph=reusable_graph(),
+                definition_kind="template",
+                allow_reuse=True,
+            ),
+        )
+        self.session.flush()
+        update_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            pipeline_id=left.id,
+            actor_id="account-1",
+            principal=self.actor,
+            registry=self.registry,
+            payload=PipelineUpdateRequest(
+                name=left.name,
+                graph=referencing_graph(
+                    f"pipeline:{right.id}",
+                    1,
+                    input_binding=True,
+                ),
+                status="draft",
+                expected_revision=1,
+                definition_kind="template",
+                allow_reuse=True,
+            ),
+        )
+
+        with self.assertRaisesRegex(DataflowConflictError, "cannot reference itself"):
+            update_pipeline(
+                self.session,
+                tenant_id="tenant-1",
+                pipeline_id=right.id,
+                actor_id="account-1",
+                principal=self.actor,
+                registry=self.registry,
+                payload=PipelineUpdateRequest(
+                    name=right.name,
+                    graph=referencing_graph(
+                        f"pipeline:{left.id}",
+                        2,
+                        input_binding=True,
+                    ),
+                    status="draft",
+                    expected_revision=1,
+                    definition_kind="template",
+                    allow_reuse=True,
+                ),
+            )
+
     def test_derived_limits_cannot_be_broadened_transitively(self) -> None:
         template = create_pipeline(
             self.session,
@@ -549,6 +776,128 @@ class DataflowTriggerTests(unittest.TestCase):
         self.assertFalse(grandchild.allow_automation)
         self.assertFalse(grandchild.inherit_to_lower_scopes)
 
+    def test_source_update_is_detected_and_rebased_as_reviewed_revision(
+        self,
+    ) -> None:
+        template = create_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="account-1",
+            payload=PipelineCreateRequest(
+                name="Reusable import",
+                graph=sample_graph(),
+                definition_kind="template",
+                allow_reuse=True,
+                allow_automation=True,
+            ),
+        )
+        derived = derive_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="account-1",
+            principal=self.actor,
+            registry=self.registry,
+            source_pipeline_id=template.id,
+            payload=PipelineDeriveRequest(
+                name="Tenant import",
+                allow_run=True,
+                allow_automation=True,
+            ),
+        )
+        self.session.commit()
+
+        before = pipeline_response(
+            self.session,
+            derived,
+            principal=self.actor,
+            registry=self.registry,
+        )
+        self.assertTrue(before.governance.source_available)
+        self.assertFalse(before.governance.update_available)
+        original_child_hash = derived.revisions[0].content_hash
+
+        update_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            pipeline_id=template.id,
+            actor_id="template-author",
+            payload=PipelineUpdateRequest(
+                name=template.name,
+                graph=sample_graph(minimum=20),
+                status="draft",
+                expected_revision=1,
+                definition_kind="template",
+                allow_reuse=True,
+                allow_automation=True,
+            ),
+        )
+        self.session.commit()
+        source_hash = template.revisions[-1].content_hash
+
+        available = pipeline_response(
+            self.session,
+            derived,
+            principal=self.actor,
+            registry=self.registry,
+        )
+        self.assertTrue(available.governance.update_available)
+        self.assertEqual(2, available.governance.source_current_revision)
+        self.assertEqual(source_hash, available.governance.source_current_hash)
+        self.assertEqual(original_child_hash, derived.revisions[0].content_hash)
+
+        with self.assertRaisesRegex(DataflowConflictError, "source hash"):
+            rebase_pipeline(
+                self.session,
+                tenant_id="tenant-1",
+                pipeline_id=derived.id,
+                actor_id="reviewer-1",
+                principal=self.actor,
+                registry=self.registry,
+                payload=PipelineRebaseRequest(
+                    expected_revision=1,
+                    source_revision=2,
+                    source_hash="0" * 64,
+                    reason="Reviewed the changed filter threshold.",
+                ),
+            )
+
+        rebased = rebase_pipeline(
+            self.session,
+            tenant_id="tenant-1",
+            pipeline_id=derived.id,
+            actor_id="reviewer-1",
+            principal=self.actor,
+            registry=self.registry,
+            payload=PipelineRebaseRequest(
+                expected_revision=1,
+                source_revision=2,
+                source_hash=source_hash,
+                reason="Reviewed the changed filter threshold.",
+            ),
+        )
+        self.session.commit()
+
+        self.assertEqual(2, rebased.current_revision)
+        self.assertEqual("draft", rebased.status)
+        self.assertEqual(2, rebased.derived_from_revision)
+        self.assertEqual(source_hash, rebased.derived_from_hash)
+        self.assertEqual(source_hash, rebased.revisions[-1].content_hash)
+        self.assertEqual(original_child_hash, rebased.revisions[0].content_hash)
+        history = rebased.derivation_provenance["rebase_history"]
+        self.assertEqual(1, len(history))
+        self.assertEqual("reviewer-1", history[0]["rebased_by"])
+        self.assertEqual(
+            "Reviewed the changed filter threshold.",
+            history[0]["reason"],
+        )
+        current = pipeline_response(
+            self.session,
+            rebased,
+            principal=self.actor,
+            registry=self.registry,
+        )
+        self.assertFalse(current.governance.update_available)
+
 
 if __name__ == "__main__":
     unittest.main()

webui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/dataflow-webui",
-  "version": "0.1.19",
+  "version": "0.1.23",
   "private": true,
   "type": "module",
   "main": "src/index.ts",

webui/src/api/dataflow.ts

@@ -95,6 +95,11 @@ export type PipelineGovernance = {
   derived_from_pipeline_id?: string | null;
   derived_from_revision?: number | null;
   derived_from_hash?: string | null;
+  source_available: boolean;
+  source_name?: string | null;
+  source_current_revision?: number | null;
+  source_current_hash?: string | null;
+  update_available: boolean;
   derivation_provenance: Record<string, unknown>;
   actions: Record<string, DefinitionActionDecision>;
 };
@@ -505,6 +510,23 @@ export function deriveDataflowPipeline(
   );
 }
 
+export function rebaseDataflowPipeline(
+  settings: ApiSettings,
+  pipelineId: string,
+  payload: {
+    expected_revision: number;
+    source_revision: number;
+    source_hash: string;
+    reason: string;
+  }
+): Promise<Pipeline> {
+  return apiFetch(
+    settings,
+    `/api/v1/dataflow/pipelines/${encodeURIComponent(pipelineId)}/rebase`,
+    { method: "POST", body: JSON.stringify(payload) }
+  );
+}
+
 export function dataflowScopeReferenceProvider(
   settings: ApiSettings,
   scopeType: "user" | "group"
@@ -564,7 +586,7 @@ export function deleteDataflowTrigger(
 
 export function validateDataflowPipeline(
   settings: ApiSettings,
-  payload: { graph?: PipelineGraph; sql_text?: string; source_nodes?: PipelineGraphNode[] }
+  payload: { pipeline_id?: string | null; graph?: PipelineGraph; sql_text?: string; source_nodes?: PipelineGraphNode[] }
 ): Promise<PipelineValidation> {
   return apiFetch<PipelineValidation>(settings, "/api/v1/dataflow/validate", {
     method: "POST",

webui/src/features/dataflow/DataflowPage.tsx

@@ -13,6 +13,7 @@ import {
   Code2,
   CopyPlus,
   DatabaseZap,
+  GitCompareArrows,
   ListChecks,
   Network,
   Play,
@@ -84,6 +85,7 @@ import {
   listDataflowTriggers,
   previewDataflowPipeline,
   promoteDataflowPipeline,
+  rebaseDataflowPipeline,
   recordDataflowDecision,
   runDataflowPipeline,
   renderDataflowSql,
@@ -161,6 +163,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
   const [runOpen, setRunOpen] = useState(false);
   const [definitionSettingsOpen, setDefinitionSettingsOpen] = useState(false);
   const [deriveOpen, setDeriveOpen] = useState(false);
+  const [rebaseOpen, setRebaseOpen] = useState(false);
   const [triggersOpen, setTriggersOpen] = useState(false);
   const [decisionReviewOpen, setDecisionReviewOpen] = useState(false);
   const [nodeLibrary, setNodeLibrary] = useState<NodeTypeDefinition[]>(FALLBACK_NODE_LIBRARY);
@@ -183,6 +186,14 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
     && canWrite
     && draft.governance?.actions.derive?.allowed
   );
+  const canRebase = Boolean(
+    draft?.id
+    && draft.governance?.update_available
+    && draft.governance.source_current_revision
+    && draft.governance.source_current_hash
+    && canEdit
+    && !dirty
+  );
   const canStartSavedRun = Boolean(
     draft?.id
     && draft.definitionKind === "flow"
@@ -411,10 +422,15 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
     setSuccess("");
     try {
       const response = await validateDataflowPipeline(settings, draft.editorMode === "sql"
-        ? { graph: draft.graph, sql_text: draft.sqlText, source_nodes: sourceNodes(draft.graph) }
-        : { graph: draft.graph });
+        ? { pipeline_id: draft.id, graph: draft.graph, sql_text: draft.sqlText, source_nodes: sourceNodes(draft.graph) }
+        : { pipeline_id: draft.id, graph: draft.graph });
       setDiagnostics(response.diagnostics);
-      if (response.valid) setSuccess("Pipeline definition is valid.");
+      if (response.valid) {
+        if (response.graph && draft.editorMode === "graph") {
+          updateDraft({ graph: response.graph });
+        }
+        setSuccess("Pipeline definition is valid.");
+      }
       setResultOpen(true);
       setResultTab("diagnostics");
     } catch (validationError) {
@@ -766,6 +782,26 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
                       }
                     />
                   ) : null}
+                  {draft.governance?.derived_from_pipeline_id ? (
+                    <IconButton
+                      label="Review source update"
+                      icon={<GitCompareArrows size={16} />}
+                      variant="ghost"
+                      onClick={() => setRebaseOpen(true)}
+                      disabled={!canRebase}
+                      disabledReason={
+                        dirty
+                          ? DATAFLOW_I18N.saveFirst
+                          : !canEdit
+                            ? editBlockedReason
+                            : !draft.governance.source_available
+                              ? "The source definition is no longer available."
+                              : !draft.governance.update_available
+                                ? "This copy already pins the current source revision."
+                                : undefined
+                      }
+                    />
+                  ) : null}
                   {draft.id ? (
                     <IconButton
                       label="Automation triggers"
@@ -780,6 +816,8 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
                 destructiveActions={draft.id ? (
                   <IconButton
                     label="Delete pipeline"
+                    helpContextId="dataflow.action.delete"
+                    helpModuleId="dataflow"
                     icon={<Trash2 size={16} />}
                     variant="danger"
                     onClick={() => setDeleteOpen(true)}
@@ -899,6 +937,10 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
                 <NodeInspector
                   node={selectedNode}
                   nodeLibrary={nodeLibrary}
+                  reusablePipelines={pipelines.filter((pipeline) => (
+                    pipeline.id !== draft.id
+                    && pipeline.governance.actions.reuse?.allowed
+                  ))}
                   sources={sources}
                   sourceCatalogueAvailable={sourceCatalogueAvailable}
                   readOnly={!canEdit}
@@ -1060,6 +1102,32 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
           setSuccess("Created a pinned scoped copy.");
         }}
       />
+      <RebasePipelineDialog
+        open={rebaseOpen}
+        settings={settings}
+        pipeline={draft?.id && draft.currentRevision && draft.governance ? {
+          id: draft.id,
+          name: draft.name,
+          currentRevision: draft.currentRevision,
+          governance: draft.governance
+        } : null}
+        onClose={() => setRebaseOpen(false)}
+        onRebased={(pipeline) => {
+          const next = draftFromPipeline(pipeline);
+          setPipelines((current) => [
+            pipeline,
+            ...current.filter((item) => item.id !== pipeline.id)
+          ]);
+          setDraft(next);
+          setSavedDraft(structuredClone(next));
+          setSelectedNodeId(next.graph.nodes[0]?.id ?? null);
+          setRebaseOpen(false);
+          setPreview(null);
+          setDiagnostics([]);
+          setNodeDiagnostics([]);
+          setSuccess(`Adopted source revision ${pipeline.governance.derived_from_revision} as draft revision ${pipeline.current_revision}.`);
+        }}
+      />
       <DataflowTriggersDialog
         open={triggersOpen}
         settings={settings}
@@ -1231,6 +1299,20 @@ function DefinitionSettingsDialog({
                 {draft.governance.derived_from_revision}
               </span>
               <code>{draft.governance.derived_from_hash}</code>
+              {!draft.governance.source_available ? (
+                <StatusBadge status="warning" label="Source unavailable" />
+              ) : draft.governance.update_available ? (
+                <>
+                  <StatusBadge status="warning" label="Source update available" />
+                  <span>
+                    {draft.governance.source_name ?? "Source definition"}
+                    {" · revision "}
+                    {draft.governance.source_current_revision}
+                  </span>
+                </>
+              ) : (
+                <StatusBadge status="success" label="Source revision current" />
+              )}
             </ContentSection>
           ) : null}
           {provenance.length ? (
@@ -1430,6 +1512,153 @@ function DerivePipelineDialog({
   );
 }
 
+function RebasePipelineDialog({
+  open,
+  settings,
+  pipeline,
+  onClose,
+  onRebased
+}: {
+  open: boolean;
+  settings: ApiSettings;
+  pipeline: {
+    id: string;
+    name: string;
+    currentRevision: number;
+    governance: Pipeline["governance"];
+  } | null;
+  onClose: () => void;
+  onRebased: (pipeline: Pipeline) => void;
+}) {
+  const { requestDiscard } = useUnsavedChanges();
+  const [reason, setReason] = useState("");
+  const [busy, setBusy] = useState(false);
+  const [error, setError] = useState("");
+  const governance = pipeline?.governance;
+  const sourceRevision = governance?.source_current_revision ?? null;
+  const sourceHash = governance?.source_current_hash ?? null;
+  const dirty = Boolean(open && reason);
+
+  useEffect(() => {
+    if (!open) return;
+    setReason("");
+    setError("");
+  }, [open, pipeline?.id, sourceRevision]);
+
+  const resetDraft = () => {
+    setReason("");
+    setError("");
+  };
+
+  const rebase = async (): Promise<boolean> => {
+    if (
+      !pipeline
+      || !sourceRevision
+      || !sourceHash
+      || !reason.trim()
+      || !governance?.update_available
+    ) return false;
+    setBusy(true);
+    setError("");
+    try {
+      onRebased(await rebaseDataflowPipeline(settings, pipeline.id, {
+        expected_revision: pipeline.currentRevision,
+        source_revision: sourceRevision,
+        source_hash: sourceHash,
+        reason: reason.trim()
+      }));
+      return true;
+    } catch (rebaseError) {
+      setError(apiErrorMessage(rebaseError));
+      return false;
+    } finally {
+      setBusy(false);
+    }
+  };
+
+  useUnsavedDraftGuard({
+    dirty,
+    title: "Unapplied source update",
+    message: "Apply the reviewed source update or discard the review reason before leaving.",
+    onSave: rebase,
+    onDiscard: resetDraft
+  });
+
+  const close = () => {
+    if (busy) return;
+    if (dirty) requestDiscard(onClose);
+    else onClose();
+  };
+
+  return (
+    <Dialog
+      open={open}
+      title="Review source update"
+      className="dataflow-definition-dialog"
+      closeDisabled={busy}
+      onClose={close}
+      footer={(
+        <>
+          <Button onClick={close} disabled={busy}>Cancel</Button>
+          <Button
+            variant="primary"
+            onClick={() => void rebase()}
+            disabled={
+              busy
+              || !pipeline
+              || !sourceRevision
+              || !sourceHash
+              || !reason.trim()
+              || !governance?.update_available
+            }
+          >
+            <GitCompareArrows size={16} /> Adopt source revision
+          </Button>
+        </>
+      )}
+    >
+      <div className="dataflow-definition-fields">
+        {error ? <DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert> : null}
+        <ContentSection spacing="none" surface="subtle" density="compact" layout="stack">
+          <strong>{governance?.source_name ?? "Source definition"}</strong>
+          <span>
+            Pinned revision {governance?.derived_from_revision ?? "—"}
+            {" → source revision "}
+            {sourceRevision ?? "—"}
+          </span>
+          {sourceHash ? <code>{sourceHash}</code> : null}
+        </ContentSection>
+        <DismissibleAlert
+          tone="warning"
+          resetKey={`${pipeline?.id ?? "none"}:${sourceRevision ?? "none"}`}
+        >
+          Adopting the update replaces the copy's current graph with the exact
+          reviewed source revision and returns the copy to draft. Existing
+          revisions, run evidence and rebase provenance remain immutable.
+        </DismissibleAlert>
+        <FormField
+          label="Review reason"
+          help="Record what was reviewed and why this source revision is appropriate for the scoped copy."
+          interfaceId="dataflow.field.rebase-reason"
+          helpContextId="dataflow.field.rebase-reason"
+          helpModuleId="dataflow"
+          helpTopicId="dataflow.reference.fields-and-consequences"
+          documentation={DATAFLOW_FIELDS_DOCUMENTATION}
+        >
+          <textarea
+            value={reason}
+            onChange={(event) => setReason(event.target.value)}
+            rows={4}
+            maxLength={4000}
+            disabled={busy}
+            required
+          />
+        </FormField>
+      </div>
+    </Dialog>
+  );
+}
+
 function DataflowTriggersDialog({
   open,
   settings,
@@ -1812,6 +2041,8 @@ function DataflowTriggersDialog({
               {selected.last_error ? <small className="is-error">{selected.last_error}</small> : null}
               <Button
                 variant="danger"
+                helpContextId="dataflow.action.delete"
+                helpModuleId="dataflow"
                 onClick={() => requestNavigation(() => setDeleteCandidate(selected))}
                 disabled={busy || !editable}
                 disabledReason={busy ? DATAFLOW_I18N.working : !editable ? DATAFLOW_I18N.writeReason : undefined}

webui/src/features/dataflow/NodeInspector.tsx

@@ -9,6 +9,7 @@ import {
 } from "@govoplan/core-webui";
 import type {
   NodeTypeDefinition,
+  Pipeline,
   PipelineGraphNode,
   TabularSource
 } from "../../api/dataflow";
@@ -28,6 +29,7 @@ function NodeFormField({ documentation, ...props }: NodeFormFieldProps) {
 type NodeInspectorProps = {
   node: PipelineGraphNode | null;
   nodeLibrary: NodeTypeDefinition[];
+  reusablePipelines: Pipeline[];
   sources: TabularSource[];
   sourceCatalogueAvailable: boolean;
   readOnly: boolean;
@@ -41,6 +43,7 @@ type NodeInspectorProps = {
 export default function NodeInspector({
   node,
   nodeLibrary,
+  reusablePipelines,
   sources,
   sourceCatalogueAvailable,
   readOnly,
@@ -57,7 +60,6 @@ export default function NodeInspector({
   const [rankSortText, setRankSortText] = useState("");
   const [rulesText, setRulesText] = useState("");
   const [parametersText, setParametersText] = useState("");
-  const [subflowGraphText, setSubflowGraphText] = useState("");
   const [localError, setLocalError] = useState("");
 
   useEffect(() => {
@@ -68,7 +70,6 @@ export default function NodeInspector({
     setRankSortText(node ? sortFieldsToText(node.config.order_by) : "");
     setRulesText(node ? JSON.stringify(node.config.rules ?? [], null, 2) : "");
     setParametersText(node ? JSON.stringify(node.config.parameters ?? {}, null, 2) : "");
-    setSubflowGraphText(node ? JSON.stringify(node.config.graph ?? {}, null, 2) : "");
     setLocalError("");
   }, [node?.id]);
 
@@ -84,6 +85,9 @@ export default function NodeInspector({
   }
 
   const definition = nodeLibrary.find((item) => item.type === node.type);
+  const selectedReusable = reusablePipelines.find(
+    (item) => `pipeline:${item.id}` === textValue(node.config.template_ref)
+  );
   const updateConfig = (patch: Record<string, unknown>) => {
     onChange({ ...node, config: { ...node.config, ...patch } });
   };
@@ -178,6 +182,8 @@ export default function NodeInspector({
           />
           <IconButton
             label="Delete node"
+            helpContextId="dataflow.action.delete"
+            helpModuleId="dataflow"
             icon={<Trash2 size={16} />}
             variant="danger"
             onClick={() => onDelete(node.id)}
@@ -263,16 +269,33 @@ export default function NodeInspector({
           </>
         ) : null}
         {node.type === "source.inline" ? (
-          <NodeFormField label="Rows">
-            <textarea
-              className="dataflow-json-editor"
-              value={rowsText}
-              onChange={(event) => setRowsText(event.target.value)}
-              onBlur={commitRows}
-              spellCheck={false}
-              disabled={readOnly}
-            />
-          </NodeFormField>
+          <>
+            <NodeFormField label="Rows">
+              <textarea
+                className="dataflow-json-editor"
+                value={rowsText}
+                onChange={(event) => setRowsText(event.target.value)}
+                onBlur={commitRows}
+                spellCheck={false}
+                disabled={readOnly}
+              />
+            </NodeFormField>
+            <NodeFormField
+              label="Reusable input binding"
+              help="A reusable definition must mark exactly one typed inline source as the rows supplied by its caller."
+              interfaceId="dataflow.field.reusable-input-binding"
+              helpContextId="dataflow.field.reusable-input-binding"
+              helpModuleId="dataflow"
+              helpTopicId="dataflow.reference.nodes-and-expressions"
+            >
+              <input
+                type="checkbox"
+                checked={node.config.input_binding === true}
+                onChange={(event) => updateConfig({ input_binding: event.target.checked })}
+                disabled={readOnly}
+              />
+            </NodeFormField>
+          </>
         ) : null}
         {node.type === "filter" ? (
           <>
@@ -783,19 +806,62 @@ export default function NodeInspector({
         ) : null}
         {node.type === "subflow" ? (
           <>
-            <NodeFormField label="Template reference">
-              <input
+            <NodeFormField
+              label="Reusable definition"
+              interfaceId="dataflow.field.subflow-reference"
+              helpContextId="dataflow.field.subflow-reference"
+              helpModuleId="dataflow"
+              helpTopicId="dataflow.reference.nodes-and-expressions"
+            >
+              <select
                 value={textValue(node.config.template_ref)}
-                onChange={(event) => updateConfig({ template_ref: event.target.value })}
+                onChange={(event) => {
+                  const selected = reusablePipelines.find(
+                    (item) => `pipeline:${item.id}` === event.target.value
+                  );
+                  updateConfig({
+                    template_ref: event.target.value,
+                    template_version: selected ? String(selected.current_revision) : "",
+                    template_hash: "",
+                    graph: { schema_version: 1, nodes: [], edges: [] },
+                    input_schema: [],
+                    output_schema: []
+                  });
+                }}
                 disabled={readOnly}
-              />
+              >
+                <option value="">Choose a reusable definition</option>
+                {reusablePipelines.map((pipeline) => (
+                  <option key={pipeline.id} value={`pipeline:${pipeline.id}`}>
+                    {pipeline.name} · revision {pipeline.current_revision}
+                  </option>
+                ))}
+              </select>
             </NodeFormField>
-            <NodeFormField label="Template version">
-              <input
+            <NodeFormField
+              label="Template version"
+              interfaceId="dataflow.field.subflow-revision"
+              helpContextId="dataflow.field.subflow-revision"
+              helpModuleId="dataflow"
+              helpTopicId="dataflow.reference.nodes-and-expressions"
+            >
+              <select
                 value={textValue(node.config.template_version)}
                 onChange={(event) => updateConfig({ template_version: event.target.value })}
                 disabled={readOnly}
-              />
+              >
+                {textValue(node.config.template_version)
+                  && textValue(node.config.template_version) !== String(selectedReusable?.current_revision ?? "") ? (
+                    <option value={textValue(node.config.template_version)}>
+                      Pinned revision {textValue(node.config.template_version)}
+                    </option>
+                  ) : null}
+                {selectedReusable ? (
+                  <option value={String(selectedReusable.current_revision)}>
+                    Current revision {selectedReusable.current_revision}
+                  </option>
+                ) : null}
+              </select>
             </NodeFormField>
             <NodeFormField label="Parameters">
               <textarea
@@ -807,16 +873,13 @@ export default function NodeInspector({
                 disabled={readOnly}
               />
             </NodeFormField>
-            <NodeFormField label="Pinned graph">
-              <textarea
-                className="dataflow-json-editor"
-                value={subflowGraphText}
-                onChange={(event) => setSubflowGraphText(event.target.value)}
-                onBlur={() => commitJsonConfig("graph", subflowGraphText, "object")}
-                spellCheck={false}
-                disabled={readOnly}
-              />
-            </NodeFormField>
+            {Array.isArray(node.config.input_schema) && Array.isArray(node.config.output_schema) ? (
+              <NodeFormField label="Pinned contracts">
+                <code>
+                  {node.config.input_schema.length} input · {node.config.output_schema.length} output fields
+                </code>
+              </NodeFormField>
+            ) : null}
           </>
         ) : null}
       </div>

webui/src/features/dataflow/model.ts

@@ -272,7 +272,7 @@ export const FALLBACK_NODE_LIBRARY: NodeTypeDefinition[] = [
     "transform",
     "Transform",
     "Reusable subflow",
-    "Run a pinned parameterized template snapshot.",
+    "Run a Policy-authorized, server-resolved immutable definition revision.",
     "boxes",
     input,
     output,

webui/src/i18n/generatedTranslations.ts

@@ -56,6 +56,25 @@ const en = {
   "New pipeline": "New pipeline",
   "Definition settings": "Definition settings",
   "Reuse as scoped copy": "Reuse as scoped copy",
+  "Review source update": "Review source update",
+  "Adopt source revision": "Adopt source revision",
+  "Source unavailable": "Source unavailable",
+  "Source update available": "Source update available",
+  "Source definition": "Source definition",
+  "Source revision current": "Source revision current",
+  "Unapplied source update": "Unapplied source update",
+  "Apply the reviewed source update or discard the review reason before leaving.": "Apply the reviewed source update or discard the review reason before leaving.",
+  "The source definition is no longer available.": "The source definition is no longer available.",
+  "This copy already pins the current source revision.": "This copy already pins the current source revision.",
+  "Adopting the update replaces the copy's current graph with the exact reviewed source revision and returns the copy to draft. Existing revisions, run evidence and rebase provenance remain immutable.": "Adopting the update replaces the copy's current graph with the exact reviewed source revision and returns the copy to draft. Existing revisions, run evidence and rebase provenance remain immutable.",
+  "Review reason": "Review reason",
+  "Record what was reviewed and why this source revision is appropriate for the scoped copy.": "Record what was reviewed and why this source revision is appropriate for the scoped copy.",
+  "Reusable input binding": "Reusable input binding",
+  "A reusable definition must mark exactly one typed inline source as the rows supplied by its caller.": "A reusable definition must mark exactly one typed inline source as the rows supplied by its caller.",
+  "Reusable definition": "Reusable definition",
+  "Choose a reusable definition": "Choose a reusable definition",
+  "Template version": "Template version",
+  "Pinned contracts": "Pinned contracts",
   "Automation triggers": "Automation triggers",
   "Discard changes": "Discard changes",
   "Delete pipeline": "Delete pipeline",
@@ -136,6 +155,25 @@ const de: Record<keyof typeof en, string> = {
   "New pipeline": "Neuer Datenfluss",
   "Definition settings": "Definitionseinstellungen",
   "Reuse as scoped copy": "Als eingegrenzte Kopie verwenden",
+  "Review source update": "Aktualisierung der Quelle prüfen",
+  "Adopt source revision": "Quellrevision übernehmen",
+  "Source unavailable": "Quelle nicht verfügbar",
+  "Source update available": "Aktualisierung der Quelle verfügbar",
+  "Source definition": "Quelldefinition",
+  "Source revision current": "Quellrevision aktuell",
+  "Unapplied source update": "Nicht übernommene Quellenaktualisierung",
+  "Apply the reviewed source update or discard the review reason before leaving.": "Übernehmen Sie die geprüfte Quellenaktualisierung oder verwerfen Sie die Prüfbegründung, bevor Sie den Dialog verlassen.",
+  "The source definition is no longer available.": "Die Quelldefinition ist nicht mehr verfügbar.",
+  "This copy already pins the current source revision.": "Diese Kopie ist bereits an die aktuelle Quellrevision gebunden.",
+  "Adopting the update replaces the copy's current graph with the exact reviewed source revision and returns the copy to draft. Existing revisions, run evidence and rebase provenance remain immutable.": "Die Übernahme ersetzt den aktuellen Graphen der Kopie durch die exakt geprüfte Quellrevision und setzt die Kopie auf Entwurf zurück. Bestehende Revisionen, Ausführungsnachweise und die Herkunft der Übernahme bleiben unveränderlich.",
+  "Review reason": "Prüfbegründung",
+  "Record what was reviewed and why this source revision is appropriate for the scoped copy.": "Dokumentieren Sie, was geprüft wurde und warum diese Quellrevision für die eingegrenzte Kopie geeignet ist.",
+  "Reusable input binding": "Wiederverwendbare Eingabebindung",
+  "A reusable definition must mark exactly one typed inline source as the rows supplied by its caller.": "Eine wiederverwendbare Definition muss genau eine typisierte Inline-Quelle als die vom Aufrufer gelieferten Zeilen kennzeichnen.",
+  "Reusable definition": "Wiederverwendbare Definition",
+  "Choose a reusable definition": "Wiederverwendbare Definition auswählen",
+  "Template version": "Vorlagenversion",
+  "Pinned contracts": "Gebundene Verträge",
   "Automation triggers": "Automatisierungsauslöser",
   "Discard changes": "Änderungen verwerfen",
   "Delete pipeline": "Datenfluss löschen",