feat: persist rule evidence and artifact handoff

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

@@ -0,0 +1,126 @@
+# Task 6 Report — Artifact handoff and governed rule evidence
+
+Date: 2026-07-23
+Branch: `codex/data-rule-execution-m3a-m5`
+
+Lifecycle boundary: Task 6 consumes only plans already marked `published`.
+It does not add, relax, or emulate the Task 7 publication gate.
+
+## Outcome
+
+Task 6 now provides a production-wired path from a signed Runner task to one
+server-attested `rule_runs` record, optional durable violation evidence, and a
+closed artifact-only handoff to the next node.
+
+The implementation includes:
+
+- `PostgresRuleEvidenceWriter` with canonical deployment, component, rule
+  version, plan hash, DataFlow, workflow version, node, and correlation
+  attestation;
+- a server-derived evidence key that makes a node execution idempotent across
+  different signed task tokens and Runner retries;
+- reliable `running` to `success`, `failed`, `unknown`, or `cancelled`
+  finalization with row counts, timings, commit outcome, and bounded public
+  result;
+- 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;
+- a `pending` to verified MinIO upload to `ready` sample state machine with
+  digest, schema hash, reference, expiry, and bounded TTL cleanup;
+- 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;
+- 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.
+
+`rule_runs` gains:
+
+- nullable legacy-compatible `evidence_key` with a unique constraint;
+- `commit_outcome`;
+- bounded `public_result`;
+- `failure_code` and `updated_at`.
+
+`rule_violation_samples` gains:
+
+- artifact and schema digests;
+- `legacy`, `pending`, `ready`, and `failed` handoff states;
+- failure and update fields;
+- one sample per run;
+- a strict 100-row ceiling and retention index.
+
+Downgrade is deliberately rejected because deleting execution evidence would
+break audit and replay guarantees.
+
+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.
+
+## RED / GREEN evidence
+
+The initial fail-first slice produced five expected failures:
+
+- four executor evidence tests failed because `RulePlanExecutor` did not accept
+  an evidence writer;
+- the Kestra artifact-only test failed because downstream parameters contained
+  only workflow inputs and no upstream output reference.
+
+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:
+
+- focused evidence, Runner, Polars, Kestra, migration, and real integration:
+  `27 passed`;
+- full suite: `586 passed, 26 skipped, 59 subtests passed`;
+- Ruff across every changed Python file: `All checks passed!`;
+- `git diff --check`: passed;
+- Docker Compose configuration validation: passed;
+- historical migration 140/150 diff check: empty.
+
+## Real PostgreSQL and MinIO acceptance
+
+The production-path integration uses:
+
+- source PostgreSQL and source MySQL;
+- platform PostgreSQL at migration head;
+- real MinIO;
+- the isolated Polars worker;
+- `PostgresRulePlanRepository`;
+- `PostgresArtifactResolver`;
+- `PostgresRuleEvidenceWriter`;
+- `RulePlanExecutor`;
+- signed Runner HTTP tokens and the durable task ledger.
+
+It proves:
+
+1. the first node reads the PostgreSQL and MySQL artifacts, executes
+   normalize/join/assert/deduplicate, publishes an output, and records one
+   successful run;
+2. one violating row is stored as an all-field-redacted, expiring Parquet
+   sample with ready state and exact digest;
+3. a second published rule node consumes the first node's exact
+   `output_artifact` reference and publishes a distinct output;
+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
+   acceptance.
+
+## Residual boundary
+
+Task 6 deliberately does not decide whether a compiled plan may become
+published. Task 7 must require trusted generation, test, and preflight evidence
+before changing plan lifecycle state. The evidence added here is the immutable
+input to that gate, not a bypass around it.

app/core/orchestration/compilers/kestra.py

@@ -2,15 +2,15 @@ from __future__ import annotations
 
 import hashlib
 import json
+from collections.abc import Mapping
 from dataclasses import dataclass
-from typing import Any, Mapping
+from typing import Any
 
 from app.core.orchestration.spec import (
     validate_schedule_plan,
     validate_workflow_spec,
 )
 
-
 RUNNER_TASK_URL = "http://dataops-runner:5600/v1/tasks/execute"
 ENVIRONMENTS = {"development", "test", "production"}
 INPUT_TYPES = {
@@ -160,14 +160,33 @@ def _compile_dag_tasks(
             if node_id.replace("_", "").isalnum()
             else f"{{{{ inputs.dataops_task_tokens['{node_id}'] }}}}"
         )
+        parameters = (
+            {}
+            if node["type"] in {"rule.apply", "quality.check"}
+            else {
+                key: f"{{{{ inputs.{key} }}}}"
+                for key in sorted(spec["parameters"])
+            }
+        )
+        predecessors = sorted(dependencies[node_id])
+        if len(predecessors) == 1:
+            parameters["input_artifact"] = (
+                "{{ outputs.dataops_dag.tasks."
+                f"{predecessors[0]}.body.output_artifact }}}}"
+            )
+        elif len(predecessors) > 1:
+            parameters["input_artifacts"] = {
+                predecessor: (
+                    "{{ outputs.dataops_dag.tasks."
+                    f"{predecessor}.body.output_artifact }}}}"
+                )
+                for predecessor in predecessors
+            }
         runner_payload = {
             "dataflow_uid": spec["dataflow_uid"],
             "kestra_execution_id": "{{ execution.id }}",
             "node": node,
-            "parameters": {
-                key: f"{{{{ inputs.{key} }}}}"
-                for key in sorted(spec["parameters"])
-            },
+            "parameters": parameters,
             "task_token": token_expression,
             "workflow_version": version_no,
         }

app/runner/api.py

@@ -62,6 +62,9 @@ def create_runner_app(*, verifier, ledger, registry):
                 parameters,
                 write_authorized=claims.write_authorized,
                 correlation_id=claims.correlation_id,
+                dataflow_uid=claims.dataflow_uid,
+                workflow_version=claims.workflow_version,
+                node_id=claims.node_id,
             )
         except NodeExecutionError as exc:
             recorded = finish_safely(
@@ -97,12 +100,16 @@ def create_runner_app(*, verifier, ledger, registry):
             safe_detail="task completed",
         ):
             return jsonify({"error": "task ledger is unavailable"}), 503
-        return jsonify(
-            {
-                "task_uid": claims.task_uid,
-                "correlation_id": claims.correlation_id,
-                "result": result,
-            }
-        )
+        response = {
+            "task_uid": claims.task_uid,
+            "correlation_id": claims.correlation_id,
+            "result": result,
+        }
+        if (
+            isinstance(result, dict)
+            and isinstance(result.get("output_artifact"), str)
+        ):
+            response["output_artifact"] = result["output_artifact"]
+        return jsonify(response)
 
     return app

app/runner/artifacts.py

@@ -838,6 +838,62 @@ class PostgresArtifactResolver:
             "binding_hash": str(row["catalog_binding_hash"]),
         }
 
+    def resolve_handoff(
+        self,
+        *,
+        artifact_ref: str,
+        correlation_id: str,
+    ) -> dict[str, Any]:
+        """Resolve only a ready upstream output from the same execution."""
+
+        correlation = _uid(correlation_id, "artifact correlation id")
+        key = self.artifact_store._parse_ref(artifact_ref)
+        if not key.startswith(f"rules/{correlation}/"):
+            raise ValueError(
+                "artifact handoff does not match the execution correlation"
+            )
+        with self.engine.connect() as connection:
+            row = connection.execute(
+                text(
+                    """
+                    SELECT a.artifact_digest, a.row_count, a.schema_hash,
+                           a.schema_fields, a.expires_at
+                    FROM public.rule_run_artifacts a
+                    JOIN public.dataflow_dataset_bindings b
+                      ON b.id = a.binding_id
+                    WHERE a.artifact_ref = :artifact_ref
+                      AND a.correlation_id =
+                          CAST(:correlation_id AS uuid)
+                      AND a.artifact_kind = 'output'
+                      AND a.handoff_status = 'ready'
+                      AND a.expires_at > CURRENT_TIMESTAMP
+                      AND a.binding_hash = b.binding_hash
+                    """
+                ),
+                {
+                    "artifact_ref": artifact_ref,
+                    "correlation_id": correlation,
+                },
+            ).mappings().one_or_none()
+        if row is None:
+            raise ValueError("ready upstream artifact handoff was not found")
+        fields = row["schema_fields"]
+        if isinstance(fields, str):
+            fields = json.loads(fields)
+        described = self.artifact_store.describe(artifact_ref)
+        if (
+            described["digest"] != str(row["artifact_digest"])
+            or described["row_count"] != int(row["row_count"])
+            or described["schema_hash"] != str(row["schema_hash"])
+        ):
+            raise ValueError(
+                "upstream artifact handoff metadata does not match"
+            )
+        return {
+            **described,
+            "schema_fields": _normalized_schema_fields(fields),
+        }
+
     def attest_binding(
         self,
         *,

app/runner/bootstrap.py

@@ -23,6 +23,7 @@ from app.runner.nodes import (
     SqlExecuteExecutor,
     SqlQueryExecutor,
 )
+from app.runner.rule_evidence import PostgresRuleEvidenceWriter
 from app.runner.rule_polars import PolarsRulePlanAdapter
 from app.runner.rule_sql import (
     SqlGlotQualityPlanAdapter,
@@ -78,7 +79,9 @@ class RunnerSettings:
     max_query_rows: int = 1000
 
 
-def register_artifact_cleanup_cli(app, artifact_resolver):
+def register_artifact_cleanup_cli(
+    app, artifact_resolver, evidence_writer=None
+):
     """Register the bounded production maintenance entrypoint."""
 
     @app.cli.command("reconcile-rule-artifacts")
@@ -109,6 +112,8 @@ def register_artifact_cleanup_cli(app, artifact_resolver):
     @click.option("--limit", type=click.IntRange(1, 1_000), default=100)
     def cleanup_rule_artifacts(limit):
         removed = artifact_resolver.cleanup_expired(limit=limit)
+        if evidence_writer is not None:
+            removed += evidence_writer.cleanup_expired(limit=limit)
         click.echo(f"removed {removed} expired rule artifacts")
 
 
@@ -229,6 +234,11 @@ def build_runner_application(settings=None):
         },
         artifact_ttl_seconds=settings.artifact_ttl_seconds,
     )
+    evidence_writer = PostgresRuleEvidenceWriter(
+        runtime.platform_engine,
+        artifact_store,
+        sample_ttl_seconds=settings.artifact_ttl_seconds,
+    )
     rule_executor = RulePlanExecutor(
         PostgresRulePlanRepository(runtime.platform_engine),
         adapters={
@@ -236,6 +246,7 @@ def build_runner_application(settings=None):
             "polars_batch": polars_rule_adapter,
             "quality_check": SqlGlotQualityPlanAdapter(),
         },
+        evidence_writer=evidence_writer,
     )
     registry = NodeRegistry(
         {
@@ -254,7 +265,9 @@ def build_runner_application(settings=None):
         ledger=PostgresTaskLedger(runtime.platform_engine),
         registry=registry,
     )
-    register_artifact_cleanup_cli(application, artifact_resolver)
+    register_artifact_cleanup_cli(
+        application, artifact_resolver, evidence_writer
+    )
     application.extensions["dataops_runner_runtime"] = runtime
     application.extensions["dataops_runner_settings"] = settings
     return application

app/runner/nodes.py

@@ -376,6 +376,9 @@ class NodeRegistry:
         *,
         write_authorized=False,
         correlation_id=None,
+        dataflow_uid=None,
+        workflow_version=None,
+        node_id=None,
     ):
         executor = self.executors.get(node.get("type"))
         if executor is None or not callable(getattr(executor, "execute", None)):
@@ -385,4 +388,7 @@ class NodeRegistry:
             parameters,
             write_authorized=write_authorized,
             correlation_id=correlation_id,
+            dataflow_uid=dataflow_uid,
+            workflow_version=workflow_version,
+            node_id=node_id,
         )

app/runner/polars_worker.py

@@ -205,6 +205,7 @@ def _execute_polars_job(job: dict[str, Any]) -> dict[str, Any]:
     rows_in = _count(frame)
     expressions = _ExpressionCompiler(plan["timezone"])
     violations = []
+    violation_sample = []
     metrics = {
         "rows_rejected": 0,
         "rows_filtered": 0,
@@ -251,6 +252,22 @@ def _execute_polars_job(job: dict[str, Any]) -> dict[str, Any]:
             predicate = expressions.compile(
                 operation["expression_ast"]
             ).fill_null(False)
+            remaining_sample = 100 - len(violation_sample)
+            if remaining_sample > 0:
+                redacted = [
+                    pl.when(pl.col(name).is_null())
+                    .then(None)
+                    .otherwise(pl.lit("[REDACTED]"))
+                    .alias(name)
+                    for name in frame.collect_schema().names()
+                ]
+                violation_sample.extend(
+                    frame.filter(~predicate)
+                    .head(remaining_sample)
+                    .select(redacted)
+                    .collect(engine="streaming")
+                    .to_dicts()
+                )
             invalid = int(
                 frame.select((~predicate).sum().alias("count"))
                 .collect(engine="streaming")
@@ -415,6 +432,7 @@ def _execute_polars_job(job: dict[str, Any]) -> dict[str, Any]:
         **metrics,
         "violation_count": sum(item["count"] for item in violations),
         "violations": violations,
+        "_violation_sample": violation_sample,
     }
 
 

app/runner/rule_evidence.py

@@ -0,0 +1,692 @@
+"""Transactional, server-attested evidence for governed rule execution."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import tempfile
+from contextlib import suppress
+from typing import Any
+
+import polars as pl
+from sqlalchemy import text
+
+from app.core.common.identifiers import (
+    ensure_governance_uid,
+    new_governance_uid,
+)
+
+_DIGEST = re.compile(r"^[0-9a-f]{64}$")
+_FINAL_STATUSES = {"success", "failed", "unknown", "cancelled"}
+_COMMIT_OUTCOMES = {
+    "not_applicable",
+    "not_committed",
+    "committed",
+    "unknown",
+}
+_FINISH_KEYS = {
+    "status",
+    "rows_in",
+    "rows_out",
+    "rows_rejected",
+    "rows_quarantined",
+    "commit_outcome",
+    "timings",
+    "public_result",
+    "violation_sample",
+    "sample_count",
+    "redaction_policy",
+}
+
+
+def _uid(value: Any, label: str) -> str:
+    try:
+        return ensure_governance_uid({"uid": str(value)})
+    except ValueError as exc:
+        raise ValueError(f"{label} is invalid") from exc
+
+
+def _canonical_digest(value: Any) -> str:
+    encoded = json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _bounded_count(value: Any, label: str) -> int:
+    if isinstance(value, bool):
+        raise ValueError(f"{label} is invalid")
+    try:
+        normalized = int(value or 0)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"{label} is invalid") from exc
+    if normalized < 0 or normalized > 10_000_000_000:
+        raise ValueError(f"{label} is outside the evidence limit")
+    return normalized
+
+
+def _sample_fields(sample: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    names = sorted({str(key) for row in sample for key in row})
+    if not names:
+        raise ValueError("violation sample has no fields")
+    return [
+        {"name": name, "type": "string", "nullable": True}
+        for name in names
+    ]
+
+
+class PostgresRuleEvidenceWriter:
+    """Persist one immutable run and at most one expiring violation sample."""
+
+    def __init__(
+        self,
+        engine,
+        artifact_store,
+        *,
+        sample_ttl_seconds: int = 3600,
+    ):
+        self.engine = engine
+        self.artifact_store = artifact_store
+        self.sample_ttl_seconds = int(sample_ttl_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")
+
+    def start(
+        self,
+        *,
+        component_binding_id: str,
+        rule_version_id: str,
+        plan_hash: str,
+        correlation_id: str,
+        dataflow_uid: str,
+        workflow_version: int,
+        node_id: 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")
+        if _DIGEST.fullmatch(str(plan_hash or "")) is None:
+            raise ValueError("plan hash is invalid")
+        if (
+            isinstance(workflow_version, bool)
+            or not isinstance(workflow_version, int)
+            or workflow_version < 1
+        ):
+            raise ValueError("workflow version is invalid")
+        if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,99}", str(node_id)):
+            raise ValueError("node id is invalid")
+        evidence_key = _canonical_digest(
+            {
+                "component_binding_id": component,
+                "correlation_id": correlation,
+                "dataflow_uid": dataflow,
+                "node_id": node_id,
+                "plan_hash": plan_hash,
+                "rule_version_id": rule,
+                "workflow_version": workflow_version,
+            }
+        )
+        with self.engine.begin() as connection:
+            deployments = connection.execute(
+                text(
+                    """
+                    SELECT d.id::text AS deployment_id
+                    FROM public.dataflow_component_bindings b
+                    JOIN public.dataflow_versions v
+                      ON v.id = b.dataflow_version_id
+                    JOIN public.rule_execution_plans p
+                      ON p.component_binding_id = b.id
+                    JOIN public.dataflow_deployments d
+                      ON d.dataflow_version_id = v.id
+                    WHERE b.id = CAST(:component_binding_id AS uuid)
+                      AND b.rule_version_id =
+                          CAST(:rule_version_id AS uuid)
+                      AND p.plan_hash = :plan_hash
+                      AND p.status = 'published'
+                      AND v.dataflow_uid = CAST(:dataflow_uid AS uuid)
+                      AND v.version_no = :workflow_version
+                      AND d.status IN ('canary','active')
+                    ORDER BY
+                      CASE d.status WHEN 'active' THEN 0 ELSE 1 END,
+                      d.created_at DESC
+                    LIMIT 2
+                    """
+                ),
+                {
+                    "component_binding_id": component,
+                    "rule_version_id": rule,
+                    "plan_hash": plan_hash,
+                    "dataflow_uid": dataflow,
+                    "workflow_version": workflow_version,
+                },
+            ).mappings().all()
+            if len(deployments) != 1:
+                raise ValueError(
+                    "canonical rule deployment is missing or ambiguous"
+                )
+            rule_run_id = new_governance_uid()
+            selected = connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_runs (
+                        id, deployment_id, component_binding_id,
+                        rule_version_id, plan_hash, status,
+                        correlation_id, evidence_key, started_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
+                    )
+                    ON CONFLICT (evidence_key) DO NOTHING
+                    RETURNING id::text
+                    """
+                ),
+                {
+                    "id": rule_run_id,
+                    "deployment_id": deployments[0]["deployment_id"],
+                    "component_binding_id": component,
+                    "rule_version_id": rule,
+                    "plan_hash": plan_hash,
+                    "correlation_id": correlation,
+                    "evidence_key": evidence_key,
+                },
+            ).scalar_one_or_none()
+            if selected is None:
+                selected = connection.execute(
+                    text(
+                        """
+                        SELECT id::text
+                        FROM public.rule_runs
+                        WHERE evidence_key = :evidence_key
+                        """
+                    ),
+                    {"evidence_key": evidence_key},
+                ).scalar_one()
+        return str(selected)
+
+    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:
+            row = connection.execute(
+                text(
+                    """
+                    SELECT status, commit_outcome, public_result
+                    FROM public.rule_runs
+                    WHERE id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": run_id},
+            ).mappings().one_or_none()
+        if row is None:
+            raise ValueError("rule run was not found")
+        if row["status"] in {"queued", "running"}:
+            return None
+        result = row["public_result"]
+        if isinstance(result, str):
+            result = json.loads(result)
+        return {
+            **(dict(result) if isinstance(result, dict) else {}),
+            "status": str(row["status"]),
+            "commit_outcome": str(row["commit_outcome"]),
+        }
+
+    @staticmethod
+    def _validate_finish(result: Any) -> dict[str, Any]:
+        if not isinstance(result, dict) or set(result) - _FINISH_KEYS:
+            raise ValueError("rule evidence result has unsupported fields")
+        status = result.get("status")
+        commit_outcome = result.get("commit_outcome", "not_applicable")
+        if status not in _FINAL_STATUSES:
+            raise ValueError("rule evidence status is invalid")
+        if commit_outcome not in _COMMIT_OUTCOMES:
+            raise ValueError("rule evidence commit outcome is invalid")
+        timings = result.get("timings", {})
+        if (
+            not isinstance(timings, dict)
+            or set(timings) != {"duration_ms"}
+            or isinstance(timings.get("duration_ms"), bool)
+            or not isinstance(timings.get("duration_ms"), int)
+            or timings["duration_ms"] < 0
+            or timings["duration_ms"] > 86_400_000
+        ):
+            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")
+        sample = result.get("violation_sample")
+        if sample is not None:
+            if (
+                status != "success"
+                or not isinstance(sample, list)
+                or not 1 <= len(sample) <= 100
+                or result.get("sample_count") != len(sample)
+                or result.get("redaction_policy")
+                != "rule-violation-default-v1"
+            ):
+                raise ValueError("violation sample is invalid")
+            for row in sample:
+                if not isinstance(row, dict):
+                    raise ValueError("violation sample row is invalid")
+                for value in row.values():
+                    if value not in {None, "[REDACTED]"}:
+                        raise ValueError(
+                            "violation sample contains unredacted values"
+                        )
+        return {
+            **result,
+            "rows_in": _bounded_count(result.get("rows_in"), "rows_in"),
+            "rows_out": _bounded_count(result.get("rows_out"), "rows_out"),
+            "rows_rejected": _bounded_count(
+                result.get("rows_rejected"), "rows_rejected"
+            ),
+            "rows_quarantined": _bounded_count(
+                result.get("rows_quarantined"), "rows_quarantined"
+            ),
+            "commit_outcome": commit_outcome,
+        }
+
+    def _prepare_sample(
+        self,
+        run_id: str,
+        correlation_id: str,
+        sample: list[dict[str, Any]],
+        redaction_policy: str,
+    ) -> tuple[str, dict[str, Any], str]:
+        fields = _sample_fields(sample)
+        frame = pl.DataFrame(
+            {
+                field["name"]: [
+                    row.get(field["name"]) for row in sample
+                ]
+                for field in fields
+            },
+            schema={field["name"]: pl.String for field in fields},
+        )
+        with tempfile.NamedTemporaryFile(
+            prefix="dataops-rule-violation-",
+            suffix=".parquet",
+            delete=False,
+        ) as handle:
+            path = handle.name
+        sample_id = None
+        prepared = None
+        try:
+            frame.write_parquet(path)
+            prepared = self.artifact_store.prepare_path(
+                path,
+                correlation_id,
+                self.sample_ttl_seconds,
+                schema_fields=fields,
+                limits={
+                    "max_rows": min(100, self.artifact_store.max_rows),
+                    "max_artifact_bytes": min(
+                        4 * 1024 * 1024,
+                        self.artifact_store.max_artifact_bytes,
+                    ),
+                    "memory_limit_bytes": min(
+                        16 * 1024 * 1024,
+                        self.artifact_store.memory_limit_bytes,
+                    ),
+                },
+            )
+            sample_id = new_governance_uid()
+            with self.engine.begin() as connection:
+                row = connection.execute(
+                    text(
+                        """
+                        SELECT id::text, artifact_ref, artifact_digest,
+                               schema_hash, sample_count,
+                               redaction_policy, expires_at, handoff_status
+                        FROM public.rule_violation_samples
+                        WHERE rule_run_id = CAST(:rule_run_id AS uuid)
+                        FOR UPDATE
+                        """
+                    ),
+                    {"rule_run_id": run_id},
+                ).mappings().one_or_none()
+                if row is None:
+                    connection.execute(
+                        text(
+                            """
+                            INSERT INTO public.rule_violation_samples (
+                                id, rule_run_id, artifact_ref,
+                                artifact_digest, schema_hash, sample_count,
+                                redaction_policy, expires_at,
+                                handoff_status
+                            ) VALUES (
+                                CAST(:id AS uuid),
+                                CAST(:rule_run_id AS uuid), :artifact_ref,
+                                :artifact_digest, :schema_hash,
+                                :sample_count, :redaction_policy,
+                                CAST(:expires_at AS timestamptz), 'pending'
+                            )
+                            """
+                        ),
+                        {
+                            "id": sample_id,
+                            "rule_run_id": run_id,
+                            "artifact_ref": prepared["artifact_ref"],
+                            "artifact_digest": prepared["digest"],
+                            "schema_hash": prepared["schema_hash"],
+                            "sample_count": len(sample),
+                            "redaction_policy": redaction_policy,
+                            "expires_at": prepared["expires_at"],
+                        },
+                    )
+                else:
+                    if (
+                        str(row["artifact_digest"]) != prepared["digest"]
+                        or int(row["sample_count"]) != len(sample)
+                        or str(row["redaction_policy"])
+                        != redaction_policy
+                    ):
+                        raise ValueError(
+                            "violation sample evidence is immutable"
+                        )
+                    sample_id = str(row["id"])
+                    prepared.update(
+                        {
+                            "artifact_ref": str(row["artifact_ref"]),
+                            "digest": str(row["artifact_digest"]),
+                            "schema_hash": str(row["schema_hash"]),
+                            "expires_at": str(row["expires_at"]),
+                        }
+                    )
+                    if row["handoff_status"] == "ready":
+                        stored = self.artifact_store.describe(
+                            prepared["artifact_ref"]
+                        )
+                        if (
+                            stored["digest"] != prepared["digest"]
+                            or stored["schema_hash"]
+                            != prepared["schema_hash"]
+                            or stored["row_count"] != len(sample)
+                        ):
+                            raise ValueError(
+                                "ready violation sample does not match storage"
+                            )
+                        return sample_id, prepared, path
+                    if row["handoff_status"] == "failed":
+                        raise ValueError(
+                            "violation sample handoff already failed"
+                        )
+            self.artifact_store.upload_path(
+                path,
+                prepared,
+                limits={
+                    "max_rows": min(100, self.artifact_store.max_rows),
+                    "max_artifact_bytes": min(
+                        4 * 1024 * 1024,
+                        self.artifact_store.max_artifact_bytes,
+                    ),
+                    "memory_limit_bytes": min(
+                        16 * 1024 * 1024,
+                        self.artifact_store.memory_limit_bytes,
+                    ),
+                },
+            )
+            with self.engine.begin() as connection:
+                updated = connection.execute(
+                    text(
+                        """
+                        UPDATE public.rule_violation_samples
+                        SET handoff_status = 'ready',
+                            updated_at = CURRENT_TIMESTAMP
+                        WHERE id = CAST(:id AS uuid)
+                          AND handoff_status = 'pending'
+                          AND artifact_digest = :artifact_digest
+                        """
+                    ),
+                    {
+                        "id": sample_id,
+                        "artifact_digest": prepared["digest"],
+                    },
+                )
+                if updated.rowcount != 1:
+                    state = connection.execute(
+                        text(
+                            """
+                            SELECT handoff_status
+                            FROM public.rule_violation_samples
+                            WHERE id = CAST(:id AS uuid)
+                            """
+                        ),
+                        {"id": sample_id},
+                    ).scalar_one_or_none()
+                    if state != "ready":
+                        raise RuntimeError(
+                            "violation sample finalize outcome is unknown"
+                        )
+            return sample_id, prepared, path
+        except Exception:
+            if sample_id is not None and prepared is not None:
+                try:
+                    with self.engine.connect() as connection:
+                        committed = connection.execute(
+                            text(
+                                """
+                                SELECT handoff_status, artifact_digest
+                                FROM public.rule_violation_samples
+                                WHERE id = CAST(:id AS uuid)
+                                """
+                            ),
+                            {"id": sample_id},
+                        ).mappings().one_or_none()
+                    if (
+                        committed is not None
+                        and committed["handoff_status"] == "ready"
+                        and str(committed["artifact_digest"])
+                        == prepared["digest"]
+                    ):
+                        return sample_id, prepared, path
+                except Exception:
+                    pass
+            with self.engine.begin() as connection:
+                connection.execute(
+                    text(
+                        """
+                        UPDATE public.rule_violation_samples
+                        SET handoff_status = 'failed',
+                            failure_code = 'sample_handoff_failed',
+                            updated_at = CURRENT_TIMESTAMP
+                        WHERE rule_run_id = CAST(:rule_run_id AS uuid)
+                          AND handoff_status = 'pending'
+                        """
+                    ),
+                    {"rule_run_id": run_id},
+                )
+            with suppress(FileNotFoundError):
+                os.unlink(path)
+            raise
+
+    def finish(self, rule_run_id: str, result: Any) -> None:
+        run_id = _uid(rule_run_id, "rule run id")
+        normalized = self._validate_finish(result)
+        sample_path = None
+        try:
+            with self.engine.connect() as connection:
+                current = connection.execute(
+                    text(
+                        """
+                        SELECT status, correlation_id::text
+                        FROM public.rule_runs
+                        WHERE id = CAST(:id AS uuid)
+                        """
+                    ),
+                    {"id": run_id},
+                ).mappings().one_or_none()
+            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"]:
+                    return
+                raise ValueError("rule run evidence is immutable")
+            if normalized.get("violation_sample"):
+                _sample_id, _prepared, sample_path = self._prepare_sample(
+                    run_id,
+                    str(current["correlation_id"]),
+                    normalized["violation_sample"],
+                    normalized["redaction_policy"],
+                )
+            try:
+                with self.engine.begin() as connection:
+                    updated = connection.execute(
+                        text(
+                            """
+                            UPDATE public.rule_runs
+                            SET rows_in = :rows_in,
+                                rows_out = :rows_out,
+                                rows_rejected = :rows_rejected,
+                                rows_quarantined = :rows_quarantined,
+                                status = :status,
+                                timings = CAST(:timings AS jsonb),
+                                commit_outcome = :commit_outcome,
+                                public_result = CAST(:public_result AS jsonb),
+                                failure_code = :failure_code,
+                                finished_at = CURRENT_TIMESTAMP,
+                                updated_at = CURRENT_TIMESTAMP
+                            WHERE id = CAST(:id AS uuid)
+                              AND status IN ('queued','running')
+                            """
+                        ),
+                        {
+                            "id": run_id,
+                            "rows_in": normalized["rows_in"],
+                            "rows_out": normalized["rows_out"],
+                            "rows_rejected": normalized[
+                                "rows_rejected"
+                            ],
+                            "rows_quarantined": normalized[
+                                "rows_quarantined"
+                            ],
+                            "status": normalized["status"],
+                            "timings": json.dumps(
+                                normalized["timings"]
+                            ),
+                            "commit_outcome": normalized[
+                                "commit_outcome"
+                            ],
+                            "public_result": (
+                                json.dumps(
+                                    normalized.get("public_result")
+                                )
+                                if normalized.get("public_result")
+                                is not None
+                                else None
+                            ),
+                            "failure_code": (
+                                None
+                                if normalized["status"] == "success"
+                                else (
+                                    f"execution_{normalized['status']}"
+                                )
+                            ),
+                        },
+                    )
+                    if updated.rowcount != 1:
+                        state = connection.execute(
+                            text(
+                                """
+                                SELECT status, commit_outcome
+                                FROM public.rule_runs
+                                WHERE id = CAST(:id AS uuid)
+                                """
+                            ),
+                            {"id": run_id},
+                        ).mappings().one_or_none()
+                        if (
+                            state is None
+                            or state["status"] != normalized["status"]
+                            or state["commit_outcome"]
+                            != normalized["commit_outcome"]
+                        ):
+                            raise RuntimeError(
+                                "rule run finalize outcome is unknown"
+                            )
+            except Exception as exc:
+                try:
+                    replay = self.replay(run_id)
+                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"]
+                    != normalized["commit_outcome"]
+                ):
+                    raise RuntimeError(
+                        "rule run finalize outcome is unknown"
+                    ) from exc
+        finally:
+            if sample_path is not None:
+                with suppress(FileNotFoundError):
+                    os.unlink(sample_path)
+
+    def cleanup_expired(self, *, limit: int = 100) -> int:
+        if isinstance(limit, bool) or not isinstance(limit, int):
+            raise ValueError("cleanup limit is invalid")
+        if limit < 1 or limit > 1000:
+            raise ValueError("cleanup limit is invalid")
+        removed = 0
+        with self.engine.begin() as connection:
+            rows = connection.execute(
+                text(
+                    """
+                    SELECT id::text, artifact_ref
+                    FROM public.rule_violation_samples
+                    WHERE expires_at <= CURRENT_TIMESTAMP
+                      AND handoff_status IN ('legacy','ready','failed')
+                    ORDER BY expires_at, id
+                    FOR UPDATE SKIP LOCKED
+                    LIMIT :limit
+                    """
+                ),
+                {"limit": limit},
+            ).mappings().all()
+            for row in rows:
+                try:
+                    self.artifact_store.delete(str(row["artifact_ref"]))
+                except Exception:
+                    continue
+                connection.execute(
+                    text(
+                        """
+                        DELETE FROM public.rule_violation_samples
+                        WHERE id = CAST(:id AS uuid)
+                        """
+                    ),
+                    {"id": str(row["id"])},
+                )
+                removed += 1
+        return removed
+
+
+__all__ = ["PostgresRuleEvidenceWriter"]

app/runner/rule_polars.py

@@ -42,6 +42,7 @@ _RESULT_KEYS = (
     "rows_aggregated",
     "violation_count",
     "violations",
+    "_violation_sample",
 )
 
 
@@ -137,9 +138,17 @@ class PolarsRulePlanAdapter:
             raise NodeExecutionError(
                 "governed write authorization and idempotency are required"
             )
-        if parameters not in ({}, None):
+        if parameters in ({}, None):
+            input_handoff = None
+        elif (
+            isinstance(parameters, dict)
+            and set(parameters) == {"input_artifact"}
+            and isinstance(parameters["input_artifact"], str)
+        ):
+            input_handoff = parameters["input_artifact"]
+        else:
             raise NodeExecutionError(
-                "bound Polars rule plans do not accept runtime parameters"
+                "bound Polars rule plans accept only one artifact handoff"
             )
         correlation = _uid(correlation_id, "correlation_id")
         try:
@@ -152,13 +161,29 @@ class PolarsRulePlanAdapter:
             raise NodeExecutionError(
                 "published Polars output binding no longer matches"
             ) from exc
-        source = _resolve_artifact(
-            self.artifact_resolver,
-            binding_id=normalized["input_binding_id"],
-            binding_hash=normalized["input_binding_hash"],
-            correlation_id=correlation,
-            kind="input",
-        )
+        if input_handoff is None:
+            source = _resolve_artifact(
+                self.artifact_resolver,
+                binding_id=normalized["input_binding_id"],
+                binding_hash=normalized["input_binding_hash"],
+                correlation_id=correlation,
+                kind="input",
+            )
+        else:
+            try:
+                self.artifact_resolver.attest_binding(
+                    binding_id=normalized["input_binding_id"],
+                    binding_hash=normalized["input_binding_hash"],
+                    access_mode="read",
+                )
+                source = self.artifact_resolver.resolve_handoff(
+                    artifact_ref=input_handoff,
+                    correlation_id=correlation,
+                )
+            except Exception as exc:
+                raise NodeExecutionError(
+                    "upstream Polars artifact handoff was not resolved"
+                ) from exc
         limits = normalized["resource_limits"]
         output_path = None
         try:

app/runner/rules.py

@@ -5,6 +5,8 @@ from __future__ import annotations
 import hashlib
 import json
 import re
+import time
+from asyncio import CancelledError
 from collections.abc import Mapping
 from typing import Any
 
@@ -125,9 +127,88 @@ class PostgresRulePlanRepository:
 class RulePlanExecutor:
     """Fail closed unless the exact published plan is still executable."""
 
-    def __init__(self, repository, *, adapters: Mapping[str, Any]):
+    def __init__(
+        self,
+        repository,
+        *,
+        adapters: Mapping[str, Any],
+        evidence_writer=None,
+    ):
         self.repository = repository
         self.adapters = dict(adapters)
+        self.evidence_writer = evidence_writer
+
+    @staticmethod
+    def _redacted_sample(value: Any) -> list[dict[str, Any]]:
+        if not isinstance(value, list):
+            return []
+        sample = []
+        for row in value[:100]:
+            if not isinstance(row, dict):
+                continue
+            sample.append(
+                {
+                    str(key): (
+                        None if item is None else "[REDACTED]"
+                    )
+                    for key, item in sorted(row.items())
+                }
+            )
+        return sample
+
+    def _start_evidence(
+        self,
+        *,
+        component_binding_id: str,
+        rule_version_id: str,
+        plan_hash: str,
+        execution_context: Mapping[str, Any],
+    ):
+        if self.evidence_writer is None:
+            return None, None
+        correlation_id = execution_context.get("correlation_id")
+        dataflow_uid = execution_context.get("dataflow_uid")
+        workflow_version = execution_context.get("workflow_version")
+        node_id = execution_context.get("node_id")
+        if (
+            not correlation_id
+            or not dataflow_uid
+            or isinstance(workflow_version, bool)
+            or not isinstance(workflow_version, int)
+            or workflow_version < 1
+            or not isinstance(node_id, str)
+            or not node_id
+        ):
+            raise NodeExecutionError(
+                "trusted rule evidence context is incomplete"
+            )
+        try:
+            rule_run_id = self.evidence_writer.start(
+                component_binding_id=component_binding_id,
+                rule_version_id=rule_version_id,
+                plan_hash=plan_hash,
+                correlation_id=correlation_id,
+                dataflow_uid=dataflow_uid,
+                workflow_version=workflow_version,
+                node_id=node_id,
+            )
+            replay = self.evidence_writer.replay(rule_run_id)
+        except Exception as exc:
+            raise NodeExecutionError(
+                "rule execution evidence is unavailable"
+            ) from exc
+        return rule_run_id, replay
+
+    def _finish_evidence(self, rule_run_id, result):
+        if rule_run_id is None:
+            return
+        try:
+            self.evidence_writer.finish(rule_run_id, result)
+        except Exception as exc:
+            raise NodeExecutionError(
+                "rule execution evidence commit outcome is unknown",
+                commit_outcome="unknown",
+            ) from exc
 
     def execute(
         self,
@@ -270,23 +351,200 @@ class RulePlanExecutor:
         adapter = self.adapters.get(backend)
         if adapter is None or not callable(getattr(adapter, "execute", None)):
             raise NodeExecutionError("rule plan backend is not registered")
+        adapter_parameters = parameters
+        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
+            ):
+                raise NodeExecutionError(
+                    "SQL rule handoff is not a canonical staging binding"
+                )
+            adapter_parameters = {}
+        rule_run_id, replay = self._start_evidence(
+            component_binding_id=component_binding_id,
+            rule_version_id=rule_version_id,
+            plan_hash=plan_hash,
+            execution_context=execution_context,
+        )
+        if isinstance(replay, dict):
+            status = replay.get("status")
+            if status == "success":
+                return {
+                    key: value
+                    for key, value in replay.items()
+                    if key != "status"
+                }
+            if status in {"failed", "unknown", "cancelled"}:
+                raise NodeExecutionError(
+                    "rule execution was already finalized",
+                    commit_outcome=str(
+                        replay.get("commit_outcome")
+                        or "not_applicable"
+                    ),
+                )
+            raise NodeExecutionError("rule execution is already in progress")
         adapter_context = {}
         if backend == "polars_batch":
             adapter_context["correlation_id"] = execution_context.get(
                 "correlation_id"
             )
-        result = adapter.execute(
-            plan=record["plan"],
-            node=node,
-            parameters=parameters,
-            write_authorized=write_authorized,
-            **adapter_context,
-        )
+        started = time.monotonic()
+        try:
+            result = adapter.execute(
+                plan=record["plan"],
+                node=node,
+                parameters=adapter_parameters,
+                write_authorized=write_authorized,
+                **adapter_context,
+            )
+        except CancelledError:
+            self._finish_evidence(
+                rule_run_id,
+                {
+                    "status": "cancelled",
+                    "commit_outcome": "not_applicable",
+                    "timings": {
+                        "duration_ms": int(
+                            (time.monotonic() - started) * 1000
+                        )
+                    },
+                },
+            )
+            raise
+        except NodeExecutionError as exc:
+            self._finish_evidence(
+                rule_run_id,
+                {
+                    "status": (
+                        "unknown"
+                        if exc.commit_outcome == "unknown"
+                        else "failed"
+                    ),
+                    "commit_outcome": exc.commit_outcome,
+                    "timings": {
+                        "duration_ms": int(
+                            (time.monotonic() - started) * 1000
+                        )
+                    },
+                },
+            )
+            raise
+        except Exception as exc:
+            self._finish_evidence(
+                rule_run_id,
+                {
+                    "status": "failed",
+                    "commit_outcome": "not_committed",
+                    "timings": {
+                        "duration_ms": int(
+                            (time.monotonic() - started) * 1000
+                        )
+                    },
+                },
+            )
+            raise NodeExecutionError(
+                "rule plan execution failed",
+                commit_outcome="not_committed",
+            ) from exc
         if not isinstance(result, dict):
-            raise NodeExecutionError("rule plan result must be an object")
-        return {
-            **result,
-            "component_binding_id": component_binding_id,
-            "rule_version_id": rule_version_id,
-            "execution_plan_hash": plan_hash,
+            error = NodeExecutionError(
+                "rule plan result must be an object"
+            )
+            self._finish_evidence(
+                rule_run_id,
+                {
+                    "status": "failed",
+                    "commit_outcome": error.commit_outcome,
+                    "timings": {
+                        "duration_ms": int(
+                            (time.monotonic() - started) * 1000
+                        )
+                    },
+                },
+            )
+            raise error
+        try:
+            sample = self._redacted_sample(
+                result.pop("_violation_sample", None)
+            )
+            public_result = {
+                **result,
+                "component_binding_id": component_binding_id,
+                "rule_version_id": rule_version_id,
+                "execution_plan_hash": plan_hash,
+            }
+            for metric in (
+                "rows_in",
+                "rows_out",
+                "rows_rejected",
+                "rows_quarantined",
+            ):
+                value = public_result.get(metric, 0)
+                if isinstance(value, bool):
+                    raise ValueError("boolean metric")
+                public_result[metric] = max(0, int(value))
+            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']}"
+                )
+        except (TypeError, ValueError) as exc:
+            self._finish_evidence(
+                rule_run_id,
+                {
+                    "status": "failed",
+                    "commit_outcome": "not_committed",
+                    "timings": {
+                        "duration_ms": int(
+                            (time.monotonic() - started) * 1000
+                        )
+                    },
+                },
+            )
+            raise NodeExecutionError(
+                "rule plan result is invalid",
+                commit_outcome="not_committed",
+            ) from exc
+        evidence_result = {
+            "status": "success",
+            "rows_in": max(0, int(public_result.get("rows_in", 0))),
+            "rows_out": max(0, int(public_result.get("rows_out", 0))),
+            "rows_rejected": max(
+                0, int(public_result.get("rows_rejected", 0))
+            ),
+            "rows_quarantined": max(
+                0, int(public_result.get("rows_quarantined", 0))
+            ),
+            "commit_outcome": str(
+                public_result.get("commit_outcome", "not_applicable")
+            ),
+            "timings": {
+                "duration_ms": int(
+                    (time.monotonic() - started) * 1000
+                )
+            },
+            "public_result": public_result,
         }
+        if sample:
+            evidence_result.update(
+                {
+                    "violation_sample": sample,
+                    "sample_count": len(sample),
+                    "redaction_policy": (
+                        "rule-violation-default-v1"
+                    ),
+                }
+            )
+        self._finish_evidence(rule_run_id, evidence_result)
+        return public_result

migrations/versions/20260723_160_rule_execution_evidence.py

@@ -0,0 +1,72 @@
+"""Make governed rule execution evidence durable and idempotent."""
+
+from alembic import op
+
+revision = "20260723_160"
+down_revision = "20260723_150"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE public.rule_runs
+            ADD COLUMN evidence_key CHAR(64),
+            ADD COLUMN commit_outcome VARCHAR(30) NOT NULL
+                DEFAULT 'not_applicable'
+                CHECK (commit_outcome IN (
+                    'not_applicable','not_committed','committed','unknown'
+                )),
+            ADD COLUMN public_result JSONB,
+            ADD COLUMN failure_code VARCHAR(100),
+            ADD COLUMN updated_at TIMESTAMPTZ NOT NULL
+                DEFAULT CURRENT_TIMESTAMP;
+
+        ALTER TABLE public.rule_runs
+            ADD CONSTRAINT rule_runs_evidence_key UNIQUE (evidence_key);
+
+        ALTER TABLE public.rule_violation_samples
+            ADD COLUMN artifact_digest CHAR(64),
+            ADD COLUMN schema_hash CHAR(64),
+            ADD COLUMN handoff_status VARCHAR(20) NOT NULL
+                DEFAULT 'legacy'
+                CHECK (handoff_status IN (
+                    'legacy','pending','ready','failed'
+                )),
+            ADD COLUMN failure_code VARCHAR(100),
+            ADD COLUMN updated_at TIMESTAMPTZ NOT NULL
+                DEFAULT CURRENT_TIMESTAMP;
+
+        DO $$
+        BEGIN
+            IF EXISTS (
+                SELECT 1
+                FROM public.rule_violation_samples
+                WHERE sample_count > 100
+            ) THEN
+                RAISE EXCEPTION
+                    'violation sample migration found rows above 100';
+            END IF;
+        END
+        $$;
+
+        ALTER TABLE public.rule_violation_samples
+            DROP CONSTRAINT IF EXISTS
+                rule_violation_samples_sample_count_check,
+            ADD CONSTRAINT rule_violation_samples_sample_count_check
+                CHECK (sample_count >= 0 AND sample_count <= 100),
+            ADD CONSTRAINT rule_violation_samples_rule_run_key
+                UNIQUE (rule_run_id);
+
+        CREATE INDEX idx_rule_violation_samples_retention
+            ON public.rule_violation_samples
+            (handoff_status, expires_at, id);
+        """
+    )
+
+
+def downgrade() -> None:
+    raise RuntimeError(
+        "rule execution evidence is forward-only and cannot be downgraded"
+    )

tests/core/orchestration/test_kestra_compiler.py

@@ -1,4 +1,5 @@
 import copy
+import json
 import re
 
 import pytest
@@ -123,6 +124,60 @@ def test_compiler_emits_a_disabled_dag_that_calls_only_the_dataops_runner():
         assert forbidden not in source.lower()
 
 
+def test_downstream_nodes_receive_only_closed_artifact_handoffs():
+    spec = workflow_spec()
+    spec["nodes"].append(
+        {
+            "id": "publish_result",
+            "type": "notify",
+            "config": {"channel_uid": "catalog"},
+        }
+    )
+    spec["edges"].append(
+        {"from": "read_orders", "to": "publish_result"}
+    )
+
+    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: ")
+    ]
+    by_node = {body["node"]["id"]: body for body in body_lines}
+
+    single = by_node["notify_done"]["parameters"]
+    assert single["input_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.
+    spec["edges"].append(
+        {"from": "notify_done", "to": "publish_result"}
+    )
+    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 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():
     plan = schedule_plan()
     plan["triggers"] = [{"type": "manual"}]

tests/integration/test_data_rule_polars_execution.py

@@ -110,17 +110,24 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         max_ttl_seconds=3600,
     )
     correlation_id = new_governance_uid()
+    failure_correlation_id = new_governance_uid()
+    unknown_correlation_id = new_governance_uid()
     prefix = f"rules/{correlation_id}/"
     customer_table = "task5_polars_customers"
     segment_table = "task5_polars_segments"
     rule_uid = new_governance_uid()
     rule_id = new_governance_uid()
+    downstream_rule_uid = new_governance_uid()
+    downstream_rule_id = new_governance_uid()
     dataflow_uid = new_governance_uid()
     dataflow_version_id = new_governance_uid()
     deployment_id = new_governance_uid()
     component_binding_id = new_governance_uid()
     plan_id = new_governance_uid()
+    downstream_component_binding_id = new_governance_uid()
+    downstream_plan_id = new_governance_uid()
     ledger_jti = None
+    retry_ledger_jti = None
 
     try:
         with postgres.begin() as connection:
@@ -221,9 +228,15 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         output_binding = _binding(
             output_schema,
             source_uid=new_governance_uid(),
-            access_mode="write",
+            access_mode="read_write",
             object_ref="polars-output-artifact",
         )
+        downstream_output_binding = _binding(
+            output_schema,
+            source_uid=new_governance_uid(),
+            access_mode="write",
+            object_ref="polars-downstream-output-artifact",
+        )
         spec = validate_rule_spec(
             {
                 "schema_version": "2.0",
@@ -295,6 +308,45 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 },
             },
         )
+        downstream_spec = validate_rule_spec(
+            {
+                "schema_version": "2.0",
+                "rule_uid": new_governance_uid(),
+                "name": "task6_real_artifact_handoff",
+                "input_schema_ref": output_schema["schema_ref"],
+                "output_schema_ref": output_schema["schema_ref"],
+                "steps": [
+                    {
+                        "id": "normalize_downstream_name",
+                        "op": "normalize_text",
+                        "column": "name",
+                        "trim": True,
+                    }
+                ],
+                "null_policy": "explicit",
+                "timezone": "Asia/Shanghai",
+            }
+        )
+        downstream_rule = {
+            "id": downstream_rule_id,
+            "status": "published",
+            "rule_spec": downstream_spec,
+            "spec_hash": rule_spec_hash(downstream_spec),
+        }
+        downstream_compiled = PolarsRuleCompiler().compile(
+            rule_version=downstream_rule,
+            input_schema=output_schema,
+            output_schema=output_schema,
+            input_binding=output_binding,
+            output_binding=downstream_output_binding,
+            backend={
+                "max_rows": 1_000,
+                "max_artifact_bytes": 4 * 1024 * 1024,
+                "memory_limit_bytes": 256 * 1024 * 1024,
+                "masking_policies": {},
+                "lookup_bindings": {},
+            },
+        )
         lookup_operation = compiled["plan"]["operations"][1]
         schema_hashes = {
             "rule_spec_hash": compiled["plan"]["rule_spec_hash"],
@@ -331,6 +383,21 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "name": spec["name"],
                 },
             )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.data_rules
+                    (id, rule_uid, name, category, status)
+                    VALUES (CAST(:id AS uuid), CAST(:rule_uid AS uuid),
+                            :name, 'general', 'active')
+                    """
+                ),
+                {
+                    "id": new_governance_uid(),
+                    "rule_uid": downstream_rule_uid,
+                    "name": downstream_spec["name"],
+                },
+            )
             connection.execute(
                 text(
                     """
@@ -351,6 +418,31 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "spec_hash": rule["spec_hash"],
                 },
             )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.data_rule_versions
+                    (id, rule_uid, version_no, source_text,
+                     source_language, rule_spec, spec_hash,
+                     generated_kind, status, published_at)
+                    VALUES (
+                        CAST(:id AS uuid), CAST(:rule_uid AS uuid), 1,
+                        :source_text, 'en', CAST(:rule_spec AS jsonb),
+                        :spec_hash, 'polars', 'published',
+                        CURRENT_TIMESTAMP
+                    )
+                    """
+                ),
+                {
+                    "id": downstream_rule_id,
+                    "rule_uid": downstream_rule_uid,
+                    "source_text": (
+                        "Task 6 real two-node artifact handoff"
+                    ),
+                    "rule_spec": json.dumps(downstream_spec),
+                    "spec_hash": downstream_rule["spec_hash"],
+                },
+            )
             connection.execute(
                 text(
                     """
@@ -409,6 +501,13 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     output_binding,
                     compiled["plan"]["output_binding_hash"],
                 ),
+                (
+                    "downstream",
+                    downstream_output_binding,
+                    downstream_compiled["plan"][
+                        "output_binding_hash"
+                    ],
+                ),
             ):
                 connection.execute(
                     text(
@@ -461,6 +560,34 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     ),
                 },
             )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.dataflow_component_bindings
+                    (id, dataflow_version_id, component_id,
+                     component_kind, rule_version_id, stage, order_no,
+                     idempotency, provenance)
+                    VALUES (
+                        CAST(:id AS uuid),
+                        CAST(:dataflow_version_id AS uuid),
+                        'task6_real_handoff', 'rule.apply',
+                        CAST(:rule_version_id AS uuid), 'transform', 1,
+                        CAST(:idempotency AS jsonb), '{}'::jsonb
+                    )
+                    """
+                ),
+                {
+                    "id": downstream_component_binding_id,
+                    "dataflow_version_id": dataflow_version_id,
+                    "rule_version_id": downstream_rule_id,
+                    "idempotency": json.dumps(
+                        {
+                            "strategy": "deduplication_key",
+                            "key": "customer_id",
+                        }
+                    ),
+                },
+            )
             connection.execute(
                 text(
                     """
@@ -483,6 +610,53 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "schema_hashes": json.dumps(schema_hashes),
                 },
             )
+            connection.execute(
+                text(
+                    """
+                    INSERT INTO public.rule_execution_plans
+                    (id, component_binding_id, backend,
+                     compiler_version, plan, plan_hash,
+                     schema_hashes, status)
+                    VALUES (
+                        CAST(:id AS uuid),
+                        CAST(:component_binding_id AS uuid),
+                        'polars_batch', :compiler_version,
+                        CAST(:plan AS jsonb), :plan_hash,
+                        CAST(:schema_hashes AS jsonb), 'published'
+                    )
+                    """
+                ),
+                {
+                    "id": downstream_plan_id,
+                    "component_binding_id": (
+                        downstream_component_binding_id
+                    ),
+                    "compiler_version": downstream_compiled[
+                        "compiler_version"
+                    ],
+                    "plan": json.dumps(downstream_compiled["plan"]),
+                    "plan_hash": downstream_compiled["plan_hash"],
+                    "schema_hashes": json.dumps(
+                        {
+                            "rule_spec_hash": downstream_compiled[
+                                "plan"
+                            ]["rule_spec_hash"],
+                            "input_schema_snapshot_id": output_schema[
+                                "id"
+                            ],
+                            "input_schema_hash": output_schema[
+                                "schema_hash"
+                            ],
+                            "output_schema_snapshot_id": output_schema[
+                                "id"
+                            ],
+                            "output_schema_hash": output_schema[
+                                "schema_hash"
+                            ],
+                        }
+                    ),
+                },
+            )
         customer_path = tmp_path / "customers.parquet"
         segment_path = tmp_path / "segments.parquet"
         pl.DataFrame(customer_rows).write_parquet(customer_path)
@@ -594,6 +768,7 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         from app.runner.auth import TaskTokenIssuer, TaskTokenVerifier
         from app.runner.ledger import PostgresTaskLedger
         from app.runner.nodes import NodeRegistry
+        from app.runner.rule_evidence import PostgresRuleEvidenceWriter
 
         task_secret = "task5-real-http-secret-value-32-bytes"
         verifier = TaskTokenVerifier(task_secret)
@@ -606,10 +781,36 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             write_authorized=True,
         )
         ledger_jti = verifier.verify(task_token, node=node).jti
+        retry_token = TaskTokenIssuer(task_secret).issue(
+            task_uid=new_governance_uid(),
+            dataflow_uid=dataflow_uid,
+            workflow_version=1,
+            correlation_id=correlation_id,
+            node=node,
+            write_authorized=True,
+        )
+        retry_ledger_jti = verifier.verify(
+            retry_token, node=node
+        ).jti
+        evidenced_executor = RulePlanExecutor(
+            PostgresRulePlanRepository(platform),
+            adapters={
+                "polars_batch": PolarsRulePlanAdapter(
+                    artifact_store=store,
+                    artifact_resolver=resolver,
+                    artifact_ttl_seconds=900,
+                )
+            },
+            evidence_writer=PostgresRuleEvidenceWriter(
+                platform,
+                store,
+                sample_ttl_seconds=900,
+            ),
+        )
         runner_app = create_runner_app(
             verifier=verifier,
             ledger=PostgresTaskLedger(platform),
-            registry=NodeRegistry({"rule.apply": executor}),
+            registry=NodeRegistry({"rule.apply": evidenced_executor}),
         )
         with runner_app.test_client() as client:
             http_result = client.post(
@@ -628,15 +829,256 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "parameters": {},
                 },
             )
+            retried = client.post(
+                "/v1/tasks/execute",
+                json={
+                    "task_token": retry_token,
+                    "node": node,
+                    "parameters": {},
+                },
+            )
         assert http_result.status_code == 200
+        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 retried.status_code == 200
+        assert retried.get_json()["output_artifact"] == result[
+            "artifact_ref"
+        ]
         ledger_record = PostgresTaskLedger(platform).get(ledger_jti)
         assert ledger_record is not None
         assert ledger_record.status == "success"
         assert ledger_record.commit_outcome == "committed"
+        with platform.connect() as connection:
+            run_evidence = connection.execute(
+                text(
+                    """
+                    SELECT status, commit_outcome, rows_in, rows_out,
+                           rows_rejected, rows_quarantined, public_result
+                    FROM public.rule_runs
+                    WHERE correlation_id = CAST(:correlation_id AS uuid)
+                      AND component_binding_id =
+                          CAST(:component_binding_id AS uuid)
+                    """
+                ),
+                {
+                    "correlation_id": correlation_id,
+                    "component_binding_id": component_binding_id,
+                },
+            ).mappings().one()
+            sample_evidence = connection.execute(
+                text(
+                    """
+                    SELECT s.artifact_ref, s.artifact_digest,
+                           s.sample_count, s.redaction_policy,
+                           s.expires_at, s.handoff_status
+                    FROM public.rule_violation_samples s
+                    JOIN public.rule_runs r ON r.id = s.rule_run_id
+                    WHERE r.correlation_id =
+                        CAST(:correlation_id AS uuid)
+                    """
+                ),
+                {"correlation_id": correlation_id},
+            ).mappings().one()
+        assert run_evidence["status"] == "success"
+        assert run_evidence["commit_outcome"] == "committed"
+        assert run_evidence["rows_in"] == 4
+        assert run_evidence["rows_out"] == 2
+        assert run_evidence["rows_rejected"] == 1
+        assert run_evidence["rows_quarantined"] == 0
+        assert run_evidence["public_result"]["output_artifact"] == result[
+            "artifact_ref"
+        ]
+        assert sample_evidence["artifact_digest"]
+        assert sample_evidence["sample_count"] == 1
+        assert (
+            sample_evidence["redaction_policy"]
+            == "rule-violation-default-v1"
+        )
+        assert sample_evidence["handoff_status"] == "ready"
+        assert store.read(
+            sample_evidence["artifact_ref"],
+            sample_evidence["artifact_digest"],
+            expected_schema_fields=[
+                {
+                    "name": name,
+                    "type": "string",
+                    "nullable": True,
+                }
+                for name in (
+                    "customer_id",
+                    "mobile",
+                    "name",
+                    "segment_code",
+                    "segment_name",
+                    "version_no",
+                )
+            ],
+        ).collect().to_dicts() == [
+            {
+                "customer_id": "[REDACTED]",
+                "mobile": "[REDACTED]",
+                "name": "[REDACTED]",
+                "segment_code": "[REDACTED]",
+                "segment_name": "[REDACTED]",
+                "version_no": "[REDACTED]",
+            }
+        ]
+        downstream_node = {
+            "id": "task6_real_handoff",
+            "type": "rule.apply",
+            "purpose": "write",
+            "idempotency": {
+                "strategy": "deduplication_key",
+                "key": "customer_id",
+            },
+            "config": {
+                "component_binding_id": (
+                    downstream_component_binding_id
+                ),
+                "rule_version_id": downstream_rule_id,
+                "execution_plan_hash": downstream_compiled[
+                    "plan_hash"
+                ],
+            },
+        }
+        downstream_result = evidenced_executor.execute(
+            downstream_node,
+            {"input_artifact": result["artifact_ref"]},
+            write_authorized=True,
+            correlation_id=correlation_id,
+            dataflow_uid=dataflow_uid,
+            workflow_version=1,
+            node_id="task6_real_handoff",
+        )
+        assert downstream_result["rows_in"] == 2
+        assert downstream_result["rows_out"] == 2
+        assert downstream_result["output_artifact"] != result[
+            "artifact_ref"
+        ]
+        assert store.read(
+            downstream_result["artifact_ref"],
+            downstream_result["digest"],
+            expected_schema_fields=output_schema["fields"],
+            limits=downstream_compiled["plan"]["resource_limits"],
+        ).collect().sort("customer_id").to_dicts() == (
+            output.sort("customer_id").to_dicts()
+        )
+        replayed_downstream = evidenced_executor.execute(
+            downstream_node,
+            {"input_artifact": result["artifact_ref"]},
+            write_authorized=True,
+            correlation_id=correlation_id,
+            dataflow_uid=dataflow_uid,
+            workflow_version=1,
+            node_id="task6_real_handoff",
+        )
+        assert replayed_downstream["artifact_ref"] == downstream_result[
+            "artifact_ref"
+        ]
+        with platform.connect() as connection:
+            assert connection.execute(
+                text(
+                    """
+                    SELECT COUNT(*)
+                    FROM public.rule_runs
+                    WHERE correlation_id =
+                        CAST(:correlation_id AS uuid)
+                    """
+                ),
+                {"correlation_id": correlation_id},
+            ).scalar_one() == 2
+        from app.runner.nodes import NodeExecutionError
+
+        class FailingAdapter:
+            def execute(self, **_kwargs):
+                raise NodeExecutionError(
+                    "safe downstream failure",
+                    commit_outcome="not_committed",
+                )
+
+        failed_executor = RulePlanExecutor(
+            PostgresRulePlanRepository(platform),
+            adapters={"polars_batch": FailingAdapter()},
+            evidence_writer=PostgresRuleEvidenceWriter(
+                platform,
+                store,
+                sample_ttl_seconds=900,
+            ),
+        )
+        with pytest.raises(NodeExecutionError, match="safe downstream"):
+            failed_executor.execute(
+                node,
+                {},
+                write_authorized=True,
+                correlation_id=failure_correlation_id,
+                dataflow_uid=dataflow_uid,
+                workflow_version=1,
+                node_id="task5_real_polars",
+            )
+        with platform.connect() as connection:
+            failed_evidence = connection.execute(
+                text(
+                    """
+                    SELECT status, commit_outcome
+                    FROM public.rule_runs
+                    WHERE correlation_id =
+                        CAST(:correlation_id AS uuid)
+                    """
+                ),
+                {"correlation_id": failure_correlation_id},
+            ).mappings().one()
+        assert dict(failed_evidence) == {
+            "status": "failed",
+            "commit_outcome": "not_committed",
+        }
+
+        class UnknownAdapter:
+            def execute(self, **_kwargs):
+                raise NodeExecutionError(
+                    "safe uncertain commit",
+                    commit_outcome="unknown",
+                )
+
+        unknown_executor = RulePlanExecutor(
+            PostgresRulePlanRepository(platform),
+            adapters={"polars_batch": UnknownAdapter()},
+            evidence_writer=PostgresRuleEvidenceWriter(
+                platform,
+                store,
+                sample_ttl_seconds=900,
+            ),
+        )
+        with pytest.raises(NodeExecutionError, match="uncertain commit"):
+            unknown_executor.execute(
+                node,
+                {},
+                write_authorized=True,
+                correlation_id=unknown_correlation_id,
+                dataflow_uid=dataflow_uid,
+                workflow_version=1,
+                node_id="task5_real_polars",
+            )
+        with platform.connect() as connection:
+            unknown_evidence = connection.execute(
+                text(
+                    """
+                    SELECT status, commit_outcome
+                    FROM public.rule_runs
+                    WHERE correlation_id =
+                        CAST(:correlation_id AS uuid)
+                    """
+                ),
+                {"correlation_id": unknown_correlation_id},
+            ).mappings().one()
+        assert dict(unknown_evidence) == {
+            "status": "unknown",
+            "commit_outcome": "unknown",
+        }
         conflict_path = tmp_path / "conflict.parquet"
         pl.DataFrame(
             {
@@ -682,17 +1124,20 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             ).mappings().all()
         # The repeated deterministic output has the same digest and is
         # idempotently retained as one stable catalog handoff.
-        assert len(catalog_rows) == 3
+        assert len(catalog_rows) == 4
         assert all(row["handoff_status"] == "ready" for row in catalog_rows)
         assert all(len(row["binding_hash"]) == 64 for row in catalog_rows)
-        assert next(
+        assert {
             row["artifact_ref"]
             for row in catalog_rows
             if row["artifact_kind"] == "output"
-        ) == result["artifact_ref"]
+        } == {
+            result["artifact_ref"],
+            downstream_result["artifact_ref"],
+        }
         assert len(
             list(minio.list_objects(bucket, prefix=prefix, recursive=True))
-        ) == 3
+        ) == 5
     finally:
         for item in list(
             minio.list_objects(bucket, prefix=prefix, recursive=True)
@@ -714,6 +1159,40 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     ),
                     {"jti": ledger_jti},
                 )
+            if retry_ledger_jti is not None:
+                connection.execute(
+                    text(
+                        "DELETE FROM public.runner_task_executions "
+                        "WHERE token_jti = CAST(:jti AS uuid)"
+                    ),
+                    {"jti": retry_ledger_jti},
+                )
+            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)
+                    """
+                ),
+                {"correlation_id": correlation_id},
+            )
+            connection.execute(
+                text(
+                    "DELETE FROM public.rule_runs "
+                    "WHERE correlation_id IN ("
+                    "CAST(:correlation_id AS uuid), "
+                    "CAST(:failure_correlation_id AS uuid), "
+                    "CAST(:unknown_correlation_id AS uuid))"
+                ),
+                {
+                    "correlation_id": correlation_id,
+                    "failure_correlation_id": failure_correlation_id,
+                    "unknown_correlation_id": unknown_correlation_id,
+                },
+            )
             connection.execute(
                 text(
                     "DELETE FROM public.rule_run_artifacts "
@@ -724,9 +1203,13 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             connection.execute(
                 text(
                     "DELETE FROM public.rule_execution_plans "
-                    "WHERE id = CAST(:id AS uuid)"
+                    "WHERE id IN (CAST(:id AS uuid), "
+                    "CAST(:downstream_id AS uuid))"
                 ),
-                {"id": plan_id},
+                {
+                    "id": plan_id,
+                    "downstream_id": downstream_plan_id,
+                },
             )
             connection.execute(
                 text(
@@ -738,9 +1221,13 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             connection.execute(
                 text(
                     "DELETE FROM public.dataflow_component_bindings "
-                    "WHERE id = CAST(:id AS uuid)"
+                    "WHERE id IN (CAST(:id AS uuid), "
+                    "CAST(:downstream_id AS uuid))"
                 ),
-                {"id": component_binding_id},
+                {
+                    "id": component_binding_id,
+                    "downstream_id": downstream_component_binding_id,
+                },
             )
             connection.execute(
                 text(
@@ -759,16 +1246,24 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             connection.execute(
                 text(
                     "DELETE FROM public.data_rule_versions "
-                    "WHERE id = CAST(:id AS uuid)"
+                    "WHERE id IN (CAST(:id AS uuid), "
+                    "CAST(:downstream_id AS uuid))"
                 ),
-                {"id": rule_id},
+                {
+                    "id": rule_id,
+                    "downstream_id": downstream_rule_id,
+                },
             )
             connection.execute(
                 text(
                     "DELETE FROM public.data_rules "
-                    "WHERE rule_uid = CAST(:rule_uid AS uuid)"
+                    "WHERE rule_uid IN (CAST(:rule_uid AS uuid), "
+                    "CAST(:downstream_rule_uid AS uuid))"
                 ),
-                {"rule_uid": rule_uid},
+                {
+                    "rule_uid": rule_uid,
+                    "downstream_rule_uid": downstream_rule_uid,
+                },
             )
             for schema in (input_schema, lookup_schema, output_schema):
                 connection.execute(

tests/runner/test_api.py

@@ -3,7 +3,6 @@ from app.runner.auth import TaskTokenIssuer, TaskTokenVerifier
 from app.runner.ledger import InMemoryTaskLedger
 from app.runner.nodes import NodeRegistry
 
-
 NODE = {
     "id": "read_orders",
     "type": "sql.query",
@@ -16,10 +15,16 @@ NODE = {
 class Executor:
     def __init__(self):
         self.calls = 0
+        self.context = None
 
-    def execute(self, node, parameters, **_kwargs):
+    def execute(self, node, parameters, **kwargs):
         self.calls += 1
-        return {"node": node["id"], "parameters": parameters}
+        self.context = kwargs
+        return {
+            "node": node["id"],
+            "parameters": parameters,
+            "output_artifact": "minio://dataops-rules/rules/output.parquet",
+        }
 
 
 def test_runner_accepts_one_signed_task_and_rejects_replay():
@@ -47,6 +52,12 @@ def test_runner_accepts_one_signed_task_and_rejects_replay():
 
     assert first.status_code == 200
     assert first.get_json()["result"]["node"] == "read_orders"
+    assert first.get_json()["output_artifact"].startswith("minio://")
+    assert executor.context["dataflow_uid"] == (
+        "01900000-0000-7000-8000-000000000012"
+    )
+    assert executor.context["workflow_version"] == 7
+    assert executor.context["node_id"] == "read_orders"
     assert replay.status_code == 409
     assert replay.get_json() == {"error": "task token already consumed"}
     assert executor.calls == 1

tests/runner/test_rule_evidence.py

@@ -0,0 +1,231 @@
+from __future__ import annotations
+
+import hashlib
+import json
+
+import pytest
+
+from app.core.common.identifiers import new_governance_uid
+from app.runner.nodes import NodeExecutionError
+
+PLAN = {"op": "not_null", "column": "mobile"}
+PLAN_HASH = hashlib.sha256(
+    json.dumps(PLAN, sort_keys=True, separators=(",", ":")).encode()
+).hexdigest()
+
+
+def _node():
+    return {
+        "id": "customer_mobile",
+        "type": "quality.check",
+        "purpose": "read",
+        "config": {
+            "component_binding_id": new_governance_uid(),
+            "rule_version_id": new_governance_uid(),
+            "execution_plan_hash": PLAN_HASH,
+        },
+    }
+
+
+class Repository:
+    def __init__(self, node):
+        self.record = {
+            "component_binding_id": node["config"]["component_binding_id"],
+            "rule_version_id": node["config"]["rule_version_id"],
+            "backend": "quality_check",
+            "plan": PLAN,
+            "plan_hash": PLAN_HASH,
+            "plan_status": "published",
+            "rule_status": "published",
+            "component_kind": "quality.check",
+            "binding_idempotency": None,
+        }
+
+    def load(self, **_kwargs):
+        return dict(self.record)
+
+
+class Evidence:
+    def __init__(self, replay=None):
+        self.started = []
+        self.finished = []
+        self._replay = replay
+
+    def start(self, **values):
+        self.started.append(values)
+        return new_governance_uid()
+
+    def replay(self, _rule_run_id):
+        return self._replay
+
+    def finish(self, rule_run_id, result):
+        self.finished.append((rule_run_id, result))
+
+
+class Adapter:
+    def __init__(self, result=None, error=None):
+        self.result = result
+        self.error = error
+        self.calls = 0
+
+    def execute(self, **_kwargs):
+        self.calls += 1
+        if self.error:
+            raise self.error
+        return dict(self.result or {})
+
+
+def _context():
+    return {
+        "correlation_id": new_governance_uid(),
+        "dataflow_uid": new_governance_uid(),
+        "workflow_version": 3,
+        "node_id": "customer_mobile",
+    }
+
+
+def test_rule_executor_records_success_and_bounded_violation_sample():
+    from app.runner.rules import RulePlanExecutor
+
+    node = _node()
+    evidence = Evidence()
+    adapter = Adapter(
+        {
+            "rows_in": 3,
+            "rows_out": 2,
+            "rows_rejected": 1,
+            "rows_quarantined": 1,
+            "commit_outcome": "committed",
+            "_violation_sample": [
+                {"mobile": f"13800138{index:03d}", "name": f"Person {index}"}
+                for index in range(150)
+            ],
+        }
+    )
+    executor = RulePlanExecutor(
+        Repository(node),
+        adapters={"quality_check": adapter},
+        evidence_writer=evidence,
+    )
+
+    result = executor.execute(node, {}, **_context())
+
+    assert result["rows_in"] == 3
+    assert result["rows_out"] == 2
+    assert "_violation_sample" not in result
+    assert evidence.started[0]["plan_hash"] == PLAN_HASH
+    finished = evidence.finished[0][1]
+    assert finished["status"] == "success"
+    assert finished["sample_count"] == 100
+    assert finished["redaction_policy"] == "rule-violation-default-v1"
+    assert finished["violation_sample"][0] == {
+        "mobile": "[REDACTED]",
+        "name": "[REDACTED]",
+    }
+
+
+@pytest.mark.parametrize(
+    ("error", "status", "commit_outcome"),
+    [
+        (
+            NodeExecutionError(
+                "write failed",
+                commit_outcome="not_committed",
+            ),
+            "failed",
+            "not_committed",
+        ),
+        (
+            NodeExecutionError(
+                "commit uncertain",
+                commit_outcome="unknown",
+            ),
+            "unknown",
+            "unknown",
+        ),
+    ],
+)
+def test_rule_executor_finalizes_failure_and_unknown_commit(
+    error, status, commit_outcome
+):
+    from app.runner.rules import RulePlanExecutor
+
+    node = _node()
+    evidence = Evidence()
+    executor = RulePlanExecutor(
+        Repository(node),
+        adapters={"quality_check": Adapter(error=error)},
+        evidence_writer=evidence,
+    )
+
+    with pytest.raises(NodeExecutionError):
+        executor.execute(node, {}, **_context())
+
+    assert evidence.finished[0][1]["status"] == status
+    assert evidence.finished[0][1]["commit_outcome"] == commit_outcome
+    assert "violation_sample" not in evidence.finished[0][1]
+
+
+def test_rule_executor_replays_finished_evidence_without_adapter_execution():
+    from app.runner.rules import RulePlanExecutor
+
+    node = _node()
+    prior = {
+        "status": "success",
+        "rows_in": 3,
+        "rows_out": 2,
+        "rows_rejected": 1,
+        "rows_quarantined": 1,
+        "commit_outcome": "committed",
+        "output_artifact": "minio://dataops-rules/rules/prior.parquet",
+    }
+    evidence = Evidence(replay=prior)
+    adapter = Adapter()
+    executor = RulePlanExecutor(
+        Repository(node),
+        adapters={"quality_check": adapter},
+        evidence_writer=evidence,
+    )
+
+    result = executor.execute(node, {}, **_context())
+
+    assert result["output_artifact"] == prior["output_artifact"]
+    assert adapter.calls == 0
+    assert evidence.finished == []
+
+
+def test_rule_executor_finalizes_cancellation():
+    from asyncio import CancelledError
+
+    from app.runner.rules import RulePlanExecutor
+
+    node = _node()
+    evidence = Evidence()
+    executor = RulePlanExecutor(
+        Repository(node),
+        adapters={
+            "quality_check": Adapter(error=CancelledError())
+        },
+        evidence_writer=evidence,
+    )
+
+    with pytest.raises(CancelledError):
+        executor.execute(node, {}, **_context())
+
+    assert evidence.finished[0][1]["status"] == "cancelled"
+
+
+def test_evidence_migration_is_forward_only_and_bounded():
+    from pathlib import Path
+
+    source = Path(
+        "migrations/versions/"
+        "20260723_160_rule_execution_evidence.py"
+    ).read_text(encoding="utf-8")
+
+    assert 'revision = "20260723_160"' in source
+    assert 'down_revision = "20260723_150"' in source
+    assert "UNIQUE (evidence_key)" in source
+    assert "sample_count <= 100" in source
+    assert "rule_violation_samples_rule_run_key" in source
+    assert "forward-only" in source

tests/runner/test_rule_polars.py

@@ -39,6 +39,7 @@ class Resolver:
         self.attest_error = None
         self.publish_error = None
         self.registrations = []
+        self.handoffs = {}
 
     def resolve(self, *, binding_id, correlation_id, kind):
         self.calls.append((binding_id, correlation_id))
@@ -51,6 +52,10 @@ class Resolver:
             raise self.attest_error
         return {"binding_hash": binding_hash}
 
+    def resolve_handoff(self, *, artifact_ref, correlation_id):
+        self.events.append(("handoff", artifact_ref, correlation_id))
+        return self.handoffs[artifact_ref]
+
     def publish_path(
         self,
         path,
@@ -205,6 +210,8 @@ def test_polars_adapter_reconstructs_assert_and_deduplicate_and_writes_artifact(
     assert result["violations"] == [
         {"step_id": "mobile_format", "count": 1}
     ]
+    assert len(result["_violation_sample"]) == 1
+    assert result["_violation_sample"][0]["mobile"] == "[REDACTED]"
     assert result["commit_outcome"] == "committed"
     assert "schema_fields" not in result
     assert resolver.events[0] == (
@@ -279,6 +286,66 @@ def test_polars_adapter_reports_unknown_catalog_commit_outcome():
     assert error.value.commit_outcome == "unknown"
 
 
+def test_polars_adapter_consumes_only_attested_upstream_artifact_ref():
+    from app.runner.rule_polars import PolarsRulePlanAdapter
+
+    compiled, _input_binding = _compiled_plan(
+        [
+            {
+                "id": "trim_name",
+                "op": "normalize_text",
+                "column": "name",
+                "trim": True,
+            }
+        ]
+    )
+    store = _plan_store(FakeMinio())
+    correlation_id = new_governance_uid()
+    source = store.write(
+        pl.DataFrame(
+            {
+                "customer_id": [1],
+                "name": ["Alice"],
+                "mobile": ["13800138000"],
+            }
+        ),
+        correlation_id,
+        600,
+        schema_fields=compiled["plan"]["input_fields"],
+        limits=compiled["plan"]["resource_limits"],
+    )
+    resolver = Resolver({}, store)
+    resolver.handoffs[source["artifact_ref"]] = source
+    adapter = PolarsRulePlanAdapter(
+        artifact_store=store,
+        artifact_resolver=resolver,
+        artifact_ttl_seconds=300,
+    )
+
+    result = adapter.execute(
+        plan=compiled["plan"],
+        node=_node(compiled),
+        parameters={"input_artifact": source["artifact_ref"]},
+        write_authorized=True,
+        correlation_id=correlation_id,
+    )
+
+    assert result["rows_out"] == 1
+    assert (
+        "handoff",
+        source["artifact_ref"],
+        correlation_id,
+    ) in resolver.events
+    with pytest.raises(NodeExecutionError, match="only one artifact"):
+        adapter.execute(
+            plan=compiled["plan"],
+            node=_node(compiled),
+            parameters={"rows": [{"customer_id": 1}]},
+            write_authorized=True,
+            correlation_id=correlation_id,
+        )
+
+
 def test_polars_adapter_fails_closed_for_plan_hash_binding_and_authorization():
     from app.runner.rule_polars import PolarsRulePlanAdapter