Gate historical orphan recovery by exhausted attempts, age and bounded batches

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(

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)

tests/test_dispatch_orphan_batch_recovery.py

@@ -34,14 +34,27 @@ 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),
             )
         db.commit()
 
     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)

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