fix(subscriber): preserve non-secret dispatcher claim custody during lease recovery

src/echo_certification_forge/subscriber.py

@@ -886,7 +886,7 @@
         return f"{prefix}_{secrets.token_hex(12)}"
 
     def _token_digest(self, token: str) -> str:
         return hmac.new(self._token_pepper, token.encode("utf-8"), hashlib.sha256).hexdigest()
 
     def _append_audit(
         self,
@@ -4083,6 +4083,9 @@
         rows = connection.execute(
             """
             SELECT d.run_id, d.organization_id, d.idempotency_key, d.attempts,
+                   d.lease_owner AS previous_lease_owner,
+                   d.lease_expires_at AS previous_lease_expires_at,
+                   d.claim_token AS previous_claim_token,
                    CASE WHEN r.execution_started_at IS NULL
                              AND r.consumed_at IS NULL
                              AND r.worker_id IS NULL
@@ -4180,6 +4183,24 @@
                 """,
                 (next_state, error, now_text, row["run_id"]),
             )
+            recovery_details = {
+                "recovery": next_state.lower(),
+                "attempts": int(row["attempts"]),
+                "meter_compensated": meter_compensated,
+                "ownership_review_required": ownership_review_required,
+            }
+            if ownership_review_required:
+                # Preserve non-secret lease provenance in the append-only audit
+                # before mutable claim fields are cleared. Never persist a replayable token.
+                previous_claim = row["previous_claim_token"]
+                recovery_details.update({
+                    "previous_lease_owner": row["previous_lease_owner"],
+                    "previous_lease_expires_at": row["previous_lease_expires_at"],
+                    "previous_claim_token_sha256": (
+                        hashlib.sha256(str(previous_claim).encode("utf-8")).hexdigest()
+                        if previous_claim is not None else None
+                    ),
+                })
             self._append_audit(
                 connection,
                 organization_id=str(row["organization_id"]),
@@ -4188,12 +4209,7 @@
                 resource_type="certification",
                 resource_id=str(row["run_id"]),
                 outcome="allowed",
-                details={
-                    "recovery": next_state.lower(),
-                    "attempts": int(row["attempts"]),
-                    "meter_compensated": meter_compensated,
-                    "ownership_review_required": ownership_review_required,
-                },
+                details=recovery_details,
             )
         # Runs orphaned before fleet #49604: dispatch FAILED, run still QUEUED, reservation BOUND.
         # Once terminalized they no longer match, so this pass is idempotent.

tests/test_dispatch_orphaned_runs.py

@@ -298,3 +298,93 @@ def test_expired_claim_without_ownership_preserves_existing_policy(
     governance.claim_dispatch("dispatcher-b", organization_id=org.organization_id)
     assert store.get_run(run_id, org.organization_id)["state"] == expected_state
     assert _usage(governance, org)["meters"]["certification_runs"] == expected_meter
+
+
+def test_expired_owned_dispatch_preserves_nonsecret_claim_custody(tmp_path, manifest):
+    # Disposable SQLite fixture only: no production mutation or artificial heartbeat.
+    import hashlib
+    import json
+
+    store, governance, org, run_id = _run(
+        tmp_path, manifest, "expired-custody", max_attempts=3
+    )
+    dispatch = governance.claim_dispatch(
+        "dispatcher-a", run_id=run_id, organization_id=org.organization_id
+    )
+    assert dispatch is not None
+    with sqlite3.connect(governance.db_path) as connection:
+        connection.execute(
+            "UPDATE subscriber_run_reservations SET consumed_at = ? WHERE run_id = ?",
+            ("2026-11-12T12:00:00Z", run_id),
+        )
+        connection.execute(
+            "UPDATE subscriber_run_dispatches SET lease_expires_at = ? WHERE run_id = ?",
+            ("2000-01-01T00:00:00Z", run_id),
+        )
+    assert governance.claim_dispatch(
+        "dispatcher-b", organization_id=org.organization_id
+    ) is None
+    with sqlite3.connect(governance.db_path) as connection:
+        row = connection.execute(
+            "SELECT details_json FROM subscriber_audit_events "
+            "WHERE organization_id = ? AND resource_id = ? "
+            "AND action = 'certification.dispatch_recovered' "
+            "ORDER BY event_id DESC LIMIT 1",
+            (org.organization_id, run_id),
+        ).fetchone()
+        state = connection.execute(
+            "SELECT state, claim_token, lease_owner, lease_expires_at "
+            "FROM subscriber_run_dispatches WHERE run_id = ?", (run_id,)
+        ).fetchone()
+    assert row is not None
+    details = json.loads(row[0])
+    assert details["ownership_review_required"] is True
+    assert details["meter_compensated"] is False
+    assert details["previous_lease_owner"] == "dispatcher-a"
+    assert details["previous_lease_expires_at"] == "2000-01-01T00:00:00Z"
+    assert details["previous_claim_token_sha256"] == hashlib.sha256(
+        dispatch.claim_token.encode("utf-8")
+    ).hexdigest()
+    assert dispatch.claim_token not in row[0]
+    assert state == ("FAILED", None, None, None)
+    assert store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value
+    assert _usage(governance, org)["meters"]["certification_runs"] == 1
+    principal = _owner(
+        governance, org.organization_id, org.bootstrap_api_key, Permission.AUDIT_READ
+    )
+    verification = governance.verify_audit(principal)
+    assert verification["valid"] is True
+    assert verification["invalid_event_ids"] == []
+
+
+def test_expired_unstarted_dispatch_does_not_expose_claim_token_in_audit(tmp_path, manifest):
+    import json
+
+    store, governance, org, run_id = _run(
+        tmp_path, manifest, "unstarted-custody", max_attempts=1
+    )
+    dispatch = governance.claim_dispatch(
+        "dispatcher-a", run_id=run_id, organization_id=org.organization_id
+    )
+    assert dispatch is not None
+    with sqlite3.connect(governance.db_path) as connection:
+        connection.execute(
+            "UPDATE subscriber_run_dispatches SET lease_expires_at = ? WHERE run_id = ?",
+            ("2000-01-01T00:00:00Z", run_id),
+        )
+    governance.claim_dispatch("dispatcher-b", organization_id=org.organization_id)
+    with sqlite3.connect(governance.db_path) as connection:
+        row = connection.execute(
+            "SELECT details_json FROM subscriber_audit_events "
+            "WHERE organization_id = ? AND resource_id = ? "
+            "AND action = 'certification.dispatch_recovered' "
+            "ORDER BY event_id DESC LIMIT 1",
+            (org.organization_id, run_id),
+        ).fetchone()
+    assert row is not None
+    details = json.loads(row[0])
+    assert details["ownership_review_required"] is False
+    assert "previous_claim_token_sha256" not in details
+    assert dispatch.claim_token not in row[0]
+    assert store.get_run(run_id, org.organization_id)["state"] == RunState.INFRASTRUCTURE_FAILURE.value
+    assert _usage(governance, org)["meters"]["certification_runs"] == 0