fix: harden governed rule execution evidence

.superpowers/sdd/task-6-report.md

@@ -22,6 +22,12 @@ The implementation includes:
 - reliable `running` to `success`, `failed`, `unknown`, or `cancelled`
   finalization with row counts, timings, commit outcome, and bounded public
   result;
+- a signed, exact deployment/environment task identity plus a durable lease
+  owner, heartbeat, and expiry boundary; an expired run is finalized
+  `unknown` and is never blindly re-executed;
+- same-JTI response replay from the durable ledger, including recovery from
+  terminal rule evidence when execution committed but the HTTP/ledger response
+  was lost;
 - commit-acknowledgement rechecks for both run and sample finalization;
 - a maximum 100-row violation sample, redacted before it leaves the isolated
   Polars worker and redacted again at the evidence boundary;
@@ -30,18 +36,18 @@ The implementation includes:
 - no raw sample values in Runner responses or application log calls;
 - correlation-scoped upstream output resolution that accepts only current,
   ready, unexpired, storage-attested artifacts;
-- a closed SQL staging reference for same-source pushdown handoff instead of
-  row JSON;
-- Kestra single-predecessor `input_artifact` and multi-predecessor named
-  `input_artifacts` expressions, with governed rule nodes receiving no
-  unrelated workflow parameters;
+- an opaque `dataops-staging://<uuid>` SQL receipt for same-source pushdown
+  handoff, bound to producer run, deployment, correlation, dataset binding,
+  binding hash, relation digest, commit state, and expiry;
+- bracket-safe Kestra single-predecessor `input_artifact` expressions;
+  governed rule fan-in is rejected until an explicit merge component exists;
 - a top-level Runner `output_artifact` field matching the Kestra expression
   contract while retaining the full bounded result object.
 
 ## Forward-only schema change
 
-Migration `20260723_160` extends the existing evidence tables without changing
-historical migrations 110, 140, or 150.
+Migrations `20260723_160` and `20260723_170` extend the existing evidence
+tables without changing historical migrations 110, 140, or 150.
 
 `rule_runs` gains:
 
@@ -61,9 +67,15 @@ historical migrations 110, 140, or 150.
 Downgrade is deliberately rejected because deleting execution evidence would
 break audit and replay guarantees.
 
+Migration 160 explicitly preflights duplicate legacy `rule_run_id` samples and
+legacy samples over 100 rows before adding constraints. Migration 170 adds
+leases, exact task deployment identity, bounded replay fields, evidence
+digests, and SQL staging receipts.
+
 The local Docker PostgreSQL was upgraded through the formal
-`20260723_150 -> 20260723_160` Alembic path. The existing isolated old-140
-migration acceptance also passes through current head.
+`20260723_150 -> head` Alembic path. Two isolated real PostgreSQL tests prove
+that duplicate and oversized legacy samples abort with a clear diagnostic and
+leave Alembic at revision 150.
 
 ## RED / GREEN evidence
 
@@ -77,11 +89,13 @@ The initial fail-first slice produced five expected failures:
 A second fail-first assertion proved that the Polars adapter did not expose a
 bounded violation sample before the worker implementation was added.
 
-Final verification:
+Final review verification:
 
-- focused evidence, Runner, Polars, Kestra, migration, and real integration:
-  `27 passed`;
-- full suite: `586 passed, 26 skipped, 59 subtests passed`;
+- focused Runner/rule/Kestra/schema suite: `129 passed`;
+- real PostgreSQL/MySQL/MinIO evidence acceptance: `1 passed`;
+- real SQL adapter acceptance: `4 passed`;
+- real malformed `150 -> head` migration acceptance: `2 passed`;
+- full suite: `594 passed, 28 skipped, 59 subtests passed`;
 - Ruff across every changed Python file: `All checks passed!`;
 - `git diff --check`: passed;
 - Docker Compose configuration validation: passed;
@@ -113,9 +127,23 @@ It proves:
 4. the same correlation contains exactly two successful run records;
 5. a different signed token for the same node replays the existing run and
    sample instead of executing again;
-6. a failed adapter records `failed/not_committed`;
-7. an uncertain write records `unknown/unknown`;
-8. all test-owned PostgreSQL rows and MinIO objects are removed after
+6. same-JTI retry returns the exact stored response without invoking the
+   adapter again, and a lost HTTP/ledger response is rebuilt from terminal
+   rule evidence;
+7. the sample-ready/run-finalize crash window is retried safely without
+   changing evidence or leaking raw values;
+8. the lease owner is enforced, heartbeat extends only the owned run, and an
+   expired lease becomes `unknown` instead of being reclaimed;
+9. a real opaque SQL receipt is unavailable while pending, becomes executable
+   only after a committed producer finish, and rejects forged, cross-
+   correlation, and failed-producer references;
+10. same-status/different-content finish retries are rejected by the exact
+    evidence digest;
+11. real MinIO reconciliation retains pending/ready violation evidence while
+    removing an unreferenced old object;
+12. a failed adapter records `failed/not_committed`;
+13. an uncertain write records `unknown/unknown`;
+14. all test-owned PostgreSQL rows and MinIO objects are removed after
    acceptance.
 
 ## Residual boundary

app/core/mcp/gateway.py

@@ -247,6 +247,8 @@ class SchedulingGateway:
         response = self._require_engine().deploy_disabled(compiled.yaml)
         deployment = {
             "candidate_id": candidate_id,
+            "deployment_id": candidate_id,
+            "environment": environment,
             "namespace": compiled.namespace,
             "flow_id": compiled.flow_id,
             "definition_hash": compiled.definition_hash,
@@ -320,6 +322,8 @@ class SchedulingGateway:
             task_tokens[node["id"]] = self.token_issuer.issue(
                 task_uid=str(uuid.uuid4()),
                 dataflow_uid=candidate["dataflow_uid"],
+                deployment_id=deployment["deployment_id"],
+                environment=environment,
                 workflow_version=int(candidate["version_no"]),
                 correlation_id=identity.correlation_id,
                 node=node,

app/core/orchestration/compilers/kestra.py

@@ -169,10 +169,17 @@ def _compile_dag_tasks(
             }
         )
         predecessors = sorted(dependencies[node_id])
+        if (
+            len(predecessors) > 1
+            and node["type"] in {"rule.apply", "quality.check"}
+        ):
+            raise ValueError(
+                "governed rule nodes require exactly one predecessor"
+            )
         if len(predecessors) == 1:
             parameters["input_artifact"] = (
-                "{{ outputs.dataops_dag.tasks."
-                f"{predecessors[0]}.body.output_artifact }}}}"
+                "{{ outputs.dataops_dag.tasks["
+                f"'{predecessors[0]}'].body.output_artifact }}}}"
             )
         elif len(predecessors) > 1:
             parameters["input_artifacts"] = {

app/runner/api.py

@@ -1,5 +1,8 @@
 """Minimal HTTP surface for the independent DataOps Runner."""
 
+import hashlib
+import json
+
 from flask import Flask, jsonify, request
 
 from app.runner.auth import TaskTokenInvalid
@@ -38,6 +41,8 @@ def create_runner_app(*, verifier, ledger, registry):
         binding = {
             "task_uid": claims.task_uid,
             "dataflow_uid": claims.dataflow_uid,
+            "deployment_id": claims.deployment_id,
+            "environment": claims.environment,
             "workflow_version": claims.workflow_version,
             "correlation_id": claims.correlation_id,
             "node_id": claims.node_id,
@@ -55,7 +60,87 @@ def create_runner_app(*, verifier, ledger, registry):
             app.logger.error("runner task ledger unavailable")
             return jsonify({"error": "task ledger is unavailable"}), 503
         if not claimed:
-            return jsonify({"error": "task token already consumed"}), 409
+            try:
+                existing = ledger.get(claims.jti)
+            except Exception:
+                return jsonify({"error": "task ledger is unavailable"}), 503
+            if (
+                existing is None
+                or any(
+                    existing.binding.get(key) != value
+                    for key, value in binding.items()
+                )
+            ):
+                return jsonify({"error": "task token already consumed"}), 409
+            if existing.status == "running":
+                recovered = None
+                if node.get("type") in {"rule.apply", "quality.check"}:
+                    try:
+                        recovered = registry.replay_task(
+                            node,
+                            correlation_id=claims.correlation_id,
+                            deployment_id=claims.deployment_id,
+                            task_jti=claims.jti,
+                        )
+                    except NodeExecutionError:
+                        return jsonify(
+                            {"error": "task ledger is unavailable"}
+                        ), 503
+                if isinstance(recovered, dict):
+                    replay_body = {
+                        "task_uid": claims.task_uid,
+                        "correlation_id": claims.correlation_id,
+                        "result": recovered,
+                    }
+                    output_artifact = recovered.get("output_artifact")
+                    if isinstance(output_artifact, str):
+                        replay_body["output_artifact"] = output_artifact
+                    if not finish_safely(
+                        claims.jti,
+                        status="success",
+                        commit_outcome=recovered.get(
+                            "commit_outcome",
+                            "not_applicable",
+                        ),
+                        safe_detail="task response recovered",
+                        replay_http_status=200,
+                        replay_body=replay_body,
+                    ):
+                        return jsonify(
+                            {"error": "task ledger is unavailable"}
+                        ), 503
+                    response = jsonify(replay_body)
+                    response.headers["X-Idempotent-Replay"] = "true"
+                    return response
+                response = jsonify(
+                    {
+                        "task_uid": claims.task_uid,
+                        "correlation_id": claims.correlation_id,
+                        "status": "running",
+                    }
+                )
+                response.headers["Retry-After"] = "2"
+                return response, 202
+            if (
+                existing.replay_body is None
+                or existing.replay_http_status is None
+                or existing.replay_digest is None
+            ):
+                return jsonify({"error": "task token already consumed"}), 409
+            encoded = json.dumps(
+                existing.replay_body,
+                sort_keys=True,
+                separators=(",", ":"),
+                ensure_ascii=False,
+            ).encode("utf-8")
+            if (
+                hashlib.sha256(encoded).hexdigest()
+                != existing.replay_digest
+            ):
+                return jsonify({"error": "task ledger is unavailable"}), 503
+            response = jsonify(existing.replay_body)
+            response.headers["X-Idempotent-Replay"] = "true"
+            return response, existing.replay_http_status
         try:
             result = registry.execute(
                 node,
@@ -63,8 +148,11 @@ def create_runner_app(*, verifier, ledger, registry):
                 write_authorized=claims.write_authorized,
                 correlation_id=claims.correlation_id,
                 dataflow_uid=claims.dataflow_uid,
+                deployment_id=claims.deployment_id,
+                environment=claims.environment,
                 workflow_version=claims.workflow_version,
                 node_id=claims.node_id,
+                task_jti=claims.jti,
             )
         except NodeExecutionError as exc:
             recorded = finish_safely(
@@ -93,13 +181,6 @@ def create_runner_app(*, verifier, ledger, registry):
             if isinstance(result, dict)
             else "not_applicable"
         )
-        if not finish_safely(
-            claims.jti,
-            status="success",
-            commit_outcome=commit_outcome,
-            safe_detail="task completed",
-        ):
-            return jsonify({"error": "task ledger is unavailable"}), 503
         response = {
             "task_uid": claims.task_uid,
             "correlation_id": claims.correlation_id,
@@ -110,6 +191,20 @@ def create_runner_app(*, verifier, ledger, registry):
             and isinstance(result.get("output_artifact"), str)
         ):
             response["output_artifact"] = result["output_artifact"]
+        replay_body = (
+            response
+            if node.get("type") in {"rule.apply", "quality.check"}
+            else None
+        )
+        if not finish_safely(
+            claims.jti,
+            status="success",
+            commit_outcome=commit_outcome,
+            safe_detail="task completed",
+            replay_http_status=(200 if replay_body is not None else None),
+            replay_body=replay_body,
+        ):
+            return jsonify({"error": "task ledger is unavailable"}), 503
         return jsonify(response)
 
     return app

app/runner/artifacts.py

@@ -1644,6 +1644,12 @@ class PostgresArtifactResolver:
                         FROM public.rule_run_artifacts
                         WHERE artifact_ref =
                               ANY(CAST(:artifact_refs AS text[]))
+                        UNION
+                        SELECT artifact_ref
+                        FROM public.rule_violation_samples
+                        WHERE handoff_status IN ('pending','ready')
+                          AND artifact_ref =
+                              ANY(CAST(:artifact_refs AS text[]))
                         """
                     ),
                     {"artifact_refs": candidates},

app/runner/auth.py

@@ -6,12 +6,12 @@ import hashlib
 import json
 import time
 import uuid
+from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Any, Mapping
+from typing import Any
 
 import jwt
 
-
 ISSUER = "dataops-platform"
 AUDIENCE = "dataops-runner"
 ALGORITHM = "HS256"
@@ -46,6 +46,8 @@ def _secret_bytes(secret: str) -> bytes:
 class TaskClaims:
     task_uid: str
     dataflow_uid: str
+    deployment_id: str
+    environment: str
     workflow_version: int
     correlation_id: str
     node_id: str
@@ -71,12 +73,16 @@ class TaskTokenIssuer:
         *,
         task_uid,
         dataflow_uid,
+        deployment_id,
+        environment,
         workflow_version,
         correlation_id,
         node,
         write_authorized=False,
     ):
         now = int(self._clock())
+        if environment not in {"development", "test", "production"}:
+            raise ValueError("task environment is invalid")
         purpose = str(node.get("purpose") or "read")
         if purpose == "write" and not write_authorized:
             raise ValueError("write task requires trusted write authorization")
@@ -86,6 +92,8 @@ class TaskTokenIssuer:
             "sub": str(task_uid),
             "task_uid": str(task_uid),
             "dataflow_uid": str(dataflow_uid),
+            "deployment_id": str(deployment_id),
+            "environment": str(environment),
             "workflow_version": int(workflow_version),
             "correlation_id": str(correlation_id),
             "node_id": str(node.get("id") or ""),
@@ -121,6 +129,8 @@ class TaskTokenVerifier:
         required = {
             "task_uid",
             "dataflow_uid",
+            "deployment_id",
+            "environment",
             "workflow_version",
             "correlation_id",
             "node_id",
@@ -152,9 +162,17 @@ class TaskTokenVerifier:
             or payload["purpose"] != str(node.get("purpose") or "read")
         ):
             raise TaskTokenInvalid("task token node binding does not match")
+        if payload["environment"] not in {
+            "development",
+            "test",
+            "production",
+        }:
+            raise TaskTokenInvalid("task token environment is invalid")
         return TaskClaims(
             task_uid=str(payload["task_uid"]),
             dataflow_uid=str(payload["dataflow_uid"]),
+            deployment_id=str(payload["deployment_id"]),
+            environment=str(payload["environment"]),
             workflow_version=workflow_version,
             correlation_id=str(payload["correlation_id"]),
             node_id=str(payload["node_id"]),

app/runner/ledger.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
 import threading
+from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Mapping, Optional
 
 from sqlalchemy import text
 
@@ -17,6 +19,9 @@ class TaskLedgerRecord:
     status: str = "running"
     commit_outcome: str = "not_applicable"
     safe_detail: str = ""
+    replay_http_status: int | None = None
+    replay_body: Mapping[str, object] | None = None
+    replay_digest: str | None = None
 
 
 class InMemoryTaskLedger:
@@ -42,14 +47,26 @@ class InMemoryTaskLedger:
         status,
         commit_outcome="not_applicable",
         safe_detail="",
+        replay_http_status=None,
+        replay_body=None,
     ):
         with self._lock:
             record = self._records[str(jti)]
             record.status = str(status)
             record.commit_outcome = str(commit_outcome)
             record.safe_detail = str(safe_detail)[:500]
-
-    def get(self, jti) -> Optional[TaskLedgerRecord]:
+            if replay_body is not None:
+                encoded = json.dumps(
+                    replay_body,
+                    sort_keys=True,
+                    separators=(",", ":"),
+                    ensure_ascii=False,
+                ).encode("utf-8")
+                record.replay_http_status = int(replay_http_status)
+                record.replay_body = dict(replay_body)
+                record.replay_digest = hashlib.sha256(encoded).hexdigest()
+
+    def get(self, jti) -> TaskLedgerRecord | None:
         with self._lock:
             return self._records.get(str(jti))
 
@@ -65,6 +82,8 @@ class PostgresTaskLedger:
             "jti": str(jti),
             "task_uid": binding["task_uid"],
             "dataflow_uid": binding["dataflow_uid"],
+            "deployment_id": binding["deployment_id"],
+            "environment": binding["environment"],
             "workflow_version": int(binding["workflow_version"]),
             "correlation_id": binding["correlation_id"],
             "node_id": binding["node_id"],
@@ -80,14 +99,16 @@ class PostgresTaskLedger:
                     INSERT INTO public.runner_task_executions (
                         token_jti, task_uid, dataflow_uid, workflow_version,
                         correlation_id, node_id, node_type, data_source_uid,
-                        idempotency_key, status, commit_outcome, expires_at
+                        idempotency_key, status, commit_outcome, expires_at,
+                        deployment_id, environment
                     ) VALUES (
                         CAST(:jti AS uuid), CAST(:task_uid AS uuid),
                         CAST(:dataflow_uid AS uuid), :workflow_version,
                         CAST(:correlation_id AS uuid), :node_id, :node_type,
                         CAST(:data_source_uid AS uuid), :idempotency_key,
                         'running', 'not_applicable',
-                        to_timestamp(:expires_at)
+                        to_timestamp(:expires_at),
+                        CAST(:deployment_id AS uuid), :environment
                     )
                     ON CONFLICT DO NOTHING
                     """
@@ -103,6 +124,8 @@ class PostgresTaskLedger:
         status,
         commit_outcome="not_applicable",
         safe_detail="",
+        replay_http_status=None,
+        replay_body=None,
     ):
         allowed_statuses = {"success", "failed", "unknown"}
         allowed_outcomes = {
@@ -113,6 +136,19 @@ class PostgresTaskLedger:
         }
         if status not in allowed_statuses or commit_outcome not in allowed_outcomes:
             raise ValueError("runner task outcome is invalid")
+        replay_digest = None
+        if replay_body is not None:
+            encoded = json.dumps(
+                replay_body,
+                sort_keys=True,
+                separators=(",", ":"),
+                ensure_ascii=False,
+            ).encode("utf-8")
+            if len(encoded) > 32_768:
+                raise ValueError("runner replay body exceeds the safe limit")
+            if replay_http_status != 200:
+                raise ValueError("runner replay status is invalid")
+            replay_digest = hashlib.sha256(encoded).hexdigest()
         with self.engine.begin() as connection:
             connection.execute(
                 text(
@@ -121,6 +157,9 @@ class PostgresTaskLedger:
                     SET status = :status,
                         commit_outcome = :commit_outcome,
                         safe_detail = :safe_detail,
+                        replay_http_status = :replay_http_status,
+                        replay_body = CAST(:replay_body AS jsonb),
+                        replay_digest = :replay_digest,
                         finished_at = CURRENT_TIMESTAMP
                     WHERE token_jti = CAST(:jti AS uuid)
                       AND status = 'running'
@@ -131,6 +170,13 @@ class PostgresTaskLedger:
                     "status": status,
                     "commit_outcome": commit_outcome,
                     "safe_detail": str(safe_detail)[:500],
+                    "replay_http_status": replay_http_status,
+                    "replay_body": (
+                        json.dumps(replay_body)
+                        if replay_body is not None
+                        else None
+                    ),
+                    "replay_digest": replay_digest,
                 },
             )
 
@@ -141,10 +187,16 @@ class PostgresTaskLedger:
                     text(
                         """
                         SELECT token_jti::text AS jti, task_uid::text,
+                               dataflow_uid::text, deployment_id::text,
+                               environment, workflow_version,
+                               correlation_id::text,
                                node_id, data_source_uid::text,
                                idempotency_key, status, commit_outcome,
+                               node_type,
                                safe_detail, EXTRACT(EPOCH FROM expires_at)::bigint
-                                   AS expires_at
+                                   AS expires_at,
+                               replay_http_status, replay_body,
+                               replay_digest
                         FROM public.runner_task_executions
                         WHERE token_jti = CAST(:jti AS uuid)
                         """
@@ -160,7 +212,13 @@ class PostgresTaskLedger:
             jti=row["jti"],
             binding={
                 "task_uid": row["task_uid"],
+                "dataflow_uid": row["dataflow_uid"],
+                "deployment_id": row["deployment_id"],
+                "environment": row["environment"],
+                "workflow_version": row["workflow_version"],
+                "correlation_id": row["correlation_id"],
                 "node_id": row["node_id"],
+                "node_type": row["node_type"],
                 "data_source_uid": row["data_source_uid"],
                 "idempotency_key": row["idempotency_key"],
             },
@@ -168,4 +226,11 @@ class PostgresTaskLedger:
             status=row["status"],
             commit_outcome=row["commit_outcome"],
             safe_detail=row["safe_detail"] or "",
+            replay_http_status=row["replay_http_status"],
+            replay_body=(
+                dict(row["replay_body"])
+                if isinstance(row["replay_body"], dict)
+                else row["replay_body"]
+            ),
+            replay_digest=row["replay_digest"],
         )

app/runner/nodes.py

@@ -377,8 +377,11 @@ class NodeRegistry:
         write_authorized=False,
         correlation_id=None,
         dataflow_uid=None,
+        deployment_id=None,
+        environment=None,
         workflow_version=None,
         node_id=None,
+        task_jti=None,
     ):
         executor = self.executors.get(node.get("type"))
         if executor is None or not callable(getattr(executor, "execute", None)):
@@ -389,6 +392,28 @@ class NodeRegistry:
             write_authorized=write_authorized,
             correlation_id=correlation_id,
             dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment=environment,
             workflow_version=workflow_version,
             node_id=node_id,
+            task_jti=task_jti,
+        )
+
+    def replay_task(
+        self,
+        node: Mapping[str, Any],
+        *,
+        correlation_id=None,
+        deployment_id=None,
+        task_jti=None,
+    ):
+        executor = self.executors.get(node.get("type"))
+        replay = getattr(executor, "replay_task", None)
+        if not callable(replay):
+            return None
+        return replay(
+            node=node,
+            correlation_id=correlation_id,
+            deployment_id=deployment_id,
+            task_jti=task_jti,
         )

app/runner/rule_evidence.py

@@ -7,6 +7,7 @@ import json
 import os
 import re
 import tempfile
+import uuid
 from contextlib import suppress
 from typing import Any
 
@@ -39,6 +40,75 @@ _FINISH_KEYS = {
     "sample_count",
     "redaction_policy",
 }
+_PUBLIC_RESULT_KEYS = {
+    "affected_rows",
+    "artifact_ref",
+    "commit_outcome",
+    "component_binding_id",
+    "digest",
+    "execution_plan_hash",
+    "expires_at",
+    "output_artifact",
+    "row_count",
+    "rows_aggregated",
+    "rows_deduplicated",
+    "rows_filtered",
+    "rows_in",
+    "rows_join_dropped",
+    "rows_out",
+    "rows_quarantined",
+    "rows_rejected",
+    "rule_version_id",
+    "schema_hash",
+    "violation_count",
+    "violations",
+}
+_ATTESTATION_RESULT_KEYS = {
+    "component_binding_id",
+    "execution_plan_hash",
+    "rule_version_id",
+}
+_BACKEND_PUBLIC_RESULT_KEYS = {
+    "sql_pushdown": _ATTESTATION_RESULT_KEYS
+    | {
+        "commit_outcome",
+        "output_artifact",
+        "rows_in",
+        "rows_out",
+        "rows_quarantined",
+        "rows_rejected",
+    },
+    "polars_batch": _ATTESTATION_RESULT_KEYS
+    | {
+        "artifact_ref",
+        "commit_outcome",
+        "digest",
+        "expires_at",
+        "output_artifact",
+        "row_count",
+        "rows_aggregated",
+        "rows_deduplicated",
+        "rows_filtered",
+        "rows_in",
+        "rows_join_dropped",
+        "rows_out",
+        "rows_quarantined",
+        "rows_rejected",
+        "schema_hash",
+        "violation_count",
+        "violations",
+    },
+    "quality_check": _ATTESTATION_RESULT_KEYS
+    | {
+        "commit_outcome",
+        "rows_in",
+        "rows_out",
+        "rows_quarantined",
+        "rows_rejected",
+        "violation_count",
+        "violations",
+    },
+}
 
 
 def _uid(value: Any, label: str) -> str:
@@ -48,6 +118,13 @@ def _uid(value: Any, label: str) -> str:
         raise ValueError(f"{label} is invalid") from exc
 
 
+def _uuid(value: Any, label: str) -> str:
+    try:
+        return str(uuid.UUID(str(value)))
+    except (TypeError, ValueError, AttributeError) as exc:
+        raise ValueError(f"{label} is invalid") from exc
+
+
 def _canonical_digest(value: Any) -> str:
     encoded = json.dumps(
         value,
@@ -80,6 +157,55 @@ def _sample_fields(sample: list[dict[str, Any]]) -> list[dict[str, Any]]:
     ]
 
 
+def validate_public_rule_result(
+    value: Any,
+    *,
+    backend: str | None = None,
+) -> dict[str, Any]:
+    allowed_keys = (
+        _BACKEND_PUBLIC_RESULT_KEYS.get(backend)
+        if backend is not None
+        else _PUBLIC_RESULT_KEYS
+    )
+    if (
+        not isinstance(value, dict)
+        or allowed_keys is None
+        or set(value) - allowed_keys
+        or any(str(key).startswith("_") for key in value)
+        or len(
+            json.dumps(
+                value,
+                sort_keys=True,
+                separators=(",", ":"),
+                ensure_ascii=False,
+            ).encode("utf-8")
+        )
+        > 32_768
+    ):
+        raise ValueError("public rule result is not evidence safe")
+    for key, item in value.items():
+        if key == "violations":
+            if not isinstance(item, list) or len(item) > 100:
+                raise ValueError(
+                    "public rule result is not evidence safe"
+                )
+            for summary in item:
+                if (
+                    not isinstance(summary, dict)
+                    or set(summary) != {"step_id", "count"}
+                    or not isinstance(summary["step_id"], str)
+                    or isinstance(summary["count"], bool)
+                    or not isinstance(summary["count"], int)
+                    or summary["count"] < 0
+                ):
+                    raise ValueError(
+                        "public rule result is not evidence safe"
+                    )
+        elif isinstance(item, (dict, list, tuple, set)):
+            raise ValueError("public rule result is not evidence safe")
+    return dict(value)
+
+
 class PostgresRuleEvidenceWriter:
     """Persist one immutable run and at most one expiring violation sample."""
 
@@ -89,16 +215,20 @@ class PostgresRuleEvidenceWriter:
         artifact_store,
         *,
         sample_ttl_seconds: int = 3600,
+        lease_seconds: int = 300,
     ):
         self.engine = engine
         self.artifact_store = artifact_store
         self.sample_ttl_seconds = int(sample_ttl_seconds)
+        self.lease_seconds = int(lease_seconds)
         if (
             self.sample_ttl_seconds < 1
             or self.sample_ttl_seconds
             > self.artifact_store.max_ttl_seconds
         ):
             raise ValueError("violation sample TTL is invalid")
+        if self.lease_seconds < 30 or self.lease_seconds > 900:
+            raise ValueError("rule execution lease is invalid")
 
     def start(
         self,
@@ -108,13 +238,20 @@ class PostgresRuleEvidenceWriter:
         plan_hash: str,
         correlation_id: str,
         dataflow_uid: str,
+        deployment_id: str,
+        environment: str,
         workflow_version: int,
         node_id: str,
+        lease_owner: str,
     ) -> str:
         component = _uid(component_binding_id, "component binding id")
         rule = _uid(rule_version_id, "rule version id")
         correlation = _uid(correlation_id, "correlation id")
         dataflow = _uid(dataflow_uid, "dataflow id")
+        deployment = _uid(deployment_id, "deployment id")
+        owner = _uuid(lease_owner, "lease owner")
+        if environment not in {"development", "test", "production"}:
+            raise ValueError("deployment environment is invalid")
         if _DIGEST.fullmatch(str(plan_hash or "")) is None:
             raise ValueError("plan hash is invalid")
         if (
@@ -130,6 +267,8 @@ class PostgresRuleEvidenceWriter:
                 "component_binding_id": component,
                 "correlation_id": correlation,
                 "dataflow_uid": dataflow,
+                "deployment_id": deployment,
+                "environment": environment,
                 "node_id": node_id,
                 "plan_hash": plan_hash,
                 "rule_version_id": rule,
@@ -137,7 +276,7 @@ class PostgresRuleEvidenceWriter:
             }
         )
         with self.engine.begin() as connection:
-            deployments = connection.execute(
+            canonical = connection.execute(
                 text(
                     """
                     SELECT d.id::text AS deployment_id
@@ -153,13 +292,12 @@ class PostgresRuleEvidenceWriter:
                           CAST(:rule_version_id AS uuid)
                       AND p.plan_hash = :plan_hash
                       AND p.status = 'published'
+                      AND b.component_id = :node_id
                       AND v.dataflow_uid = CAST(:dataflow_uid AS uuid)
                       AND v.version_no = :workflow_version
+                      AND d.id = CAST(:deployment_id AS uuid)
+                      AND d.environment = :environment
                       AND d.status IN ('canary','active')
-                    ORDER BY
-                      CASE d.status WHEN 'active' THEN 0 ELSE 1 END,
-                      d.created_at DESC
-                    LIMIT 2
                     """
                 ),
                 {
@@ -167,13 +305,60 @@ class PostgresRuleEvidenceWriter:
                     "rule_version_id": rule,
                     "plan_hash": plan_hash,
                     "dataflow_uid": dataflow,
+                    "deployment_id": deployment,
+                    "environment": environment,
                     "workflow_version": workflow_version,
+                    "node_id": node_id,
                 },
-            ).mappings().all()
-            if len(deployments) != 1:
-                raise ValueError(
-                    "canonical rule deployment is missing or ambiguous"
-                )
+            ).mappings().one_or_none()
+            if canonical is None:
+                raise ValueError("canonical rule deployment does not match")
+            existing = connection.execute(
+                text(
+                    """
+                    SELECT id::text, status, lease_owner::text,
+                           lease_expires_at
+                    FROM public.rule_runs
+                    WHERE evidence_key = :evidence_key
+                    FOR UPDATE
+                    """
+                ),
+                {"evidence_key": evidence_key},
+            ).mappings().one_or_none()
+            if existing is not None:
+                if (
+                    existing["status"] == "running"
+                    and existing["lease_expires_at"] is not None
+                ):
+                    expired = connection.execute(
+                        text(
+                            """
+                            SELECT :lease_expires_at <= CURRENT_TIMESTAMP
+                            """
+                        ),
+                        {
+                            "lease_expires_at": existing[
+                                "lease_expires_at"
+                            ]
+                        },
+                    ).scalar_one()
+                    if expired:
+                        connection.execute(
+                            text(
+                                """
+                                UPDATE public.rule_runs
+                                SET status = 'unknown',
+                                    commit_outcome = 'unknown',
+                                    failure_code = 'execution_lease_expired',
+                                    finished_at = CURRENT_TIMESTAMP,
+                                    updated_at = CURRENT_TIMESTAMP
+                                WHERE id = CAST(:id AS uuid)
+                                  AND status = 'running'
+                                """
+                            ),
+                            {"id": existing["id"]},
+                        )
+                return str(existing["id"])
             rule_run_id = new_governance_uid()
             selected = connection.execute(
                 text(
@@ -182,12 +367,18 @@ class PostgresRuleEvidenceWriter:
                         id, deployment_id, component_binding_id,
                         rule_version_id, plan_hash, status,
                         correlation_id, evidence_key, started_at
+                        , attempt_no, lease_owner, lease_expires_at,
+                        heartbeat_at
                     ) VALUES (
                         CAST(:id AS uuid), CAST(:deployment_id AS uuid),
                         CAST(:component_binding_id AS uuid),
                         CAST(:rule_version_id AS uuid), :plan_hash,
                         'running', CAST(:correlation_id AS uuid),
-                        :evidence_key, CURRENT_TIMESTAMP
+                        :evidence_key, CURRENT_TIMESTAMP, 1,
+                        CAST(:lease_owner AS uuid),
+                        CURRENT_TIMESTAMP
+                            + make_interval(secs => :lease_seconds),
+                        CURRENT_TIMESTAMP
                     )
                     ON CONFLICT (evidence_key) DO NOTHING
                     RETURNING id::text
@@ -195,12 +386,14 @@ class PostgresRuleEvidenceWriter:
                 ),
                 {
                     "id": rule_run_id,
-                    "deployment_id": deployments[0]["deployment_id"],
+                    "deployment_id": canonical["deployment_id"],
                     "component_binding_id": component,
                     "rule_version_id": rule,
                     "plan_hash": plan_hash,
                     "correlation_id": correlation,
                     "evidence_key": evidence_key,
+                    "lease_owner": owner,
+                    "lease_seconds": self.lease_seconds,
                 },
             ).scalar_one_or_none()
             if selected is None:
@@ -216,6 +409,205 @@ class PostgresRuleEvidenceWriter:
                 ).scalar_one()
         return str(selected)
 
+    def heartbeat(self, rule_run_id: str, lease_owner: str) -> None:
+        run_id = _uid(rule_run_id, "rule run id")
+        owner = _uuid(lease_owner, "lease owner")
+        with self.engine.begin() as connection:
+            updated = connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_runs
+                    SET heartbeat_at = CURRENT_TIMESTAMP,
+                        lease_expires_at = CURRENT_TIMESTAMP
+                            + make_interval(secs => :lease_seconds),
+                        updated_at = CURRENT_TIMESTAMP
+                    WHERE id = CAST(:id AS uuid)
+                      AND status = 'running'
+                      AND lease_owner = CAST(:lease_owner AS uuid)
+                    """
+                ),
+                {
+                    "id": run_id,
+                    "lease_owner": owner,
+                    "lease_seconds": self.lease_seconds,
+                },
+            )
+            if updated.rowcount != 1:
+                raise ValueError("rule execution lease is not owned")
+
+    def stage_sql_output(
+        self,
+        rule_run_id: str,
+        *,
+        output_binding_id: str,
+        ttl_seconds: int | None = None,
+    ) -> str:
+        run_id = _uid(rule_run_id, "rule run id")
+        binding_id = _uid(output_binding_id, "output binding id")
+        if ttl_seconds is None:
+            ttl_seconds = min(3600, self.sample_ttl_seconds)
+        if (
+            isinstance(ttl_seconds, bool)
+            or not isinstance(ttl_seconds, int)
+            or ttl_seconds < 1
+            or ttl_seconds > self.sample_ttl_seconds
+        ):
+            raise ValueError("SQL staging TTL is invalid")
+        receipt_id = new_governance_uid()
+        with self.engine.begin() as connection:
+            row = connection.execute(
+                text(
+                    """
+                    SELECT r.deployment_id::text, r.correlation_id::text,
+                           b.binding_hash, b.object_ref, b.object_kind,
+                           b.access_mode
+                    FROM public.rule_runs r
+                    JOIN public.dataflow_dataset_bindings b
+                      ON b.id = CAST(:binding_id AS uuid)
+                     AND b.dataflow_deployment_id = r.deployment_id
+                    WHERE r.id = CAST(:run_id AS uuid)
+                      AND r.status = 'running'
+                    FOR SHARE OF r, b
+                    """
+                ),
+                {"run_id": run_id, "binding_id": binding_id},
+            ).mappings().one_or_none()
+            if (
+                row is None
+                or row["object_kind"] not in {"table", "view"}
+                or row["access_mode"] not in {"write", "read_write"}
+            ):
+                raise ValueError("SQL staging output binding is invalid")
+            relation_digest = _canonical_digest(
+                {
+                    "binding_hash": str(row["binding_hash"]),
+                    "object_kind": str(row["object_kind"]),
+                    "object_ref": str(row["object_ref"]),
+                }
+            )
+            selected = connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_sql_staging_receipts (
+                        id, producer_rule_run_id, deployment_id,
+                        correlation_id, output_binding_id,
+                        output_binding_hash, relation_ref,
+                        relation_digest, commit_outcome, status,
+                        expires_at
+                    ) VALUES (
+                        CAST(:id AS uuid), CAST(:run_id AS uuid),
+                        CAST(:deployment_id AS uuid),
+                        CAST(:correlation_id AS uuid),
+                        CAST(:binding_id AS uuid), :binding_hash,
+                        :relation_ref, :relation_digest, 'committed',
+                        'pending', CURRENT_TIMESTAMP
+                            + make_interval(secs => :ttl_seconds)
+                    )
+                    ON CONFLICT (
+                        producer_rule_run_id, output_binding_id
+                    ) DO NOTHING
+                    RETURNING id::text
+                    """
+                ),
+                {
+                    "id": receipt_id,
+                    "run_id": run_id,
+                    "deployment_id": row["deployment_id"],
+                    "correlation_id": row["correlation_id"],
+                    "binding_id": binding_id,
+                    "binding_hash": str(row["binding_hash"]),
+                    "relation_ref": str(row["object_ref"]),
+                    "relation_digest": relation_digest,
+                    "ttl_seconds": ttl_seconds,
+                },
+            ).scalar_one_or_none()
+            if selected is None:
+                selected = connection.execute(
+                    text(
+                        """
+                        SELECT id::text
+                        FROM public.rule_sql_staging_receipts
+                        WHERE producer_rule_run_id = CAST(:run_id AS uuid)
+                          AND output_binding_id =
+                              CAST(:binding_id AS uuid)
+                        """
+                    ),
+                    {"run_id": run_id, "binding_id": binding_id},
+                ).scalar_one()
+        return f"dataops-staging://{selected}"
+
+    def resolve_sql_staging(
+        self,
+        receipt_ref: str,
+        *,
+        deployment_id: str,
+        correlation_id: str,
+        input_binding_id: str,
+    ) -> dict[str, str]:
+        match = re.fullmatch(
+            r"dataops-staging://([0-9a-f-]{36})",
+            str(receipt_ref or ""),
+        )
+        if match is None:
+            raise ValueError("SQL staging receipt is invalid")
+        receipt_id = _uid(match.group(1), "SQL staging receipt id")
+        deployment = _uid(deployment_id, "deployment id")
+        correlation = _uid(correlation_id, "correlation id")
+        binding_id = _uid(input_binding_id, "input binding id")
+        with self.engine.connect() as connection:
+            row = connection.execute(
+                text(
+                    """
+                    SELECT s.relation_ref, s.relation_digest,
+                           s.output_binding_hash, b.binding_hash,
+                           b.object_ref, b.object_kind, b.access_mode
+                    FROM public.rule_sql_staging_receipts s
+                    JOIN public.rule_runs r
+                      ON r.id = s.producer_rule_run_id
+                    JOIN public.dataflow_dataset_bindings b
+                      ON b.id = s.output_binding_id
+                    WHERE s.id = CAST(:id AS uuid)
+                      AND s.deployment_id =
+                          CAST(:deployment_id AS uuid)
+                      AND s.correlation_id =
+                          CAST(:correlation_id AS uuid)
+                      AND s.output_binding_id =
+                          CAST(:input_binding_id AS uuid)
+                      AND s.status = 'ready'
+                      AND s.expires_at > CURRENT_TIMESTAMP
+                      AND s.commit_outcome = 'committed'
+                      AND r.status = 'success'
+                      AND r.commit_outcome = 'committed'
+                      AND b.binding_hash = s.output_binding_hash
+                      AND b.access_mode IN ('read','read_write')
+                    """
+                ),
+                {
+                    "id": receipt_id,
+                    "deployment_id": deployment,
+                    "correlation_id": correlation,
+                    "input_binding_id": binding_id,
+                },
+            ).mappings().one_or_none()
+        if row is None:
+            raise ValueError("SQL staging receipt is not executable")
+        expected_digest = _canonical_digest(
+            {
+                "binding_hash": str(row["binding_hash"]),
+                "object_kind": str(row["object_kind"]),
+                "object_ref": str(row["object_ref"]),
+            }
+        )
+        if (
+            expected_digest != str(row["relation_digest"])
+            or str(row["relation_ref"]) != str(row["object_ref"])
+        ):
+            raise ValueError("SQL staging receipt attestation does not match")
+        return {
+            "relation_ref": str(row["relation_ref"]),
+            "relation_digest": str(row["relation_digest"]),
+        }
+
     def replay(self, rule_run_id: str) -> dict[str, Any] | None:
         run_id = _uid(rule_run_id, "rule run id")
         with self.engine.connect() as connection:
@@ -242,6 +634,57 @@ class PostgresRuleEvidenceWriter:
             "commit_outcome": str(row["commit_outcome"]),
         }
 
+    def replay_by_lease_owner(
+        self,
+        *,
+        lease_owner: str,
+        deployment_id: str,
+        correlation_id: str,
+        component_binding_id: str,
+        rule_version_id: str,
+        plan_hash: str,
+    ) -> dict[str, Any] | None:
+        owner = _uuid(lease_owner, "lease owner")
+        deployment = _uid(deployment_id, "deployment id")
+        correlation = _uid(correlation_id, "correlation id")
+        component = _uid(
+            component_binding_id,
+            "component binding id",
+        )
+        rule = _uid(rule_version_id, "rule version id")
+        if _DIGEST.fullmatch(str(plan_hash or "")) is None:
+            raise ValueError("plan hash is invalid")
+        with self.engine.connect() as connection:
+            row = connection.execute(
+                text(
+                    """
+                    SELECT id::text
+                    FROM public.rule_runs
+                    WHERE lease_owner = CAST(:lease_owner AS uuid)
+                      AND deployment_id =
+                          CAST(:deployment_id AS uuid)
+                      AND correlation_id =
+                          CAST(:correlation_id AS uuid)
+                      AND component_binding_id =
+                          CAST(:component_binding_id AS uuid)
+                      AND rule_version_id =
+                          CAST(:rule_version_id AS uuid)
+                      AND plan_hash = :plan_hash
+                    """
+                ),
+                {
+                    "lease_owner": owner,
+                    "deployment_id": deployment,
+                    "correlation_id": correlation,
+                    "component_binding_id": component,
+                    "rule_version_id": rule,
+                    "plan_hash": plan_hash,
+                },
+            ).scalar_one_or_none()
+        if row is None:
+            return None
+        return self.replay(str(row))
+
     @staticmethod
     def _validate_finish(result: Any) -> dict[str, Any]:
         if not isinstance(result, dict) or set(result) - _FINISH_KEYS:
@@ -263,21 +706,8 @@ class PostgresRuleEvidenceWriter:
         ):
             raise ValueError("rule evidence timings are invalid")
         public_result = result.get("public_result")
-        if public_result is not None and (
-            not isinstance(public_result, dict)
-            or any(str(key).startswith("_") for key in public_result)
-            or "rows" in public_result
-            or len(
-                json.dumps(
-                    public_result,
-                    sort_keys=True,
-                    separators=(",", ":"),
-                    ensure_ascii=False,
-                ).encode("utf-8")
-            )
-            > 32_768
-        ):
-            raise ValueError("public rule result is not evidence safe")
+        if public_result is not None:
+            validate_public_rule_result(public_result)
         sample = result.get("violation_sample")
         if sample is not None:
             if (
@@ -527,13 +957,15 @@ class PostgresRuleEvidenceWriter:
     def finish(self, rule_run_id: str, result: Any) -> None:
         run_id = _uid(rule_run_id, "rule run id")
         normalized = self._validate_finish(result)
+        evidence_digest = _canonical_digest(normalized)
         sample_path = None
         try:
             with self.engine.connect() as connection:
                 current = connection.execute(
                     text(
                         """
-                        SELECT status, correlation_id::text
+                        SELECT status, correlation_id::text,
+                               evidence_digest
                         FROM public.rule_runs
                         WHERE id = CAST(:id AS uuid)
                         """
@@ -543,8 +975,11 @@ class PostgresRuleEvidenceWriter:
             if current is None:
                 raise ValueError("rule run was not found")
             if current["status"] not in {"queued", "running"}:
-                replay = self.replay(run_id)
-                if replay and replay["status"] == normalized["status"]:
+                if (
+                    current["status"] == normalized["status"]
+                    and str(current["evidence_digest"] or "")
+                    == evidence_digest
+                ):
                     return
                 raise ValueError("rule run evidence is immutable")
             if normalized.get("violation_sample"):
@@ -569,6 +1004,8 @@ class PostgresRuleEvidenceWriter:
                                 commit_outcome = :commit_outcome,
                                 public_result = CAST(:public_result AS jsonb),
                                 failure_code = :failure_code,
+                                evidence_digest = :evidence_digest,
+                                lease_expires_at = NULL,
                                 finished_at = CURRENT_TIMESTAMP,
                                 updated_at = CURRENT_TIMESTAMP
                             WHERE id = CAST(:id AS uuid)
@@ -607,13 +1044,15 @@ class PostgresRuleEvidenceWriter:
                                     f"execution_{normalized['status']}"
                                 )
                             ),
+                            "evidence_digest": evidence_digest,
                         },
                     )
                     if updated.rowcount != 1:
                         state = connection.execute(
                             text(
                                 """
-                                SELECT status, commit_outcome
+                                SELECT status, commit_outcome,
+                                       evidence_digest
                                 FROM public.rule_runs
                                 WHERE id = CAST(:id AS uuid)
                                 """
@@ -625,22 +1064,69 @@ class PostgresRuleEvidenceWriter:
                             or state["status"] != normalized["status"]
                             or state["commit_outcome"]
                             != normalized["commit_outcome"]
+                            or str(state["evidence_digest"] or "")
+                            != evidence_digest
                         ):
                             raise RuntimeError(
                                 "rule run finalize outcome is unknown"
                             )
+                    receipt_status = (
+                        "ready"
+                        if normalized["status"] == "success"
+                        and normalized["commit_outcome"] == "committed"
+                        else "failed"
+                    )
+                    connection.execute(
+                        text(
+                            """
+                            UPDATE public.rule_sql_staging_receipts
+                            SET status = :status,
+                                ready_at = CASE
+                                    WHEN :status = 'ready'
+                                    THEN CURRENT_TIMESTAMP
+                                    ELSE ready_at
+                                END,
+                                commit_outcome = CASE
+                                    WHEN :status = 'ready'
+                                    THEN 'committed'
+                                    ELSE 'unknown'
+                                END,
+                                updated_at = CURRENT_TIMESTAMP
+                            WHERE producer_rule_run_id =
+                                CAST(:run_id AS uuid)
+                              AND status = 'pending'
+                            """
+                        ),
+                        {
+                            "run_id": run_id,
+                            "status": receipt_status,
+                        },
+                    )
             except Exception as exc:
                 try:
-                    replay = self.replay(run_id)
+                    with self.engine.connect() as connection:
+                        terminal = connection.execute(
+                            text(
+                                """
+                                SELECT status, commit_outcome,
+                                       evidence_digest
+                                FROM public.rule_runs
+                                WHERE id = CAST(:id AS uuid)
+                                """
+                            ),
+                            {"id": run_id},
+                        ).mappings().one_or_none()
                 except Exception as recheck_exc:
                     raise RuntimeError(
                         "rule run finalize outcome is unknown"
                     ) from recheck_exc
                 if (
-                    replay is None
-                    or replay["status"] != normalized["status"]
-                    or replay["commit_outcome"]
+                    terminal is None
+                    or terminal["status"] != normalized["status"]
+                    or terminal["commit_outcome"]
                     != normalized["commit_outcome"]
+                    or str(terminal["evidence_digest"] or "")
+                    != evidence_digest
                 ):
                     raise RuntimeError(
                         "rule run finalize outcome is unknown"
@@ -689,4 +1175,7 @@ class PostgresRuleEvidenceWriter:
         return removed
 
 
-__all__ = ["PostgresRuleEvidenceWriter"]
+__all__ = [
+    "PostgresRuleEvidenceWriter",
+    "validate_public_rule_result",
+]

app/runner/rules.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+import logging
 import re
 import time
 from asyncio import CancelledError
@@ -26,6 +27,7 @@ from app.core.data_rules.compilers.sql import (
     validate_bound_sql_plan,
 )
 from app.runner.nodes import NodeExecutionError
+from app.runner.rule_evidence import validate_public_rule_result
 
 CONFIG_KEYS = {
     "component_binding_id",
@@ -38,6 +40,7 @@ IDEMPOTENCY_STRATEGIES = {
     "upsert",
     "deduplication_key",
 }
+LOGGER = logging.getLogger(__name__)
 
 
 def _canonical_hash(value: Any) -> str:
@@ -168,11 +171,17 @@ class RulePlanExecutor:
             return None, None
         correlation_id = execution_context.get("correlation_id")
         dataflow_uid = execution_context.get("dataflow_uid")
+        deployment_id = execution_context.get("deployment_id")
+        environment = execution_context.get("environment")
         workflow_version = execution_context.get("workflow_version")
         node_id = execution_context.get("node_id")
+        task_jti = execution_context.get("task_jti")
         if (
             not correlation_id
             or not dataflow_uid
+            or not deployment_id
+            or environment
+            not in {"development", "test", "production"}
             or isinstance(workflow_version, bool)
             or not isinstance(workflow_version, int)
             or workflow_version < 1
@@ -189,11 +198,15 @@ class RulePlanExecutor:
                 plan_hash=plan_hash,
                 correlation_id=correlation_id,
                 dataflow_uid=dataflow_uid,
+                deployment_id=deployment_id,
+                environment=environment,
                 workflow_version=workflow_version,
                 node_id=node_id,
+                lease_owner=task_jti,
             )
             replay = self.evidence_writer.replay(rule_run_id)
         except Exception as exc:
+            LOGGER.exception("governed rule evidence start failed")
             raise NodeExecutionError(
                 "rule execution evidence is unavailable"
             ) from exc
@@ -210,6 +223,48 @@ class RulePlanExecutor:
                 commit_outcome="unknown",
             ) from exc
 
+    def replay_task(
+        self,
+        *,
+        node,
+        correlation_id,
+        deployment_id,
+        task_jti,
+    ):
+        if self.evidence_writer is None:
+            return None
+        config = node.get("config")
+        if not isinstance(config, dict):
+            return None
+        try:
+            replay = self.evidence_writer.replay_by_lease_owner(
+                lease_owner=task_jti,
+                deployment_id=deployment_id,
+                correlation_id=correlation_id,
+                component_binding_id=_uid(
+                    config.get("component_binding_id"),
+                    "component_binding_id",
+                ),
+                rule_version_id=_uid(
+                    config.get("rule_version_id"),
+                    "rule_version_id",
+                ),
+                plan_hash=str(
+                    config.get("execution_plan_hash") or ""
+                ),
+            )
+        except Exception as exc:
+            raise NodeExecutionError(
+                "rule execution evidence is unavailable"
+            ) from exc
+        if replay is None or replay.get("status") != "success":
+            return None
+        return {
+            key: value
+            for key, value in replay.items()
+            if key != "status"
+        }
+
     def execute(
         self,
         node,
@@ -352,20 +407,17 @@ class RulePlanExecutor:
         if adapter is None or not callable(getattr(adapter, "execute", None)):
             raise NodeExecutionError("rule plan backend is not registered")
         adapter_parameters = parameters
+        staging_input = None
         if backend == "sql_pushdown" and parameters not in ({}, None):
-            expected = (
-                "dataops-staging://"
-                f"{execution_context.get('correlation_id')}/"
-                f"{plan['input_binding_id']}"
-            )
             if (
                 not isinstance(parameters, dict)
                 or set(parameters) != {"input_artifact"}
-                or parameters["input_artifact"] != expected
+                or not isinstance(parameters["input_artifact"], str)
             ):
                 raise NodeExecutionError(
-                    "SQL rule handoff is not a canonical staging binding"
+                    "SQL rule handoff must contain one opaque receipt"
                 )
+            staging_input = parameters["input_artifact"]
             adapter_parameters = {}
         rule_run_id, replay = self._start_evidence(
             component_binding_id=component_binding_id,
@@ -390,6 +442,44 @@ class RulePlanExecutor:
                     ),
                 )
             raise NodeExecutionError("rule execution is already in progress")
+        if self.evidence_writer is not None:
+            try:
+                self.evidence_writer.heartbeat(
+                    rule_run_id,
+                    execution_context.get("task_jti"),
+                )
+            except Exception as exc:
+                raise NodeExecutionError(
+                    "rule execution lease is unavailable"
+                ) from exc
+        if staging_input is not None:
+            if self.evidence_writer is None:
+                raise NodeExecutionError(
+                    "SQL staging receipt resolver is unavailable"
+                )
+            try:
+                self.evidence_writer.resolve_sql_staging(
+                    staging_input,
+                    deployment_id=execution_context.get(
+                        "deployment_id"
+                    ),
+                    correlation_id=execution_context.get(
+                        "correlation_id"
+                    ),
+                    input_binding_id=plan["input_binding_id"],
+                )
+            except Exception as exc:
+                self._finish_evidence(
+                    rule_run_id,
+                    {
+                        "status": "failed",
+                        "commit_outcome": "not_committed",
+                        "timings": {"duration_ms": 0},
+                    },
+                )
+                raise NodeExecutionError(
+                    "SQL staging receipt is not executable"
+                ) from exc
         adapter_context = {}
         if backend == "polars_batch":
             adapter_context["correlation_id"] = execution_context.get(
@@ -470,6 +560,30 @@ class RulePlanExecutor:
                 },
             )
             raise error
+        staging_output = None
+        if backend == "sql_pushdown" and self.evidence_writer is not None:
+            try:
+                staging_output = self.evidence_writer.stage_sql_output(
+                    rule_run_id,
+                    output_binding_id=plan["output_binding_id"],
+                )
+            except Exception as exc:
+                self._finish_evidence(
+                    rule_run_id,
+                    {
+                        "status": "unknown",
+                        "commit_outcome": "unknown",
+                        "timings": {
+                            "duration_ms": int(
+                                (time.monotonic() - started) * 1000
+                            )
+                        },
+                    },
+                )
+                raise NodeExecutionError(
+                    "SQL staging receipt outcome is unknown",
+                    commit_outcome="unknown",
+                ) from exc
         try:
             sample = self._redacted_sample(
                 result.pop("_violation_sample", None)
@@ -493,12 +607,15 @@ class RulePlanExecutor:
             artifact_ref = public_result.get("artifact_ref")
             if isinstance(artifact_ref, str):
                 public_result["output_artifact"] = artifact_ref
-            elif backend == "sql_pushdown":
-                public_result["output_artifact"] = (
-                    "dataops-staging://"
-                    f"{execution_context.get('correlation_id')}/"
-                    f"{plan['output_binding_id']}"
-                )
+            elif (
+                backend == "sql_pushdown"
+                and staging_output is not None
+            ):
+                public_result["output_artifact"] = staging_output
+            public_result = validate_public_rule_result(
+                public_result,
+                backend=backend,
+            )
         except (TypeError, ValueError) as exc:
             self._finish_evidence(
                 rule_run_id,

migrations/versions/20260723_160_rule_execution_evidence.py

@@ -40,6 +40,15 @@ def upgrade() -> None:
 
         DO $$
         BEGIN
+            IF EXISTS (
+                SELECT 1
+                FROM public.rule_violation_samples
+                GROUP BY rule_run_id
+                HAVING COUNT(*) > 1
+            ) THEN
+                RAISE EXCEPTION
+                    'violation sample migration found duplicate rule_run_id';
+            END IF;
             IF EXISTS (
                 SELECT 1
                 FROM public.rule_violation_samples

migrations/versions/20260723_170_rule_execution_attempts.py

@@ -0,0 +1,74 @@
+"""Add exact task identity, leases, replay, and SQL staging receipts."""
+
+from alembic import op
+
+revision = "20260723_170"
+down_revision = "20260723_160"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE public.rule_runs
+            ADD COLUMN attempt_no INTEGER NOT NULL DEFAULT 1
+                CHECK (attempt_no > 0),
+            ADD COLUMN lease_owner UUID,
+            ADD COLUMN lease_expires_at TIMESTAMPTZ,
+            ADD COLUMN heartbeat_at TIMESTAMPTZ,
+            ADD COLUMN evidence_digest CHAR(64);
+
+        CREATE INDEX idx_rule_runs_lease
+            ON public.rule_runs(status, lease_expires_at, id);
+
+        ALTER TABLE public.runner_task_executions
+            ADD COLUMN deployment_id UUID,
+            ADD COLUMN environment VARCHAR(20)
+                CHECK (environment IN (
+                    'development','test','production'
+                )),
+            ADD COLUMN replay_http_status SMALLINT
+                CHECK (replay_http_status BETWEEN 200 AND 599),
+            ADD COLUMN replay_body JSONB,
+            ADD COLUMN replay_digest CHAR(64);
+
+        CREATE INDEX idx_runner_task_execution_identity
+            ON public.runner_task_executions
+            (deployment_id, environment, correlation_id, node_id);
+
+        CREATE TABLE public.rule_sql_staging_receipts (
+            id UUID PRIMARY KEY,
+            producer_rule_run_id UUID NOT NULL
+                REFERENCES public.rule_runs(id) ON DELETE RESTRICT,
+            deployment_id UUID NOT NULL
+                REFERENCES public.dataflow_deployments(id)
+                ON DELETE RESTRICT,
+            correlation_id UUID NOT NULL,
+            output_binding_id UUID NOT NULL
+                REFERENCES public.dataflow_dataset_bindings(id)
+                ON DELETE RESTRICT,
+            output_binding_hash CHAR(64) NOT NULL,
+            relation_ref VARCHAR(1000) NOT NULL,
+            relation_digest CHAR(64) NOT NULL,
+            commit_outcome VARCHAR(30) NOT NULL
+                CHECK (commit_outcome IN ('committed','unknown')),
+            status VARCHAR(20) NOT NULL
+                CHECK (status IN ('pending','ready','failed','expired')),
+            expires_at TIMESTAMPTZ NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            ready_at TIMESTAMPTZ,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (producer_rule_run_id, output_binding_id)
+        );
+        CREATE INDEX idx_rule_sql_staging_receipt_lookup
+            ON public.rule_sql_staging_receipts
+            (id, deployment_id, correlation_id, status, expires_at);
+        """
+    )
+
+
+def downgrade() -> None:
+    raise RuntimeError(
+        "rule attempt leases and staging receipts are forward-only"
+    )

tests/core/orchestration/test_kestra_compiler.py

@@ -147,35 +147,40 @@ def test_downstream_nodes_receive_only_closed_artifact_handoffs():
 
     single = by_node["notify_done"]["parameters"]
     assert single["input_artifact"] == (
-        "{{ outputs.dataops_dag.tasks.read_orders.body.output_artifact }}"
+        "{{ outputs.dataops_dag.tasks['read_orders'].body.output_artifact }}"
     )
     assert "rows" not in json.dumps(single)
 
-    # Both edges target the same logical node only after adding another
-    # predecessor; the compiler must name every source instead of guessing.
+    # Governed rules support exactly one materialized input today.
+    spec["nodes"][-1]["type"] = "rule.apply"
+    spec["nodes"][-1]["purpose"] = "write"
+    spec["nodes"][-1]["config"] = {
+        "component_binding_id": new_governance_uid(),
+        "rule_version_id": new_governance_uid(),
+        "execution_plan_hash": "a" * 64,
+    }
+    spec["nodes"][-1]["idempotency"] = {
+        "strategy": "upsert",
+        "key": "id",
+    }
     spec["edges"].append(
         {"from": "notify_done", "to": "publish_result"}
     )
+    with pytest.raises(ValueError, match="exactly one"):
+        compile_kestra_flow(spec, schedule_plan(), "test", 1)
+
+
+def test_artifact_expression_is_bracket_safe_for_hyphenated_node_id():
+    spec = workflow_spec()
+    spec["nodes"][0]["id"] = "read-orders"
+    spec["edges"][0]["from"] = "read-orders"
+
     source = compile_kestra_flow(spec, schedule_plan(), "test", 1).yaml
-    body_lines = [
-        json.loads(json.loads(line.split("body: ", 1)[1]))
-        for line in source.splitlines()
-        if line.strip().startswith("body: ")
-    ]
-    parameters = next(
-        body["parameters"]
-        for body in body_lines
-        if body["node"]["id"] == "publish_result"
+
+    assert (
+        "{{ outputs.dataops_dag.tasks['read-orders']"
+        ".body.output_artifact }}" in source
     )
-    assert parameters["input_artifacts"] == {
-        "notify_done": (
-            "{{ outputs.dataops_dag.tasks.notify_done.body.output_artifact }}"
-        ),
-        "read_orders": (
-            "{{ outputs.dataops_dag.tasks.read_orders.body.output_artifact }}"
-        ),
-    }
-    assert "input_artifact" not in parameters
 
 
 def test_manual_plan_creates_no_automatic_trigger():

tests/integration/test_data_rule_polars_execution.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 import re
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 import polars as pl
@@ -112,6 +113,11 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
     correlation_id = new_governance_uid()
     failure_correlation_id = new_governance_uid()
     unknown_correlation_id = new_governance_uid()
+    lease_correlation_id = new_governance_uid()
+    sample_crash_correlation_id = new_governance_uid()
+    receipt_correlation_id = new_governance_uid()
+    failed_receipt_correlation_id = new_governance_uid()
+    sql_binding_id = new_governance_uid()
     prefix = f"rules/{correlation_id}/"
     customer_table = "task5_polars_customers"
     segment_table = "task5_polars_segments"
@@ -775,6 +781,8 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         task_token = TaskTokenIssuer(task_secret).issue(
             task_uid=new_governance_uid(),
             dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
             workflow_version=1,
             correlation_id=correlation_id,
             node=node,
@@ -784,6 +792,8 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         retry_token = TaskTokenIssuer(task_secret).issue(
             task_uid=new_governance_uid(),
             dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
             workflow_version=1,
             correlation_id=correlation_id,
             node=node,
@@ -837,14 +847,16 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "parameters": {},
                 },
             )
-        assert http_result.status_code == 200
+        assert http_result.status_code == 200, http_result.get_json()
         assert http_result.get_json()["output_artifact"] == result[
             "artifact_ref"
         ]
         assert http_result.get_json()["result"]["artifact_ref"] == result[
             "artifact_ref"
         ]
-        assert replay.status_code == 409
+        assert replay.status_code == 200
+        assert replay.headers["X-Idempotent-Replay"] == "true"
+        assert replay.get_json() == http_result.get_json()
         assert retried.status_code == 200
         assert retried.get_json()["output_artifact"] == result[
             "artifact_ref"
@@ -928,6 +940,32 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 "version_no": "[REDACTED]",
             }
         ]
+        orphan = store.write(
+            pl.DataFrame({"value": ["orphan"]}),
+            correlation_id,
+            900,
+            schema_fields=[
+                {
+                    "name": "value",
+                    "type": "string",
+                    "nullable": True,
+                }
+            ],
+        )
+        original_clock = store.clock
+        store.clock = lambda: datetime.now(UTC) + timedelta(seconds=60)
+        try:
+            reconciliation = resolver.reconcile(
+                limit=20,
+                grace_seconds=30,
+            )
+        finally:
+            store.clock = original_clock
+        assert reconciliation["orphans_deleted"] >= 1
+        assert store.describe_optional(orphan["artifact_ref"]) is None
+        assert store.describe(sample_evidence["artifact_ref"])[
+            "digest"
+        ] == sample_evidence["artifact_digest"]
         downstream_node = {
             "id": "task6_real_handoff",
             "type": "rule.apply",
@@ -952,8 +990,11 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             write_authorized=True,
             correlation_id=correlation_id,
             dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
             workflow_version=1,
             node_id="task6_real_handoff",
+            task_jti=new_governance_uid(),
         )
         assert downstream_result["rows_in"] == 2
         assert downstream_result["rows_out"] == 2
@@ -974,8 +1015,11 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             write_authorized=True,
             correlation_id=correlation_id,
             dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
             workflow_version=1,
             node_id="task6_real_handoff",
+            task_jti=new_governance_uid(),
         )
         assert replayed_downstream["artifact_ref"] == downstream_result[
             "artifact_ref"
@@ -1017,8 +1061,11 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 write_authorized=True,
                 correlation_id=failure_correlation_id,
                 dataflow_uid=dataflow_uid,
+                deployment_id=deployment_id,
+                environment="test",
                 workflow_version=1,
                 node_id="task5_real_polars",
+                task_jti=new_governance_uid(),
             )
         with platform.connect() as connection:
             failed_evidence = connection.execute(
@@ -1060,8 +1107,11 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 write_authorized=True,
                 correlation_id=unknown_correlation_id,
                 dataflow_uid=dataflow_uid,
+                deployment_id=deployment_id,
+                environment="test",
                 workflow_version=1,
                 node_id="task5_real_polars",
+                task_jti=new_governance_uid(),
             )
         with platform.connect() as connection:
             unknown_evidence = connection.execute(
@@ -1079,6 +1129,288 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             "status": "unknown",
             "commit_outcome": "unknown",
         }
+        evidence_writer = PostgresRuleEvidenceWriter(
+            platform,
+            store,
+            sample_ttl_seconds=900,
+            lease_seconds=30,
+        )
+        lease_owner = new_governance_uid()
+        lease_run_id = evidence_writer.start(
+            component_binding_id=component_binding_id,
+            rule_version_id=rule_id,
+            plan_hash=compiled["plan_hash"],
+            correlation_id=lease_correlation_id,
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            node_id="task5_real_polars",
+            lease_owner=lease_owner,
+        )
+        with pytest.raises(ValueError, match="not owned"):
+            evidence_writer.heartbeat(
+                lease_run_id,
+                new_governance_uid(),
+            )
+        evidence_writer.heartbeat(lease_run_id, lease_owner)
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_runs
+                    SET lease_expires_at =
+                        CURRENT_TIMESTAMP - INTERVAL '1 second'
+                    WHERE id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": lease_run_id},
+            )
+        assert evidence_writer.start(
+            component_binding_id=component_binding_id,
+            rule_version_id=rule_id,
+            plan_hash=compiled["plan_hash"],
+            correlation_id=lease_correlation_id,
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            node_id="task5_real_polars",
+            lease_owner=new_governance_uid(),
+        ) == lease_run_id
+        assert evidence_writer.replay(lease_run_id)["status"] == "unknown"
+
+        class FailOnceConnection:
+            def __init__(self, connection, state):
+                self.connection = connection
+                self.state = state
+
+            def execute(self, statement, parameters=None):
+                sql = str(statement)
+                if (
+                    self.state["armed"]
+                    and "UPDATE public.rule_runs" in sql
+                    and "rows_in = :rows_in" in sql
+                ):
+                    self.state["armed"] = False
+                    raise RuntimeError("simulated response loss")
+                return self.connection.execute(statement, parameters)
+
+        class FailOnceBegin:
+            def __init__(self, context, state):
+                self.context = context
+                self.state = state
+
+            def __enter__(self):
+                return FailOnceConnection(
+                    self.context.__enter__(),
+                    self.state,
+                )
+
+            def __exit__(self, *args):
+                return self.context.__exit__(*args)
+
+        class FailOnceEngine:
+            def __init__(self, engine):
+                self.engine = engine
+                self.state = {"armed": True}
+
+            def connect(self):
+                return self.engine.connect()
+
+            def begin(self):
+                return FailOnceBegin(
+                    self.engine.begin(),
+                    self.state,
+                )
+
+        sample_crash_writer = PostgresRuleEvidenceWriter(
+            FailOnceEngine(platform),
+            store,
+            sample_ttl_seconds=900,
+        )
+        sample_crash_run_id = sample_crash_writer.start(
+            component_binding_id=component_binding_id,
+            rule_version_id=rule_id,
+            plan_hash=compiled["plan_hash"],
+            correlation_id=sample_crash_correlation_id,
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            node_id="task5_real_polars",
+            lease_owner=new_governance_uid(),
+        )
+        sample_crash_evidence = {
+            "status": "success",
+            "rows_in": 1,
+            "rows_out": 0,
+            "rows_rejected": 1,
+            "rows_quarantined": 0,
+            "commit_outcome": "committed",
+            "timings": {"duration_ms": 1},
+            "violation_sample": [{"mobile": "[REDACTED]"}],
+            "sample_count": 1,
+            "redaction_policy": "rule-violation-default-v1",
+        }
+        with pytest.raises(RuntimeError, match="outcome is unknown"):
+            sample_crash_writer.finish(
+                sample_crash_run_id,
+                sample_crash_evidence,
+            )
+        with platform.connect() as connection:
+            assert connection.execute(
+                text(
+                    """
+                    SELECT handoff_status
+                    FROM public.rule_violation_samples
+                    WHERE rule_run_id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": sample_crash_run_id},
+            ).scalar_one() == "ready"
+        evidence_writer.finish(
+            sample_crash_run_id,
+            sample_crash_evidence,
+        )
+        assert evidence_writer.replay(sample_crash_run_id)[
+            "status"
+        ] == "success"
+
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.dataflow_dataset_bindings
+                    (id, dataflow_deployment_id, logical_ref,
+                     data_source_uid, object_kind, object_ref,
+                     schema_snapshot_id, dialect, access_mode, write_mode,
+                     binding_hash)
+                    VALUES (
+                        CAST(:id AS uuid), CAST(:deployment_id AS uuid),
+                        'sql_receipt', CAST(:source_uid AS uuid), 'table',
+                        'public.task6_sql_receipt',
+                        CAST(:schema_snapshot_id AS uuid), 'postgresql',
+                        'read_write', 'upsert', :binding_hash
+                    )
+                    """
+                ),
+                {
+                    "id": sql_binding_id,
+                    "deployment_id": deployment_id,
+                    "source_uid": output_binding["data_source_uid"],
+                    "schema_snapshot_id": output_schema["id"],
+                    "binding_hash": "b" * 64,
+                },
+            )
+        receipt_run_id = evidence_writer.start(
+            component_binding_id=component_binding_id,
+            rule_version_id=rule_id,
+            plan_hash=compiled["plan_hash"],
+            correlation_id=receipt_correlation_id,
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            node_id="task5_real_polars",
+            lease_owner=new_governance_uid(),
+        )
+        receipt = evidence_writer.stage_sql_output(
+            receipt_run_id,
+            output_binding_id=sql_binding_id,
+        )
+        assert re.fullmatch(
+            r"dataops-staging://[0-9a-f-]{36}",
+            receipt,
+        )
+        with pytest.raises(ValueError, match="not executable"):
+            evidence_writer.resolve_sql_staging(
+                receipt,
+                deployment_id=deployment_id,
+                correlation_id=receipt_correlation_id,
+                input_binding_id=sql_binding_id,
+            )
+        receipt_evidence = {
+            "status": "success",
+            "rows_in": 2,
+            "rows_out": 2,
+            "rows_rejected": 0,
+            "rows_quarantined": 0,
+            "commit_outcome": "committed",
+            "timings": {"duration_ms": 1},
+            "public_result": {
+                "rows_in": 2,
+                "rows_out": 2,
+                "rows_rejected": 0,
+                "rows_quarantined": 0,
+                "commit_outcome": "committed",
+            },
+        }
+        evidence_writer.finish(receipt_run_id, receipt_evidence)
+        evidence_writer.finish(receipt_run_id, receipt_evidence)
+        with pytest.raises(ValueError, match="immutable"):
+            evidence_writer.finish(
+                receipt_run_id,
+                {
+                    **receipt_evidence,
+                    "rows_out": 1,
+                },
+            )
+        resolved_receipt = evidence_writer.resolve_sql_staging(
+            receipt,
+            deployment_id=deployment_id,
+            correlation_id=receipt_correlation_id,
+            input_binding_id=sql_binding_id,
+        )
+        assert resolved_receipt["relation_ref"] == (
+            "public.task6_sql_receipt"
+        )
+        assert len(resolved_receipt["relation_digest"]) == 64
+        with pytest.raises(ValueError, match="not executable"):
+            evidence_writer.resolve_sql_staging(
+                receipt,
+                deployment_id=deployment_id,
+                correlation_id=new_governance_uid(),
+                input_binding_id=sql_binding_id,
+            )
+        with pytest.raises(ValueError, match="invalid"):
+            evidence_writer.resolve_sql_staging(
+                "dataops-staging://public.task6_sql_receipt",
+                deployment_id=deployment_id,
+                correlation_id=receipt_correlation_id,
+                input_binding_id=sql_binding_id,
+            )
+        failed_receipt_run_id = evidence_writer.start(
+            component_binding_id=component_binding_id,
+            rule_version_id=rule_id,
+            plan_hash=compiled["plan_hash"],
+            correlation_id=failed_receipt_correlation_id,
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            node_id="task5_real_polars",
+            lease_owner=new_governance_uid(),
+        )
+        failed_receipt = evidence_writer.stage_sql_output(
+            failed_receipt_run_id,
+            output_binding_id=sql_binding_id,
+        )
+        evidence_writer.finish(
+            failed_receipt_run_id,
+            {
+                "status": "failed",
+                "commit_outcome": "not_committed",
+                "timings": {"duration_ms": 1},
+            },
+        )
+        with pytest.raises(ValueError, match="not executable"):
+            evidence_writer.resolve_sql_staging(
+                failed_receipt,
+                deployment_id=deployment_id,
+                correlation_id=failed_receipt_correlation_id,
+                input_binding_id=sql_binding_id,
+            )
         conflict_path = tmp_path / "conflict.parquet"
         pl.DataFrame(
             {
@@ -1167,17 +1499,41 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     ),
                     {"jti": retry_ledger_jti},
                 )
+            connection.execute(
+                text(
+                    """
+                    DELETE FROM public.rule_sql_staging_receipts
+                    WHERE correlation_id IN (
+                        CAST(:receipt_correlation_id AS uuid),
+                        CAST(:failed_receipt_correlation_id AS uuid)
+                    )
+                    """
+                ),
+                {
+                    "receipt_correlation_id": receipt_correlation_id,
+                    "failed_receipt_correlation_id": (
+                        failed_receipt_correlation_id
+                    ),
+                },
+            )
             connection.execute(
                 text(
                     """
                     DELETE FROM public.rule_violation_samples s
                     USING public.rule_runs r
                     WHERE s.rule_run_id = r.id
-                      AND r.correlation_id =
-                          CAST(:correlation_id AS uuid)
+                      AND r.correlation_id IN (
+                          CAST(:correlation_id AS uuid),
+                          CAST(:sample_crash_correlation_id AS uuid)
+                      )
                     """
                 ),
-                {"correlation_id": correlation_id},
+                {
+                    "correlation_id": correlation_id,
+                    "sample_crash_correlation_id": (
+                        sample_crash_correlation_id
+                    ),
+                },
             )
             connection.execute(
                 text(
@@ -1185,14 +1541,33 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "WHERE correlation_id IN ("
                     "CAST(:correlation_id AS uuid), "
                     "CAST(:failure_correlation_id AS uuid), "
-                    "CAST(:unknown_correlation_id AS uuid))"
+                    "CAST(:unknown_correlation_id AS uuid), "
+                    "CAST(:lease_correlation_id AS uuid), "
+                    "CAST(:sample_crash_correlation_id AS uuid), "
+                    "CAST(:receipt_correlation_id AS uuid), "
+                    "CAST(:failed_receipt_correlation_id AS uuid))"
                 ),
                 {
                     "correlation_id": correlation_id,
                     "failure_correlation_id": failure_correlation_id,
                     "unknown_correlation_id": unknown_correlation_id,
+                    "lease_correlation_id": lease_correlation_id,
+                    "sample_crash_correlation_id": (
+                        sample_crash_correlation_id
+                    ),
+                    "receipt_correlation_id": receipt_correlation_id,
+                    "failed_receipt_correlation_id": (
+                        failed_receipt_correlation_id
+                    ),
                 },
             )
+            connection.execute(
+                text(
+                    "DELETE FROM public.dataflow_dataset_bindings "
+                    "WHERE id = CAST(:id AS uuid)"
+                ),
+                {"id": sql_binding_id},
+            )
             connection.execute(
                 text(
                     "DELETE FROM public.rule_run_artifacts "

tests/integration/test_kestra_runner_execution.py

@@ -22,7 +22,6 @@ from app.core.orchestration.compilers import compile_kestra_flow
 from app.core.orchestration.engines import KestraAdapter
 from app.runner.auth import TaskTokenIssuer
 
-
 pytestmark = pytest.mark.integration
 
 
@@ -59,6 +58,8 @@ def test_kestra_passes_one_signed_task_to_the_runner():
     ).issue(
         task_uid=task_uid,
         dataflow_uid=dataflow_uid,
+        deployment_id="01900000-0000-7000-8000-000000000099",
+        environment="test",
         workflow_version=1,
         correlation_id=new_governance_uid(),
         node=node,

tests/integration/test_runner_datasource_pool.py

@@ -19,7 +19,6 @@ from app.core.data_source.models import (
 )
 from app.runner.auth import TaskTokenIssuer
 
-
 pytestmark = pytest.mark.integration
 
 
@@ -171,6 +170,8 @@ def test_runner_reuses_governed_pools_and_rejects_duplicate_submit():
             token = issuer.issue(
                 task_uid=task_uid,
                 dataflow_uid=dataflow_uid,
+                deployment_id=new_governance_uid(),
+                environment="test",
                 workflow_version=1,
                 correlation_id=new_governance_uid(),
                 node=node,

tests/runner/test_api.py

@@ -11,6 +11,21 @@ NODE = {
     "config": {"statement": "SELECT 1", "parameters": {}},
 }
 
+RULE_NODE = {
+    "id": "apply_orders",
+    "type": "rule.apply",
+    "purpose": "write",
+    "idempotency": {
+        "strategy": "upsert",
+        "key": "orders:2026-07-23",
+    },
+    "config": {
+        "component_binding_id": "01900000-0000-7000-8000-000000000021",
+        "rule_version_id": "01900000-0000-7000-8000-000000000022",
+        "execution_plan_hash": "a" * 64,
+    },
+}
+
 
 class Executor:
     def __init__(self):
@@ -40,6 +55,8 @@ def test_runner_accepts_one_signed_task_and_rejects_replay():
     token = issuer.issue(
         task_uid="01900000-0000-7000-8000-000000000011",
         dataflow_uid="01900000-0000-7000-8000-000000000012",
+        deployment_id="01900000-0000-7000-8000-000000000014",
+        environment="test",
         workflow_version=7,
         correlation_id="01900000-0000-7000-8000-000000000013",
         node=NODE,
@@ -91,6 +108,8 @@ def test_runner_fails_closed_when_the_durable_ledger_is_unavailable():
     token = issuer.issue(
         task_uid="01900000-0000-7000-8000-000000000011",
         dataflow_uid="01900000-0000-7000-8000-000000000012",
+        deployment_id="01900000-0000-7000-8000-000000000014",
+        environment="test",
         workflow_version=7,
         correlation_id="01900000-0000-7000-8000-000000000013",
         node=NODE,
@@ -109,3 +128,134 @@ def test_runner_fails_closed_when_the_durable_ledger_is_unavailable():
     assert response.status_code == 503
     assert response.get_json() == {"error": "task ledger is unavailable"}
     assert "password" not in response.get_data(as_text=True).lower()
+
+
+def _rule_token(issuer):
+    return issuer.issue(
+        task_uid="01900000-0000-7000-8000-000000000031",
+        dataflow_uid="01900000-0000-7000-8000-000000000032",
+        deployment_id="01900000-0000-7000-8000-000000000033",
+        environment="production",
+        workflow_version=7,
+        correlation_id="01900000-0000-7000-8000-000000000034",
+        node=RULE_NODE,
+        write_authorized=True,
+    )
+
+
+def test_governed_rule_same_jti_replays_durable_response_once():
+    issuer = TaskTokenIssuer("x" * 32, clock=lambda: 1_000)
+    executor = Executor()
+    ledger = InMemoryTaskLedger()
+    app = create_runner_app(
+        verifier=TaskTokenVerifier("x" * 32, clock=lambda: 1_000),
+        ledger=ledger,
+        registry=NodeRegistry({"rule.apply": executor}),
+    )
+    payload = {
+        "task_token": _rule_token(issuer),
+        "node": RULE_NODE,
+        "parameters": {},
+    }
+
+    with app.test_client() as client:
+        first = client.post("/v1/tasks/execute", json=payload)
+        replay = client.post("/v1/tasks/execute", json=payload)
+
+    assert first.status_code == replay.status_code == 200
+    assert first.get_json() == replay.get_json()
+    assert replay.headers["X-Idempotent-Replay"] == "true"
+    assert executor.calls == 1
+
+
+def test_governed_rule_recovers_terminal_evidence_after_response_loss():
+    class LostFirstFinishLedger(InMemoryTaskLedger):
+        def __init__(self):
+            super().__init__()
+            self.finish_calls = 0
+
+        def finish(self, jti, **outcome):
+            self.finish_calls += 1
+            if self.finish_calls == 1:
+                return
+            return super().finish(jti, **outcome)
+
+    class RecoverableExecutor(Executor):
+        def replay_task(self, **_context):
+            return {
+                "node": RULE_NODE["id"],
+                "parameters": {},
+                "rows_out": 9,
+            }
+
+    issuer = TaskTokenIssuer("x" * 32, clock=lambda: 1_000)
+    executor = RecoverableExecutor()
+    ledger = LostFirstFinishLedger()
+    app = create_runner_app(
+        verifier=TaskTokenVerifier("x" * 32, clock=lambda: 1_000),
+        ledger=ledger,
+        registry=NodeRegistry({"rule.apply": executor}),
+    )
+    payload = {
+        "task_token": _rule_token(issuer),
+        "node": RULE_NODE,
+        "parameters": {},
+    }
+
+    with app.test_client() as client:
+        first = client.post("/v1/tasks/execute", json=payload)
+        recovered = client.post("/v1/tasks/execute", json=payload)
+        replay = client.post("/v1/tasks/execute", json=payload)
+
+    assert first.status_code == recovered.status_code == replay.status_code == 200
+    assert recovered.get_json()["result"]["rows_out"] == 9
+    assert replay.get_json() == recovered.get_json()
+    assert executor.calls == 1
+
+
+def test_running_rule_without_terminal_evidence_returns_retryable_202():
+    class RunningExecutor(Executor):
+        def replay_task(self, **_context):
+            return None
+
+    issuer = TaskTokenIssuer("x" * 32, clock=lambda: 1_000)
+    executor = RunningExecutor()
+    ledger = InMemoryTaskLedger()
+    token = _rule_token(issuer)
+    verifier = TaskTokenVerifier("x" * 32, clock=lambda: 1_000)
+    claims = verifier.verify(token, node=RULE_NODE)
+    ledger.claim(
+        claims.jti,
+        {
+            "task_uid": claims.task_uid,
+            "dataflow_uid": claims.dataflow_uid,
+            "deployment_id": claims.deployment_id,
+            "environment": claims.environment,
+            "workflow_version": claims.workflow_version,
+            "correlation_id": claims.correlation_id,
+            "node_id": claims.node_id,
+            "node_type": claims.node_type,
+            "data_source_uid": None,
+            "idempotency_key": RULE_NODE["idempotency"]["key"],
+        },
+        expires_at=claims.expires_at,
+    )
+    app = create_runner_app(
+        verifier=verifier,
+        ledger=ledger,
+        registry=NodeRegistry({"rule.apply": executor}),
+    )
+
+    with app.test_client() as client:
+        response = client.post(
+            "/v1/tasks/execute",
+            json={
+                "task_token": token,
+                "node": RULE_NODE,
+                "parameters": {},
+            },
+        )
+
+    assert response.status_code == 202
+    assert response.headers["Retry-After"] == "2"
+    assert executor.calls == 0

tests/runner/test_artifact_handoff.py

@@ -338,9 +338,17 @@ class _ReconcileConnection:
                 None,
             )
             return _Result(copy.deepcopy(row))
-        if "SELECT artifact_ref" in sql and "handoff_status" not in sql:
+        if (
+            "SELECT artifact_ref" in sql
+            and "ANY(CAST(:artifact_refs AS text[]))" in sql
+        ):
             return _Result(
-                [{"artifact_ref": row["artifact_ref"]} for row in self.engine.rows]
+                [
+                    {"artifact_ref": row["artifact_ref"]}
+                    for row in self.engine.rows
+                    if row["artifact_ref"]
+                    in parameters["artifact_refs"]
+                ]
             )
         row = next(
             (

tests/runner/test_postgres_ledger.py

@@ -41,6 +41,8 @@ def test_postgres_ledger_claim_is_atomic_and_persists_only_bounded_fields():
     binding = {
         "task_uid": "01900000-0000-7000-8000-000000000011",
         "dataflow_uid": "01900000-0000-7000-8000-000000000012",
+        "deployment_id": "01900000-0000-7000-8000-000000000016",
+        "environment": "production",
         "workflow_version": 7,
         "correlation_id": "01900000-0000-7000-8000-000000000013",
         "node_id": "write_orders",
@@ -60,6 +62,19 @@ def test_postgres_ledger_claim_is_atomic_and_persists_only_bounded_fields():
     assert "encrypted_payload" not in sql
     assert "password" not in sql
     assert parameters["idempotency_key"] == "orders:2026-07-19"
+    assert parameters["deployment_id"] == binding["deployment_id"]
+    assert parameters["environment"] == "production"
+
+    ledger.finish(
+        "01900000-0000-7000-8000-000000000015",
+        status="success",
+        replay_http_status=200,
+        replay_body={"result": {"rows_out": 3}},
+    )
+    finish_sql, finish_parameters = engine.connection.calls[1]
+    assert "replay_digest" in finish_sql
+    assert finish_parameters["replay_http_status"] == 200
+    assert len(finish_parameters["replay_digest"]) == 64
 
 
 def test_v52_migration_has_single_use_and_commit_outcome_constraints():

tests/runner/test_rule_evidence.py

@@ -58,6 +58,9 @@ class Evidence:
     def replay(self, _rule_run_id):
         return self._replay
 
+    def heartbeat(self, _rule_run_id, _lease_owner):
+        return None
+
     def finish(self, rule_run_id, result):
         self.finished.append((rule_run_id, result))
 
@@ -79,8 +82,11 @@ def _context():
     return {
         "correlation_id": new_governance_uid(),
         "dataflow_uid": new_governance_uid(),
+        "deployment_id": new_governance_uid(),
+        "environment": "test",
         "workflow_version": 3,
         "node_id": "customer_mobile",
+        "task_jti": new_governance_uid(),
     }
 
 
@@ -228,4 +234,55 @@ def test_evidence_migration_is_forward_only_and_bounded():
     assert "UNIQUE (evidence_key)" in source
     assert "sample_count <= 100" in source
     assert "rule_violation_samples_rule_run_key" in source
+    assert "duplicate rule_run_id" in source
     assert "forward-only" in source
+
+
+def test_public_result_rejects_nested_records_and_unclosed_fields():
+    from app.runner.rule_evidence import PostgresRuleEvidenceWriter
+
+    base = {
+        "status": "success",
+        "rows_in": 1,
+        "rows_out": 1,
+        "rows_rejected": 0,
+        "rows_quarantined": 0,
+        "commit_outcome": "committed",
+        "timings": {"duration_ms": 1},
+    }
+    for public_result in (
+        {"rows": [{"secret": "value"}]},
+        {"records": [{"secret": "value"}]},
+        {"arbitrary": {"nested": "value"}},
+        {"violations": [{"step_id": "x", "count": 1, "raw": "secret"}]},
+    ):
+        with pytest.raises(ValueError, match="evidence safe"):
+            PostgresRuleEvidenceWriter._validate_finish(
+                {**base, "public_result": public_result}
+            )
+
+
+def test_public_result_contract_is_backend_specific():
+    from app.runner.rule_evidence import validate_public_rule_result
+
+    with pytest.raises(ValueError, match="evidence safe"):
+        validate_public_rule_result(
+            {
+                "rows_in": 1,
+                "rows_out": 1,
+                "artifact_ref": "minio://unexpected",
+            },
+            backend="sql_pushdown",
+        )
+    assert validate_public_rule_result(
+        {
+            "rows_in": 1,
+            "rows_out": 1,
+            "commit_outcome": "committed",
+            "output_artifact": (
+                "dataops-staging://"
+                "01900000-0000-7000-8000-000000000099"
+            ),
+        },
+        backend="sql_pushdown",
+    )["rows_out"] == 1

tests/runner/test_task_tokens.py

@@ -30,6 +30,8 @@ def test_task_token_is_short_lived_and_bound_to_one_node():
     token = issuer.issue(
         task_uid="01900000-0000-7000-8000-000000000011",
         dataflow_uid="01900000-0000-7000-8000-000000000012",
+        deployment_id="01900000-0000-7000-8000-000000000014",
+        environment="test",
         workflow_version=7,
         correlation_id="01900000-0000-7000-8000-000000000013",
         node=NODE,
@@ -37,6 +39,8 @@ def test_task_token_is_short_lived_and_bound_to_one_node():
     claims = verifier.verify(token, node=NODE)
 
     assert claims.node_id == "read_orders"
+    assert claims.deployment_id == "01900000-0000-7000-8000-000000000014"
+    assert claims.environment == "test"
     assert claims.node_type == "sql.query"
     assert claims.purpose == "read"
     assert claims.node_digest == node_digest(NODE)
@@ -52,6 +56,8 @@ def test_task_token_rejects_expiry_tampering_and_other_node():
     token = issuer.issue(
         task_uid="01900000-0000-7000-8000-000000000011",
         dataflow_uid="01900000-0000-7000-8000-000000000012",
+        deployment_id="01900000-0000-7000-8000-000000000014",
+        environment="test",
         workflow_version=7,
         correlation_id="01900000-0000-7000-8000-000000000013",
         node=NODE,

tests/test_data_rule_schema.py

@@ -36,6 +36,18 @@ ARTIFACT_HANDOFF_MIGRATION = (
     / "versions"
     / "20260723_150_rule_artifact_handoff_state.py"
 )
+RULE_EVIDENCE_MIGRATION = (
+    ROOT
+    / "migrations"
+    / "versions"
+    / "20260723_160_rule_execution_evidence.py"
+)
+RULE_ATTEMPT_MIGRATION = (
+    ROOT
+    / "migrations"
+    / "versions"
+    / "20260723_170_rule_execution_attempts.py"
+)
 
 EXPECTED_TABLES = {
     "data_rules",
@@ -192,3 +204,38 @@ def test_artifact_handoff_state_migration_upgrades_old_140_forward_only():
     spec.loader.exec_module(module)
     with pytest.raises(RuntimeError, match="forward-only|cannot downgrade"):
         module.downgrade()
+
+
+def test_rule_evidence_upgrade_preflights_legacy_duplicates_and_oversize():
+    source = RULE_EVIDENCE_MIGRATION.read_text(encoding="utf-8")
+
+    duplicate_check = source.index("HAVING COUNT(*) > 1")
+    unique_constraint = source.index(
+        "ADD CONSTRAINT rule_violation_samples_rule_run_key"
+    )
+    assert duplicate_check < unique_constraint
+    assert "duplicate rule_run_id" in source
+    assert "sample_count > 100" in source
+    assert "rows above 100" in source
+
+
+def test_rule_attempt_migration_adds_exact_identity_replay_lease_and_receipts():
+    source = RULE_ATTEMPT_MIGRATION.read_text(encoding="utf-8")
+
+    assert 'revision = "20260723_170"' in source
+    assert 'down_revision = "20260723_160"' in source
+    for expected in (
+        "lease_owner UUID",
+        "lease_expires_at TIMESTAMPTZ",
+        "evidence_digest CHAR(64)",
+        "deployment_id UUID",
+        "environment VARCHAR(20)",
+        "replay_body JSONB",
+        "replay_digest CHAR(64)",
+        "CREATE TABLE public.rule_sql_staging_receipts",
+        "producer_rule_run_id UUID NOT NULL",
+        "relation_digest CHAR(64) NOT NULL",
+        "UNIQUE (producer_rule_run_id, output_binding_id)",
+    ):
+        assert expected in source
+    assert "DROP TABLE" not in source.split("def downgrade()", 1)[1].upper()

tests/test_database_migrations.py

@@ -10,7 +10,6 @@ import pytest
 from sqlalchemy import create_engine, inspect
 from sqlalchemy.engine import make_url
 
-
 ROOT = Path(__file__).resolve().parents[1]
 EXPECTED_BASELINE_TABLES = {
     "data_orders",
@@ -35,6 +34,7 @@ EXPECTED_UPGRADED_TABLES = {
     "workflow_dual_runs",
     "workflow_reconciliation_reports",
     "workflow_cutover_operations",
+    "rule_sql_staging_receipts",
 }
 
 
@@ -89,7 +89,7 @@ def test_alembic_upgrade_is_repeatable_and_downgrade_preserves_tables():
         try:
             tables = set(inspect(engine).get_table_names(schema="public"))
             assert EXPECTED_BASELINE_TABLES | {"alembic_version"} <= tables
-            assert EXPECTED_UPGRADED_TABLES <= tables
+            assert tables >= EXPECTED_UPGRADED_TABLES
         finally:
             engine.dispose()
 
@@ -98,7 +98,7 @@ def test_alembic_upgrade_is_repeatable_and_downgrade_preserves_tables():
         engine = create_engine(target_url)
         try:
             tables = set(inspect(engine).get_table_names(schema="public"))
-            assert EXPECTED_BASELINE_TABLES <= tables
+            assert tables >= EXPECTED_BASELINE_TABLES
         finally:
             engine.dispose()
     finally:
@@ -110,3 +110,113 @@ def test_alembic_upgrade_is_repeatable_and_downgrade_preserves_tables():
             )
             cursor.execute(f'DROP DATABASE IF EXISTS "{database_name}"')
         connection.close()
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    ("legacy_rows", "diagnostic"),
+    (
+        ((1, 1), "duplicate rule_run_id"),
+        ((101,), "rows above 100"),
+    ),
+)
+def test_upgrade_from_150_rejects_malformed_legacy_samples(
+    legacy_rows,
+    diagnostic,
+):
+    admin_url = os.environ.get("TEST_POSTGRES_ADMIN_URL")
+    if not admin_url:
+        pytest.skip("TEST_POSTGRES_ADMIN_URL is not configured")
+
+    parsed = make_url(admin_url)
+    database_name = f"dataops_evidence_{uuid.uuid4().hex[:12]}"
+    target_url = parsed.set(database=database_name).render_as_string(
+        hide_password=False
+    )
+    admin = psycopg2.connect(admin_url)
+    admin.autocommit = True
+    try:
+        with admin.cursor() as cursor:
+            cursor.execute(f'CREATE DATABASE "{database_name}"')
+        env = os.environ.copy()
+        env["SQLALCHEMY_DATABASE_URI"] = target_url
+        command = [
+            str(ROOT / ".venv" / "bin" / "alembic"),
+            "-c",
+            "alembic.ini",
+        ]
+        subprocess.run(
+            command + ["upgrade", "20260723_150"],
+            cwd=ROOT,
+            env=env,
+            check=True,
+        )
+        database = psycopg2.connect(target_url)
+        try:
+            database.autocommit = True
+            with database.cursor() as cursor:
+                cursor.execute("SET session_replication_role = replica")
+                run_id = str(uuid.uuid4())
+                cursor.execute(
+                    """
+                    INSERT INTO public.rule_runs (
+                        id, deployment_id, component_binding_id,
+                        rule_version_id, plan_hash, status, correlation_id
+                    ) VALUES (%s, %s, %s, %s, %s, 'failed', %s)
+                    """,
+                    (
+                        run_id,
+                        str(uuid.uuid4()),
+                        str(uuid.uuid4()),
+                        str(uuid.uuid4()),
+                        "a" * 64,
+                        str(uuid.uuid4()),
+                    ),
+                )
+                for count in legacy_rows:
+                    cursor.execute(
+                        """
+                        INSERT INTO public.rule_violation_samples (
+                            id, rule_run_id, artifact_ref, sample_count,
+                            redaction_policy, expires_at
+                        ) VALUES (%s, %s, %s, %s, 'legacy-v1', NOW())
+                        """,
+                        (
+                            str(uuid.uuid4()),
+                            run_id,
+                            f"minio://legacy/{uuid.uuid4()}",
+                            count,
+                        ),
+                    )
+                cursor.execute("SET session_replication_role = origin")
+        finally:
+            database.close()
+
+        failed = subprocess.run(
+            command + ["upgrade", "head"],
+            cwd=ROOT,
+            env=env,
+            text=True,
+            capture_output=True,
+        )
+        assert failed.returncode != 0
+        assert diagnostic in (failed.stdout + failed.stderr)
+
+        database = psycopg2.connect(target_url)
+        try:
+            with database.cursor() as cursor:
+                cursor.execute(
+                    "SELECT version_num FROM public.alembic_version"
+                )
+                assert cursor.fetchone()[0] == "20260723_150"
+        finally:
+            database.close()
+    finally:
+        with admin.cursor() as cursor:
+            cursor.execute(
+                "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
+                "WHERE datname = %s AND pid <> pg_backend_pid()",
+                (database_name,),
+            )
+            cursor.execute(f'DROP DATABASE IF EXISTS "{database_name}"')
+        admin.close()