fix: close rule replay and cleanup leases

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

@@ -46,10 +46,11 @@ The implementation includes:
 
 ## Forward-only schema change
 
-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.
+Migrations `20260723_160`, `20260723_170`, `20260723_180`, and
+`20260723_190` 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, and the third review leaves migration 180 unchanged.
+Each additional schema revision is isolated in its own forward-only migration.
 
 `rule_runs` gains:
 
@@ -74,7 +75,9 @@ 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. Migration 180 adds the Runner ledger lease,
 bounded cleanup claims, retained sample schema fields, and the
-`unknown`/`expired` reconciliation states.
+`unknown`/`expired` reconciliation states. Migration 190 adds expiring cleanup
+claim leases for violation samples and SQL staging receipts, allowing a crashed
+cleanup worker to be taken over through compare-and-set after lease expiry.
 
 The local Docker PostgreSQL was upgraded through the formal
 `20260723_150 -> head` Alembic path. Two isolated real PostgreSQL tests prove
@@ -158,6 +161,49 @@ Second-review verification:
 The full-repository Ruff invocation still reports 430 pre-existing findings
 outside this change set. No changed file contributes a Ruff finding.
 
+## Third-review closed replay and cleanup takeover
+
+The third review makes terminal evidence and cleanup ownership explicit:
+
+- Runner recovery accepts only three exact replay shapes: `running`,
+  `missing`, or a fully closed `terminal` response. A terminal success must
+  carry a matching canonical result digest, evidence digest, status, and commit
+  outcome before the durable ledger may be finalized;
+- a genuinely running execution always returns HTTP 202 and never stores a
+  success response in the ledger. An expired replay-only token pointing to
+  running evidence is rejected with HTTP 401;
+- terminal replay responses returned by the PostgreSQL evidence writer include
+  both the canonical result digest and the persisted evidence digest, so the
+  API does not infer success from an arbitrary dictionary;
+- violation-sample and SQL-receipt cleanup claims now have explicit expiry.
+  Workers claim rows with compare-and-set semantics, cannot steal an active
+  claim, and can take over a claim left behind by a crashed worker after its
+  lease expires;
+- violation sample reconciliation checks object existence through the storage
+  adapter before promotion. A confirmed missing object becomes failed and is
+  eligible for cleanup, while transient storage/network errors retain the
+  original pending or unknown state and release the claim for retry;
+- object-store operations remain outside database transactions, and every
+  success, failure, and retry path clears both the claim owner and expiry.
+
+Third-review verification:
+
+- focused Runner, evidence, token, ledger, schema, and real integration suite:
+  `41 passed`;
+- real PostgreSQL/MySQL/MinIO integration acceptance, including live running
+  replay, closed terminal digest recovery, cleanup crash/takeover, confirmed
+  missing object, and transient storage failure: `1 passed`;
+- full suite: `605 passed, 28 skipped, 59 subtests passed`;
+- Ruff across every file changed by the third review:
+  `All checks passed!`;
+- `git diff --check`: passed;
+- migration 180 diff against the second-review commit: empty;
+- real local PostgreSQL upgrade: `20260723_180 -> 20260723_190 (head)`;
+- rebuilt local Docker backend and Runner: both healthy;
+- Runner `/health`: `{"status":"ok"}`;
+- Alembic inside the rebuilt backend container:
+  `20260723_190 (head)`.
+
 ## Real PostgreSQL and MinIO acceptance
 
 The production-path integration uses:
@@ -196,11 +242,13 @@ It proves:
    correlation, and failed-producer references;
 10. same-status/different-content finish retries are rejected by the exact
     evidence digest;
-11. real MinIO reconciliation retains pending/ready violation evidence while
-    removing an unreferenced old object;
-12. a failed adapter records `failed/not_committed`;
-13. an uncertain write records `unknown/unknown`;
-14. all test-owned PostgreSQL rows and MinIO objects are removed after
+11. real MinIO reconciliation prevents active-claim theft, takes over an
+    expired claim, marks confirmed-missing evidence failed, preserves transient
+    failures for retry, and removes an unreferenced old object;
+12. cleanup claim takeover also applies to opaque SQL staging receipts;
+13. a failed adapter records `failed/not_committed`;
+14. an uncertain write records `unknown/unknown`;
+15. all test-owned PostgreSQL rows and MinIO objects are removed after
    acceptance.
 
 ## Residual boundary

app/runner/api.py

@@ -2,6 +2,7 @@
 
 import hashlib
 import json
+import re
 
 from flask import Flask, jsonify, request
 
@@ -105,11 +106,31 @@ def create_runner_app(*, verifier, ledger, registry):
                     if isinstance(recovered, dict)
                     else None
                 )
+                if isinstance(recovered, dict):
+                    expected_fields = {
+                        "running": {"state"},
+                        "missing": {"state"},
+                        "terminal": {
+                            "state",
+                            "status",
+                            "commit_outcome",
+                            "result",
+                            "result_digest",
+                            "evidence_digest",
+                        },
+                    }.get(recovered_state)
+                    if (
+                        expected_fields is None
+                        or set(recovered) != expected_fields
+                    ):
+                        return jsonify(
+                            {"error": "task ledger is unavailable"}
+                        ), 503
                 recovered_result = (
                     recovered.get("result")
                     if recovered_state == "terminal"
                     and recovered.get("status") == "success"
-                    else recovered
+                    else None
                 )
                 if (
                     recovered_state == "terminal"
@@ -136,10 +157,44 @@ def create_runner_app(*, verifier, ledger, registry):
                             )
                         }
                     ), 409
-                if (
-                    isinstance(recovered_result, dict)
-                    and recovered_state != "missing"
-                ):
+                if recovered_state == "terminal":
+                    result_digest = str(
+                        recovered.get("result_digest") or ""
+                    )
+                    evidence_digest = str(
+                        recovered.get("evidence_digest") or ""
+                    )
+                    if (
+                        not isinstance(recovered_result, dict)
+                        or recovered.get("commit_outcome")
+                        != recovered_result.get("commit_outcome")
+                        or re.fullmatch(
+                            r"[0-9a-f]{64}",
+                            result_digest,
+                        )
+                        is None
+                        or re.fullmatch(
+                            r"[0-9a-f]{64}",
+                            evidence_digest,
+                        )
+                        is None
+                    ):
+                        return jsonify(
+                            {"error": "task ledger is unavailable"}
+                        ), 503
+                    encoded_result = json.dumps(
+                        recovered_result,
+                        sort_keys=True,
+                        separators=(",", ":"),
+                        ensure_ascii=False,
+                    ).encode("utf-8")
+                    if (
+                        hashlib.sha256(encoded_result).hexdigest()
+                        != result_digest
+                    ):
+                        return jsonify(
+                            {"error": "task ledger is unavailable"}
+                        ), 503
                     replay_body = {
                         "task_uid": claims.task_uid,
                         "correlation_id": claims.correlation_id,
@@ -188,6 +243,10 @@ def create_runner_app(*, verifier, ledger, registry):
                                 )
                             }
                         ), 409
+                elif recovered_state != "running":
+                    return jsonify(
+                        {"error": "task ledger is unavailable"}
+                    ), 503
                 if replay_only:
                     return jsonify(
                         {"error": "expired task is not replayable"}

app/runner/rule_evidence.py

@@ -216,11 +216,13 @@ class PostgresRuleEvidenceWriter:
         *,
         sample_ttl_seconds: int = 3600,
         lease_seconds: int = 300,
+        cleanup_claim_seconds: int = 300,
     ):
         self.engine = engine
         self.artifact_store = artifact_store
         self.sample_ttl_seconds = int(sample_ttl_seconds)
         self.lease_seconds = int(lease_seconds)
+        self.cleanup_claim_seconds = int(cleanup_claim_seconds)
         if (
             self.sample_ttl_seconds < 1
             or self.sample_ttl_seconds
@@ -229,6 +231,11 @@ 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")
+        if (
+            self.cleanup_claim_seconds < 30
+            or self.cleanup_claim_seconds > 3600
+        ):
+            raise ValueError("rule cleanup lease is invalid")
         self.heartbeat_interval_seconds = max(
             5.0,
             min(60.0, self.lease_seconds / 3),
@@ -721,7 +728,8 @@ class PostgresRuleEvidenceWriter:
             row = connection.execute(
                 text(
                     """
-                    SELECT id::text, status, lease_expires_at
+                    SELECT id::text, status, lease_expires_at,
+                           evidence_digest
                     FROM public.rule_runs
                     WHERE lease_owner = CAST(:lease_owner AS uuid)
                       AND deployment_id =
@@ -787,15 +795,18 @@ class PostgresRuleEvidenceWriter:
                 return {"state": "running"}
             run_id = str(row["id"])
         replay = self.replay(run_id)
+        result = {
+            key: value
+            for key, value in replay.items()
+            if key != "status"
+        }
         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"
-            },
+            "result": result,
+            "result_digest": _canonical_digest(result),
+            "evidence_digest": str(row["evidence_digest"] or ""),
         }
 
     @staticmethod
@@ -1271,7 +1282,7 @@ class PostgresRuleEvidenceWriter:
             raise ValueError("cleanup limit is invalid")
         claim = new_governance_uid()
         with self.engine.begin() as connection:
-            rows = connection.execute(
+            candidates = connection.execute(
                 text(
                     """
                     SELECT id::text, artifact_ref, artifact_digest,
@@ -1279,7 +1290,10 @@ class PostgresRuleEvidenceWriter:
                            handoff_status,
                            expires_at <= CURRENT_TIMESTAMP AS expired
                     FROM public.rule_violation_samples
-                    WHERE cleanup_claim IS NULL
+                    WHERE (
+                        cleanup_claim IS NULL
+                        OR cleanup_claim_expires_at <= CURRENT_TIMESTAMP
+                    )
                       AND (
                           handoff_status IN ('pending','unknown')
                           OR (
@@ -1296,19 +1310,35 @@ class PostgresRuleEvidenceWriter:
                 ),
                 {"limit": limit},
             ).mappings().all()
-            for row in rows:
-                connection.execute(
+            rows = []
+            for row in candidates:
+                updated = connection.execute(
                     text(
                         """
                         UPDATE public.rule_violation_samples
                         SET cleanup_claim = CAST(:claim AS uuid),
+                            cleanup_claim_expires_at =
+                                CURRENT_TIMESTAMP
+                                + make_interval(
+                                    secs => :claim_seconds
+                                ),
                             updated_at = CURRENT_TIMESTAMP
                         WHERE id = CAST(:id AS uuid)
-                          AND cleanup_claim IS NULL
+                          AND (
+                              cleanup_claim IS NULL
+                              OR cleanup_claim_expires_at
+                                  <= CURRENT_TIMESTAMP
+                          )
                         """
                     ),
-                    {"id": str(row["id"]), "claim": claim},
+                    {
+                        "id": str(row["id"]),
+                        "claim": claim,
+                        "claim_seconds": self.cleanup_claim_seconds,
+                    },
                 )
+                if int(updated.rowcount or 0) == 1:
+                    rows.append(row)
         metrics = {
             "claimed": len(rows),
             "ready": 0,
@@ -1327,6 +1357,7 @@ class PostgresRuleEvidenceWriter:
                                 """
                                 UPDATE public.rule_violation_samples
                                 SET cleanup_claim = NULL,
+                                    cleanup_claim_expires_at = NULL,
                                     updated_at = CURRENT_TIMESTAMP
                                 WHERE id = CAST(:id AS uuid)
                                   AND cleanup_claim =
@@ -1361,6 +1392,23 @@ class PostgresRuleEvidenceWriter:
                     raise ValueError(
                         "violation sample schema is unavailable"
                     )
+                described = self.artifact_store.describe_optional(
+                    str(row["artifact_ref"])
+                )
+                if described is None:
+                    raise ValueError(
+                        "violation sample object is missing"
+                    )
+                if (
+                    described["digest"]
+                    != str(row["artifact_digest"])
+                    or described["schema_hash"] != row["schema_hash"]
+                    or described["row_count"]
+                    != int(row["sample_count"])
+                ):
+                    raise ValueError(
+                        "violation sample attestation does not match"
+                    )
                 with self.artifact_store.stage(
                     str(row["artifact_ref"]),
                     str(row["artifact_digest"]),
@@ -1381,17 +1429,6 @@ class PostgresRuleEvidenceWriter:
                     },
                 ):
                     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:
@@ -1408,6 +1445,7 @@ class PostgresRuleEvidenceWriter:
                         SET handoff_status = :handoff_status,
                             failure_code = :failure_code,
                             cleanup_claim = NULL,
+                            cleanup_claim_expires_at = NULL,
                             updated_at = CURRENT_TIMESTAMP
                         WHERE id = CAST(:id AS uuid)
                           AND cleanup_claim = CAST(:claim AS uuid)
@@ -1432,14 +1470,18 @@ class PostgresRuleEvidenceWriter:
             raise ValueError("cleanup limit is invalid")
         claim = new_governance_uid()
         with self.engine.begin() as connection:
-            rows = connection.execute(
+            candidates = 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
+                      AND (
+                          cleanup_claim IS NULL
+                          OR cleanup_claim_expires_at
+                              <= CURRENT_TIMESTAMP
+                      )
                     ORDER BY expires_at, id
                     FOR UPDATE SKIP LOCKED
                     LIMIT :limit
@@ -1447,19 +1489,35 @@ class PostgresRuleEvidenceWriter:
                 ),
                 {"limit": limit},
             ).mappings().all()
-            for row in rows:
-                connection.execute(
+            rows = []
+            for row in candidates:
+                updated = connection.execute(
                     text(
                         """
                         UPDATE public.rule_sql_staging_receipts
                         SET cleanup_claim = CAST(:claim AS uuid),
+                            cleanup_claim_expires_at =
+                                CURRENT_TIMESTAMP
+                                + make_interval(
+                                    secs => :claim_seconds
+                                ),
                             updated_at = CURRENT_TIMESTAMP
                         WHERE id = CAST(:id AS uuid)
-                          AND cleanup_claim IS NULL
+                          AND (
+                              cleanup_claim IS NULL
+                              OR cleanup_claim_expires_at
+                                  <= CURRENT_TIMESTAMP
+                          )
                         """
                     ),
-                    {"id": str(row["id"]), "claim": claim},
+                    {
+                        "id": str(row["id"]),
+                        "claim": claim,
+                        "claim_seconds": self.cleanup_claim_seconds,
+                    },
                 )
+                if int(updated.rowcount or 0) == 1:
+                    rows.append(row)
         finalized = 0
         for row in rows:
             with self.engine.begin() as connection:
@@ -1469,6 +1527,7 @@ class PostgresRuleEvidenceWriter:
                         UPDATE public.rule_sql_staging_receipts
                         SET status = 'expired',
                             cleanup_claim = NULL,
+                            cleanup_claim_expires_at = NULL,
                             updated_at = CURRENT_TIMESTAMP
                         WHERE id = CAST(:id AS uuid)
                           AND cleanup_claim = CAST(:claim AS uuid)

migrations/versions/20260723_190_rule_cleanup_leases.py

@@ -0,0 +1,34 @@
+"""Add expiring claims for governed-rule cleanup workers."""
+
+from alembic import op
+
+revision = "20260723_190"
+down_revision = "20260723_180"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE public.rule_violation_samples
+            ADD COLUMN cleanup_claim_expires_at TIMESTAMPTZ;
+
+        CREATE INDEX idx_rule_violation_sample_cleanup_lease
+            ON public.rule_violation_samples(
+                cleanup_claim_expires_at, expires_at, id
+            );
+
+        ALTER TABLE public.rule_sql_staging_receipts
+            ADD COLUMN cleanup_claim_expires_at TIMESTAMPTZ;
+
+        CREATE INDEX idx_rule_sql_receipt_cleanup_lease
+            ON public.rule_sql_staging_receipts(
+                cleanup_claim_expires_at, expires_at, id
+            );
+        """
+    )
+
+
+def downgrade() -> None:
+    raise RuntimeError("rule cleanup leases are forward-only")

tests/integration/test_data_rule_polars_execution.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import re
 from datetime import UTC, datetime, timedelta
@@ -119,6 +120,7 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
     failed_receipt_correlation_id = new_governance_uid()
     no_run_crash_correlation_id = new_governance_uid()
     evidence_crash_correlation_id = new_governance_uid()
+    active_retry_correlation_id = new_governance_uid()
     sql_binding_id = new_governance_uid()
     prefix = f"rules/{correlation_id}/"
     customer_table = "task5_polars_customers"
@@ -138,6 +140,7 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
     retry_ledger_jti = None
     no_run_crash_jti = None
     evidence_crash_jti = None
+    active_retry_jti = None
     gateway_candidate_id = None
 
     try:
@@ -952,6 +955,21 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             node=node,
         )
         evidence_crash_jti = evidence_crash_claims.jti
+        active_retry_token = TaskTokenIssuer(task_secret).issue(
+            task_uid=new_governance_uid(),
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            correlation_id=active_retry_correlation_id,
+            node=node,
+            write_authorized=True,
+        )
+        active_retry_claims = verifier.verify(
+            active_retry_token,
+            node=node,
+        )
+        active_retry_jti = active_retry_claims.jti
 
         def task_binding(claims):
             return {
@@ -977,6 +995,11 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             task_binding(evidence_crash_claims),
             expires_at=evidence_crash_claims.expires_at,
         )
+        assert real_ledger.claim(
+            active_retry_jti,
+            task_binding(active_retry_claims),
+            expires_at=active_retry_claims.expires_at,
+        )
         real_evidence_writer.start(
             component_binding_id=component_binding_id,
             rule_version_id=rule_id,
@@ -989,6 +1012,18 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             node_id=node["id"],
             lease_owner=evidence_crash_jti,
         )
+        real_evidence_writer.start(
+            component_binding_id=component_binding_id,
+            rule_version_id=rule_id,
+            plan_hash=compiled["plan_hash"],
+            correlation_id=active_retry_correlation_id,
+            dataflow_uid=dataflow_uid,
+            deployment_id=deployment_id,
+            environment="test",
+            workflow_version=1,
+            node_id=node["id"],
+            lease_owner=active_retry_jti,
+        )
         with platform.begin() as connection:
             connection.execute(
                 text(
@@ -1059,6 +1094,14 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "parameters": {},
                 },
             )
+            active_retry = client.post(
+                "/v1/tasks/execute",
+                json={
+                    "task_token": active_retry_token,
+                    "node": node,
+                    "parameters": {},
+                },
+            )
         assert http_result.status_code == 200, http_result.get_json()
         assert http_result.get_json()["output_artifact"] == result[
             "artifact_ref"
@@ -1081,6 +1124,10 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         assert evidence_crash.get_json() == {
             "error": "task execution outcome is unknown"
         }
+        assert active_retry.status_code == 202
+        assert active_retry.headers["Retry-After"] == "2"
+        assert real_ledger.get(active_retry_jti).status == "running"
+        assert real_ledger.get(active_retry_jti).replay_body is None
         assert real_ledger.get(no_run_crash_jti).status == "unknown"
         assert real_ledger.get(evidence_crash_jti).status == "unknown"
         with platform.connect() as connection:
@@ -1110,6 +1157,28 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             "status": "unknown",
             "commit_outcome": "unknown",
         }
+        terminal_replay = real_evidence_writer.reconcile_expired_lease(
+            lease_owner=ledger_jti,
+            deployment_id=deployment_id,
+            correlation_id=correlation_id,
+            component_binding_id=component_binding_id,
+            rule_version_id=rule_id,
+            plan_hash=compiled["plan_hash"],
+        )
+        assert terminal_replay["state"] == "terminal"
+        assert terminal_replay["status"] == "success"
+        assert terminal_replay["result_digest"] == hashlib.sha256(
+            json.dumps(
+                terminal_replay["result"],
+                sort_keys=True,
+                separators=(",", ":"),
+                ensure_ascii=False,
+            ).encode("utf-8")
+        ).hexdigest()
+        assert re.fullmatch(
+            r"[0-9a-f]{64}",
+            terminal_replay["evidence_digest"],
+        )
         ledger_record = PostgresTaskLedger(platform).get(ledger_jti)
         assert ledger_record is not None
         assert ledger_record.status == "success"
@@ -1523,7 +1592,29 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 text(
                     """
                     UPDATE public.rule_violation_samples
-                    SET handoff_status = 'unknown'
+                    SET handoff_status = 'unknown',
+                        cleanup_claim = CAST(:cleanup_claim AS uuid),
+                        cleanup_claim_expires_at =
+                            CURRENT_TIMESTAMP + INTERVAL '5 minutes'
+                    WHERE rule_run_id = CAST(:id AS uuid)
+                    """
+                ),
+                {
+                    "id": sample_crash_run_id,
+                    "cleanup_claim": new_governance_uid(),
+                },
+            )
+        before_claim_expiry = evidence_writer.reconcile_samples(
+            limit=10
+        )
+        assert before_claim_expiry["claimed"] == 0
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_violation_samples
+                    SET cleanup_claim_expires_at =
+                        CURRENT_TIMESTAMP - INTERVAL '1 second'
                     WHERE rule_run_id = CAST(:id AS uuid)
                     """
                 ),
@@ -1532,6 +1623,7 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         sample_reconciliation = evidence_writer.reconcile_samples(
             limit=10
         )
+        assert sample_reconciliation["claimed"] == 1
         assert sample_reconciliation["ready"] >= 1
         evidence_writer.finish(
             sample_crash_run_id,
@@ -1540,6 +1632,102 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
         assert evidence_writer.replay(sample_crash_run_id)[
             "status"
         ] == "success"
+        with platform.connect() as connection:
+            missing_sample_ref = connection.execute(
+                text(
+                    """
+                    SELECT artifact_ref
+                    FROM public.rule_violation_samples
+                    WHERE rule_run_id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": sample_crash_run_id},
+            ).scalar_one()
+        store.delete(missing_sample_ref)
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_violation_samples
+                    SET handoff_status = 'unknown',
+                        cleanup_claim = NULL,
+                        cleanup_claim_expires_at = NULL
+                    WHERE rule_run_id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": sample_crash_run_id},
+            )
+        missing_reconciliation = evidence_writer.reconcile_samples(
+            limit=10
+        )
+        assert missing_reconciliation["failed"] == 1
+        with platform.connect() as connection:
+            assert connection.execute(
+                text(
+                    """
+                    SELECT handoff_status
+                    FROM public.rule_violation_samples
+                    WHERE rule_run_id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": sample_crash_run_id},
+            ).scalar_one() == "failed"
+
+        class TransientArtifactStore:
+            def __init__(self, wrapped):
+                self.wrapped = wrapped
+
+            def __getattr__(self, name):
+                return getattr(self.wrapped, name)
+
+            def describe_optional(self, _ref):
+                raise TimeoutError("temporary MinIO timeout")
+
+        with platform.begin() as connection:
+            transient_sample_id = connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_violation_samples s
+                    SET handoff_status = 'unknown',
+                        cleanup_claim = NULL,
+                        cleanup_claim_expires_at = NULL
+                    FROM public.rule_runs r
+                    WHERE s.rule_run_id = r.id
+                      AND r.correlation_id =
+                          CAST(:correlation_id AS uuid)
+                    RETURNING s.id::text
+                    """
+                ),
+                {"correlation_id": correlation_id},
+            ).scalar_one()
+        transient_writer = PostgresRuleEvidenceWriter(
+            platform,
+            TransientArtifactStore(store),
+            sample_ttl_seconds=900,
+        )
+        transient_result = transient_writer.reconcile_samples(
+            limit=10
+        )
+        assert transient_result["claimed"] == 1
+        assert transient_result["ready"] == 0
+        assert transient_result["failed"] == 0
+        with platform.connect() as connection:
+            transient_state = connection.execute(
+                text(
+                    """
+                    SELECT handoff_status,
+                           cleanup_claim,
+                           cleanup_claim_expires_at
+                    FROM public.rule_violation_samples
+                    WHERE id = CAST(:id AS uuid)
+                    """
+                ),
+                {"id": transient_sample_id},
+            ).mappings().one()
+        assert transient_state["handoff_status"] == "unknown"
+        assert transient_state["cleanup_claim"] is None
+        assert transient_state["cleanup_claim_expires_at"] is None
+        assert evidence_writer.reconcile_samples(limit=10)["ready"] == 1
 
         with platform.begin() as connection:
             connection.execute(
@@ -1681,7 +1869,20 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     """
                     UPDATE public.rule_sql_staging_receipts
                     SET expires_at =
-                        CURRENT_TIMESTAMP - INTERVAL '1 second'
+                        CURRENT_TIMESTAMP - INTERVAL '1 second',
+                        cleanup_claim = CASE
+                            WHEN correlation_id =
+                                CAST(:ready AS uuid)
+                            THEN CAST(:cleanup_claim AS uuid)
+                            ELSE NULL
+                        END,
+                        cleanup_claim_expires_at = CASE
+                            WHEN correlation_id =
+                                CAST(:ready AS uuid)
+                            THEN CURRENT_TIMESTAMP
+                                + INTERVAL '5 minutes'
+                            ELSE NULL
+                        END
                     WHERE correlation_id IN (
                         CAST(:ready AS uuid),
                         CAST(:failed AS uuid)
@@ -1691,9 +1892,34 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                 {
                     "ready": receipt_correlation_id,
                     "failed": failed_receipt_correlation_id,
+                    "cleanup_claim": new_governance_uid(),
                 },
             )
-        assert evidence_writer.cleanup_sql_staging(limit=10) == 2
+        assert evidence_writer.cleanup_sql_staging(limit=10) == 1
+        with platform.connect() as connection:
+            assert connection.execute(
+                text(
+                    """
+                    SELECT status
+                    FROM public.rule_sql_staging_receipts
+                    WHERE correlation_id = CAST(:ready AS uuid)
+                    """
+                ),
+                {"ready": receipt_correlation_id},
+            ).scalar_one() == "ready"
+        with platform.begin() as connection:
+            connection.execute(
+                text(
+                    """
+                    UPDATE public.rule_sql_staging_receipts
+                    SET cleanup_claim_expires_at =
+                        CURRENT_TIMESTAMP - INTERVAL '1 second'
+                    WHERE correlation_id = CAST(:ready AS uuid)
+                    """
+                ),
+                {"ready": receipt_correlation_id},
+            )
+        assert evidence_writer.cleanup_sql_staging(limit=10) == 1
         with platform.connect() as connection:
             assert connection.execute(
                 text(
@@ -1803,6 +2029,7 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
             for crash_jti in (
                 no_run_crash_jti,
                 evidence_crash_jti,
+                active_retry_jti,
             ):
                 if crash_jti is not None:
                     connection.execute(
@@ -1860,7 +2087,8 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "CAST(: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))"
+                    "CAST(:evidence_crash_correlation_id AS uuid), "
+                    "CAST(:active_retry_correlation_id AS uuid))"
                 ),
                 {
                     "correlation_id": correlation_id,
@@ -1880,6 +2108,9 @@ def test_real_postgres_mysql_minio_polars_cross_source_execution(tmp_path):
                     "evidence_crash_correlation_id": (
                         evidence_crash_correlation_id
                     ),
+                    "active_retry_correlation_id": (
+                        active_retry_correlation_id
+                    ),
                 },
             )
             connection.execute(

tests/runner/test_api.py

@@ -1,3 +1,6 @@
+import hashlib
+import json
+
 from app.runner.api import create_runner_app
 from app.runner.auth import TaskTokenIssuer, TaskTokenVerifier
 from app.runner.ledger import InMemoryTaskLedger
@@ -42,6 +45,16 @@ class Executor:
         }
 
 
+def _digest(value):
+    encoded = json.dumps(
+        value,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
 def test_runner_accepts_one_signed_task_and_rejects_replay():
     issuer = TaskTokenIssuer("x" * 32, clock=lambda: 1_000)
     verifier = TaskTokenVerifier("x" * 32, clock=lambda: 1_000)
@@ -182,10 +195,19 @@ def test_governed_rule_recovers_terminal_evidence_after_response_loss():
 
     class RecoverableExecutor(Executor):
         def replay_task(self, **_context):
-            return {
+            result = {
                 "node": RULE_NODE["id"],
                 "parameters": {},
                 "rows_out": 9,
+                "commit_outcome": "committed",
+            }
+            return {
+                "state": "terminal",
+                "status": "success",
+                "commit_outcome": "committed",
+                "result": result,
+                "result_digest": _digest(result),
+                "evidence_digest": "b" * 64,
             }
 
     issuer = TaskTokenIssuer("x" * 32, clock=lambda: 1_000)
@@ -261,6 +283,169 @@ def test_running_rule_without_terminal_evidence_returns_retryable_202():
     assert executor.calls == 0
 
 
+def test_explicit_running_evidence_never_finalizes_ledger_success():
+    class RunningExecutor(Executor):
+        def replay_task(self, **_context):
+            return {"state": "running"}
+
+    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()
+    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)
+    executor = RunningExecutor()
+    app = create_runner_app(
+        verifier=verifier,
+        ledger=ledger,
+        registry=NodeRegistry({"rule.apply": executor}),
+    )
+
+    with app.test_client() as client:
+        response = client.post(
+            "/v1/tasks/execute",
+            json={
+                "task_token": token,
+                "node": RULE_NODE,
+                "parameters": {},
+            },
+        )
+
+    assert response.status_code == 202
+    assert ledger.get(claims.jti).status == "running"
+    assert ledger.get(claims.jti).replay_body is None
+    assert executor.calls == 0
+
+
+def test_terminal_success_requires_matching_result_digest():
+    class MismatchedExecutor(Executor):
+        def replay_task(self, **_context):
+            result = {
+                "rows_out": 9,
+                "commit_outcome": "committed",
+            }
+            return {
+                "state": "terminal",
+                "status": "success",
+                "commit_outcome": "committed",
+                "result": result,
+                "result_digest": "0" * 64,
+                "evidence_digest": "b" * 64,
+            }
+
+    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()
+    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": MismatchedExecutor()}
+        ),
+    )
+
+    with app.test_client() as client:
+        response = client.post(
+            "/v1/tasks/execute",
+            json={
+                "task_token": token,
+                "node": RULE_NODE,
+                "parameters": {},
+            },
+        )
+
+    assert response.status_code == 503
+    assert ledger.get(claims.jti).status == "running"
+    assert ledger.get(claims.jti).replay_body is None
+
+
+def test_terminal_replay_rejects_unclosed_state_fields():
+    class UnclosedExecutor(Executor):
+        def replay_task(self, **_context):
+            result = {
+                "rows_out": 9,
+                "commit_outcome": "committed",
+            }
+            return {
+                "state": "terminal",
+                "status": "success",
+                "commit_outcome": "committed",
+                "result": result,
+                "result_digest": _digest(result),
+                "evidence_digest": "b" * 64,
+                "raw_internal_row": {"secret": "must-not-pass"},
+            }
+
+    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": UnclosedExecutor()}
+        ),
+    )
+
+    with app.test_client() as client:
+        response = client.post(
+            "/v1/tasks/execute",
+            json={
+                "task_token": token,
+                "node": RULE_NODE,
+                "parameters": {},
+            },
+        )
+
+    assert response.status_code == 503
+    assert ledger.get(claims.jti).status == "running"
+
+
 def test_running_ledger_without_rule_run_expires_to_unknown():
     class StaleLedger(InMemoryTaskLedger):
         def reconcile_running(self, jti, binding):
@@ -320,10 +505,14 @@ def test_running_ledger_without_rule_run_expires_to_unknown():
 def test_expired_rule_run_is_reconciled_and_ledger_finalized_unknown():
     class ExpiredEvidenceExecutor(Executor):
         def replay_task(self, **_context):
+            result = {"commit_outcome": "unknown"}
             return {
                 "state": "terminal",
                 "status": "unknown",
                 "commit_outcome": "unknown",
+                "result": result,
+                "result_digest": _digest(result),
+                "evidence_digest": "",
             }
 
     issuer = TaskTokenIssuer("x" * 32, clock=lambda: 1_000)
@@ -370,6 +559,10 @@ def test_expired_rule_run_is_reconciled_and_ledger_finalized_unknown():
 
 
 def test_expired_token_is_replay_only_and_never_starts_execution():
+    class RunningReplayExecutor(Executor):
+        def replay_task(self, **_context):
+            return {"state": "running"}
+
     now = [1_000]
     issuer = TaskTokenIssuer(
         "x" * 32,
@@ -379,7 +572,7 @@ def test_expired_token_is_replay_only_and_never_starts_execution():
     verifier = TaskTokenVerifier("x" * 32, clock=lambda: now[0])
     token = _rule_token(issuer)
     ledger = InMemoryTaskLedger()
-    executor = Executor()
+    executor = RunningReplayExecutor()
     app = create_runner_app(
         verifier=verifier,
         ledger=ledger,

tests/test_data_rule_schema.py

@@ -54,6 +54,12 @@ RULE_RECONCILIATION_MIGRATION = (
     / "versions"
     / "20260723_180_rule_reconciliation.py"
 )
+RULE_CLEANUP_LEASE_MIGRATION = (
+    ROOT
+    / "migrations"
+    / "versions"
+    / "20260723_190_rule_cleanup_leases.py"
+)
 
 EXPECTED_TABLES = {
     "data_rules",
@@ -262,3 +268,14 @@ def test_rule_reconciliation_migration_extends_170_without_rewriting_it():
     ):
         assert expected in source
     assert "forward-only" in source
+
+
+def test_rule_cleanup_lease_migration_extends_180_forward_only():
+    source = RULE_CLEANUP_LEASE_MIGRATION.read_text(encoding="utf-8")
+
+    assert 'revision = "20260723_190"' in source
+    assert 'down_revision = "20260723_180"' in source
+    assert source.count("cleanup_claim_expires_at TIMESTAMPTZ") == 2
+    assert "rule_violation_samples" in source
+    assert "rule_sql_staging_receipts" in source
+    assert "forward-only" in source