diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index e358df2..57b0a45 100644 --- a/src/echo_certification_forge/subscriber.py +++ b/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 diff --git a/tests/test_dispatch_orphaned_runs.py b/tests/test_dispatch_orphaned_runs.py index a80578e..fee0364 100644 --- a/tests/test_dispatch_orphaned_runs.py +++ b/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