diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index fef7c65..e0839e9 100644 --- a/src/echo_certification_forge/subscriber.py +++ b/src/echo_certification_forge/subscriber.py @@ -4254,7 +4254,23 @@ def _recover_expired_dispatches( AND d.claim_token IS NULL AND d.lease_owner IS NULL AND d.lease_expires_at IS NULL - """ + -- Only exhausted, known pre-execution failures are safe for + -- automatic historical compensation. Recent/unknown failures + -- remain untouched for independent reconciliation. + AND d.attempts >= ? + AND d.last_error IN ( + 'subscriber_worker_claim_denied', + 'dispatch_attempts_exhausted' + ) + AND d.updated_at <= ? + ORDER BY d.created_at, d.run_id + LIMIT ? + """, + ( + self.policy.max_dispatch_attempts, + to_utc_iso(now - timedelta(hours=24)), + 4, + ), ).fetchall() for row in orphans: self._terminalize_unstarted_run( diff --git a/tests/test_dispatch_orphan_audit_redaction.py b/tests/test_dispatch_orphan_audit_redaction.py index db925e9..70880c7 100644 --- a/tests/test_dispatch_orphan_audit_redaction.py +++ b/tests/test_dispatch_orphan_audit_redaction.py @@ -40,10 +40,16 @@ def test_orphan_recovery_audit_redacts_untrusted_last_error( assert governance.claim_dispatch( "audit-recovery", organization_id=org.organization_id ) is None + # Only a recognized, exhausted pre-execution failure can be compensated. + # Unknown or untrusted last_error values are retained for independent review. + recognized = incoming_error == "subscriber_worker_claim_denied" assert store.get_run(run_id, org.organization_id)["state"] == ( - RunState.INFRASTRUCTURE_FAILURE.value + RunState.INFRASTRUCTURE_FAILURE.value if recognized + else RunState.QUEUED.value + ) + assert _usage(governance, org)["meters"]["certification_runs"] == ( + 0 if recognized else 1 ) - assert _usage(governance, org)["meters"]["certification_runs"] == 0 with sqlite3.connect(governance.db_path) as connection: audit_rows = connection.execute( @@ -52,7 +58,15 @@ def test_orphan_recovery_audit_redacts_untrusted_last_error( "AND action = 'certification.dispatch_orphan_reconciled'", (org.organization_id, run_id), ).fetchall() - assert len(audit_rows) == 1 - recorded = audit_rows[0][0] - assert json.loads(recorded)["last_error"] == audited_error - assert "synthetic-do-not-store-000" not in recorded + immutable_details = [row[0] for row in connection.execute( + "SELECT details_json FROM subscriber_audit_events " + "WHERE organization_id = ? AND resource_id = ?", + (org.organization_id, run_id), + )] + if recognized: + assert len(audit_rows) == 1 + assert json.loads(audit_rows[0][0])["last_error"] == audited_error + else: + # A rejected candidate cannot produce a misleading compensation audit. + assert audit_rows == [] + assert "synthetic-do-not-store-000" not in "".join(immutable_details) diff --git a/tests/test_dispatch_orphan_batch_recovery.py b/tests/test_dispatch_orphan_batch_recovery.py index 7706f21..5710f4c 100644 --- a/tests/test_dispatch_orphan_batch_recovery.py +++ b/tests/test_dispatch_orphan_batch_recovery.py @@ -34,7 +34,8 @@ def test_multiple_tenant_orphans_reconciled_once(tmp_path, manifest): "UPDATE subscriber_run_dispatches " "SET state='FAILED', attempts=3, claim_token=NULL, " "lease_owner=NULL, lease_expires_at=NULL, " - "last_error='subscriber_worker_claim_denied' " + "last_error='subscriber_worker_claim_denied', " + "updated_at='2026-10-01T00:00:00Z' " "WHERE run_id=? AND organization_id=?", (run_id, org.organization_id), ) @@ -42,6 +43,18 @@ def test_multiple_tenant_orphans_reconciled_once(tmp_path, manifest): assert all(_usage(governance, org)["meters"]["certification_runs"] == 1 for org, _ in pairs) assert governance.claim_dispatch("batch-recovery") is None + recovered_first = sum( + store.get_run(run_id, org.organization_id)["state"] + == RunState.INFRASTRUCTURE_FAILURE.value for org, run_id in pairs + ) + # Four records at most per transaction, with the fifth held intact. + assert recovered_first == 4 + assert sum( + store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value + for org, run_id in pairs + ) == 1 + + assert governance.claim_dispatch("batch-recovery-follow-up") is None for org, run_id in pairs: assert store.get_run(run_id, org.organization_id)["state"] == RunState.INFRASTRUCTURE_FAILURE.value usage = _usage(governance, org) diff --git a/tests/test_dispatch_orphaned_runs.py b/tests/test_dispatch_orphaned_runs.py index 3baea28..883e20e 100644 --- a/tests/test_dispatch_orphaned_runs.py +++ b/tests/test_dispatch_orphaned_runs.py @@ -81,7 +81,8 @@ def _orphan_the_old_way(governance, org, run_id) -> None: """ UPDATE subscriber_run_dispatches SET state = 'FAILED', attempts = 3, claim_token = NULL, lease_owner = NULL, - lease_expires_at = NULL, last_error = 'subscriber_worker_claim_denied' + lease_expires_at = NULL, last_error = 'subscriber_worker_claim_denied', + updated_at = '2026-10-01T00:00:00Z' WHERE run_id = ? AND organization_id = ? """, (run_id, org.organization_id), @@ -391,3 +392,95 @@ def test_expired_unstarted_dispatch_does_not_expose_claim_token_in_audit(tmp_pat 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 + +@pytest.mark.parametrize(("column","value"), [ + ("attempts", 1), + ("last_error", "unrecognized_failure"), + ("updated_at", "2099-01-01T00:00:00Z"), +]) +def test_historical_recovery_holds_nonexhausted_unknown_or_recent( + tmp_path, manifest, column, value +): + store, governance, org, run_id = _run( + tmp_path, manifest, "orphan-scope-guard", max_attempts=3 + ) + _orphan_the_old_way(governance, org, run_id) + with sqlite3.connect(governance.db_path) as connection: + # Only a disposable synthetic fixture; production is never mutated. + connection.execute( + f"UPDATE subscriber_run_dispatches SET {column} = ? WHERE run_id = ?", + (value, run_id), + ) + assert governance.claim_dispatch( + "dispatcher-guard", 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 + + +def test_historical_orphans_recovered_in_bounded_batches(tmp_path, manifest) -> None: + store, governance, org, first_id = _run( + tmp_path, manifest, "orphan-batch-scope", max_attempts=3 + ) + _orphan_the_old_way(governance, org, first_id) + run_ids = [first_id] + # Seed a synthetic historical backlog without hitting normal API quota/rate + # limits. Only this disposable test SQLite database is ever written. + with sqlite3.connect(governance.db_path) as connection: + connection.row_factory = sqlite3.Row + original_dispatch = connection.execute( + "SELECT idempotency_key FROM subscriber_run_dispatches WHERE run_id=?", + (first_id,), + ).fetchone() + original_key = original_dispatch["idempotency_key"] + for i in range(4): + run_id = f"cert_synthetic_backlog_{i:02d}" + key = f"batch-fixture-{i:02d}" + for table in ( + "runs", + "subscriber_run_reservations", + "subscriber_run_dispatches", + "subscriber_usage_events", + ): + where = ( + "run_id = ?" if table == "runs" + else "run_id = ?" if table != "subscriber_usage_events" + else "idempotency_key = ? AND meter = 'certification_runs' AND quantity > 0" + ) + source_id = original_key if table == "subscriber_usage_events" else first_id + source = connection.execute( + f"SELECT * FROM {table} WHERE {where}", (source_id,), + ).fetchone() + assert source is not None + values = dict(source) + values.pop("usage_event_id", None) + if "run_id" in values: + values["run_id"] = run_id + if "idempotency_key" in values: + values["idempotency_key"] = key + columns = list(values) + placeholders = ", ".join("?" for _ in columns) + connection.execute( + f"INSERT INTO {table} ({', '.join(columns)}) " + f"VALUES ({placeholders})", + tuple(values.values()), + ) + run_ids.append(run_id) + assert governance.claim_dispatch( + "dispatcher-first-batch", organization_id=org.organization_id + ) is None + first_count = sum( + store.get_run(run_id, org.organization_id)["state"] + == RunState.INFRASTRUCTURE_FAILURE.value for run_id in run_ids + ) + assert first_count == 4 + assert governance.claim_dispatch( + "dispatcher-second-batch", organization_id=org.organization_id + ) is None + assert all( + store.get_run(run_id, org.organization_id)["state"] + == RunState.INFRASTRUCTURE_FAILURE.value for run_id in run_ids + ) + assert _usage(governance, org)["meters"]["certification_runs"] == 0