fix(agent): run failed apply replay on broker

apps/api/tests/test_awooop_truth_chain_service.py

@@ -25,6 +25,7 @@ from src.services.awooop_ansible_check_mode_service import (
     _execution_capability_timeout_seconds,
     _expire_stale_ansible_execution_capabilities,
     _issue_ansible_execution_capability,
+    _load_open_failed_apply_retry_row,
     _load_pre_decision_context_runtime_stage_receipts,
     _post_apply_km_path_type,
     _post_apply_verification_result,
@@ -1871,6 +1872,17 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims
     async def no_stale_claims(**_kwargs):
         return []
 
+    async def no_failed_apply_retry(**_kwargs):
+        return {
+            "scanned": 0,
+            "replayed": 0,
+            "check_mode_passed": 0,
+            "check_mode_failed": 0,
+            "runtime_stage_receipt_written": 0,
+            "blockers": [],
+            "error": None,
+        }
+
     async def failed_catalog_replay(**_kwargs):
         raise RuntimeError("catalog replay query canceled")
 
@@ -1891,6 +1903,11 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims
         no_transport_blockers,
     )
     monkeypatch.setattr(service, "claim_stale_pending_check_modes", no_stale_claims)
+    monkeypatch.setattr(
+        service,
+        "run_failed_apply_check_mode_replay_once",
+        no_failed_apply_retry,
+    )
     monkeypatch.setattr(
         service,
         "claim_catalog_drift_failed_check_modes",
@@ -1906,6 +1923,79 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims
     assert result["catalog_replay_error"] == "RuntimeError"
 
 
+@pytest.mark.asyncio
+async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_claims(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from src.services import awooop_ansible_check_mode_service as service
+
+    claim_attempted = False
+
+    async def no_expired_capabilities(**_kwargs):
+        return 0
+
+    async def no_transport_blockers(**_kwargs):
+        return []
+
+    async def replay_open_retry(**_kwargs):
+        return {
+            "scanned": 1,
+            "replayed": 1,
+            "check_mode_passed": 1,
+            "check_mode_failed": 0,
+            "runtime_stage_receipt_written": 1,
+            "blockers": [],
+            "error": None,
+        }
+
+    async def candidate_claim_must_not_run(**_kwargs):
+        nonlocal claim_attempted
+        claim_attempted = True
+        return []
+
+    monkeypatch.setattr(
+        service,
+        "_expire_stale_ansible_execution_capabilities",
+        no_expired_capabilities,
+    )
+    monkeypatch.setattr(service, "_runtime_blockers", lambda: [])
+    monkeypatch.setattr(
+        service,
+        "recent_ansible_transport_blockers",
+        no_transport_blockers,
+    )
+    monkeypatch.setattr(
+        service,
+        "run_failed_apply_check_mode_replay_once",
+        replay_open_retry,
+    )
+    monkeypatch.setattr(
+        service,
+        "claim_stale_pending_check_modes",
+        candidate_claim_must_not_run,
+    )
+    monkeypatch.setattr(
+        service,
+        "claim_catalog_drift_failed_check_modes",
+        candidate_claim_must_not_run,
+    )
+    monkeypatch.setattr(
+        service,
+        "claim_pending_check_modes",
+        candidate_claim_must_not_run,
+    )
+
+    result = await service.run_pending_check_modes_once(limit=1)
+
+    assert claim_attempted is False
+    assert result["claimed"] == 0
+    assert result["failed_apply_retry_scanned"] == 1
+    assert result["failed_apply_retry_replayed"] == 1
+    assert result["failed_apply_retry_check_mode_passed"] == 1
+    assert result["failed_apply_retry_stage_receipt_written"] == 1
+    assert result["failed_apply_retry_priority_tick"] is True
+
+
 def test_check_mode_worker_error_backoff_is_short_and_bounded() -> None:
     from src.jobs import awooop_ansible_check_mode_job as job
 
@@ -2108,12 +2198,13 @@ def test_ansible_live_controlled_apply_sends_telegram_receipt_but_backfill_does_
 
 def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None:
     source = inspect.getsource(run_failed_apply_check_mode_replay_once)
+    preflight_source = inspect.getsource(_load_open_failed_apply_retry_row)
 
-    assert "FOR UPDATE SKIP LOCKED" in source
+    assert "FOR UPDATE SKIP LOCKED" in preflight_source
     assert "controlled_retry_check_mode_replay" in source
     assert "build_ansible_check_mode_command" in source
-    assert "verifier.incident_id = coalesce(" in source
-    assert "apply.input ->> 'incident_id'" in source
+    assert "verifier.incident_id = coalesce(" in preflight_source
+    assert "apply.input ->> 'incident_id'" in preflight_source
     assert "controlled_apply_allowed=True" in source
     assert '"approval_required_before_apply": False' in source
     assert '"owner_review_required": False' in source
@@ -2121,7 +2212,8 @@ def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None:
     assert "queue_ai_playbook_or_transport_repair_candidate" in source
     assert "retry_requires_repair_and_new_apply_gate" not in source
     assert "_record_retry_runtime_stage_receipt" in source
-    assert "NOT EXISTS" in source
+    assert "NOT EXISTS" in preflight_source
+    assert "failed_apply_retry_already_claimed" in source
     assert "run_controlled_apply_for_claim" not in source
 
     receipt_source = inspect.getsource(_record_retry_runtime_stage_receipt)