fix(subscriber): preserve execution ownership during historical orphan recovery

src/echo_certification_forge/subscriber.py

@@ -4083,6 +4083,17 @@ def _recover_expired_dispatches(
         rows = connection.execute(
             """
             SELECT d.run_id, d.organization_id, d.idempotency_key, d.attempts,
+                   CASE WHEN r.execution_started_at IS NULL
+                             AND r.consumed_at IS NULL
+                             AND r.worker_id IS NULL
+                             AND r.worker_attestation_sha256 IS NULL
+                             AND r.execution_location IS NULL
+                             AND r.signing_authority IS NULL
+                             AND r.signing_key_id IS NULL
+                             AND r.claim_token IS NULL
+                             AND r.lease_expires_at IS NULL
+                             AND r.last_heartbeat_at IS NULL
+                        THEN 1 ELSE 0 END AS unstarted_without_ownership,
                    COALESCE(
                        (
                            SELECT u.occurred_at
@@ -4111,7 +4122,14 @@ def _recover_expired_dispatches(
             (now_text,),
         ).fetchall()
         for row in rows:
-            if int(row["attempts"]) >= self.policy.max_dispatch_attempts:
+            ownership_review_required = int(row["unstarted_without_ownership"]) != 1
+            meter_compensated = False
+            if ownership_review_required:
+                # Expired dispatcher lease cannot prove that live worker work stopped.
+                # Hold the run and reservation for independent ownership review.
+                next_state = "FAILED"
+                error = "dispatch_lease_expired_ownership_review"
+            elif int(row["attempts"]) >= self.policy.max_dispatch_attempts:
                 self._transition_run_with_connection(
                     connection,
                     str(row["run_id"]),
@@ -4149,6 +4167,7 @@ def _recover_expired_dispatches(
                 )
                 next_state = "FAILED"
                 error = "dispatch_attempts_exhausted"
+                meter_compensated = True
             else:
                 next_state = "PENDING"
                 error = "dispatch_lease_expired"
@@ -4172,7 +4191,8 @@ def _recover_expired_dispatches(
                 details={
                     "recovery": next_state.lower(),
                     "attempts": int(row["attempts"]),
-                    "meter_compensated": next_state == "FAILED",
+                    "meter_compensated": meter_compensated,
+                    "ownership_review_required": ownership_review_required,
                 },
             )
         # Runs orphaned before fleet #49604: dispatch FAILED, run still QUEUED, reservation BOUND.
@@ -4202,6 +4222,22 @@ def _recover_expired_dispatches(
             WHERE d.state = 'FAILED'
               AND r.state = 'BOUND'
               AND runs.state = 'QUEUED'
+              -- Legacy recovery may only compensate work with no execution or
+              -- ownership evidence. Inconsistent state is held for human review.
+              AND r.execution_started_at IS NULL
+              -- A previously consumed reservation is not proven unstarted.
+              AND r.consumed_at IS NULL
+              AND r.worker_id IS NULL
+              AND r.worker_attestation_sha256 IS NULL
+              AND r.execution_location IS NULL
+              AND r.signing_authority IS NULL
+              AND r.signing_key_id IS NULL
+              AND r.claim_token IS NULL
+              AND r.lease_expires_at IS NULL
+              AND r.last_heartbeat_at IS NULL
+              AND d.claim_token IS NULL
+              AND d.lease_owner IS NULL
+              AND d.lease_expires_at IS NULL
             """
         ).fetchall()
         for row in orphans:
@@ -4371,6 +4407,17 @@ def finish_dispatch(
                 """
                 SELECT d.state, d.attempts, d.idempotency_key,
                        runs.state AS run_state, r.state AS reservation_state,
+                       CASE WHEN r.execution_started_at IS NULL
+                             AND r.consumed_at IS NULL
+                             AND r.worker_id IS NULL
+                             AND r.worker_attestation_sha256 IS NULL
+                             AND r.execution_location IS NULL
+                             AND r.signing_authority IS NULL
+                             AND r.signing_key_id IS NULL
+                             AND r.claim_token IS NULL
+                             AND r.lease_expires_at IS NULL
+                             AND r.last_heartbeat_at IS NULL
+                            THEN 1 ELSE 0 END AS unstarted_without_ownership,
                        COALESCE(
                            (
                                SELECT u.occurred_at
@@ -4436,8 +4483,14 @@ def finish_dispatch(
                 )
                 state = "FAILED"
             elif error is not None and int(row["attempts"]) < self.policy.max_dispatch_attempts:
-                state = "PENDING"
-            elif error is not None and run_state == "QUEUED" and row["reservation_state"] == "BOUND":
+                # Never requeue if the worker may already have started execution.
+                state = "PENDING" if int(row["unstarted_without_ownership"]) == 1 else "FAILED"
+            elif (
+                error is not None
+                and run_state == "QUEUED"
+                and row["reservation_state"] == "BOUND"
+                and int(row["unstarted_without_ownership"]) == 1
+            ):
                 # Last attempt failed before execution started (for example
                 # subscriber_worker_claim_denied). Marking only the dispatch FAILED left the run
                 # QUEUED forever and uncompensated (fleet #49604).
@@ -4454,6 +4507,22 @@ def finish_dispatch(
                 state = "FAILED"
             else:
                 state = "FAILED"
+            if (
+                error is not None
+                and run_state == "QUEUED"
+                and row["reservation_state"] == "BOUND"
+                and int(row["unstarted_without_ownership"]) != 1
+            ):
+                self._append_audit(
+                    connection,
+                    organization_id=dispatch.organization_id,
+                    actor_ref="certforge.dispatcher",
+                    action="certification.dispatch_ownership_review_required",
+                    resource_type="certification",
+                    resource_id=dispatch.run_id,
+                    outcome="denied",
+                    details={"attempts": int(row["attempts"]), "meter_compensated": False},
+                )
             connection.execute(
                 """
                 UPDATE subscriber_run_dispatches

tests/test_dispatch_orphaned_runs.py

@@ -115,3 +115,186 @@ def test_a_failed_dispatch_whose_run_already_finished_is_left_alone(tmp_path, ma
     before = _usage(governance, org)["meters"]["certification_runs"]
     assert governance.claim_dispatch("dispatcher-b", organization_id=org.organization_id) is None
     assert _usage(governance, org)["meters"]["certification_runs"] == before == 0
+
+
+# Synthetic contradictory records must not lose live execution ownership,
+# immutable evidence or metered reservations during historical recovery.
+import pytest
+
+
+@pytest.mark.parametrize(
+    ("table", "column", "value"),
+    [
+        ("subscriber_run_reservations", "execution_started_at", "2026-11-12T12:00:00Z"),
+        ("subscriber_run_reservations", "consumed_at", "2026-11-12T12:01:00Z"),
+        ("subscriber_run_reservations", "worker_id", "synthetic-test-worker"),
+        ("subscriber_run_reservations", "worker_attestation_sha256", "1" * 64),
+        ("subscriber_run_reservations", "execution_location", "synthetic-test-location"),
+        ("subscriber_run_reservations", "signing_authority", "synthetic-test-authority"),
+        ("subscriber_run_reservations", "signing_key_id", "synthetic-test-key"),
+        ("subscriber_run_reservations", "claim_token", "synthetic-test-claim"),
+        ("subscriber_run_reservations", "lease_expires_at", "2026-11-12T13:00:00Z"),
+        ("subscriber_run_reservations", "last_heartbeat_at", "2026-11-12T12:00:00Z"),
+        ("subscriber_run_dispatches", "claim_token", "synthetic-test-dispatch-claim"),
+        ("subscriber_run_dispatches", "lease_owner", "synthetic-test-dispatch-owner"),
+        ("subscriber_run_dispatches", "lease_expires_at", "2026-11-12T13:00:00Z"),
+    ],
+)
+def test_orphan_recovery_preserves_conflicting_ownership(
+    tmp_path, manifest, table, column, value
+) -> None:
+    store, governance, org, run_id = _run(
+        tmp_path, manifest, "contradictory-owner", max_attempts=3
+    )
+    _orphan_the_old_way(governance, org, run_id)
+    # Only disposable test SQLite, never production. Parameters are test constants.
+    with sqlite3.connect(governance.db_path) as connection:
+        connection.execute(
+            f"UPDATE {table} SET {column} = ? WHERE run_id = ? AND organization_id = ?",
+            (value, run_id, org.organization_id),
+        )
+    assert governance.claim_dispatch(
+        "dispatcher-reconciler", organization_id=org.organization_id
+    ) is None
+    assert store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value
+    usage = _usage(governance, org)
+    assert usage["meters"]["certification_runs"] == 1
+    assert usage["active_run_reservations"] == 1
+    with sqlite3.connect(governance.db_path) as connection:
+        state = connection.execute(
+            "SELECT state FROM subscriber_run_reservations "
+            "WHERE run_id = ? AND organization_id = ?",
+            (run_id, org.organization_id),
+        ).fetchone()[0]
+        assert state == "BOUND"
+
+
+@pytest.mark.parametrize(
+    ("column", "value"),
+    [
+        ("consumed_at", "2026-11-12T14:00:00Z"),
+        ("execution_started_at", "2026-11-12T14:00:00Z"),
+        ("worker_id", "synthetic-live-worker"),
+    ],
+)
+def test_final_dispatch_error_preserves_conflicting_execution_evidence(
+    tmp_path, manifest, column, value
+) -> None:
+    """An exhausted claimed dispatch cannot erase contradictory ownership."""
+    store, governance, org, run_id = _run(
+        tmp_path, manifest, "final-attempt-owner", max_attempts=1
+    )
+    dispatch = governance.claim_dispatch(
+        "dispatcher-a", run_id=run_id, organization_id=org.organization_id
+    )
+    assert dispatch is not None
+    # Synthetic state mutation is confined to the disposable test fixture DB.
+    with sqlite3.connect(governance.db_path) as connection:
+        connection.execute(
+            f"UPDATE subscriber_run_reservations SET {column} = ? "
+            "WHERE run_id = ? AND organization_id = ?",
+            (value, run_id, org.organization_id),
+        )
+    assert governance.finish_dispatch(
+        dispatch, error="subscriber_worker_claim_denied"
+    ) == "FAILED"
+    assert store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value
+    assert governance.dispatch_status(run_id, org.organization_id)["state"] == "FAILED"
+    usage = _usage(governance, org)
+    assert usage["meters"]["certification_runs"] == 1
+    assert usage["active_run_reservations"] == 1
+    # No historical reconcile may reinterpret the contradictory record as unstarted.
+    assert governance.claim_dispatch("dispatcher-b", organization_id=org.organization_id) is None
+    assert store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value
+
+
+@pytest.mark.parametrize("max_attempts", [1, 3])
+@pytest.mark.parametrize(("column", "value"), [
+    ("consumed_at", "2026-11-12T12:00:00Z"),
+    ("execution_started_at", "2026-11-12T12:00:00Z"),
+    ("worker_id", "synthetic-worker"),
+])
+def test_expired_claim_with_ownership_requires_review(
+    tmp_path, manifest, max_attempts, column, value
+):
+    store, governance, org, run_id = _run(
+        tmp_path, manifest, "expired-dispatch-owner", max_attempts=max_attempts
+    )
+    dispatch = governance.claim_dispatch(
+        "dispatcher-a", run_id=run_id, organization_id=org.organization_id
+    )
+    assert dispatch is not None
+    # Deliberate contradictory state exists only in this disposable test database.
+    with sqlite3.connect(governance.db_path) as connection:
+        connection.execute(
+            f"UPDATE subscriber_run_reservations SET {column} = ? WHERE run_id = ?",
+            (value, run_id),
+        )
+        connection.execute(
+            "UPDATE subscriber_run_dispatches "
+            "SET lease_expires_at = '2000-01-01T00:00:00Z', attempts = ? "
+            "WHERE run_id = ?", (max_attempts, run_id)
+        )
+    assert governance.claim_dispatch(
+        "dispatcher-b", organization_id=org.organization_id
+    ) is None
+    assert store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value
+    assert governance.dispatch_status(run_id, org.organization_id)["state"] == "FAILED"
+    usage = _usage(governance, org)
+    assert usage["meters"]["certification_runs"] == 1
+    assert usage["active_run_reservations"] == 1
+
+@pytest.mark.parametrize(("column", "value"), [
+    ("consumed_at", "2026-11-12T12:00:00Z"),
+    ("worker_id", "synthetic-worker"),
+])
+def test_retryable_dispatch_with_ownership_is_not_requeued(
+    tmp_path, manifest, column, value
+):
+    store, governance, org, run_id = _run(
+        tmp_path, manifest, "retryable-dispatch-owner", 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(
+            f"UPDATE subscriber_run_reservations SET {column} = ? WHERE run_id = ?",
+            (value, run_id),
+        )
+    assert governance.finish_dispatch(
+        dispatch, error="subscriber_worker_claim_denied"
+    ) == "FAILED"
+    assert store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value
+    usage = _usage(governance, org)
+    assert usage["meters"]["certification_runs"] == 1
+    assert usage["active_run_reservations"] == 1
+    assert governance.claim_dispatch(
+        "dispatcher-b", organization_id=org.organization_id
+    ) is None
+
+
+@pytest.mark.parametrize("max_attempts,expected_state,expected_meter", [
+    (1, RunState.INFRASTRUCTURE_FAILURE.value, 0),
+    (3, RunState.QUEUED.value, 1),
+])
+def test_expired_claim_without_ownership_preserves_existing_policy(
+    tmp_path, manifest, max_attempts, expected_state, expected_meter
+):
+    store, governance, org, run_id = _run(
+        tmp_path, manifest, "expired-unstarted-proof", max_attempts=max_attempts
+    )
+    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 = '2000-01-01T00:00:00Z' WHERE run_id = ?",
+            (run_id,)
+        )
+    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