fix(security): isolate reference execution and bound source inputs

src/govoplan_dataflow/backend/backends/base.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+import math
 from typing import Any, Mapping, Protocol, runtime_checkable
 
 from govoplan_dataflow.backend.batches import TypedBatch
@@ -33,8 +34,8 @@ class ExecutionBudget:
             raise ValueError("Execution output row limit must be positive.")
         if self.max_batch_bytes < 1:
             raise ValueError("Execution byte limit must be positive.")
-        if self.max_wall_seconds <= 0:
-            raise ValueError("Execution time limit must be positive.")
+        if not math.isfinite(self.max_wall_seconds) or self.max_wall_seconds <= 0:
+            raise ValueError("Execution time limit must be finite and positive.")
         if self.max_memory_bytes < 64 * 1024 * 1024:
             raise ValueError("Execution memory limit must be at least 64 MiB.")
         if self.max_concurrency < 1:
@@ -106,11 +107,19 @@ class BackendExecutionError(RuntimeError):
         code: str = "backend.execution",
         node_id: str | None = None,
         diagnostics: tuple[DataflowDiagnostic, ...] = (),
+        node_diagnostics: tuple[NodePreviewDiagnostic, ...] = (),
+        source_fingerprints: tuple[dict[str, Any], ...] = (),
+        input_row_count: int = 0,
+        node_preview: NodePreviewResult | None = None,
     ) -> None:
         super().__init__(message)
         self.code = code
         self.node_id = node_id
         self.diagnostics = diagnostics
+        self.node_diagnostics = node_diagnostics
+        self.source_fingerprints = source_fingerprints
+        self.input_row_count = input_row_count
+        self.node_preview = node_preview
 
 
 @runtime_checkable

src/govoplan_dataflow/backend/backends/reference.py

@@ -1,10 +1,25 @@
 from __future__ import annotations
 
+from dataclasses import asdict
+import hashlib
+import json
+import math
+
+from govoplan_core.security.bounded_process import (
+    ProcessBudgetError,
+    ProcessLimits,
+    run_bounded_operation,
+)
+from govoplan_core.security.worker_payload import (
+    decode_worker_payload,
+    encode_worker_payload,
+)
 from govoplan_dataflow.backend.backends.base import (
     BackendExecutionError,
     BackendExecutionRequest,
     BackendExecutionResult,
     BackendSource,
+    ExecutionBudget,
     canonical_result_schema,
 )
 from govoplan_dataflow.backend.batches import TypedBatch
@@ -14,8 +29,17 @@ from govoplan_dataflow.backend.executor import (
     ResolvedSource,
     execute_preview,
 )
-from govoplan_dataflow.backend.ir import IrExecutionResult, ir_to_graph
-from govoplan_dataflow.backend.schemas import GraphNode
+from govoplan_dataflow.backend.ir import IrExecutionResult, IrSchema, TypedGraphIr, ir_to_graph
+from govoplan_dataflow.backend.planner import ExecutionPlan
+from govoplan_dataflow.backend.schemas import (
+    DataflowDiagnostic,
+    GraphNode,
+    NodePreviewDiagnostic,
+    NodePreviewResult,
+)
+
+
+_TRANSPORT_BYTES = 32 * 1024 * 1024
 
 
 class ReferenceExecutionBackend:
@@ -32,6 +56,52 @@ class ReferenceExecutionBackend:
         self,
         request: BackendExecutionRequest,
     ) -> BackendExecutionResult:
+        _validate_source_batches(request)
+        try:
+            limits = ProcessLimits(
+                wall_seconds=request.budget.max_wall_seconds,
+                cpu_seconds=max(1, math.ceil(request.budget.max_wall_seconds)),
+                memory_bytes=request.budget.max_memory_bytes,
+                input_bytes=_TRANSPORT_BYTES,
+                output_bytes=_TRANSPORT_BYTES,
+            )
+            payload = encode_worker_payload(_request_payload(request), max_bytes=_TRANSPORT_BYTES)
+            response = decode_worker_payload(
+                run_bounded_operation(_execute_reference_worker, payload, limits=limits),
+                max_bytes=_TRANSPORT_BYTES,
+            )
+        except ProcessBudgetError as exc:
+            raise BackendExecutionError(str(exc), code=f"backend.process.{exc.code}") from exc
+        except ValueError as exc:
+            raise BackendExecutionError(
+                "Reference execution exceeds its supported process/transport budget.",
+                code="backend.budget",
+            ) from exc
+        if "error" in response:
+            error = response["error"]
+            raise BackendExecutionError(
+                error["message"], code=error["code"], node_id=error["node_id"],
+                diagnostics=tuple(DataflowDiagnostic.model_validate(item) for item in error["diagnostics"]),
+                node_diagnostics=tuple(NodePreviewDiagnostic.model_validate(item) for item in error["node_diagnostics"]),
+                source_fingerprints=error["source_fingerprints"],
+                input_row_count=error["input_row_count"],
+                node_preview=NodePreviewResult.model_validate(error["node_preview"]) if error["node_preview"] else None,
+            )
+        batch = _batch_from_payload(response["batch"])
+        batch.ensure_within(
+            max_rows=request.budget.max_output_rows,
+            max_bytes=request.budget.max_batch_bytes,
+        )
+        return BackendExecutionResult(
+            contract=IrExecutionResult.model_validate(response["contract"]),
+            batch=batch,
+            node_diagnostics=tuple(NodePreviewDiagnostic.model_validate(item) for item in response["node_diagnostics"]),
+            node_preview=NodePreviewResult.model_validate(response["node_preview"]) if response["node_preview"] else None,
+            metadata=response["metadata"],
+        )
+
+    def _execute_in_process(self, request: BackendExecutionRequest) -> BackendExecutionResult:
+        """Pure reference evaluation, called only inside the disposable worker."""
         _validate_source_batches(request)
         try:
             result = execute_preview(
@@ -50,6 +120,10 @@ class ReferenceExecutionBackend:
                 code="backend.reference",
                 node_id=exc.node_id,
                 diagnostics=tuple(exc.diagnostics),
+                node_diagnostics=tuple(exc.node_diagnostics),
+                source_fingerprints=tuple(exc.source_fingerprints),
+                input_row_count=exc.input_row_count,
+                node_preview=exc.node_preview,
             ) from exc
         observed_batch = TypedBatch.from_rows(result.rows)
         batch = TypedBatch.from_rows(
@@ -97,6 +171,90 @@ class ReferenceExecutionBackend:
         )
 
 
+def _batch_payload(batch: TypedBatch) -> dict:
+    return {
+        "schema": batch.schema.model_dump(mode="python"),
+        "columns": dict(batch.columns),
+        "row_count": batch.row_count,
+        "byte_count": batch.byte_count,
+    }
+
+
+def _batch_from_payload(value: dict) -> TypedBatch:
+    return TypedBatch(
+        schema=IrSchema.model_validate(value["schema"]), columns=value["columns"],
+        row_count=value["row_count"], byte_count=value["byte_count"],
+    )
+
+
+def _request_payload(request: BackendExecutionRequest) -> dict:
+    plan = request.plan
+    return {
+        "plan": {
+            "graph": plan.graph.model_dump(mode="python"),
+            "ordered_node_ids": plan.ordered_node_ids,
+            "diagnostics": tuple(item.model_dump(mode="python") for item in plan.diagnostics),
+            "generated_sql": plan.generated_sql,
+            "sql_diagnostics": tuple(item.model_dump(mode="python") for item in plan.sql_diagnostics),
+            "semantic_hash": plan.semantic_hash,
+        },
+        "budget": asdict(request.budget),
+        "preview_node_id": request.preview_node_id,
+        "sources": {
+            key: {
+                "node_id": source.node_id, "batch": _batch_payload(source.batch),
+                "source_ref": source.source_ref, "provider": source.provider,
+                "fingerprint": source.fingerprint, "total_rows": source.total_rows,
+                "truncated": source.truncated, "source_name": source.source_name,
+                "kind": source.kind,
+            }
+            for key, source in request.sources.items()
+        },
+    }
+
+
+def _execute_reference_worker(payload: bytes) -> bytes:
+    """Data-only worker boundary; no database, provider callback or principal."""
+    value = decode_worker_payload(payload, max_bytes=_TRANSPORT_BYTES)
+    plan = value["plan"]
+    request = BackendExecutionRequest(
+        plan=ExecutionPlan(
+            graph=TypedGraphIr.model_validate(plan["graph"]),
+            ordered_node_ids=plan["ordered_node_ids"],
+            diagnostics=tuple(DataflowDiagnostic.model_validate(item) for item in plan["diagnostics"]),
+            generated_sql=plan["generated_sql"],
+            sql_diagnostics=tuple(DataflowDiagnostic.model_validate(item) for item in plan["sql_diagnostics"]),
+            semantic_hash=plan["semantic_hash"],
+        ),
+        budget=ExecutionBudget(**value["budget"]),
+        preview_node_id=value["preview_node_id"],
+        sources={
+            key: BackendSource(**{**source, "batch": _batch_from_payload(source["batch"])})
+            for key, source in value["sources"].items()
+        },
+    )
+    try:
+        result = ReferenceExecutionBackend()._execute_in_process(request)
+    except BackendExecutionError as exc:
+        response = {"error": {
+            "message": str(exc), "code": exc.code, "node_id": exc.node_id,
+            "diagnostics": tuple(item.model_dump(mode="python") for item in exc.diagnostics),
+            "node_diagnostics": tuple(item.model_dump(mode="python") for item in exc.node_diagnostics),
+            "source_fingerprints": exc.source_fingerprints,
+            "input_row_count": exc.input_row_count,
+            "node_preview": exc.node_preview.model_dump(mode="python") if exc.node_preview else None,
+        }}
+    else:
+        response = {
+            "contract": result.contract.model_dump(mode="python"),
+            "batch": _batch_payload(result.batch),
+            "node_diagnostics": tuple(item.model_dump(mode="python") for item in result.node_diagnostics),
+            "node_preview": result.node_preview.model_dump(mode="python") if result.node_preview else None,
+            "metadata": dict(result.metadata),
+        }
+    return encode_worker_payload(response, max_bytes=_TRANSPORT_BYTES)
+
+
 def _validate_source_batches(request: BackendExecutionRequest) -> None:
     for source in request.sources.values():
         try:
@@ -139,6 +297,7 @@ def _source_for_node(
     node: GraphNode,
 ) -> BackendSource | None:
     candidates = (
+        reference_source_key(node),
         node.id,
         str(node.config.get("source_ref") or ""),
         str(node.config.get("source_name") or ""),
@@ -153,4 +312,10 @@ def _source_for_node(
     )
 
 
+def reference_source_key(node: GraphNode) -> str:
+    """Nested graphs may reuse node IDs; bind resolved data to the full config."""
+    content = json.dumps(node.config, sort_keys=True, separators=(",", ":"), default=str)
+    return "reference-config:" + hashlib.sha256(content.encode("utf-8")).hexdigest()
+
+
 __all__ = ["ReferenceExecutionBackend"]

src/govoplan_dataflow/backend/manifest.py

@@ -154,6 +154,40 @@ ROLE_TEMPLATES = (
 )
 
 DOCUMENTATION = localize_documentation_topics((
+    DocumentationTopic(
+        id="dataflow.reference-worker-limits",
+        title="Reference execution process limits",
+        summary="Contain expensive expressions and intermediate allocations without changing datasource authority.",
+        body=(
+            "Reference previews and reference development runs evaluate in a fresh disposable process, including regex and aggregate intermediate allocations. "
+            "The existing row, node and per-result byte checks remain. The process additionally enforces the request's wall-clock and memory budgets "
+            "(default preview: 2 seconds and 256 MiB virtual address space), rounded-up CPU seconds, no regular-file output, and 32 MiB per data-only input/result transport. "
+            "Supported process budgets are at most 600 seconds and 8 GiB; unsupported controls or exceeded limits fail with structured backend.process diagnostics, never inline fallback. "
+            "Datasource authorization and bounded source reads remain in the parent, including nested subflow sources; sessions and credentials are not passed to the child. "
+            "Reference source collection also checks a cumulative 32 MiB typed-data budget before constructing another columnar copy; providers retain their separate per-read limits. "
+            "Completed node diagnostics survive ordinary evaluation errors; a killed worker returns no partial rows or invented node progress. "
+            "GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY limits shared isolated-work admission per API/worker process, default 1; busy capacity is retryable. "
+            "It is not a fleet-wide quota or arbitrary-code sandbox. Cancellation checks before/after reference runs remain; hard wall limits stop an unresponsive expression. "
+            "Staging/production still require the separate DuckDB backend and are not converted to reference execution."
+        ),
+        layer="static", documentation_types=("user", "admin"), audience=("user", "module_admin", "operator"), order=6,
+        translations={"de": {
+            "title": "Prozessgrenzen der Referenzausführung",
+            "summary": "Aufwendige Ausdrücke und Zwischenspeicher begrenzen, ohne Datenquellenrechte zu verändern.",
+            "body": (
+                "Referenz-Vorschauen und Referenz-Entwicklungsläufe werten Ausdrücke in einem frischen, kurzlebigen Prozess aus, einschließlich regulärer Ausdrücke und großer Zwischenergebnisse. "
+                "Bestehende Zeilen-, Knoten- und Ergebnis-Bytegrenzen bleiben bestehen. Zusätzlich gelten das Laufzeit- und Speicherbudget der Anfrage "
+                "(Vorschau standardmäßig 2 Sekunden und 256 MiB virtueller Adressraum), aufgerundete CPU-Sekunden, keine regulären Ausgabedateien und je 32 MiB für den reinen Datentransport. "
+                "Prozessbudgets unterstützen höchstens 600 Sekunden und 8 GiB. Fehlende Betriebssystemkontrollen oder überschrittene Grenzen erzeugen strukturierte backend.process-Diagnosen, ohne Ausweichbetrieb im Hauptprozess. "
+                "Datenquellenrechte und begrenzte Quellabrufe werden im Hauptprozess geprüft, auch für verschachtelte Teilflüsse; Sitzungen und Zugangsdaten gelangen nicht in den Kindprozess. "
+                "Referenz-Quellabrufe prüfen außerdem zusammen höchstens 32 MiB typisierte Daten, bevor eine weitere spaltenweise Kopie entsteht; getrennte Abrufgrenzen der Anbieter bleiben bestehen. "
+                "Gewöhnliche Auswertungsfehler behalten bereits abgeschlossene Knotendiagnosen. Ein gestoppter Prozess liefert keine Teilzeilen und keinen erfundenen Knotenfortschritt. "
+                "GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY begrenzt gemeinsam genutzte isolierte Arbeit je API-/Worker-Prozess, standardmäßig 1; bei Auslastung ist ein erneuter Versuch möglich. "
+                "Dies ist weder eine systemweite Quote noch eine Sandbox für beliebigen Code. Abbruchprüfungen vor und nach Referenzläufen bleiben erhalten; harte Laufzeitgrenzen stoppen hängende Ausdrücke. "
+                "Staging und Produktion benötigen weiterhin das gesonderte DuckDB-Backend und wechseln nicht zur Referenzausführung."
+            ),
+        }},
+    ),
     DocumentationTopic(
         id="dataflow.workspace-layout",
         title="Dataflow workspace actions",

src/govoplan_dataflow/backend/service.py

@@ -28,6 +28,7 @@ from govoplan_core.core.datasources import (
     datasource_publication,
 )
 from govoplan_core.db.base import utcnow
+from govoplan_core.security.worker_payload import WorkerPayloadError, encode_worker_payload
 from govoplan_dataflow.backend.backends import (
     BackendExecutionError,
     BackendSource,
@@ -35,6 +36,7 @@ from govoplan_dataflow.backend.backends import (
     execute_typed_graph,
 )
 from govoplan_dataflow.backend.batches import TypedBatch
+from govoplan_dataflow.backend.backends.reference import reference_source_key
 from govoplan_dataflow.backend.db.models import (
     DataflowPipeline,
     DataflowPipelineDeployment,
@@ -48,7 +50,6 @@ from govoplan_dataflow.backend.executor import (
     PipelineExecutionError,
     PipelineExecutionResult,
     ResolvedSource,
-    execute_preview,
 )
 from govoplan_dataflow.backend.governance import (
     definition_governance_payload,
@@ -1122,20 +1123,11 @@ def _execute_pipeline_preview(
         principal=principal,
         registry=registry,
     )
-    if backend == "reference":
-        return (
-            execute_preview(
-                graph,
-                row_limit=row_limit,
-                source_resolver=source_resolver,
-                preview_node_id=preview_node_id,
-            ),
-            EXECUTOR_VERSION,
-        )
     sources = _typed_backend_sources(
         graph,
         source_resolver=source_resolver,
-        source_limit=max(MAX_SOURCE_ROWS, row_limit),
+        source_limit=MAX_SOURCE_ROWS if backend == "reference" else max(MAX_SOURCE_ROWS, row_limit),
+        include_subflows=backend == "reference",
     )
     try:
         result = execute_typed_graph(
@@ -1149,8 +1141,14 @@ def _execute_pipeline_preview(
         raise PipelineExecutionError(
             str(exc),
             node_id=exc.node_id,
-            diagnostics=tuple(exc.diagnostics),
-            retryable=exc.code == "backend.capacity",
+            diagnostics=(*exc.diagnostics, DataflowDiagnostic(
+                severity="error", code=exc.code, message=str(exc), node_id=exc.node_id,
+            )),
+            node_diagnostics=exc.node_diagnostics,
+            source_fingerprints=exc.source_fingerprints,
+            input_row_count=exc.input_row_count,
+            node_preview=exc.node_preview,
+            retryable=exc.code in {"backend.capacity", "backend.process.busy"},
         ) from exc
     columns = [
         PreviewColumn(
@@ -1227,13 +1225,51 @@ def _typed_backend_sources(
     *,
     source_resolver,
     source_limit: int = MAX_SOURCE_ROWS,
+    include_subflows: bool = False,
+    _depth: int = 0,
+    _remaining_source_bytes: list[int] | None = None,
 ) -> dict[str, BackendSource]:
+    if _depth > 5:
+        raise PipelineExecutionError("Subflows are limited to five nested levels.")
+    if _remaining_source_bytes is None:
+        _remaining_source_bytes = [32 * 1024 * 1024]
     sources: dict[str, BackendSource] = {}
     for node in graph.nodes:
+        if include_subflows and node.type == "subflow":
+            from govoplan_dataflow.backend.subflows import substitute_parameters
+
+            parameters = node.config.get("parameters")
+            nested = PipelineGraph.model_validate(substitute_parameters(
+                node.config.get("graph"), parameters if isinstance(parameters, dict) else {},
+            ))
+            sources.update(_typed_backend_sources(
+                nested, source_resolver=source_resolver, source_limit=source_limit,
+                include_subflows=True, _depth=_depth + 1,
+                _remaining_source_bytes=_remaining_source_bytes,
+            ))
         if node.type != "source.reference":
             continue
+        if include_subflows and _remaining_source_bytes[0] <= 0:
+            raise PipelineExecutionError(
+                "Combined source data exceeds the 32 MiB transfer budget.", node_id=node.id,
+            )
         resolved = source_resolver(node, source_limit)
-        sources[node.id] = BackendSource(
+        if include_subflows:
+            try:
+                # Check before constructing another columnar copy. The provider
+                # still owns bounds on its individual authorized read; do not keep
+                # accumulating individually valid batches before the worker gate.
+                encoded_size = len(encode_worker_payload(
+                    tuple(dict(row) for row in resolved.rows), max_bytes=_remaining_source_bytes[0],
+                ))
+            except WorkerPayloadError as exc:
+                raise PipelineExecutionError(
+                    "Combined source data exceeds the 32 MiB transfer budget or contains unsupported values.",
+                    node_id=node.id,
+                ) from exc
+            _remaining_source_bytes[0] -= encoded_size
+        key = reference_source_key(node) if include_subflows else node.id
+        sources[key] = BackendSource(
             node_id=node.id,
             batch=TypedBatch.from_rows(resolved.rows),
             source_ref=resolved.source_ref,

tests/test_reference_process.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+from decimal import Decimal
+import time
+import unittest
+from unittest.mock import patch
+
+from govoplan_core.security.bounded_process import ProcessBudgetError
+from govoplan_dataflow.backend.backends import (
+    BackendExecutionError, BackendSource, ExecutionBudget, execute_typed_graph,
+)
+from govoplan_dataflow.backend.backends.reference import ReferenceExecutionBackend
+from govoplan_dataflow.backend.batches import TypedBatch
+from govoplan_dataflow.backend.executor import PipelineExecutionError, ResolvedSource
+from govoplan_dataflow.backend.manifest import get_manifest
+from govoplan_dataflow.backend.schemas import GraphEdge, GraphNode, GraphPosition, PipelineGraph
+from govoplan_dataflow.backend.service import _execute_pipeline_preview, _typed_backend_sources
+
+
+def graph_for(expression: str = "value", rows: list | None = None) -> PipelineGraph:
+    nodes = [
+        GraphNode(id="source", type="source.inline", label="Source", position=GraphPosition(x=0, y=0),
+                  config={"source_name": "records", "rows": rows or [{"value": "normal"}]}),
+        GraphNode(id="expression", type="expression", label="Expression", position=GraphPosition(x=100, y=0),
+                  config={"target_column": "result", "expression": expression, "result_type": "unknown"}),
+        GraphNode(id="output", type="output", label="Output", position=GraphPosition(x=200, y=0), config={}),
+    ]
+    return PipelineGraph(nodes=nodes, edges=[
+        GraphEdge(id="first", source="source", target="expression"),
+        GraphEdge(id="second", source="expression", target="output"),
+    ])
+
+
+class ReferenceProcessTests(unittest.TestCase):
+    def test_non_finite_deadlines_are_rejected_before_registry_wait(self) -> None:
+        for value in (float("nan"), float("inf")):
+            with self.subTest(value=value), self.assertRaisesRegex(ValueError, "finite"):
+                ExecutionBudget(max_wall_seconds=value)
+
+    def test_real_child_not_parent_helper_and_typed_result_preserved(self) -> None:
+        row = {"value": Decimal("1.20"), "date": date(2026, 9, 8),
+               "when": datetime(2026, 9, 8, tzinfo=timezone.utc), "binary": b"\x00\xff"}
+        graph = graph_for()
+        graph.nodes[0] = graph.nodes[0].model_copy(update={"type": "source.reference", "config": {
+            "source_ref": "datasource:fixture", "source_name": "records",
+        }})
+        source = BackendSource(node_id="source", batch=TypedBatch.from_rows([row]),
+                               source_ref="datasource:fixture", provider="test", fingerprint="pinned", total_rows=1)
+        with patch.object(ReferenceExecutionBackend, "_execute_in_process", side_effect=AssertionError("parent evaluation")):
+            result = execute_typed_graph(graph, backend="reference", sources={"source": source})
+        self.assertEqual(result.rows, [{**row, "result": Decimal("1.20")}])
+        self.assertEqual(result.contract.lineage.source_fingerprints[0]["fingerprint"], "pinned")
+
+    def test_real_pathological_regex_is_stopped_with_structured_failure(self) -> None:
+        graph = graph_for("regexp_full_match(value, '(a+)+$')", [{"value": "a" * 100 + "!"}])
+        started = time.monotonic()
+        with self.assertRaises(BackendExecutionError) as caught:
+            execute_typed_graph(graph, backend="reference", budget=ExecutionBudget(max_wall_seconds=2))
+        self.assertIn(caught.exception.code, {"backend.process.timeout", "backend.process.cpu_limit"})
+        self.assertLess(time.monotonic() - started, 4)
+
+    def test_aggregate_padding_allocation_is_contained_by_child_memory_limit(self) -> None:
+        graph = graph_for("lpad(value, 900000, '0')", [{"value": "x"} for _ in range(250)])
+        with self.assertRaises(BackendExecutionError) as caught:
+            execute_typed_graph(graph, backend="reference", budget=ExecutionBudget(
+                max_wall_seconds=5, max_memory_bytes=128 * 1024 * 1024,
+            ))
+        self.assertEqual(caught.exception.code, "backend.process.memory_limit")
+
+    def test_busy_preview_is_retryable_and_never_evaluates_inline(self) -> None:
+        with patch("govoplan_dataflow.backend.backends.reference.run_bounded_operation", side_effect=ProcessBudgetError("busy")):
+            with self.assertRaises(PipelineExecutionError) as caught:
+                _execute_pipeline_preview(graph_for(), session=None, principal=None, registry=None,
+                                          backend="reference", row_limit=10, preview_node_id=None)
+        self.assertTrue(caught.exception.retryable)
+        self.assertEqual(caught.exception.diagnostics[-1].code, "backend.process.busy")
+
+    def test_nested_source_ids_do_not_alias_different_authorized_data(self) -> None:
+        outer = graph_for()
+        inner = graph_for()
+        for graph, ref in ((outer, "datasource:outer"), (inner, "datasource:inner")):
+            graph.nodes[0] = graph.nodes[0].model_copy(update={"type": "source.reference", "config": {
+                "source_ref": ref, "source_name": "records",
+            }})
+        outer.nodes[1] = outer.nodes[1].model_copy(update={"type": "subflow", "config": {
+            "graph": inner.model_dump(mode="python"), "parameters": {},
+        }})
+        def resolve(node, limit):
+            return ResolvedSource(rows=({"value": node.config["source_ref"]},),
+                                  source_ref=node.config["source_ref"], provider="test",
+                                  fingerprint=node.config["source_ref"], total_rows=1)
+        sources = _typed_backend_sources(outer, source_resolver=resolve, include_subflows=True)
+        self.assertEqual({source.source_ref for source in sources.values()}, {"datasource:outer", "datasource:inner"})
+        self.assertEqual(len(sources), 2)
+
+    def test_static_worker_documentation_is_bilingual(self) -> None:
+        topic = next(item for item in get_manifest().documentation if item.id == "dataflow.reference-worker-limits")
+        for body in (topic.body, topic.translations["de"]["body"]):
+            self.assertIn("GOVOPLAN_ISOLATED_PROCESS_CONCURRENCY", body)
+            self.assertIn("32 MiB", body)
+        self.assertEqual(set(topic.documentation_types), {"user", "admin"})
+
+    def test_cumulative_source_budget_stops_before_reading_further_sources(self) -> None:
+        graph = graph_for()
+        graph.nodes = [graph.nodes[0].model_copy(update={
+            "id": f"source-{index}", "type": "source.reference", "config": {"source_ref": f"fixture:{index}"},
+        }) for index in range(3)]
+        calls = []
+        def resolve(node, limit):
+            calls.append(node.id)
+            return ResolvedSource(rows=({"value": "x" * 100},), source_ref=node.config["source_ref"],
+                                  provider="test", fingerprint="fixed", total_rows=1)
+        with self.assertRaisesRegex(PipelineExecutionError, "Combined source data"):
+            _typed_backend_sources(graph, source_resolver=resolve, include_subflows=True,
+                                   _remaining_source_bytes=[200])
+        self.assertEqual(calls, ["source-0", "source-1"])

tests/test_reference_subflow_process.py

@@ -0,0 +1,194 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock, patch
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.datasources import (
+    CAPABILITY_DATASOURCE_CATALOGUE,
+    DatasourceAccessError,
+    DatasourceCatalogueProvider,
+    DatasourceDescriptor,
+    DatasourceReadRequest,
+    DatasourceReadResult,
+)
+from govoplan_core.security.bounded_process import run_bounded_operation
+from govoplan_dataflow.backend.backends.reference import ReferenceExecutionBackend
+from govoplan_dataflow.backend.executor import EXECUTOR_VERSION, PipelineExecutionError
+from govoplan_dataflow.backend.graph import validate_graph
+from govoplan_dataflow.backend.schemas import GraphEdge, GraphNode, GraphPosition, PipelineGraph
+from govoplan_dataflow.backend.service import _execute_pipeline_preview
+
+
+OUTER_ROWS = ({"id": "outer", "amount": 15}, {"id": "outer-low", "amount": 5})
+INNER_ROWS = ({"id": "inner", "amount": 25}, {"id": "inner-low", "amount": 2})
+
+
+def node(node_id: str, node_type: str, config: dict) -> GraphNode:
+    return GraphNode(
+        id=node_id, type=node_type, label=node_id,
+        position=GraphPosition(x=0, y=0), config=config,
+    )
+
+
+def nested_source_graph() -> PipelineGraph:
+    # Both external sources deliberately share their node ID and logical name.
+    # The pinned subflow still has exactly one distinct inline input binding.
+    nested = PipelineGraph(
+        nodes=[
+            node("input", "source.inline", {
+                "source_name": "bound_input", "rows": [], "input_binding": True,
+            }),
+            node("shared", "source.reference", {
+                "source_name": "records", "source_ref": {"$parameter": "source_ref"},
+                "expected_fingerprint": "inner-pinned", "consistency": "frozen",
+            }),
+            node("union", "combine.union", {"mode": "all"}),
+            node("minimum", "filter.expression", {"expression": "amount >= ${minimum}"}),
+            node("output", "output", {}),
+        ],
+        edges=[
+            GraphEdge(id="input-union", source="input", target="union"),
+            GraphEdge(id="shared-union", source="shared", target="union"),
+            GraphEdge(id="union-minimum", source="union", target="minimum"),
+            GraphEdge(id="minimum-output", source="minimum", target="output"),
+        ],
+    )
+    return PipelineGraph(
+        nodes=[
+            node("shared", "source.reference", {
+                "source_name": "records", "source_ref": "datasource:outer",
+                "expected_fingerprint": "outer-pinned", "consistency": "current",
+            }),
+            node("nested", "subflow", {
+                "template_ref": "fixture-nested-source", "template_version": "1",
+                "parameters": {"source_ref": "datasource:inner", "minimum": 10},
+                "graph": nested.model_dump(mode="python"),
+            }),
+            node("output", "output", {}),
+        ],
+        edges=[
+            GraphEdge(id="shared-nested", source="shared", target="nested"),
+            GraphEdge(id="nested-output", source="nested", target="output"),
+        ],
+    )
+
+
+class ReferenceSubflowProcessTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.graph = nested_source_graph()
+        self.assertEqual([], [item.model_dump() for item in validate_graph(self.graph) if item.severity == "error"])
+        self.session = Mock(spec=Session)
+        self.principal = ApiPrincipal(
+            principal=PrincipalRef(
+                account_id="fixture-account", membership_id="fixture-membership",
+                tenant_id="fixture-tenant", scopes=frozenset(),
+            ),
+            account=object(), user=object(),
+        )
+        self.provider = Mock(spec=DatasourceCatalogueProvider)
+        self.provider.read_datasource.side_effect = self.read_source
+        self.registry = SimpleNamespace(
+            has_capability=lambda name: name == CAPABILITY_DATASOURCE_CATALOGUE,
+            capability=lambda name: self.provider if name == CAPABILITY_DATASOURCE_CATALOGUE else None,
+        )
+        self.reads: list[tuple[int, DatasourceReadRequest]] = []
+        self.denied = False
+
+    def read_source(self, session, principal, *, request: DatasourceReadRequest) -> DatasourceReadResult:
+        self.assertIs(session, self.session)
+        self.assertIs(principal, self.principal)
+        self.reads.append((os.getpid(), request))
+        if request.datasource_ref == "datasource:inner" and self.denied:
+            raise DatasourceAccessError("Current principal cannot read datasource:inner.")
+        rows, fingerprint = {
+            "datasource:outer": (OUTER_ROWS, "outer-pinned"),
+            "datasource:inner": (INNER_ROWS, "inner-pinned"),
+        }[request.datasource_ref]
+        self.assertEqual(fingerprint, request.expected_fingerprint)
+        return DatasourceReadResult(
+            datasource=DatasourceDescriptor(
+                ref=request.datasource_ref, source_name="records", name="Fixture source",
+                kind="custom", mode="static", shape="tabular",
+                fingerprint=fingerprint, provider="fixture-catalogue",
+            ),
+            rows=rows, total_rows=len(rows), truncated=False,
+        )
+
+    def preview(self, *, row_limit: int = 10):
+        return _execute_pipeline_preview(
+            self.graph, session=self.session, principal=self.principal,
+            registry=self.registry, backend="reference", row_limit=row_limit,
+            preview_node_id="nested",
+        )
+
+    def test_nested_parameterized_sources_execute_in_real_child_without_id_aliasing(self) -> None:
+        with (
+            patch.object(ReferenceExecutionBackend, "_execute_in_process", side_effect=AssertionError("parent evaluation")),
+            patch("govoplan_dataflow.backend.backends.reference.run_bounded_operation", wraps=run_bounded_operation) as worker,
+        ):
+            result, version = self.preview()
+        worker.assert_called_once()
+        self.assertEqual(EXECUTOR_VERSION, version)
+        self.assertEqual([OUTER_ROWS[0], INNER_ROWS[0]], result.rows)
+        self.assertEqual(2, result.total_rows)
+        self.assertFalse(result.truncated)
+        self.assertEqual(2, result.input_row_count)  # Root input count, not the nested binding again.
+        self.assertIsNotNone(result.node_preview)
+        self.assertEqual("nested", result.node_preview.node_id)
+        self.assertEqual(result.rows, result.node_preview.rows)
+        self.assertEqual(2, result.node_preview.total_rows)
+        self.assertEqual(
+            [("shared", "succeeded", 0, 2), ("nested", "succeeded", 2, 2), ("output", "succeeded", 2, 2)],
+            [(item.node_id, item.status, item.input_rows, item.output_rows) for item in result.node_diagnostics],
+        )
+        binding_hash = hashlib.sha256(json.dumps(list(OUTER_ROWS), sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+        self.assertEqual([
+            {"node_id": "shared", "source_ref": "datasource:outer", "source_name": "records", "kind": "datasource", "provider": "fixture-catalogue", "fingerprint": "outer-pinned", "row_count": 2, "preview_rows": 2, "truncated": False},
+            {"node_id": "input", "source_name": "bound_input", "kind": "inline", "fingerprint": binding_hash, "row_count": 2, "subflow_node_id": "nested"},
+            {"node_id": "shared", "source_ref": "datasource:inner", "source_name": "records", "kind": "datasource", "provider": "fixture-catalogue", "fingerprint": "inner-pinned", "row_count": 2, "preview_rows": 2, "truncated": False, "subflow_node_id": "nested"},
+        ], result.source_fingerprints)
+        self.assertEqual(
+            [("datasource:outer", "current", "outer-pinned"), ("datasource:inner", "frozen", "inner-pinned")],
+            [(request.datasource_ref, request.consistency, request.expected_fingerprint) for _, request in self.reads],
+        )
+        self.assertTrue(all(pid == os.getpid() for pid, _ in self.reads))
+        self.assertTrue(all(request.limit <= 500 and request.offset == 0 for _, request in self.reads))
+        self.assertEqual([], self.session.mock_calls)
+
+    def test_nested_result_and_node_preview_keep_full_totals_when_output_is_bounded(self) -> None:
+        result, _ = self.preview(row_limit=1)
+        self.assertEqual([OUTER_ROWS[0]], result.rows)
+        self.assertEqual(2, result.total_rows)
+        self.assertTrue(result.truncated)
+        self.assertEqual([OUTER_ROWS[0]], result.node_preview.rows)
+        self.assertEqual(2, result.node_preview.total_rows)
+        self.assertTrue(result.node_preview.truncated)
+        self.assertEqual({"outer-pinned", "inner-pinned"}, {
+            item["fingerprint"] for item in result.source_fingerprints if item["kind"] == "datasource"
+        })
+        self.assertEqual([], self.session.mock_calls)
+
+    def test_denied_nested_datasource_stops_before_worker_or_persistence(self) -> None:
+        self.denied = True
+        with (
+            patch("govoplan_dataflow.backend.service.execute_typed_graph") as execute,
+            patch("govoplan_dataflow.backend.backends.reference.run_bounded_operation") as worker,
+            self.assertRaises(PipelineExecutionError) as caught,
+        ):
+            self.preview()
+        execute.assert_not_called()
+        worker.assert_not_called()
+        self.assertEqual("shared", caught.exception.node_id)
+        self.assertEqual("Current principal cannot read datasource:inner.", str(caught.exception))
+        self.assertFalse(caught.exception.retryable)
+        self.assertIsInstance(caught.exception.__cause__, DatasourceAccessError)
+        self.assertEqual(["datasource:outer", "datasource:inner"], [request.datasource_ref for _, request in self.reads])
+        self.assertEqual([], self.session.mock_calls)