fix: reconcile governed rule execution crashes

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

@@ -46,8 +46,10 @@ The implementation includes:
 
 ## Forward-only schema change
 
-Migrations `20260723_160` and `20260723_170` extend the existing evidence
-tables without changing historical migrations 110, 140, or 150.
+Migrations `20260723_160`, `20260723_170`, and `20260723_180` extend the
+existing evidence tables without changing historical migrations 110, 140, or
+150. The second review did not rewrite already-applied migrations 160 or 170;
+all additional schema is isolated in 180.
 
 `rule_runs` gains:
 
@@ -70,7 +72,9 @@ break audit and replay guarantees.
 Migration 160 explicitly preflights duplicate legacy `rule_run_id` samples and
 legacy samples over 100 rows before adding constraints. Migration 170 adds
 leases, exact task deployment identity, bounded replay fields, evidence
-digests, and SQL staging receipts.
+digests, and SQL staging receipts. Migration 180 adds the Runner ledger lease,
+bounded cleanup claims, retained sample schema fields, and the
+`unknown`/`expired` reconciliation states.
 
 The local Docker PostgreSQL was upgraded through the formal
 `20260723_150 -> head` Alembic path. Two isolated real PostgreSQL tests prove
@@ -89,7 +93,7 @@ The initial fail-first slice produced five expected failures:
 A second fail-first assertion proved that the Polars adapter did not expose a
 bounded violation sample before the worker implementation was added.
 
-Final review verification:
+First review verification:
 
 - focused Runner/rule/Kestra/schema suite: `129 passed`;
 - real PostgreSQL/MySQL/MinIO evidence acceptance: `1 passed`;
@@ -101,6 +105,59 @@ Final review verification:
 - Docker Compose configuration validation: passed;
 - historical migration 140/150 diff check: empty.
 
+## Second-review reconciliation hardening
+
+The second review closes the remaining crash and identity gaps:
+
+- the Runner task ledger has a bounded lease. A crash after claiming a token
+  but before creating `rule_runs` atomically converges to `unknown`;
+- an existing `rule_runs` row with an expired evidence lease also converges to
+  `unknown`, finalizes the exact ledger JTI, and cannot stay at HTTP 202
+  permanently;
+- expired task tokens are replay-only for a bounded window. Only the exact
+  stored JTI, node digest, deployment/environment/workflow binding, and
+  response digest may return a prior success. Missing, running, out-of-window,
+  or tampered requests are rejected without execution;
+- the scheduling gateway now resolves and persists the canonical
+  `dataflow_deployments.id`. A workflow candidate ID is never used as a rule
+  deployment ID;
+- the real gateway path binds the released governance DataFlow version to the
+  exact workflow version, schedule, and environment before issuing canary
+  tokens;
+- rule adapters run with a periodic lease heartbeat. Heartbeat failure records
+  `unknown` and preserves a known committed adapter outcome in evidence;
+- violation-sample reconciliation claims at most a bounded batch, performs
+  MinIO work outside database transactions, verifies digest/schema/count, and
+  safely releases retryable unknown claims;
+- expired sample deletion is a claim, object deletion, and conditional
+  database deletion sequence, so no object network call occurs while holding
+  a database transaction;
+- SQL staging receipts have equivalent bounded expiry claims and immutable
+  audit retention;
+- all Kestra predecessor references, including non-rule fan-in, use
+  bracket-safe expressions.
+
+Second-review verification:
+
+- targeted rule, Runner, scheduling, Kestra, and schema suite:
+  `52 passed`;
+- task-token and Runner replay boundary suite: `12 passed`;
+- real PostgreSQL/MySQL/MinIO/Gateway/Runner/evidence acceptance:
+  `1 passed`;
+- full suite: `601 passed, 28 skipped, 59 subtests passed`;
+- Ruff across every file changed by the second review:
+  `All checks passed!`;
+- `git diff --check`: passed;
+- migrations 160 and 170 diff against the first-review commit: empty;
+- real local PostgreSQL upgrade: `20260723_170 -> 20260723_180 (head)`;
+- rebuilt local Docker backend and Runner: both healthy;
+- Runner `/health`: `{"status":"ok"}`;
+- Alembic inside the rebuilt backend container:
+  `20260723_180 (head)`.
+
+The full-repository Ruff invocation still reports 430 pre-existing findings
+outside this change set. No changed file contributes a Ruff finding.
+
 ## Real PostgreSQL and MinIO acceptance
 
 The production-path integration uses:

app/core/mcp/gateway.py

@@ -245,10 +245,16 @@ class SchedulingGateway:
             int(candidate["version_no"]),
         )
         response = self._require_engine().deploy_disabled(compiled.yaml)
+        deployment_identity = (
+            self.plans.ensure_rule_deployment_identity(
+                candidate_id,
+                environment,
+                status="disabled",
+            )
+        )
         deployment = {
             "candidate_id": candidate_id,
-            "deployment_id": candidate_id,
-            "environment": environment,
+            **deployment_identity,
             "namespace": compiled.namespace,
             "flow_id": compiled.flow_id,
             "definition_hash": compiled.definition_hash,
@@ -315,6 +321,23 @@ class SchedulingGateway:
                 f"canary inputs contain unknown parameters: {', '.join(unknown_inputs)}"
             )
         deployment = self._deployment(candidate_id, business_domain, environment)
+        deployment_identity = (
+            self.plans.ensure_rule_deployment_identity(
+                candidate_id,
+                environment,
+                status="canary",
+            )
+        )
+        if (
+            deployment_identity.get("workflow_version_id")
+            != candidate_id
+            or deployment_identity.get("environment") != environment
+            or deployment_identity.get("deployment_id")
+            != deployment.get("deployment_id")
+        ):
+            raise RuntimeError(
+                "canonical rule deployment identity does not match"
+            )
         if self.token_issuer is None:
             raise RuntimeError("runner task token issuer is not configured")
         task_tokens = {}
@@ -322,7 +345,7 @@ class SchedulingGateway:
             task_tokens[node["id"]] = self.token_issuer.issue(
                 task_uid=str(uuid.uuid4()),
                 dataflow_uid=candidate["dataflow_uid"],
-                deployment_id=deployment["deployment_id"],
+                deployment_id=deployment_identity["deployment_id"],
                 environment=environment,
                 workflow_version=int(candidate["version_no"]),
                 correlation_id=identity.correlation_id,

app/core/mcp/persistence.py

@@ -222,6 +222,159 @@ class PostgresSchedulingPlanStore:
             )
         return dict(deployment)
 
+    def ensure_rule_deployment_identity(
+        self,
+        candidate_id,
+        environment,
+        *,
+        status,
+    ):
+        if status not in {"disabled", "canary"}:
+            raise ValueError("rule deployment status is invalid")
+        deployment_id = new_governance_uid()
+        with self.engine.begin() as connection:
+            target = connection.execute(
+                text(
+                    """
+                    SELECT w.id::text AS workflow_version_id,
+                           v.id::text AS dataflow_version_id,
+                           s.id::text AS schedule_plan_id
+                    FROM public.dataflow_workflow_versions w
+                    JOIN public.dataflow_versions v
+                      ON v.dataflow_uid = w.dataflow_uid
+                     AND v.version_no = w.version_no
+                     AND v.status = 'released'
+                    LEFT JOIN public.workflow_schedules s
+                      ON s.workflow_version_id = w.id
+                    WHERE w.id = CAST(:candidate_id AS uuid)
+                      AND w.environment = :environment
+                    """
+                ),
+                {
+                    "candidate_id": candidate_id,
+                    "environment": environment,
+                },
+            ).mappings().one_or_none()
+            if target is None:
+                raise RuntimeError(
+                    "released dataflow version for workflow was not found"
+                )
+            existing = connection.execute(
+                text(
+                    """
+                    SELECT id::text AS deployment_id,
+                           workflow_version_id::text,
+                           environment, status
+                    FROM public.dataflow_deployments
+                    WHERE dataflow_version_id =
+                        CAST(:dataflow_version_id AS uuid)
+                      AND environment = :environment
+                    FOR UPDATE
+                    """
+                ),
+                {
+                    "dataflow_version_id": target[
+                        "dataflow_version_id"
+                    ],
+                    "environment": environment,
+                },
+            ).mappings().one_or_none()
+            if existing is None:
+                existing = connection.execute(
+                    text(
+                        """
+                        INSERT INTO public.dataflow_deployments (
+                            id, dataflow_version_id, environment,
+                            workflow_version_id, schedule_plan_id,
+                            deployment_config, status, updated_at
+                        ) VALUES (
+                            CAST(:id AS uuid),
+                            CAST(:dataflow_version_id AS uuid),
+                            :environment,
+                            CAST(:workflow_version_id AS uuid),
+                            CAST(:schedule_plan_id AS uuid),
+                            '{}'::jsonb, :status, CURRENT_TIMESTAMP
+                        )
+                        RETURNING id::text AS deployment_id,
+                                  workflow_version_id::text,
+                                  environment, status
+                        """
+                    ),
+                    {
+                        "id": deployment_id,
+                        "dataflow_version_id": target[
+                            "dataflow_version_id"
+                        ],
+                        "environment": environment,
+                        "workflow_version_id": target[
+                            "workflow_version_id"
+                        ],
+                        "schedule_plan_id": target["schedule_plan_id"],
+                        "status": status,
+                    },
+                ).mappings().one()
+            else:
+                if existing["workflow_version_id"] is None:
+                    if existing["status"] != "disabled":
+                        raise RuntimeError(
+                            "unbound deployment is not disabled"
+                        )
+                    existing = connection.execute(
+                        text(
+                            """
+                            UPDATE public.dataflow_deployments
+                            SET workflow_version_id =
+                                    CAST(:workflow_version_id AS uuid),
+                                schedule_plan_id =
+                                    CAST(:schedule_plan_id AS uuid),
+                                updated_at = CURRENT_TIMESTAMP
+                            WHERE id = CAST(:id AS uuid)
+                              AND workflow_version_id IS NULL
+                              AND status = 'disabled'
+                            RETURNING id::text AS deployment_id,
+                                      workflow_version_id::text,
+                                      environment, status
+                            """
+                        ),
+                        {
+                            "id": existing["deployment_id"],
+                            "workflow_version_id": target[
+                                "workflow_version_id"
+                            ],
+                            "schedule_plan_id": target[
+                                "schedule_plan_id"
+                            ],
+                        },
+                    ).mappings().one()
+                elif (
+                    existing["workflow_version_id"]
+                    != target["workflow_version_id"]
+                ):
+                    raise RuntimeError(
+                        "deployment belongs to another workflow version"
+                    )
+                if status == "canary" and existing["status"] == "disabled":
+                    existing = connection.execute(
+                        text(
+                            """
+                            UPDATE public.dataflow_deployments
+                            SET status = 'canary',
+                                updated_at = CURRENT_TIMESTAMP
+                            WHERE id = CAST(:id AS uuid)
+                              AND status = 'disabled'
+                            RETURNING id::text AS deployment_id,
+                                      workflow_version_id::text,
+                                      environment, status
+                            """
+                        ),
+                        {"id": existing["deployment_id"]},
+                    ).mappings().one()
+                elif existing["status"] not in {status, "canary"}:
+                    raise RuntimeError(
+                        "deployment state cannot enter canary"
+                    )
+        return dict(existing)
+
     def get_deployment(self, candidate_id):
         with self.engine.connect() as connection:
             metadata = connection.execute(

app/core/orchestration/compilers/kestra.py

@@ -184,8 +184,8 @@ def _compile_dag_tasks(
         elif len(predecessors) > 1:
             parameters["input_artifacts"] = {
                 predecessor: (
-                    "{{ outputs.dataops_dag.tasks."
-                    f"{predecessor}.body.output_artifact }}}}"
+                    "{{ outputs.dataops_dag.tasks["
+                    f"'{predecessor}'].body.output_artifact }}}}"
                 )
                 for predecessor in predecessors
             }

app/runner/api.py

@@ -5,7 +5,7 @@ import json
 
 from flask import Flask, jsonify, request
 
-from app.runner.auth import TaskTokenInvalid
+from app.runner.auth import TaskTokenExpired, TaskTokenInvalid
 from app.runner.nodes import NodeExecutionError
 
 
@@ -34,8 +34,18 @@ def create_runner_app(*, verifier, ledger, registry):
         parameters = payload.get("parameters", {})
         if not isinstance(node, dict) or not isinstance(parameters, dict):
             return jsonify({"error": "invalid task request"}), 400
+        replay_only = False
         try:
             claims = verifier.verify(payload.get("task_token"), node=node)
+        except TaskTokenExpired:
+            try:
+                claims = verifier.verify_for_replay(
+                    payload.get("task_token"),
+                    node=node,
+                )
+            except TaskTokenInvalid:
+                return jsonify({"error": "task token is invalid"}), 401
+            replay_only = True
         except TaskTokenInvalid:
             return jsonify({"error": "task token is invalid"}), 401
         binding = {
@@ -50,15 +60,18 @@ def create_runner_app(*, verifier, ledger, registry):
             "data_source_uid": node.get("data_source_uid"),
             "idempotency_key": (node.get("idempotency") or {}).get("key"),
         }
-        try:
-            claimed = ledger.claim(
-                claims.jti,
-                binding,
-                expires_at=claims.expires_at,
-            )
-        except Exception:
-            app.logger.error("runner task ledger unavailable")
-            return jsonify({"error": "task ledger is unavailable"}), 503
+        if replay_only:
+            claimed = False
+        else:
+            try:
+                claimed = ledger.claim(
+                    claims.jti,
+                    binding,
+                    expires_at=claims.expires_at,
+                )
+            except Exception:
+                app.logger.error("runner task ledger unavailable")
+                return jsonify({"error": "task ledger is unavailable"}), 503
         if not claimed:
             try:
                 existing = ledger.get(claims.jti)
@@ -71,7 +84,8 @@ def create_runner_app(*, verifier, ledger, registry):
                     for key, value in binding.items()
                 )
             ):
-                return jsonify({"error": "task token already consumed"}), 409
+                status = 401 if replay_only else 409
+                return jsonify({"error": "task token already consumed"}), status
             if existing.status == "running":
                 recovered = None
                 if node.get("type") in {"rule.apply", "quality.check"}:
@@ -86,19 +100,60 @@ def create_runner_app(*, verifier, ledger, registry):
                         return jsonify(
                             {"error": "task ledger is unavailable"}
                         ), 503
-                if isinstance(recovered, dict):
+                recovered_state = (
+                    recovered.get("state")
+                    if isinstance(recovered, dict)
+                    else None
+                )
+                recovered_result = (
+                    recovered.get("result")
+                    if recovered_state == "terminal"
+                    and recovered.get("status") == "success"
+                    else recovered
+                )
+                if (
+                    recovered_state == "terminal"
+                    and recovered.get("status") != "success"
+                ):
+                    if not finish_safely(
+                        claims.jti,
+                        status="unknown",
+                        commit_outcome=str(
+                            recovered.get("commit_outcome")
+                            or "unknown"
+                        ),
+                        safe_detail=(
+                            "task execution evidence is terminal"
+                        ),
+                    ):
+                        return jsonify(
+                            {"error": "task ledger is unavailable"}
+                        ), 503
+                    return jsonify(
+                        {
+                            "error": (
+                                "task execution outcome is unknown"
+                            )
+                        }
+                    ), 409
+                if (
+                    isinstance(recovered_result, dict)
+                    and recovered_state != "missing"
+                ):
                     replay_body = {
                         "task_uid": claims.task_uid,
                         "correlation_id": claims.correlation_id,
-                        "result": recovered,
+                        "result": recovered_result,
                     }
-                    output_artifact = recovered.get("output_artifact")
+                    output_artifact = recovered_result.get(
+                        "output_artifact"
+                    )
                     if isinstance(output_artifact, str):
                         replay_body["output_artifact"] = output_artifact
                     if not finish_safely(
                         claims.jti,
                         status="success",
-                        commit_outcome=recovered.get(
+                        commit_outcome=recovered_result.get(
                             "commit_outcome",
                             "not_applicable",
                         ),
@@ -112,6 +167,31 @@ def create_runner_app(*, verifier, ledger, registry):
                     response = jsonify(replay_body)
                     response.headers["X-Idempotent-Replay"] = "true"
                     return response
+                if recovered_state in {None, "missing"}:
+                    try:
+                        existing = ledger.reconcile_running(
+                            claims.jti,
+                            binding,
+                        )
+                    except Exception:
+                        return jsonify(
+                            {"error": "task ledger is unavailable"}
+                        ), 503
+                    if (
+                        existing is not None
+                        and existing.status == "unknown"
+                    ):
+                        return jsonify(
+                            {
+                                "error": (
+                                    "task execution outcome is unknown"
+                                )
+                            }
+                        ), 409
+                if replay_only:
+                    return jsonify(
+                        {"error": "expired task is not replayable"}
+                    ), 401
                 response = jsonify(
                     {
                         "task_uid": claims.task_uid,

app/runner/auth.py

@@ -109,11 +109,23 @@ class TaskTokenIssuer:
 
 
 class TaskTokenVerifier:
-    def __init__(self, secret, *, clock=None):
+    def __init__(
+        self,
+        secret,
+        *,
+        clock=None,
+        replay_window_seconds=900,
+    ):
         self._secret = _secret_bytes(secret)
         self._clock = clock or time.time
+        self._replay_window_seconds = int(replay_window_seconds)
+        if (
+            self._replay_window_seconds < 1
+            or self._replay_window_seconds > 3600
+        ):
+            raise ValueError("task replay window is invalid")
 
-    def verify(self, token, *, node):
+    def _verify(self, token, *, node, replay_only):
         try:
             payload = jwt.decode(
                 token,
@@ -150,10 +162,15 @@ class TaskTokenVerifier:
             workflow_version = int(payload["workflow_version"])
         except (TypeError, ValueError) as exc:
             raise TaskTokenInvalid("task token claims are invalid") from exc
-        if expires_at < now:
-            raise TaskTokenExpired("task token has expired")
         if issued_at > now + 5 or expires_at - issued_at > 300:
             raise TaskTokenInvalid("task token lifetime is invalid")
+        if expires_at < now:
+            if not replay_only:
+                raise TaskTokenExpired("task token has expired")
+            if now - expires_at > self._replay_window_seconds:
+                raise TaskTokenExpired(
+                    "task token replay window has expired"
+                )
         if node_digest(node) != payload["node_digest"]:
             raise TaskTokenInvalid("task token node binding does not match")
         if (
@@ -184,3 +201,9 @@ class TaskTokenVerifier:
             expires_at=expires_at,
             jti=str(payload["jti"]),
         )
+
+    def verify(self, token, *, node):
+        return self._verify(token, node=node, replay_only=False)
+
+    def verify_for_replay(self, token, *, node):
+        return self._verify(token, node=node, replay_only=True)

app/runner/ledger.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import hashlib
 import json
 import threading
+import time
 from collections.abc import Mapping
 from dataclasses import dataclass
 
@@ -22,12 +23,15 @@ class TaskLedgerRecord:
     replay_http_status: int | None = None
     replay_body: Mapping[str, object] | None = None
     replay_digest: str | None = None
+    lease_expires_at: float = 0
 
 
 class InMemoryTaskLedger:
-    def __init__(self):
+    def __init__(self, *, clock=None, lease_seconds=300):
         self._records = {}
         self._lock = threading.Lock()
+        self._clock = clock or time.time
+        self._lease_seconds = int(lease_seconds)
 
     def claim(self, jti, binding, *, expires_at):
         with self._lock:
@@ -37,6 +41,9 @@ class InMemoryTaskLedger:
                 jti=str(jti),
                 binding=dict(binding),
                 expires_at=int(expires_at),
+                lease_expires_at=(
+                    float(self._clock()) + self._lease_seconds
+                ),
             )
             return True
 
@@ -70,12 +77,29 @@ class InMemoryTaskLedger:
         with self._lock:
             return self._records.get(str(jti))
 
+    def reconcile_running(self, jti, binding):
+        with self._lock:
+            record = self._records.get(str(jti))
+            if (
+                record is not None
+                and record.status == "running"
+                and record.binding == dict(binding)
+                and record.lease_expires_at <= float(self._clock())
+            ):
+                record.status = "unknown"
+                record.commit_outcome = "unknown"
+                record.safe_detail = "task execution lease expired"
+            return record
+
 
 class PostgresTaskLedger:
     """Durable, cross-worker single-use ledger backed by the platform DB."""
 
-    def __init__(self, engine):
+    def __init__(self, engine, *, lease_seconds=300):
         self.engine = engine
+        self.lease_seconds = int(lease_seconds)
+        if self.lease_seconds < 30 or self.lease_seconds > 900:
+            raise ValueError("runner task lease is invalid")
 
     def claim(self, jti, binding, *, expires_at):
         parameters = {
@@ -91,6 +115,7 @@ class PostgresTaskLedger:
             "data_source_uid": binding.get("data_source_uid"),
             "idempotency_key": binding.get("idempotency_key"),
             "expires_at": int(expires_at),
+            "lease_seconds": self.lease_seconds,
         }
         with self.engine.begin() as connection:
             result = connection.execute(
@@ -100,7 +125,7 @@ class PostgresTaskLedger:
                         token_jti, task_uid, dataflow_uid, workflow_version,
                         correlation_id, node_id, node_type, data_source_uid,
                         idempotency_key, status, commit_outcome, expires_at,
-                        deployment_id, environment
+                        deployment_id, environment, lease_expires_at
                     ) VALUES (
                         CAST(:jti AS uuid), CAST(:task_uid AS uuid),
                         CAST(:dataflow_uid AS uuid), :workflow_version,
@@ -108,7 +133,9 @@ class PostgresTaskLedger:
                         CAST(:data_source_uid AS uuid), :idempotency_key,
                         'running', 'not_applicable',
                         to_timestamp(:expires_at),
-                        CAST(:deployment_id AS uuid), :environment
+                        CAST(:deployment_id AS uuid), :environment,
+                        CURRENT_TIMESTAMP
+                            + make_interval(secs => :lease_seconds)
                     )
                     ON CONFLICT DO NOTHING
                     """
@@ -117,6 +144,46 @@ class PostgresTaskLedger:
             )
         return int(result.rowcount or 0) == 1
 
+    def reconcile_running(self, jti, binding):
+        parameters = {
+            "jti": str(jti),
+            "task_uid": binding["task_uid"],
+            "dataflow_uid": binding["dataflow_uid"],
+            "deployment_id": binding["deployment_id"],
+            "environment": binding["environment"],
+            "workflow_version": int(binding["workflow_version"]),
+            "correlation_id": binding["correlation_id"],
+            "node_id": binding["node_id"],
+            "node_type": binding["node_type"],
+        }
+        with self.engine.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.runner_task_executions
+                    SET status = 'unknown',
+                        commit_outcome = 'unknown',
+                        safe_detail = 'task execution lease expired',
+                        finished_at = CURRENT_TIMESTAMP
+                    WHERE token_jti = CAST(:jti AS uuid)
+                      AND task_uid = CAST(:task_uid AS uuid)
+                      AND dataflow_uid = CAST(:dataflow_uid AS uuid)
+                      AND deployment_id =
+                          CAST(:deployment_id AS uuid)
+                      AND environment = :environment
+                      AND workflow_version = :workflow_version
+                      AND correlation_id =
+                          CAST(:correlation_id AS uuid)
+                      AND node_id = :node_id
+                      AND node_type = :node_type
+                      AND status = 'running'
+                      AND lease_expires_at <= CURRENT_TIMESTAMP
+                    """
+                ),
+                parameters,
+            )
+        return self.get(jti)
+
     def finish(
         self,
         jti,
@@ -195,6 +262,9 @@ class PostgresTaskLedger:
                                node_type,
                                safe_detail, EXTRACT(EPOCH FROM expires_at)::bigint
                                    AS expires_at,
+                               EXTRACT(
+                                   EPOCH FROM lease_expires_at
+                               )::double precision AS lease_expires_at,
                                replay_http_status, replay_body,
                                replay_digest
                         FROM public.runner_task_executions
@@ -233,4 +303,5 @@ class PostgresTaskLedger:
                 else row["replay_body"]
             ),
             replay_digest=row["replay_digest"],
+            lease_expires_at=float(row["lease_expires_at"]),
         )

app/runner/rule_evidence.py

@@ -229,6 +229,10 @@ class PostgresRuleEvidenceWriter:
             raise ValueError("violation sample TTL is invalid")
         if self.lease_seconds < 30 or self.lease_seconds > 900:
             raise ValueError("rule execution lease is invalid")
+        self.heartbeat_interval_seconds = max(
+            5.0,
+            min(60.0, self.lease_seconds / 3),
+        )
 
     def start(
         self,
@@ -685,6 +689,115 @@ class PostgresRuleEvidenceWriter:
             return None
         return self.replay(str(row))
 
+    def reconcile_expired_lease(
+        self,
+        *,
+        lease_owner: str,
+        deployment_id: str,
+        correlation_id: str,
+        component_binding_id: str,
+        rule_version_id: str,
+        plan_hash: str,
+    ) -> dict[str, Any]:
+        owner = _uuid(lease_owner, "lease owner")
+        deployment = _uid(deployment_id, "deployment id")
+        correlation = _uid(correlation_id, "correlation id")
+        component = _uid(
+            component_binding_id,
+            "component binding id",
+        )
+        rule = _uid(rule_version_id, "rule version id")
+        if _DIGEST.fullmatch(str(plan_hash or "")) is None:
+            raise ValueError("plan hash is invalid")
+        parameters = {
+            "lease_owner": owner,
+            "deployment_id": deployment,
+            "correlation_id": correlation,
+            "component_binding_id": component,
+            "rule_version_id": rule,
+            "plan_hash": plan_hash,
+        }
+        with self.engine.begin() as connection:
+            row = connection.execute(
+                text(
+                    """
+                    SELECT id::text, status, lease_expires_at
+                    FROM public.rule_runs
+                    WHERE lease_owner = CAST(:lease_owner AS uuid)
+                      AND deployment_id =
+                          CAST(:deployment_id AS uuid)
+                      AND correlation_id =
+                          CAST(:correlation_id AS uuid)
+                      AND component_binding_id =
+                          CAST(:component_binding_id AS uuid)
+                      AND rule_version_id =
+                          CAST(:rule_version_id AS uuid)
+                      AND plan_hash = :plan_hash
+                    FOR UPDATE
+                    """
+                ),
+                parameters,
+            ).mappings().one_or_none()
+            if row is None:
+                return {"state": "missing"}
+            if (
+                row["status"] == "running"
+                and row["lease_expires_at"] is not None
+            ):
+                expired = connection.execute(
+                    text(
+                        "SELECT :lease_expires_at <= CURRENT_TIMESTAMP"
+                    ),
+                    {"lease_expires_at": row["lease_expires_at"]},
+                ).scalar_one()
+                if expired:
+                    connection.execute(
+                        text(
+                            """
+                            UPDATE public.rule_runs
+                            SET status = 'unknown',
+                                commit_outcome = 'unknown',
+                                failure_code =
+                                    'execution_lease_expired',
+                                lease_expires_at = NULL,
+                                finished_at = CURRENT_TIMESTAMP,
+                                updated_at = CURRENT_TIMESTAMP
+                            WHERE id = CAST(:id AS uuid)
+                              AND status = 'running'
+                            """
+                        ),
+                        {"id": row["id"]},
+                    )
+                    connection.execute(
+                        text(
+                            """
+                            UPDATE public.rule_sql_staging_receipts
+                            SET status = 'failed',
+                                commit_outcome = 'unknown',
+                                updated_at = CURRENT_TIMESTAMP
+                            WHERE producer_rule_run_id =
+                                CAST(:id AS uuid)
+                              AND status = 'pending'
+                            """
+                        ),
+                        {"id": row["id"]},
+                    )
+                    row = {**row, "status": "unknown"}
+            if row["status"] == "running":
+                return {"state": "running"}
+            run_id = str(row["id"])
+        replay = self.replay(run_id)
+        return {
+            "state": "terminal",
+            "status": str(replay["status"]),
+            "commit_outcome": str(replay["commit_outcome"]),
+            "result": {
+                key: value
+                for key, value in replay.items()
+                if key != "status"
+            },
+        }
+
     @staticmethod
     def _validate_finish(result: Any) -> dict[str, Any]:
         if not isinstance(result, dict) or set(result) - _FINISH_KEYS:
@@ -765,6 +878,7 @@ class PostgresRuleEvidenceWriter:
             path = handle.name
         sample_id = None
         prepared = None
+        uploaded = False
         try:
             frame.write_parquet(path)
             prepared = self.artifact_store.prepare_path(
@@ -807,13 +921,14 @@ class PostgresRuleEvidenceWriter:
                                 id, rule_run_id, artifact_ref,
                                 artifact_digest, schema_hash, sample_count,
                                 redaction_policy, expires_at,
-                                handoff_status
+                                handoff_status, schema_fields
                             ) 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'
+                                CAST(:expires_at AS timestamptz), 'pending',
+                                CAST(:schema_fields AS jsonb)
                             )
                             """
                         ),
@@ -826,6 +941,7 @@ class PostgresRuleEvidenceWriter:
                             "sample_count": len(sample),
                             "redaction_policy": redaction_policy,
                             "expires_at": prepared["expires_at"],
+                            "schema_fields": json.dumps(fields),
                         },
                     )
                 else:
@@ -880,6 +996,7 @@ class PostgresRuleEvidenceWriter:
                     ),
                 },
             )
+            uploaded = True
             with self.engine.begin() as connection:
                 updated = connection.execute(
                     text(
@@ -941,18 +1058,29 @@ class PostgresRuleEvidenceWriter:
                     text(
                         """
                         UPDATE public.rule_violation_samples
-                        SET handoff_status = 'failed',
-                            failure_code = 'sample_handoff_failed',
+                        SET handoff_status = :handoff_status,
+                            failure_code = :failure_code,
                             updated_at = CURRENT_TIMESTAMP
                         WHERE rule_run_id = CAST(:rule_run_id AS uuid)
                           AND handoff_status = 'pending'
                         """
                     ),
-                    {"rule_run_id": run_id},
+                    {
+                        "rule_run_id": run_id,
+                        "handoff_status": (
+                            "unknown" if uploaded else "failed"
+                        ),
+                        "failure_code": (
+                            "sample_finalize_unknown"
+                            if uploaded
+                            else "sample_handoff_failed"
+                        ),
+                    },
                 )
+            raise
+        finally:
             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")
@@ -1136,20 +1264,31 @@ class PostgresRuleEvidenceWriter:
                 with suppress(FileNotFoundError):
                     os.unlink(sample_path)
 
-    def cleanup_expired(self, *, limit: int = 100) -> int:
+    def reconcile_samples(self, *, limit: int = 100) -> dict[str, 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
+        claim = new_governance_uid()
         with self.engine.begin() as connection:
             rows = connection.execute(
                 text(
                     """
-                    SELECT id::text, artifact_ref
+                    SELECT id::text, artifact_ref, artifact_digest,
+                           schema_hash, schema_fields, sample_count,
+                           handoff_status,
+                           expires_at <= CURRENT_TIMESTAMP AS expired
                     FROM public.rule_violation_samples
-                    WHERE expires_at <= CURRENT_TIMESTAMP
-                      AND handoff_status IN ('legacy','ready','failed')
+                    WHERE cleanup_claim IS NULL
+                      AND (
+                          handoff_status IN ('pending','unknown')
+                          OR (
+                              expires_at <= CURRENT_TIMESTAMP
+                              AND handoff_status IN (
+                                  'legacy','ready','failed'
+                              )
+                          )
+                      )
                     ORDER BY expires_at, id
                     FOR UPDATE SKIP LOCKED
                     LIMIT :limit
@@ -1158,21 +1297,193 @@ class PostgresRuleEvidenceWriter:
                 {"limit": limit},
             ).mappings().all()
             for row in rows:
+                connection.execute(
+                    text(
+                        """
+                        UPDATE public.rule_violation_samples
+                        SET cleanup_claim = CAST(:claim AS uuid),
+                            updated_at = CURRENT_TIMESTAMP
+                        WHERE id = CAST(:id AS uuid)
+                          AND cleanup_claim IS NULL
+                        """
+                    ),
+                    {"id": str(row["id"]), "claim": claim},
+                )
+        metrics = {
+            "claimed": len(rows),
+            "ready": 0,
+            "failed": 0,
+            "expired_deleted": 0,
+        }
+        for row in rows:
+            row_id = str(row["id"])
+            if bool(row["expired"]):
                 try:
                     self.artifact_store.delete(str(row["artifact_ref"]))
                 except Exception:
+                    with self.engine.begin() as connection:
+                        connection.execute(
+                            text(
+                                """
+                                UPDATE public.rule_violation_samples
+                                SET cleanup_claim = NULL,
+                                    updated_at = CURRENT_TIMESTAMP
+                                WHERE id = CAST(:id AS uuid)
+                                  AND cleanup_claim =
+                                      CAST(:claim AS uuid)
+                                """
+                            ),
+                            {"id": row_id, "claim": claim},
+                        )
                     continue
+                with self.engine.begin() as connection:
+                    deleted = connection.execute(
+                        text(
+                            """
+                            DELETE FROM public.rule_violation_samples
+                            WHERE id = CAST(:id AS uuid)
+                              AND cleanup_claim = CAST(:claim AS uuid)
+                            """
+                        ),
+                        {"id": row_id, "claim": claim},
+                    )
+                metrics["expired_deleted"] += int(
+                    deleted.rowcount or 0
+                )
+                continue
+            next_status = row["handoff_status"]
+            failure_code = None
+            try:
+                fields = row["schema_fields"]
+                if isinstance(fields, str):
+                    fields = json.loads(fields)
+                if not isinstance(fields, list):
+                    raise ValueError(
+                        "violation sample schema is unavailable"
+                    )
+                with self.artifact_store.stage(
+                    str(row["artifact_ref"]),
+                    str(row["artifact_digest"]),
+                    expected_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,
+                        ),
+                    },
+                ):
+                    pass
+                described = self.artifact_store.describe(
+                    str(row["artifact_ref"])
+                )
+                if (
+                    described["schema_hash"] != row["schema_hash"]
+                    or described["row_count"]
+                    != int(row["sample_count"])
+                ):
+                    raise ValueError(
+                        "violation sample attestation does not match"
+                    )
+                next_status = "ready"
+                metrics["ready"] += 1
+            except ValueError:
+                next_status = "failed"
+                failure_code = "sample_reconcile_invalid"
+                metrics["failed"] += 1
+            except Exception:
+                next_status = row["handoff_status"]
+            with self.engine.begin() as connection:
+                connection.execute(
+                    text(
+                        """
+                        UPDATE public.rule_violation_samples
+                        SET handoff_status = :handoff_status,
+                            failure_code = :failure_code,
+                            cleanup_claim = NULL,
+                            updated_at = CURRENT_TIMESTAMP
+                        WHERE id = CAST(:id AS uuid)
+                          AND cleanup_claim = CAST(:claim AS uuid)
+                        """
+                    ),
+                    {
+                        "id": row_id,
+                        "claim": claim,
+                        "handoff_status": next_status,
+                        "failure_code": failure_code,
+                    },
+                )
+        return metrics
+
+    def cleanup_sql_staging(self, *, limit: int = 100) -> int:
+        if (
+            isinstance(limit, bool)
+            or not isinstance(limit, int)
+            or limit < 1
+            or limit > 1000
+        ):
+            raise ValueError("cleanup limit is invalid")
+        claim = new_governance_uid()
+        with self.engine.begin() as connection:
+            rows = connection.execute(
+                text(
+                    """
+                    SELECT id::text
+                    FROM public.rule_sql_staging_receipts
+                    WHERE expires_at <= CURRENT_TIMESTAMP
+                      AND status IN ('pending','ready','failed')
+                      AND cleanup_claim IS NULL
+                    ORDER BY expires_at, id
+                    FOR UPDATE SKIP LOCKED
+                    LIMIT :limit
+                    """
+                ),
+                {"limit": limit},
+            ).mappings().all()
+            for row in rows:
                 connection.execute(
                     text(
                         """
-                        DELETE FROM public.rule_violation_samples
+                        UPDATE public.rule_sql_staging_receipts
+                        SET cleanup_claim = CAST(:claim AS uuid),
+                            updated_at = CURRENT_TIMESTAMP
+                        WHERE id = CAST(:id AS uuid)
+                          AND cleanup_claim IS NULL
+                        """
+                    ),
+                    {"id": str(row["id"]), "claim": claim},
+                )
+        finalized = 0
+        for row in rows:
+            with self.engine.begin() as connection:
+                updated = connection.execute(
+                    text(
+                        """
+                        UPDATE public.rule_sql_staging_receipts
+                        SET status = 'expired',
+                            cleanup_claim = NULL,
+                            updated_at = CURRENT_TIMESTAMP
                         WHERE id = CAST(:id AS uuid)
+                          AND cleanup_claim = CAST(:claim AS uuid)
+                          AND expires_at <= CURRENT_TIMESTAMP
                         """
                     ),
-                    {"id": str(row["id"])},
+                    {"id": str(row["id"]), "claim": claim},
                 )
-                removed += 1
-        return removed
+            finalized += int(updated.rowcount or 0)
+        return finalized
+
+    def cleanup_expired(self, *, limit: int = 100) -> int:
+        samples = self.reconcile_samples(limit=limit)
+        receipts = self.cleanup_sql_staging(limit=limit)
+        return samples["expired_deleted"] + receipts
 
 
 __all__ = [

app/runner/rules.py

@@ -6,6 +6,7 @@ import hashlib
 import json
 import logging
 import re
+import threading
 import time
 from asyncio import CancelledError
 from collections.abc import Mapping
@@ -237,7 +238,7 @@ class RulePlanExecutor:
         if not isinstance(config, dict):
             return None
         try:
-            replay = self.evidence_writer.replay_by_lease_owner(
+            replay = self.evidence_writer.reconcile_expired_lease(
                 lease_owner=task_jti,
                 deployment_id=deployment_id,
                 correlation_id=correlation_id,
@@ -257,13 +258,7 @@ class RulePlanExecutor:
             raise NodeExecutionError(
                 "rule execution evidence is unavailable"
             ) from exc
-        if replay is None or replay.get("status") != "success":
-            return None
-        return {
-            key: value
-            for key, value in replay.items()
-            if key != "status"
-        }
+        return replay
 
     def execute(
         self,
@@ -486,6 +481,35 @@ class RulePlanExecutor:
                 "correlation_id"
             )
         started = time.monotonic()
+        heartbeat_stop = threading.Event()
+        heartbeat_failures = []
+        heartbeat_thread = None
+        if self.evidence_writer is not None:
+            interval = float(
+                getattr(
+                    self.evidence_writer,
+                    "heartbeat_interval_seconds",
+                    10.0,
+                )
+            )
+
+            def renew_lease():
+                while not heartbeat_stop.wait(interval):
+                    try:
+                        self.evidence_writer.heartbeat(
+                            rule_run_id,
+                            execution_context.get("task_jti"),
+                        )
+                    except Exception as exc:
+                        heartbeat_failures.append(exc)
+                        heartbeat_stop.set()
+
+            heartbeat_thread = threading.Thread(
+                target=renew_lease,
+                name=f"rule-heartbeat-{rule_run_id}",
+                daemon=True,
+            )
+            heartbeat_thread.start()
         try:
             result = adapter.execute(
                 plan=record["plan"],
@@ -543,6 +567,37 @@ class RulePlanExecutor:
                 "rule plan execution failed",
                 commit_outcome="not_committed",
             ) from exc
+        finally:
+            heartbeat_stop.set()
+            if heartbeat_thread is not None:
+                heartbeat_thread.join(timeout=5)
+        if heartbeat_failures:
+            result_outcome = (
+                result.get("commit_outcome")
+                if isinstance(result, dict)
+                else None
+            )
+            evidence_outcome = (
+                "committed"
+                if result_outcome == "committed"
+                else "unknown"
+            )
+            self._finish_evidence(
+                rule_run_id,
+                {
+                    "status": "unknown",
+                    "commit_outcome": evidence_outcome,
+                    "timings": {
+                        "duration_ms": int(
+                            (time.monotonic() - started) * 1000
+                        )
+                    },
+                },
+            )
+            raise NodeExecutionError(
+                "rule execution lease outcome is unknown",
+                commit_outcome="unknown",
+            )
         if not isinstance(result, dict):
             error = NodeExecutionError(
                 "rule plan result must be an object"
@@ -560,6 +615,9 @@ class RulePlanExecutor:
                 },
             )
             raise error
+        adapter_commit_outcome = str(
+            result.get("commit_outcome", "not_applicable")
+        )
         staging_output = None
         if backend == "sql_pushdown" and self.evidence_writer is not None:
             try:
@@ -572,7 +630,11 @@ class RulePlanExecutor:
                     rule_run_id,
                     {
                         "status": "unknown",
-                        "commit_outcome": "unknown",
+                        "commit_outcome": (
+                            "committed"
+                            if adapter_commit_outcome == "committed"
+                            else "unknown"
+                        ),
                         "timings": {
                             "duration_ms": int(
                                 (time.monotonic() - started) * 1000
@@ -617,11 +679,23 @@ class RulePlanExecutor:
                 backend=backend,
             )
         except (TypeError, ValueError) as exc:
+            post_commit_unknown = adapter_commit_outcome in {
+                "committed",
+                "unknown",
+            }
             self._finish_evidence(
                 rule_run_id,
                 {
-                    "status": "failed",
-                    "commit_outcome": "not_committed",
+                    "status": (
+                        "unknown"
+                        if post_commit_unknown
+                        else "failed"
+                    ),
+                    "commit_outcome": (
+                        adapter_commit_outcome
+                        if post_commit_unknown
+                        else "not_committed"
+                    ),
                     "timings": {
                         "duration_ms": int(
                             (time.monotonic() - started) * 1000
@@ -631,7 +705,11 @@ class RulePlanExecutor:
             )
             raise NodeExecutionError(
                 "rule plan result is invalid",
-                commit_outcome="not_committed",
+                commit_outcome=(
+                    "unknown"
+                    if post_commit_unknown
+                    else "not_committed"
+                ),
             ) from exc
         evidence_result = {
             "status": "success",

migrations/versions/20260723_180_rule_reconciliation.py

@@ -0,0 +1,49 @@
+"""Add bounded reconciliation claims for Runner and rule evidence."""
+
+from alembic import op
+
+revision = "20260723_180"
+down_revision = "20260723_170"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE public.runner_task_executions
+            ADD COLUMN lease_expires_at TIMESTAMPTZ NOT NULL
+                DEFAULT CURRENT_TIMESTAMP;
+
+        ALTER TABLE public.rule_violation_samples
+            DROP CONSTRAINT
+                rule_violation_samples_handoff_status_check,
+            ADD CONSTRAINT
+                rule_violation_samples_handoff_status_check
+                CHECK (handoff_status IN (
+                    'legacy','pending','ready','failed',
+                    'unknown','expired'
+                )),
+            ADD COLUMN cleanup_claim UUID,
+            ADD COLUMN schema_fields JSONB;
+
+        CREATE INDEX idx_rule_violation_sample_cleanup_claim
+            ON public.rule_violation_samples(
+                cleanup_claim, expires_at, id
+            );
+
+        ALTER TABLE public.rule_sql_staging_receipts
+            ADD COLUMN cleanup_claim UUID;
+
+        CREATE INDEX idx_rule_sql_receipt_cleanup_claim
+            ON public.rule_sql_staging_receipts(
+                cleanup_claim, expires_at, id
+            );
+        """
+    )
+
+
+def downgrade() -> None:
+    raise RuntimeError(
+        "rule reconciliation claims are forward-only"
+    )

tests/integration/test_data_rule_polars_execution.py

@@ -117,6 +117,8 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
     sample_crash_correlation_id = new_governance_uid()
     receipt_correlation_id = new_governance_uid()
     failed_receipt_correlation_id = new_governance_uid()
+    no_run_crash_correlation_id = new_governance_uid()
+    evidence_crash_correlation_id = new_governance_uid()
     sql_binding_id = new_governance_uid()
     prefix = f"rules/{correlation_id}/"
     customer_table = "task5_polars_customers"
@@ -134,6 +136,9 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
     downstream_plan_id = new_governance_uid()
     ledger_jti = None
     retry_ledger_jti = None
+    no_run_crash_jti = None
+    evidence_crash_jti = None
+    gateway_candidate_id = None
 
     try:
         with postgres.begin() as connection:
@@ -480,10 +485,10 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     """
                     INSERT INTO public.dataflow_deployments
                     (id, dataflow_version_id, environment, deployment_config,
-                     status, activated_at)
+                     status)
                     VALUES (CAST(:id AS uuid),
                             CAST(:dataflow_version_id AS uuid), 'test',
-                            '{}'::jsonb, 'active', CURRENT_TIMESTAMP)
+                            '{}'::jsonb, 'disabled')
                     """
                 ),
                 {
@@ -702,6 +707,120 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 "execution_plan_hash": compiled["plan_hash"],
             },
         }
+        from app.core.mcp.gateway import SchedulingGateway
+        from app.core.mcp.identity import AgentIdentity
+        from app.core.mcp.persistence import PostgresSchedulingPlanStore
+        from app.runner.api import create_runner_app
+        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
+
+        class Audit:
+            def __init__(self):
+                self.events = []
+
+            def record(self, event):
+                self.events.append(event)
+
+        class CanaryEngine:
+            def __init__(self):
+                self.calls = []
+
+            def deploy_disabled(self, _definition):
+                return {"revision": "task6-real-gateway"}
+
+            def activate(self, namespace, flow_id):
+                self.calls.append(("activate", namespace, flow_id))
+
+            def execute(self, namespace, flow_id, inputs=None, **_options):
+                self.calls.append(
+                    ("execute", namespace, flow_id, dict(inputs or {}))
+                )
+                return {"id": f"task6-canary-{correlation_id}"}
+
+            def deactivate(self, namespace, flow_id):
+                self.calls.append(("deactivate", namespace, flow_id))
+
+        task_secret = "task5-real-http-secret-value-32-bytes"
+        verifier = TaskTokenVerifier(task_secret)
+        gateway_engine = CanaryEngine()
+        gateway_store = PostgresSchedulingPlanStore(platform)
+        gateway = SchedulingGateway(
+            plans=gateway_store,
+            audit=Audit(),
+            engine=gateway_engine,
+            token_issuer=TaskTokenIssuer(task_secret),
+        )
+        gateway_identity = AgentIdentity(
+            subject="task6-real-scheduler",
+            roles=frozenset({"scheduler"}),
+            business_domains=frozenset({"sales"}),
+            environments=frozenset({"test"}),
+            correlation_id=correlation_id,
+        )
+        candidate = gateway.create_candidate_plan(
+            gateway_identity,
+            business_domain="sales",
+            environment="test",
+            workflow_spec={
+                "schema_version": "1.0",
+                "dataflow_uid": dataflow_uid,
+                "name": "Task 6 governed rule canary",
+                "nodes": [node],
+                "edges": [],
+                "parameters": {},
+            },
+            schedule_plan={
+                "schema_version": "1.0",
+                "timezone": "Asia/Shanghai",
+                "triggers": [{"type": "manual"}],
+                "max_concurrency": 1,
+                "conflict_policy": "skip",
+                "timeout_seconds": 600,
+                "retry": {
+                    "max_attempts": 1,
+                    "delay_seconds": 1,
+                },
+                "backfill": {"max_days": 1, "max_runs": 1},
+            },
+        )
+        gateway_candidate_id = candidate["candidate_id"]
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.dataflow_workflow_versions
+                    SET write_authorized = TRUE
+                    WHERE id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": gateway_candidate_id},
+            )
+        deployed = gateway.deploy_disabled_version(
+            gateway_identity,
+            candidate_id=gateway_candidate_id,
+            business_domain="sales",
+            environment="test",
+        )
+        assert deployed["deployment_id"] == deployment_id
+        assert deployed["candidate_id"] != deployed["deployment_id"]
+        gateway.run_canary(
+            gateway_identity,
+            candidate_id=gateway_candidate_id,
+            business_domain="sales",
+            environment="test",
+            inputs={},
+        )
+        execution_call = next(
+            call for call in gateway_engine.calls if call[0] == "execute"
+        )
+        task_token = execution_call[3]["dataops_task_tokens"][node["id"]]
+        issued_claims = verifier.verify(task_token, node=node)
+        assert issued_claims.deployment_id == deployment_id
+        assert issued_claims.workflow_version == 1
+        assert issued_claims.environment == "test"
+
         executor = RulePlanExecutor(
             PostgresRulePlanRepository(platform),
             adapters={
@@ -770,24 +889,6 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         assert repeated["artifact_ref"] == result["artifact_ref"]
         assert "schema_fields" not in result
 
-        from app.runner.api import create_runner_app
-        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)
-        task_token = TaskTokenIssuer(task_secret).issue(
-            task_uid=new_governance_uid(),
-            dataflow_uid=dataflow_uid,
-            deployment_id=deployment_id,
-            environment="test",
-            workflow_version=1,
-            correlation_id=correlation_id,
-            node=node,
-            write_authorized=True,
-        )
         ledger_jti = verifier.verify(task_token, node=node).jti
         retry_token = TaskTokenIssuer(task_secret).issue(
             task_uid=new_governance_uid(),
@@ -802,6 +903,11 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         retry_ledger_jti = verifier.verify(
             retry_token, node=node
         ).jti
+        real_evidence_writer = PostgresRuleEvidenceWriter(
+            platform,
+            store,
+            sample_ttl_seconds=900,
+        )
         evidenced_executor = RulePlanExecutor(
             PostgresRulePlanRepository(platform),
             adapters={
@@ -811,17 +917,107 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     artifact_ttl_seconds=900,
                 )
             },
-            evidence_writer=PostgresRuleEvidenceWriter(
-                platform,
-                store,
-                sample_ttl_seconds=900,
-            ),
+            evidence_writer=real_evidence_writer,
         )
+        real_ledger = PostgresTaskLedger(platform, lease_seconds=30)
         runner_app = create_runner_app(
             verifier=verifier,
-            ledger=PostgresTaskLedger(platform),
+            ledger=real_ledger,
             registry=NodeRegistry({"rule.apply": evidenced_executor}),
         )
+        no_run_crash_token = TaskTokenIssuer(task_secret).issue(
+            task_uid=new_governance_uid(),
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            correlation_id=no_run_crash_correlation_id,
+            node=node,
+            write_authorized=True,
+        )
+        no_run_claims = verifier.verify(no_run_crash_token, node=node)
+        no_run_crash_jti = no_run_claims.jti
+        evidence_crash_token = TaskTokenIssuer(task_secret).issue(
+            task_uid=new_governance_uid(),
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            correlation_id=evidence_crash_correlation_id,
+            node=node,
+            write_authorized=True,
+        )
+        evidence_crash_claims = verifier.verify(
+            evidence_crash_token,
+            node=node,
+        )
+        evidence_crash_jti = evidence_crash_claims.jti
+
+        def task_binding(claims):
+            return {
+                "task_uid": claims.task_uid,
+                "dataflow_uid": claims.dataflow_uid,
+                "deployment_id": claims.deployment_id,
+                "environment": claims.environment,
+                "workflow_version": claims.workflow_version,
+                "correlation_id": claims.correlation_id,
+                "node_id": claims.node_id,
+                "node_type": claims.node_type,
+                "data_source_uid": None,
+                "idempotency_key": "customer_id",
+            }
+
+        assert real_ledger.claim(
+            no_run_crash_jti,
+            task_binding(no_run_claims),
+            expires_at=no_run_claims.expires_at,
+        )
+        assert real_ledger.claim(
+            evidence_crash_jti,
+            task_binding(evidence_crash_claims),
+            expires_at=evidence_crash_claims.expires_at,
+        )
+        real_evidence_writer.start(
+            component_binding_id=component_binding_id,
+            rule_version_id=rule_id,
+            plan_hash=compiled["plan_hash"],
+            correlation_id=evidence_crash_correlation_id,
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            node_id=node["id"],
+            lease_owner=evidence_crash_jti,
+        )
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.runner_task_executions
+                    SET lease_expires_at =
+                        CURRENT_TIMESTAMP - INTERVAL '1 second'
+                    WHERE token_jti IN (
+                        CAST(:no_run_jti AS uuid),
+                        CAST(:evidence_jti AS uuid)
+                    )
+                    """
+                ),
+                {
+                    "no_run_jti": no_run_crash_jti,
+                    "evidence_jti": evidence_crash_jti,
+                },
+            )
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_runs
+                    SET lease_expires_at =
+                        CURRENT_TIMESTAMP - INTERVAL '1 second'
+                    WHERE lease_owner = CAST(:jti AS uuid)
+                    """
+                ),
+                {"jti": evidence_crash_jti},
+            )
         with runner_app.test_client() as client:
             http_result = client.post(
                 "/v1/tasks/execute",
@@ -847,6 +1043,22 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "parameters": {},
                 },
             )
+            no_run_crash = client.post(
+                "/v1/tasks/execute",
+                json={
+                    "task_token": no_run_crash_token,
+                    "node": node,
+                    "parameters": {},
+                },
+            )
+            evidence_crash = client.post(
+                "/v1/tasks/execute",
+                json={
+                    "task_token": evidence_crash_token,
+                    "node": node,
+                    "parameters": {},
+                },
+            )
         assert http_result.status_code == 200, http_result.get_json()
         assert http_result.get_json()["output_artifact"] == result[
             "artifact_ref"
@@ -861,6 +1073,43 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         assert retried.get_json()["output_artifact"] == result[
             "artifact_ref"
         ]
+        assert no_run_crash.status_code == 409
+        assert no_run_crash.get_json() == {
+            "error": "task execution outcome is unknown"
+        }
+        assert evidence_crash.status_code == 409
+        assert evidence_crash.get_json() == {
+            "error": "task execution outcome is unknown"
+        }
+        assert real_ledger.get(no_run_crash_jti).status == "unknown"
+        assert real_ledger.get(evidence_crash_jti).status == "unknown"
+        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": no_run_crash_correlation_id},
+            ).scalar_one() == 0
+            crash_evidence = connection.execute(
+                text(
+                    """
+                    SELECT status, commit_outcome
+                    FROM public.rule_runs
+                    WHERE correlation_id =
+                        CAST(:correlation_id AS uuid)
+                    """
+                ),
+                {"correlation_id": evidence_crash_correlation_id},
+            ).mappings().one()
+        assert dict(crash_evidence) == {
+            "status": "unknown",
+            "commit_outcome": "unknown",
+        }
         ledger_record = PostgresTaskLedger(platform).get(ledger_jti)
         assert ledger_record is not None
         assert ledger_record.status == "success"
@@ -1269,6 +1518,21 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 ),
                 {"id": sample_crash_run_id},
             ).scalar_one() == "ready"
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_violation_samples
+                    SET handoff_status = 'unknown'
+                    WHERE rule_run_id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": sample_crash_run_id},
+            )
+        sample_reconciliation = evidence_writer.reconcile_samples(
+            limit=10
+        )
+        assert sample_reconciliation["ready"] >= 1
         evidence_writer.finish(
             sample_crash_run_id,
             sample_crash_evidence,
@@ -1411,6 +1675,43 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 correlation_id=failed_receipt_correlation_id,
                 input_binding_id=sql_binding_id,
             )
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_sql_staging_receipts
+                    SET expires_at =
+                        CURRENT_TIMESTAMP - INTERVAL '1 second'
+                    WHERE correlation_id IN (
+                        CAST(:ready AS uuid),
+                        CAST(:failed AS uuid)
+                    )
+                    """
+                ),
+                {
+                    "ready": receipt_correlation_id,
+                    "failed": failed_receipt_correlation_id,
+                },
+            )
+        assert evidence_writer.cleanup_sql_staging(limit=10) == 2
+        with platform.connect() as connection:
+            assert connection.execute(
+                text(
+                    """
+                    SELECT COUNT(*)
+                    FROM public.rule_sql_staging_receipts
+                    WHERE correlation_id IN (
+                        CAST(:ready AS uuid),
+                        CAST(:failed AS uuid)
+                    )
+                      AND status = 'expired'
+                    """
+                ),
+                {
+                    "ready": receipt_correlation_id,
+                    "failed": failed_receipt_correlation_id,
+                },
+            ).scalar_one() == 2
         conflict_path = tmp_path / "conflict.parquet"
         pl.DataFrame(
             {
@@ -1499,6 +1800,18 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     ),
                     {"jti": retry_ledger_jti},
                 )
+            for crash_jti in (
+                no_run_crash_jti,
+                evidence_crash_jti,
+            ):
+                if crash_jti is not None:
+                    connection.execute(
+                        text(
+                            "DELETE FROM public.runner_task_executions "
+                            "WHERE token_jti = CAST(:jti AS uuid)"
+                        ),
+                        {"jti": crash_jti},
+                    )
             connection.execute(
                 text(
                     """
@@ -1545,7 +1858,9 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "CAST(:lease_correlation_id AS uuid), "
                     "CAST(:sample_crash_correlation_id AS uuid), "
                     "CAST(:receipt_correlation_id AS uuid), "
-                    "CAST(:failed_receipt_correlation_id AS uuid))"
+                    "CAST(:failed_receipt_correlation_id AS uuid), "
+                    "CAST(:no_run_crash_correlation_id AS uuid), "
+                    "CAST(:evidence_crash_correlation_id AS uuid))"
                 ),
                 {
                     "correlation_id": correlation_id,
@@ -1559,6 +1874,12 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "failed_receipt_correlation_id": (
                         failed_receipt_correlation_id
                     ),
+                    "no_run_crash_correlation_id": (
+                        no_run_crash_correlation_id
+                    ),
+                    "evidence_crash_correlation_id": (
+                        evidence_crash_correlation_id
+                    ),
                 },
             )
             connection.execute(
@@ -1611,6 +1932,35 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 ),
                 {"id": deployment_id},
             )
+            if gateway_candidate_id is not None:
+                connection.execute(
+                    text(
+                        "DELETE FROM public.workflow_canary_evidence "
+                        "WHERE workflow_version_id = CAST(:id AS uuid)"
+                    ),
+                    {"id": gateway_candidate_id},
+                )
+                connection.execute(
+                    text(
+                        "DELETE FROM public.workflow_schedules "
+                        "WHERE workflow_version_id = CAST(:id AS uuid)"
+                    ),
+                    {"id": gateway_candidate_id},
+                )
+                connection.execute(
+                    text(
+                        "DELETE FROM public.workflow_engine_bindings "
+                        "WHERE workflow_version_id = CAST(:id AS uuid)"
+                    ),
+                    {"id": gateway_candidate_id},
+                )
+                connection.execute(
+                    text(
+                        "DELETE FROM public.dataflow_workflow_versions "
+                        "WHERE id = CAST(:id AS uuid)"
+                    ),
+                    {"id": gateway_candidate_id},
+                )
             connection.execute(
                 text(
                     "DELETE FROM public.dataflow_versions "

tests/mcp/test_scheduling_gateway.py

@@ -28,6 +28,9 @@ class Plans:
         self.operation_claims = []
         self.completed_operations = []
         self.active_deployment = None
+        self.rule_deployment_id = (
+            "01900000-0000-7000-8000-000000000099"
+        )
 
     def create_candidate(self, record):
         result = {**record, "candidate_id": "candidate-1", "version_no": 1}
@@ -41,6 +44,22 @@ class Plans:
         self.deployments[candidate_id] = deployment
         return deployment
 
+    def ensure_rule_deployment_identity(
+        self,
+        candidate_id,
+        environment,
+        *,
+        status,
+    ):
+        assert environment == "test"
+        assert status in {"disabled", "canary"}
+        return {
+            "deployment_id": self.rule_deployment_id,
+            "workflow_version_id": candidate_id,
+            "environment": environment,
+            "status": status,
+        }
+
     def get_deployment(self, candidate_id):
         return self.deployments[candidate_id]
 
@@ -457,6 +476,11 @@ def test_canary_tokens_are_issued_inside_gateway_and_cannot_be_supplied_by_ai():
 
     assert result["status"] == "started"
     assert issuer.calls[0]["write_authorized"] is False
+    assert issuer.calls[0]["deployment_id"] == (
+        plans.rule_deployment_id
+    )
+    assert issuer.calls[0]["deployment_id"] != "candidate-1"
+    assert issuer.calls[0]["environment"] == "test"
     action_calls = [
         call[0]
         for call in engine.calls

tests/runner/test_api.py

@@ -259,3 +259,209 @@ def test_running_rule_without_terminal_evidence_returns_retryable_202():
     assert response.status_code == 202
     assert response.headers["Retry-After"] == "2"
     assert executor.calls == 0
+
+
+def test_running_ledger_without_rule_run_expires_to_unknown():
+    class StaleLedger(InMemoryTaskLedger):
+        def reconcile_running(self, jti, binding):
+            record = self.get(jti)
+            assert record.binding == binding
+            record.status = "unknown"
+            record.commit_outcome = "unknown"
+            return record
+
+    class MissingEvidenceExecutor(Executor):
+        def replay_task(self, **_context):
+            return {"state": "missing"}
+
+    issuer = TaskTokenIssuer("x" * 32, clock=lambda: 1_000)
+    verifier = TaskTokenVerifier("x" * 32, clock=lambda: 1_000)
+    token = _rule_token(issuer)
+    claims = verifier.verify(token, node=RULE_NODE)
+    ledger = StaleLedger()
+    binding = {
+        "task_uid": claims.task_uid,
+        "dataflow_uid": claims.dataflow_uid,
+        "deployment_id": claims.deployment_id,
+        "environment": claims.environment,
+        "workflow_version": claims.workflow_version,
+        "correlation_id": claims.correlation_id,
+        "node_id": claims.node_id,
+        "node_type": claims.node_type,
+        "data_source_uid": None,
+        "idempotency_key": RULE_NODE["idempotency"]["key"],
+    }
+    ledger.claim(claims.jti, binding, expires_at=claims.expires_at)
+    app = create_runner_app(
+        verifier=verifier,
+        ledger=ledger,
+        registry=NodeRegistry(
+            {"rule.apply": MissingEvidenceExecutor()}
+        ),
+    )
+
+    with app.test_client() as client:
+        response = client.post(
+            "/v1/tasks/execute",
+            json={
+                "task_token": token,
+                "node": RULE_NODE,
+                "parameters": {},
+            },
+        )
+
+    assert response.status_code == 409
+    assert response.get_json() == {
+        "error": "task execution outcome is unknown"
+    }
+    assert ledger.get(claims.jti).status == "unknown"
+
+
+def test_expired_rule_run_is_reconciled_and_ledger_finalized_unknown():
+    class ExpiredEvidenceExecutor(Executor):
+        def replay_task(self, **_context):
+            return {
+                "state": "terminal",
+                "status": "unknown",
+                "commit_outcome": "unknown",
+            }
+
+    issuer = TaskTokenIssuer("x" * 32, clock=lambda: 1_000)
+    verifier = TaskTokenVerifier("x" * 32, clock=lambda: 1_000)
+    token = _rule_token(issuer)
+    claims = verifier.verify(token, node=RULE_NODE)
+    ledger = InMemoryTaskLedger()
+    ledger.claim(
+        claims.jti,
+        {
+            "task_uid": claims.task_uid,
+            "dataflow_uid": claims.dataflow_uid,
+            "deployment_id": claims.deployment_id,
+            "environment": claims.environment,
+            "workflow_version": claims.workflow_version,
+            "correlation_id": claims.correlation_id,
+            "node_id": claims.node_id,
+            "node_type": claims.node_type,
+            "data_source_uid": None,
+            "idempotency_key": RULE_NODE["idempotency"]["key"],
+        },
+        expires_at=claims.expires_at,
+    )
+    app = create_runner_app(
+        verifier=verifier,
+        ledger=ledger,
+        registry=NodeRegistry(
+            {"rule.apply": ExpiredEvidenceExecutor()}
+        ),
+    )
+
+    with app.test_client() as client:
+        response = client.post(
+            "/v1/tasks/execute",
+            json={
+                "task_token": token,
+                "node": RULE_NODE,
+                "parameters": {},
+            },
+        )
+
+    assert response.status_code == 409
+    assert ledger.get(claims.jti).status == "unknown"
+
+
+def test_expired_token_is_replay_only_and_never_starts_execution():
+    now = [1_000]
+    issuer = TaskTokenIssuer(
+        "x" * 32,
+        clock=lambda: now[0],
+        ttl_seconds=10,
+    )
+    verifier = TaskTokenVerifier("x" * 32, clock=lambda: now[0])
+    token = _rule_token(issuer)
+    ledger = InMemoryTaskLedger()
+    executor = Executor()
+    app = create_runner_app(
+        verifier=verifier,
+        ledger=ledger,
+        registry=NodeRegistry({"rule.apply": executor}),
+    )
+    payload = {
+        "task_token": token,
+        "node": RULE_NODE,
+        "parameters": {},
+    }
+    with app.test_client() as client:
+        first = client.post("/v1/tasks/execute", json=payload)
+        now[0] = 1_011
+        replay = client.post("/v1/tasks/execute", json=payload)
+
+    assert first.status_code == replay.status_code == 200
+    assert replay.headers["X-Idempotent-Replay"] == "true"
+    assert executor.calls == 1
+
+    fresh_token = _rule_token(issuer)
+    now[0] = 1_022
+    with app.test_client() as client:
+        rejected = client.post(
+            "/v1/tasks/execute",
+            json={
+                "task_token": fresh_token,
+                "node": RULE_NODE,
+                "parameters": {},
+            },
+        )
+    assert rejected.status_code == 401
+    assert executor.calls == 1
+
+    now[0] = 1_030
+    running_token = _rule_token(issuer)
+    running_claims = verifier.verify(
+        running_token,
+        node=RULE_NODE,
+    )
+    ledger.claim(
+        running_claims.jti,
+        {
+            "task_uid": running_claims.task_uid,
+            "dataflow_uid": running_claims.dataflow_uid,
+            "deployment_id": running_claims.deployment_id,
+            "environment": running_claims.environment,
+            "workflow_version": running_claims.workflow_version,
+            "correlation_id": running_claims.correlation_id,
+            "node_id": running_claims.node_id,
+            "node_type": running_claims.node_type,
+            "data_source_uid": None,
+            "idempotency_key": RULE_NODE["idempotency"]["key"],
+        },
+        expires_at=running_claims.expires_at,
+    )
+    now[0] = 1_041
+    header, body, signature = running_token.split(".")
+    changed_signature = (
+        ("A" if signature[0] != "A" else "B") + signature[1:]
+    )
+    with app.test_client() as client:
+        running = client.post(
+            "/v1/tasks/execute",
+            json={
+                "task_token": running_token,
+                "node": RULE_NODE,
+                "parameters": {},
+            },
+        )
+        tampered = client.post(
+            "/v1/tasks/execute",
+            json={
+                "task_token": ".".join(
+                    (header, body, changed_signature)
+                ),
+                "node": RULE_NODE,
+                "parameters": {},
+            },
+        )
+    assert running.status_code == 401
+    assert running.get_json() == {
+        "error": "expired task is not replayable"
+    }
+    assert tampered.status_code == 401
+    assert executor.calls == 1

tests/runner/test_rule_evidence.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+import time
 
 import pytest
 
@@ -50,6 +51,8 @@ class Evidence:
         self.started = []
         self.finished = []
         self._replay = replay
+        self.heartbeat_interval_seconds = 0.01
+        self.heartbeats = 0
 
     def start(self, **values):
         self.started.append(values)
@@ -59,6 +62,7 @@ class Evidence:
         return self._replay
 
     def heartbeat(self, _rule_run_id, _lease_owner):
+        self.heartbeats += 1
         return None
 
     def finish(self, rule_run_id, result):
@@ -286,3 +290,95 @@ def test_public_result_contract_is_backend_specific():
         },
         backend="sql_pushdown",
     )["rows_out"] == 1
+
+
+def test_rule_executor_renews_lease_while_adapter_is_running():
+    from app.runner.rules import RulePlanExecutor
+
+    class SlowAdapter(Adapter):
+        def execute(self, **_kwargs):
+            time.sleep(0.04)
+            return {
+                "rows_in": 1,
+                "rows_out": 1,
+                "rows_rejected": 0,
+                "rows_quarantined": 0,
+                "commit_outcome": "committed",
+            }
+
+    node = _node()
+    evidence = Evidence()
+    result = RulePlanExecutor(
+        Repository(node),
+        adapters={"quality_check": SlowAdapter()},
+        evidence_writer=evidence,
+    ).execute(node, {}, **_context())
+
+    assert result["rows_out"] == 1
+    assert evidence.heartbeats >= 2
+
+
+def test_heartbeat_failure_after_committed_adapter_is_unknown_committed():
+    from app.runner.rules import RulePlanExecutor
+
+    class FailingHeartbeatEvidence(Evidence):
+        def heartbeat(self, _rule_run_id, _lease_owner):
+            self.heartbeats += 1
+            if self.heartbeats > 1:
+                raise RuntimeError("postgres unavailable")
+
+    class SlowCommittedAdapter(Adapter):
+        def execute(self, **_kwargs):
+            time.sleep(0.04)
+            return {
+                "rows_in": 1,
+                "rows_out": 1,
+                "rows_rejected": 0,
+                "rows_quarantined": 0,
+                "commit_outcome": "committed",
+            }
+
+    node = _node()
+    evidence = FailingHeartbeatEvidence()
+    executor = RulePlanExecutor(
+        Repository(node),
+        adapters={"quality_check": SlowCommittedAdapter()},
+        evidence_writer=evidence,
+    )
+
+    with pytest.raises(NodeExecutionError) as error:
+        executor.execute(node, {}, **_context())
+
+    assert error.value.commit_outcome == "unknown"
+    assert evidence.finished[-1][1]["status"] == "unknown"
+    assert evidence.finished[-1][1]["commit_outcome"] == "committed"
+
+
+def test_committed_adapter_with_invalid_public_result_is_unknown_committed():
+    from app.runner.rules import RulePlanExecutor
+
+    node = _node()
+    evidence = Evidence()
+    executor = RulePlanExecutor(
+        Repository(node),
+        adapters={
+            "quality_check": Adapter(
+                {
+                    "rows_in": 1,
+                    "rows_out": 1,
+                    "rows_rejected": 0,
+                    "rows_quarantined": 0,
+                    "commit_outcome": "committed",
+                    "raw_records": [{"secret": "value"}],
+                }
+            )
+        },
+        evidence_writer=evidence,
+    )
+
+    with pytest.raises(NodeExecutionError) as error:
+        executor.execute(node, {}, **_context())
+
+    assert error.value.commit_outcome == "unknown"
+    assert evidence.finished[-1][1]["status"] == "unknown"
+    assert evidence.finished[-1][1]["commit_outcome"] == "committed"

tests/runner/test_task_tokens.py

@@ -75,6 +75,12 @@ def test_task_token_rejects_expiry_tampering_and_other_node():
     now[0] = 1_011
     with pytest.raises(TaskTokenExpired):
         verifier.verify(token, node=NODE)
+    replay_claims = verifier.verify_for_replay(token, node=NODE)
+    assert replay_claims.jti
+
+    now[0] = 1_912
+    with pytest.raises(TaskTokenExpired, match="replay window"):
+        verifier.verify_for_replay(token, node=NODE)
 
 
 def test_task_token_secret_has_a_minimum_strength():

tests/test_data_rule_schema.py

@@ -48,6 +48,12 @@ RULE_ATTEMPT_MIGRATION = (
     / "versions"
     / "20260723_170_rule_execution_attempts.py"
 )
+RULE_RECONCILIATION_MIGRATION = (
+    ROOT
+    / "migrations"
+    / "versions"
+    / "20260723_180_rule_reconciliation.py"
+)
 
 EXPECTED_TABLES = {
     "data_rules",
@@ -239,3 +245,20 @@ def test_rule_attempt_migration_adds_exact_identity_replay_lease_and_receipts():
     ):
         assert expected in source
     assert "DROP TABLE" not in source.split("def downgrade()", 1)[1].upper()
+
+
+def test_rule_reconciliation_migration_extends_170_without_rewriting_it():
+    source = RULE_RECONCILIATION_MIGRATION.read_text(encoding="utf-8")
+
+    assert 'revision = "20260723_180"' in source
+    assert 'down_revision = "20260723_170"' in source
+    for expected in (
+        "runner_task_executions",
+        "lease_expires_at",
+        "'unknown','expired'",
+        "cleanup_claim UUID",
+        "schema_fields JSONB",
+        "rule_sql_staging_receipts",
+    ):
+        assert expected in source
+    assert "forward-only" in source