From 272bea3674f6cdecd4ca0423b30b745cbf755cab Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Thu, 12 Nov 2026 06:40:55 -0600 Subject: [PATCH 1/6] fix(subscriber): fail closed on orphaned dispatch with execution or ownership evidence --- src/echo_certification_forge/subscriber.py | 14 ++++++ tests/test_dispatch_orphaned_runs.py | 51 ++++++++++++++++++++++ 2 files changed, 65 insertions(+) diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index e358df2..66ff83d 100644 --- a/src/echo_certification_forge/subscriber.py +++ b/src/echo_certification_forge/subscriber.py @@ -4202,6 +4202,20 @@ 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 + 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: diff --git a/tests/test_dispatch_orphaned_runs.py b/tests/test_dispatch_orphaned_runs.py index a80578e..7286c2b 100644 --- a/tests/test_dispatch_orphaned_runs.py +++ b/tests/test_dispatch_orphaned_runs.py @@ -115,3 +115,54 @@ 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", "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" From 0a9bbc041bb3ded5f1bde5aba1e41591e755ef88 Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Thu, 12 Nov 2026 08:28:50 -0600 Subject: [PATCH 2/6] fix(subscriber): exclude consumed reservations from orphan compensation --- src/echo_certification_forge/subscriber.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index 66ff83d..f3b2196 100644 --- a/src/echo_certification_forge/subscriber.py +++ b/src/echo_certification_forge/subscriber.py @@ -4205,6 +4205,8 @@ def _recover_expired_dispatches( -- 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 From 0af5f6fe3047394aeb11f1c921940a6dfef7055d Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Thu, 12 Nov 2026 08:28:56 -0600 Subject: [PATCH 3/6] test(subscriber): preserve consumed orphan reservation and meter --- tests/test_dispatch_orphaned_runs.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_dispatch_orphaned_runs.py b/tests/test_dispatch_orphaned_runs.py index 7286c2b..8dfef4f 100644 --- a/tests/test_dispatch_orphaned_runs.py +++ b/tests/test_dispatch_orphaned_runs.py @@ -126,6 +126,7 @@ def test_a_failed_dispatch_whose_run_already_finished_is_left_alone(tmp_path, ma ("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"), From 7119fe77808634effba8aa5607bdb1422a152ed3 Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Thu, 12 Nov 2026 08:33:02 -0600 Subject: [PATCH 4/6] fix(subscriber): fail closed on contradictory ownership at final dispatch --- src/echo_certification_forge/subscriber.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index f3b2196..4cca4bf 100644 --- a/src/echo_certification_forge/subscriber.py +++ b/src/echo_certification_forge/subscriber.py @@ -4387,6 +4387,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 @@ -4453,7 +4464,12 @@ 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": + 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). From b98d1cedd6532ac29fd7dfb33adb340a953975fc Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Thu, 12 Nov 2026 08:33:10 -0600 Subject: [PATCH 5/6] test(subscriber): block final-attempt compensation with active evidence --- tests/test_dispatch_orphaned_runs.py | 39 ++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/tests/test_dispatch_orphaned_runs.py b/tests/test_dispatch_orphaned_runs.py index 8dfef4f..5dc8298 100644 --- a/tests/test_dispatch_orphaned_runs.py +++ b/tests/test_dispatch_orphaned_runs.py @@ -167,3 +167,42 @@ def test_orphan_recovery_preserves_conflicting_ownership( (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 From c95dd7d2a971ea26d827382a4f1d5ec96fa327d6 Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Thu, 12 Nov 2026 09:36:21 -0600 Subject: [PATCH 6/6] fix(subscriber): hold expired dispatch leases with execution evidence Fail closed on lease expiry and retry errors when worker ownership, reservation consumption or execution markers exist. Preserve run/reservation/metering evidence and add synthetic negative and positive recovery controls. Reviewed locally: 30 focused pytest cases passed; Git diff check clean. --- src/echo_certification_forge/subscriber.py | 43 +++++++++- tests/test_dispatch_orphaned_runs.py | 92 ++++++++++++++++++++++ 2 files changed, 132 insertions(+), 3 deletions(-) diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index 4cca4bf..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. @@ -4463,7 +4483,8 @@ def finish_dispatch( ) state = "FAILED" elif error is not None and int(row["attempts"]) < self.policy.max_dispatch_attempts: - state = "PENDING" + # 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" @@ -4486,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 5dc8298..fee0364 100644 --- a/tests/test_dispatch_orphaned_runs.py +++ b/tests/test_dispatch_orphaned_runs.py @@ -206,3 +206,95 @@ def test_final_dispatch_error_preserves_conflicting_execution_evidence( # 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