Fence and reconcile Dataflow runs

src/govoplan_dataflow/backend/manifest.py

@@ -176,6 +176,11 @@ DOCUMENTATION = (
                 "aggregate, sort, limit, output, revisioning, and bounded preview."
             ),
             "sql_safety": "Constrained AST compilation only; no pass-through execution.",
+            "run_recovery": (
+                "Database-only runs commit atomically with Core recovery evidence. "
+                "Output publication uses forward recovery and blocks blind retry "
+                "when provider acknowledgement is uncertain."
+            ),
         },
     ),
 )
@@ -432,10 +437,12 @@ manifest = ModuleManifest(
         maturity="vertical_slice",
         documentation_ref="README.md",
         test_ref="tests/test_golden_flows.py",
-        known_limits=("Execution adapters and operator recovery evidence do not yet cover every declared node family.",),
+        known_limits=(
+            "Execution adapters do not yet cover every declared node family.",
+        ),
         owned_concepts=("dataflow definition", "dataflow revision", "dataflow run", "transformation graph"),
         non_owned_concepts=("datasource binding", "connector transport", "report presentation", "workflow task"),
-        recovery_docs=("README.md",),
+        recovery_docs=("README.md", "docs/DURABLE_RUN_RECOVERY.md"),
         security_docs=("README.md",),
         operations_docs=("README.md",),
     ),

src/govoplan_dataflow/backend/recovery.py

@@ -0,0 +1,587 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import json
+from typing import Mapping, Sequence
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.recovery import (
+    RecoveryGuaranteeError,
+    RecoveryMode,
+    RecoveryOperation,
+    RecoveryPlan,
+    RecoveryStatus,
+)
+from govoplan_core.core.recovery_runtime import (
+    DurableRecoveryOperation,
+    RecoveryOperationBusy,
+    RecoveryOperationStateConflict,
+    begin_durable_recovery_operation,
+    claim_durable_recovery_operation,
+)
+from govoplan_core.core.runtime_coordination import process_runtime_identity
+from govoplan_dataflow.backend.db.models import DataflowRun
+
+
+class DataflowRecoveryError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class DataflowRecoveryDeclaration:
+    operation_type: str
+    mode: RecoveryMode
+    boundaries: tuple[Mapping[str, object], ...]
+    verification: tuple[str, ...]
+
+
+DATAFLOW_RECOVERY_OPERATIONS = (
+    DataflowRecoveryDeclaration(
+        operation_type="run.database-only",
+        mode=RecoveryMode.ATOMIC,
+        boundaries=(
+            {
+                "name": "source-resolution",
+                "classification": "recomputable",
+                "resume": "restart with the pinned definition and verified source fingerprints",
+            },
+            {
+                "name": "typed-execution",
+                "classification": "ephemeral",
+                "resume": "restart; intermediate rows are not persisted",
+            },
+            {
+                "name": "run-finalization",
+                "classification": "atomic",
+                "resume": "the run row and terminal recovery checkpoint commit together",
+            },
+        ),
+        verification=(
+            "verify the pinned definition and canonical request hashes",
+            "verify source fingerprints, diagnostics, and bounded row counts",
+            "commit the run result and terminal recovery checkpoint atomically",
+        ),
+    ),
+    DataflowRecoveryDeclaration(
+        operation_type="run.publish-output",
+        mode=RecoveryMode.FORWARD_RECOVERY,
+        boundaries=(
+            {
+                "name": "source-resolution",
+                "classification": "recomputable",
+                "resume": "restart only before output dispatch",
+            },
+            {
+                "name": "typed-execution",
+                "classification": "checkpointed",
+                "resume": "the output digest and source fingerprints identify the computed result",
+            },
+            {
+                "name": "output-publication",
+                "classification": "outcome-unknown",
+                "resume": "verify the sink by idempotency key; never publish blindly after a lost acknowledgement",
+            },
+        ),
+        verification=(
+            "verify the pinned definition, request, source, and output hashes",
+            "verify the publication, datasource, and materialization references",
+            "reconcile an uncertain sink outcome before any new attempt",
+        ),
+    ),
+)
+
+
+_DECLARATIONS = {
+    item.operation_type: item for item in DATAFLOW_RECOVERY_OPERATIONS
+}
+
+
+def dataflow_session_factory(session: Session) -> sessionmaker[Session]:
+    bind = session.get_bind()
+    if bind is None:
+        raise DataflowRecoveryError("Dataflow recovery requires a database bind")
+    return sessionmaker(bind=bind, expire_on_commit=False)
+
+
+def _canonical_sha256(value: object) -> str:
+    encoded = json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=True,
+        default=str,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _operation_type(run: DataflowRun) -> str:
+    return (
+        "run.publish-output"
+        if isinstance(run.request_.get("publication"), Mapping)
+        else "run.database-only"
+    )
+
+
+@dataclass(slots=True)
+class DataflowRunRecovery:
+    operation: DurableRecoveryOperation | None
+    operation_id: str
+    operation_type: str
+    mode: RecoveryMode
+    attempt: int
+    replayed: bool
+    publication_started: bool = False
+
+    def prepare_publication(
+        self,
+        session: Session,
+        *,
+        run: DataflowRun,
+        rows: Sequence[Mapping[str, object]],
+    ) -> None:
+        if self.mode != RecoveryMode.FORWARD_RECOVERY:
+            raise DataflowRecoveryError(
+                "Only forward-recovery runs may cross a publication boundary"
+            )
+        if self.operation is None:
+            raise DataflowRecoveryError(
+                "A replayed recovery operation cannot publish output again"
+            )
+        output_sha256 = _canonical_sha256([dict(row) for row in rows])
+        source_sha256 = _canonical_sha256(run.source_fingerprints)
+        _set_run_recovery_metadata(
+            run,
+            operation_id=self.operation_id,
+            mode=self.mode,
+            status=RecoveryStatus.RUNNING.value,
+            attempt=self.attempt,
+            boundary="output-publication",
+            output_sha256=output_sha256,
+        )
+        try:
+            session.commit()
+            self.operation.checkpoint(
+                kind="output-publication-dispatch",
+                summary="The computed output is durable before sink dispatch",
+                evidence={
+                    "effect_started": False,
+                    "output_sha256": output_sha256,
+                    "source_fingerprints_sha256": source_sha256,
+                    "output_row_count": run.output_row_count,
+                    "publication_idempotency_sha256": _canonical_sha256(
+                        f"{run.pipeline_id}:{run.idempotency_key or run.id}"
+                    ),
+                },
+            )
+        except Exception as exc:
+            session.rollback()
+            raise DataflowRecoveryError(
+                "Dataflow output could not be checkpointed before publication"
+            ) from exc
+        self.publication_started = True
+
+    def finish(self, session: Session, *, run: DataflowRun) -> None:
+        if self.operation is None:
+            return
+        evidence = _terminal_evidence(run, publication_started=self.publication_started)
+        try:
+            if self.mode == RecoveryMode.ATOMIC:
+                _set_run_recovery_metadata(
+                    run,
+                    operation_id=self.operation_id,
+                    mode=self.mode,
+                    status=(
+                        RecoveryStatus.SUCCEEDED.value
+                        if run.status == "succeeded"
+                        else RecoveryStatus.FAILED.value
+                    ),
+                    attempt=self.attempt,
+                    boundary="run-finalization",
+                )
+                if run.status == "succeeded":
+                    self.operation.commit_atomic_success(session, evidence=evidence)
+                else:
+                    self.operation.commit_atomic_failure(
+                        session,
+                        summary=run.error or f"Dataflow run ended as {run.status}",
+                        evidence=evidence,
+                    )
+                return
+
+            if self.publication_started and run.status != "succeeded":
+                run.status = "outcome_unknown"
+                run.progress_phase = "outcome_unknown"
+                run.progress_percent = min(run.progress_percent, 95)
+                run.error = (
+                    "The output provider may have accepted the publication; "
+                    "verify the sink before retrying."
+                )
+                _set_run_recovery_metadata(
+                    run,
+                    operation_id=self.operation_id,
+                    mode=self.mode,
+                    status=RecoveryStatus.OUTCOME_UNKNOWN.value,
+                    attempt=self.attempt,
+                    boundary="output-publication",
+                )
+                session.commit()
+                self.operation.unresolved(
+                    status=RecoveryStatus.OUTCOME_UNKNOWN,
+                    summary="The output acknowledgement was not conclusive",
+                    evidence={
+                        **evidence,
+                        "effect_started": True,
+                    },
+                    failure_summary=(
+                        "Inspect the publication target by stable idempotency key "
+                        "before any new output attempt"
+                    ),
+                )
+                return
+
+            terminal_status = (
+                RecoveryStatus.SUCCEEDED.value
+                if run.status == "succeeded"
+                else RecoveryStatus.RECOVERED.value
+            )
+            _set_run_recovery_metadata(
+                run,
+                operation_id=self.operation_id,
+                mode=self.mode,
+                status=terminal_status,
+                attempt=self.attempt,
+                boundary=(
+                    "output-publication"
+                    if run.status == "succeeded"
+                    else "typed-execution"
+                ),
+            )
+            if run.status == "succeeded":
+                self.operation.commit_verified_success(
+                    session,
+                    evidence=evidence,
+                )
+            else:
+                session.commit()
+                failure_summary = (
+                    run.error or f"Dataflow run ended as {run.status}"
+                )
+                self.operation.compensate(
+                    failure_summary=failure_summary,
+                    failure_evidence={
+                        **evidence,
+                        "effect_started": False,
+                    },
+                    recovery_evidence={
+                        "verified": True,
+                        "checks": {
+                            "effect_started": False,
+                            "publication_absent": True,
+                        },
+                    },
+                )
+        except Exception as exc:
+            session.rollback()
+            raise DataflowRecoveryError(
+                "Dataflow run state and recovery evidence could not be finalized"
+            ) from exc
+
+
+def begin_dataflow_run_recovery(
+    session: Session,
+    *,
+    run: DataflowRun,
+    lease_ttl_seconds: int,
+) -> DataflowRunRecovery:
+    if not run.id or not run.request_hash or not run.definition_hash:
+        raise DataflowRecoveryError(
+            "Dataflow recovery requires a run ID, request hash, and definition hash"
+        )
+    operation_type = _operation_type(run)
+    declaration = _DECLARATIONS[operation_type]
+    attempt = max(1, int(run.attempts))
+    operation_key = f"dataflow-run:{run.id}:attempt:{attempt}"
+    try:
+        started = begin_durable_recovery_operation(
+            dataflow_session_factory(session),
+            identity=process_runtime_identity(),
+            module_id="dataflow",
+            operation_type=operation_type,
+            idempotency_key=operation_key,
+            request={
+                "tenant_id": run.tenant_id,
+                "run_id": run.id,
+                "pipeline_id": run.pipeline_id,
+                "pipeline_revision_id": run.pipeline_revision_id,
+                "definition_hash": run.definition_hash,
+                "request_hash": run.request_hash,
+                "invocation_kind": run.invocation_kind,
+                "trigger_delivery_id": run.trigger_delivery_id,
+                "attempt": attempt,
+                "publication": operation_type == "run.publish-output",
+            },
+            recovery_plan=RecoveryPlan(
+                mode=declaration.mode,
+                preconditions=(
+                    "the pinned pipeline revision and canonical request hashes are present",
+                    "the current authorization and deployment are revalidated",
+                    "the run attempt owns a distributed execution fence",
+                ),
+                forward_recovery_steps=(
+                    "inspect the output sink by stable publication idempotency key",
+                    "record whether the expected output digest was accepted",
+                    "resume only after accepted or absent state is proven",
+                )
+                if declaration.mode == RecoveryMode.FORWARD_RECOVERY
+                else (),
+                verification_steps=declaration.verification,
+            ),
+            precondition_evidence={
+                "definition_hash": run.definition_hash,
+                "request_hash": run.request_hash,
+                "access_context_sha256": _canonical_sha256(run.authorization_),
+                "environment": run.environment,
+                "attempt": attempt,
+                "publication": operation_type == "run.publish-output",
+            },
+            lease_resource_key=f"dataflow:run:{run.id}",
+            lease_ttl_seconds=max(60, int(lease_ttl_seconds)),
+            resource_type="dataflow_run",
+            resource_id=run.id,
+            metadata={
+                "resources": (
+                    ["postgresql", "queue", "external-sink"]
+                    if declaration.mode == RecoveryMode.FORWARD_RECOVERY
+                    else ["postgresql"]
+                ),
+                "boundaries": [dict(item) for item in declaration.boundaries],
+                "attempt": attempt,
+            },
+        )
+    except RecoveryOperationBusy as exc:
+        raise DataflowRecoveryError(
+            "Another runtime owns this Dataflow run attempt"
+        ) from exc
+    except RecoveryOperationStateConflict as exc:
+        raise DataflowRecoveryError(
+            "This Dataflow run attempt is already active or unresolved"
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError) as exc:
+        raise DataflowRecoveryError(
+            "The recovery ledger is unavailable; Dataflow execution did not start"
+        ) from exc
+    _set_run_recovery_metadata(
+        run,
+        operation_id=started.operation_id,
+        mode=declaration.mode,
+        status=started.status,
+        attempt=attempt,
+        boundary="source-resolution",
+    )
+    return DataflowRunRecovery(
+        operation=started.operation,
+        operation_id=started.operation_id,
+        operation_type=operation_type,
+        mode=declaration.mode,
+        attempt=attempt,
+        replayed=started.replayed,
+    )
+
+
+def dataflow_run_recovery_state(
+    session: Session,
+    *,
+    run_id: str,
+) -> dict[str, object] | None:
+    return dataflow_run_recovery_states(session, run_ids=(run_id,)).get(run_id)
+
+
+def dataflow_run_recovery_states(
+    session: Session,
+    *,
+    run_ids: Sequence[str],
+) -> dict[str, dict[str, object]]:
+    normalized_ids = tuple(dict.fromkeys(str(run_id) for run_id in run_ids if run_id))
+    if not normalized_ids:
+        return {}
+    operations = session.scalars(
+        select(RecoveryOperation)
+        .where(
+            RecoveryOperation.module_id == "dataflow",
+            RecoveryOperation.resource_type == "dataflow_run",
+            RecoveryOperation.resource_id.in_(normalized_ids),
+        )
+        .order_by(RecoveryOperation.created_at.desc(), RecoveryOperation.id.desc())
+    )
+    states: dict[str, dict[str, object]] = {}
+    for operation in operations:
+        run_id = str(operation.resource_id or "")
+        if run_id and run_id not in states:
+            states[run_id] = _recovery_state(operation)
+    return states
+
+
+def _recovery_state(operation: RecoveryOperation) -> dict[str, object]:
+    status = str(operation.status)
+    return {
+        "operation_id": operation.id,
+        "operation_type": operation.operation_type,
+        "mode": operation.mode,
+        "status": status,
+        "checkpoint_count": operation.checkpoint_count,
+        "requires_attention": status
+        in {
+            RecoveryStatus.OUTCOME_UNKNOWN.value,
+            RecoveryStatus.RECOVERY_REQUIRED.value,
+            RecoveryStatus.MANUAL_INTERVENTION.value,
+        },
+        "explanation": _recovery_explanation(operation),
+    }
+
+
+def claim_stale_dataflow_recovery(
+    session: Session,
+    *,
+    operation_id: str,
+    lease_ttl_seconds: int,
+    effect_started: bool,
+) -> str:
+    try:
+        handle = claim_durable_recovery_operation(
+            dataflow_session_factory(session),
+            identity=process_runtime_identity(),
+            operation_id=operation_id,
+            lease_ttl_seconds=max(60, int(lease_ttl_seconds)),
+        )
+    except RecoveryOperationStateConflict as exc:
+        return exc.status
+    except RecoveryOperationBusy as exc:
+        raise DataflowRecoveryError(
+            "Another runtime still owns the stale Dataflow recovery fence"
+        ) from exc
+    except (RecoveryGuaranteeError, RuntimeError) as exc:
+        raise DataflowRecoveryError(
+            "The stale Dataflow recovery fence could not be claimed"
+        ) from exc
+    if effect_started:
+        handle.release_unresolved()
+    else:
+        handle.resolve_unknown(
+            effect_occurred=False,
+            summary="No output dispatch checkpoint was recorded before lease expiry",
+            evidence={
+                "verified": True,
+                "checks": {"effect_started": False},
+            },
+        )
+    session.expire_all()
+    state = dataflow_run_recovery_state(
+        session,
+        run_id=_operation_run_id(session, operation_id),
+    )
+    return str(state["status"] if state is not None else RecoveryStatus.OUTCOME_UNKNOWN.value)
+
+
+def _operation_run_id(session: Session, operation_id: str) -> str:
+    operation = session.get(RecoveryOperation, operation_id)
+    return str(operation.resource_id or "") if operation is not None else ""
+
+
+def _terminal_evidence(
+    run: DataflowRun,
+    *,
+    publication_started: bool,
+) -> dict[str, object]:
+    succeeded = run.status == "succeeded"
+    publication_expected = isinstance(run.request_.get("publication"), Mapping)
+    publication_verified = bool(
+        not publication_expected
+        or (
+            run.output_publication_ref
+            and run.output_datasource_ref
+            and run.output_materialization_ref
+        )
+    )
+    return {
+        "verified": bool((succeeded and publication_verified) or not succeeded),
+        "checks": {
+            "run_status": run.status,
+            "definition_hash": run.definition_hash,
+            "request_hash": run.request_hash,
+            "source_fingerprints_sha256": _canonical_sha256(
+                run.source_fingerprints
+            ),
+            "result_schema_sha256": _canonical_sha256(run.result_schema),
+            "input_row_count": run.input_row_count,
+            "output_row_count": run.output_row_count,
+            "publication_expected": publication_expected,
+            "publication_started": publication_started,
+            "publication_verified": publication_verified,
+            "output_publication_ref": run.output_publication_ref,
+            "output_datasource_ref": run.output_datasource_ref,
+            "output_materialization_ref": run.output_materialization_ref,
+        },
+    }
+
+
+def _set_run_recovery_metadata(
+    run: DataflowRun,
+    *,
+    operation_id: str,
+    mode: RecoveryMode,
+    status: str,
+    attempt: int,
+    boundary: str,
+    output_sha256: str | None = None,
+) -> None:
+    authorization = dict(run.authorization_)
+    recovery = {
+        "operation_id": operation_id,
+        "mode": mode.value,
+        "status": status,
+        "attempt": attempt,
+        "boundary": boundary,
+    }
+    if output_sha256:
+        recovery["output_sha256"] = output_sha256
+    authorization["recovery"] = recovery
+    run.authorization_ = authorization
+
+
+def _recovery_explanation(operation: RecoveryOperation) -> str:
+    status = str(operation.status)
+    if status == RecoveryStatus.OUTCOME_UNKNOWN.value:
+        return (
+            "The output may have been accepted. Verify the sink using the "
+            "recorded idempotency key before starting another publication."
+        )
+    if status in {
+        RecoveryStatus.RECOVERY_REQUIRED.value,
+        RecoveryStatus.RECOVERING.value,
+    }:
+        return operation.failure_summary or "The run requires explicit recovery."
+    if status == RecoveryStatus.SUCCEEDED.value:
+        return "The run outcome and declared output references were verified."
+    if status == RecoveryStatus.FAILED.value:
+        return operation.failure_summary or "The run failed before an uncertain output."
+    if status == RecoveryStatus.RECOVERED.value:
+        return "The run stopped before publication and no external output was created."
+    if status == RecoveryStatus.REJECTED.value:
+        return operation.failure_summary or "The run was definitively rejected."
+    return "The run is fenced and has not reached a verified terminal state."
+
+
+__all__ = [
+    "DATAFLOW_RECOVERY_OPERATIONS",
+    "DataflowRecoveryDeclaration",
+    "DataflowRecoveryError",
+    "DataflowRunRecovery",
+    "begin_dataflow_run_recovery",
+    "claim_stale_dataflow_recovery",
+    "dataflow_run_recovery_state",
+    "dataflow_run_recovery_states",
+    "dataflow_session_factory",
+]

src/govoplan_dataflow/backend/router.py

@@ -110,6 +110,7 @@ from govoplan_dataflow.backend.service import (
     validate_draft,
 )
 from govoplan_dataflow.backend.run_worker import run_metrics
+from govoplan_dataflow.backend.recovery import dataflow_run_recovery_states
 from govoplan_dataflow.backend.triggers import (
     create_trigger,
     delete_trigger,
@@ -982,8 +983,19 @@ def api_list_pipeline_runs(
             pipeline_id=pipeline_id,
             limit=limit,
         )
+        recovery_states = dataflow_run_recovery_states(
+            session,
+            run_ids=tuple(run.id for run in runs),
+        )
         return PipelineRunListResponse(
-            runs=[pipeline_run_response(session, run) for run in runs]
+            runs=[
+                pipeline_run_response(
+                    session,
+                    run,
+                    recovery_state=recovery_states.get(run.id, {}),
+                )
+                for run in runs
+            ]
         )
     except PermissionError as exc:
         raise _governance_http_error(exc) from exc

src/govoplan_dataflow/backend/run_worker.py

@@ -25,6 +25,13 @@ from govoplan_dataflow.backend.db.models import (
     DataflowTrigger,
 )
 from govoplan_dataflow.backend.governance import require_definition_action
+from govoplan_dataflow.backend.recovery import (
+    DataflowRecoveryError,
+    DataflowRunRecovery,
+    begin_dataflow_run_recovery,
+    claim_stale_dataflow_recovery,
+    dataflow_run_recovery_state,
+)
 from govoplan_dataflow.backend.service import (
     _execute_pipeline_run,
     _require_deployed_revision,
@@ -55,7 +62,10 @@ def dispatch_pending_runs(
         if worker_id and str(worker_id).strip()
         else socket.gethostname()
     )
-    recovered = _recover_expired_leases(session, now=current)
+    recovered, recovered_outcome_unknown = _recover_expired_leases(
+        session,
+        now=current,
+    )
     session.commit()
     summary: dict[str, object] = {
         "claimed": 0,
@@ -63,6 +73,7 @@ def dispatch_pending_runs(
         "retrying": 0,
         "failed": 0,
         "cancelled": 0,
+        "outcome_unknown": recovered_outcome_unknown,
         "recovered": recovered,
         "runs": [],
     }
@@ -193,7 +204,7 @@ def _recover_expired_leases(
     session: Session,
     *,
     now: datetime,
-) -> int:
+) -> tuple[int, int]:
     runs = list(
         session.scalars(
             select(DataflowRun)
@@ -205,13 +216,46 @@ def _recover_expired_leases(
             .with_for_update(skip_locked=True)
         )
     )
+    recovered = 0
+    outcome_unknown = 0
     for run in runs:
+        recovery = dataflow_run_recovery_state(session, run_id=run.id)
+        recovery_status: str | None = None
+        recovery_metadata = dict(run.authorization_).get("recovery")
+        effect_started = bool(
+            isinstance(recovery_metadata, Mapping)
+            and recovery_metadata.get("boundary") == "output-publication"
+        )
+        if recovery is not None and recovery.get("operation_id"):
+            try:
+                recovery_status = claim_stale_dataflow_recovery(
+                    session,
+                    operation_id=str(recovery["operation_id"]),
+                    lease_ttl_seconds=DEFAULT_LEASE_SECONDS,
+                    effect_started=effect_started,
+                )
+            except DataflowRecoveryError:
+                logger.warning(
+                    "Could not claim stale Dataflow recovery operation %s",
+                    recovery["operation_id"],
+                    exc_info=True,
+                )
+                continue
         run.worker_id = None
         run.claimed_at = None
         run.lease_expires_at = None
         run.heartbeat_at = None
         if run.cancellation_requested_at is not None:
             _cancel_run(run, now=now)
+        elif recovery_status == "outcome_unknown" or effect_started:
+            run.status = "outcome_unknown"
+            run.finished_at = now
+            run.progress_phase = "outcome_unknown"
+            run.error = (
+                "The worker lease expired after output publication may have "
+                "started. Verify the sink before retrying."
+            )
+            outcome_unknown += 1
         elif run.attempts < run.max_attempts:
             run.status = "retrying"
             run.available_at = now
@@ -223,7 +267,8 @@ def _recover_expired_leases(
                 now=now,
                 message="The worker lease expired after the final attempt.",
             )
-    return len(runs)
+        recovered += 1
+    return recovered, outcome_unknown
 
 
 def _claim_next_run(
@@ -291,6 +336,7 @@ def _execute_claimed_run(
     if run.cancellation_requested_at is not None:
         _cancel_run(run, now=now)
         return "cancelled"
+    recovery: DataflowRunRecovery | None = None
     try:
         principal, provenance = _resolve_principal(
             session,
@@ -329,6 +375,16 @@ def _execute_claimed_run(
             revision=revision,
             environment=run.environment,
         )
+        session.commit()
+        recovery = begin_dataflow_run_recovery(
+            session,
+            run=run,
+            lease_ttl_seconds=_lease_seconds(run),
+        )
+        if recovery.replayed:
+            session.rollback()
+            run = session.get(DataflowRun, run_id)
+            return run.status if run is not None else "failed"
         _notify_run(
             session,
             registry=registry,
@@ -344,6 +400,7 @@ def _execute_claimed_run(
             request=pipeline_run_request(run),
             principal=principal,
             registry=registry,
+            recovery=recovery,
         )
         if (
             run.status == "failed"
@@ -359,7 +416,47 @@ def _execute_claimed_run(
             run=run,
             pipeline_name=pipeline.name,
         )
+        recovery.finish(session, run=run)
         return run.status
+    except DataflowRecoveryError:
+        logger.exception(
+            "Dataflow recovery finalization failed for run %s",
+            run_id,
+        )
+        session.rollback()
+        run = session.get(DataflowRun, run_id)
+        if run is not None:
+            if recovery is not None and recovery.publication_started:
+                run.status = "outcome_unknown"
+                run.finished_at = now
+                run.progress_phase = "outcome_unknown"
+                run.error = (
+                    "Output publication completed without verifiable terminal "
+                    "recovery evidence; reconcile the sink."
+                )
+                _finish_claim(run)
+                return "outcome_unknown"
+            if recovery is None:
+                _fail_run(
+                    run,
+                    now=now,
+                    message=(
+                        "The recovery ledger was unavailable; no Dataflow "
+                        "execution started."
+                    ),
+                )
+                _finish_claim(run)
+            else:
+                _fail_run(
+                    run,
+                    now=now,
+                    message=(
+                        "The database-only run could not commit verifiable "
+                        "recovery evidence."
+                    ),
+                )
+                _finish_claim(run)
+        return "failed"
     except (DataflowWorkerError, PermissionError, ValueError) as exc:
         _fail_run(run, now=now, message=str(exc))
         _finish_claim(run)
@@ -370,6 +467,8 @@ def _execute_claimed_run(
             run=run,
             pipeline_name="Dataflow",
         )
+        if recovery is not None:
+            recovery.finish(session, run=run)
         return "failed"
     except Exception as exc:
         logger.exception(
@@ -380,6 +479,24 @@ def _execute_claimed_run(
         run = session.get(DataflowRun, run_id)
         if run is None:
             return "failed"
+        if recovery is not None and recovery.publication_started:
+            run.status = "outcome_unknown"
+            run.finished_at = now
+            run.error = (
+                "The output provider failed after dispatch began; verify the "
+                "sink before retrying."
+            )
+            run.progress_phase = "outcome_unknown"
+            _finish_claim(run)
+            try:
+                recovery.finish(session, run=run)
+            except DataflowRecoveryError:
+                logger.exception(
+                    "Could not persist the uncertain output state for run %s",
+                    run_id,
+                )
+                session.rollback()
+            return "outcome_unknown"
         _fail_run(
             run,
             now=now,
@@ -389,6 +506,15 @@ def _execute_claimed_run(
             ),
         )
         _finish_claim(run)
+        if recovery is not None:
+            try:
+                recovery.finish(session, run=run)
+            except DataflowRecoveryError:
+                logger.exception(
+                    "Could not finalize failed Dataflow recovery for run %s",
+                    run_id,
+                )
+                session.rollback()
         return "failed"
 
 

src/govoplan_dataflow/backend/schemas.py

@@ -31,6 +31,7 @@ DataflowRunStatus = Literal[
     "succeeded",
     "failed",
     "cancelled",
+    "outcome_unknown",
 ]
 
 
@@ -335,6 +336,12 @@ class PipelineRunResponse(BaseModel):
     finished_at: datetime | None
     created_by: str | None
     created_at: datetime
+    recovery_operation_id: str | None = None
+    recovery_operation_type: str | None = None
+    recovery_mode: str | None = None
+    recovery_status: str | None = None
+    recovery_requires_attention: bool = False
+    recovery_explanation: str | None = None
     replayed: bool = False
 
 

src/govoplan_dataflow/backend/service.py

@@ -40,6 +40,7 @@ from govoplan_dataflow.backend.db.models import (
     DataflowPipelineDeployment,
     DataflowPipelineRevision,
     DataflowRun,
+    new_uuid,
 )
 from govoplan_dataflow.backend.executor import (
     EXECUTOR_VERSION,
@@ -79,6 +80,12 @@ from govoplan_dataflow.backend.schemas import (
     PipelineValidationResponse,
     PreviewColumn,
 )
+from govoplan_dataflow.backend.recovery import (
+    DataflowRecoveryError,
+    DataflowRunRecovery,
+    begin_dataflow_run_recovery,
+    dataflow_run_recovery_state,
+)
 from govoplan_dataflow.backend.sql_compiler import (
     SqlCompilationError,
     compile_sql,
@@ -1071,14 +1078,27 @@ def start_pipeline_run(
         principal=principal,
         defer_execution=defer_execution,
     )
-    session.add(run)
-    session.flush()
     run.authorization_ = {
         **dict(run.authorization_),
         "authorization_ref": (
             request.invocation.trigger_ref or f"dataflow-run:{run.id}"
         ),
     }
+    recovery: DataflowRunRecovery | None = None
+    if not defer_execution:
+        try:
+            recovery = begin_dataflow_run_recovery(
+                session,
+                run=run,
+                lease_ttl_seconds=int(
+                    float(run.resource_budget.get("max_wall_seconds") or 30.0)
+                )
+                + 60,
+            )
+        except DataflowRecoveryError as exc:
+            raise DataflowConflictError(str(exc)) from exc
+    session.add(run)
+    session.flush()
     if defer_execution:
         session.flush()
         return run, False
@@ -1090,8 +1110,27 @@ def start_pipeline_run(
         request=request,
         principal=principal,
         registry=registry,
+        recovery=recovery,
     )
-    session.flush()
+    if recovery is not None:
+        try:
+            recovery.finish(session, run=run)
+        except DataflowRecoveryError as exc:
+            session.rollback()
+            if recovery.publication_started:
+                persisted = session.get(DataflowRun, run.id)
+                if persisted is not None:
+                    persisted.status = "outcome_unknown"
+                    persisted.finished_at = utcnow()
+                    persisted.progress_phase = "outcome_unknown"
+                    persisted.error = (
+                        "Output publication completed without verifiable "
+                        "terminal recovery evidence; reconcile the sink."
+                    )
+                    session.commit()
+            raise DataflowConflictError(str(exc)) from exc
+    else:
+        session.flush()
     return run, False
 
 
@@ -1256,6 +1295,7 @@ def _new_pipeline_run(
     now = utcnow()
     budget = _run_resource_budget(request)
     return DataflowRun(
+        id=new_uuid(),
         tenant_id=tenant_id,
         pipeline_id=pipeline.id,
         pipeline_revision_id=revision.id,
@@ -1360,6 +1400,7 @@ def _execute_pipeline_run(
     request: DataflowRunRequest,
     principal: ApiPrincipal,
     registry: object | None,
+    recovery: DataflowRunRecovery | None = None,
 ) -> bool:
     try:
         if run.cancellation_requested_at is not None:
@@ -1405,6 +1446,7 @@ def _execute_pipeline_run(
                 result=result,
                 principal=principal,
                 registry=registry,
+                recovery=recovery,
             )
         run.status = "succeeded"
         run.finished_at = utcnow()
@@ -1414,6 +1456,10 @@ def _execute_pipeline_run(
         return False
     except (DatasourceError, PipelineExecutionError) as exc:
         _mark_pipeline_run_failed(run, exc)
+        if recovery is not None and recovery.publication_started:
+            run.status = "outcome_unknown"
+            run.progress_phase = "outcome_unknown"
+            return False
         return isinstance(exc, DatasourceUnavailableError) or bool(
             getattr(exc, "retryable", False)
         )
@@ -1469,6 +1515,7 @@ def _publish_pipeline_result(
     result: PipelineExecutionResult,
     principal: ApiPrincipal,
     registry: object | None,
+    recovery: DataflowRunRecovery | None = None,
 ) -> None:
     publisher = datasource_publication(registry)
     if publisher is None:
@@ -1479,33 +1526,51 @@ def _publish_pipeline_result(
     target = request.publication
     if target is None:
         return
-    publication = publisher.publish_rows(
+    if recovery is None:
+        raise PipelineExecutionError(
+            "Publishing Dataflow output requires a durable recovery operation."
+        )
+    recovery.prepare_publication(
         session,
-        principal,
-        request=DatasourcePublicationRequest(
-            producer_module="dataflow",
-            producer_run_ref=f"dataflow-run:{run.id}",
-            idempotency_key=f"{pipeline.id}:{request.idempotency_key.strip()}",
-            rows=tuple(dict(row) for row in result.rows),
-            target_datasource_ref=target.target_datasource_ref,
-            name=target.name or f"{pipeline.name} output",
-            source_name=target.source_name,
-            description=target.description,
-            freeze=target.freeze,
-            frozen_label=target.frozen_label,
-            set_current=target.set_current,
-            provenance={
-                "pipeline_ref": f"pipeline:{pipeline.id}",
-                "pipeline_revision": revision.revision,
-                "definition_hash": revision.content_hash,
-                "source_fingerprints": result.source_fingerprints,
-            },
-            metadata={
-                **dict(target.metadata),
-                "dataflow_run_ref": f"dataflow-run:{run.id}",
-            },
-        ),
+        run=run,
+        rows=tuple(dict(row) for row in result.rows),
     )
+    try:
+        publication = publisher.publish_rows(
+            session,
+            principal,
+            request=DatasourcePublicationRequest(
+                producer_module="dataflow",
+                producer_run_ref=f"dataflow-run:{run.id}",
+                idempotency_key=(
+                    f"{pipeline.id}:{request.idempotency_key.strip()}"
+                ),
+                rows=tuple(dict(row) for row in result.rows),
+                target_datasource_ref=target.target_datasource_ref,
+                name=target.name or f"{pipeline.name} output",
+                source_name=target.source_name,
+                description=target.description,
+                freeze=target.freeze,
+                frozen_label=target.frozen_label,
+                set_current=target.set_current,
+                provenance={
+                    "pipeline_ref": f"pipeline:{pipeline.id}",
+                    "pipeline_revision": revision.revision,
+                    "definition_hash": revision.content_hash,
+                    "source_fingerprints": result.source_fingerprints,
+                },
+                metadata={
+                    **dict(target.metadata),
+                    "dataflow_run_ref": f"dataflow-run:{run.id}",
+                },
+            ),
+        )
+    except DatasourceError:
+        raise
+    except Exception as exc:
+        raise PipelineExecutionError(
+            "The output provider failed after publication dispatch began."
+        ) from exc
     run.output_publication_ref = publication.ref
     run.output_datasource_ref = publication.datasource.ref
     run.output_materialization_ref = publication.materialization.ref
@@ -1573,10 +1638,16 @@ def pipeline_run_response(
     run: DataflowRun,
     *,
     replayed: bool = False,
+    recovery_state: Mapping[str, object] | None = None,
 ) -> PipelineRunResponse:
     revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id)
     if revision is None:
         raise DataflowNotFoundError("Dataflow pipeline revision not found")
+    recovery = (
+        dict(recovery_state) or None
+        if recovery_state is not None
+        else dataflow_run_recovery_state(session, run_id=run.id)
+    )
     return PipelineRunResponse(
         ref=f"dataflow-run:{run.id}",
         pipeline_id=run.pipeline_id,
@@ -1622,6 +1693,27 @@ def pipeline_run_response(
         finished_at=run.finished_at,
         created_by=run.created_by,
         created_at=run.created_at,
+        recovery_operation_id=(
+            str(recovery["operation_id"]) if recovery is not None else None
+        ),
+        recovery_operation_type=(
+            str(recovery["operation_type"]) if recovery is not None else None
+        ),
+        recovery_mode=(
+            str(recovery["mode"]) if recovery is not None else None
+        ),
+        recovery_status=(
+            str(recovery["status"]) if recovery is not None else None
+        ),
+        recovery_requires_attention=(
+            bool(recovery["requires_attention"])
+            or run.status == "outcome_unknown"
+            if recovery is not None
+            else run.status == "outcome_unknown"
+        ),
+        recovery_explanation=(
+            str(recovery["explanation"]) if recovery is not None else None
+        ),
         replayed=replayed,
     )
 
@@ -1635,6 +1727,7 @@ def pipeline_run_descriptor(
     revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id)
     if revision is None:
         raise DataflowNotFoundError("Dataflow pipeline revision not found")
+    recovery = dataflow_run_recovery_state(session, run_id=run.id)
     return DataflowRunDescriptor(
         ref=f"dataflow-run:{run.id}",
         pipeline_ref=f"pipeline:{run.pipeline_id}",
@@ -1670,6 +1763,7 @@ def pipeline_run_descriptor(
             "progress_phase": run.progress_phase,
             "source_fingerprints": list(run.source_fingerprints),
             "diagnostics": list(run.diagnostics),
+            "recovery": dict(recovery) if recovery is not None else None,
         },
     )