fix(agent): run failed apply replay on broker

apps/api/src/services/awooop_ansible_check_mode_service.py

@@ -2146,34 +2146,13 @@ async def backfill_missing_auto_repair_execution_receipts_once(
     return stats
 
 
-async def run_failed_apply_check_mode_replay_once(
+async def _load_open_failed_apply_retry_row(
     *,
-    project_id: str = "awoooi",
+    project_id: str,
-    window_hours: int = 24,
+    window_hours: int,
-    timeout_seconds: int | None = None,
+) -> dict[str, Any] | None:
-) -> dict[str, Any]:
+    """Read one durable failed-apply retry candidate before transport checks."""
-    """Replay one failed apply in no-write check mode and persist its terminal receipt."""
 
-    stats: dict[str, Any] = {
-        "scanned": 0,
-        "replayed": 0,
-        "check_mode_passed": 0,
-        "check_mode_failed": 0,
-        "runtime_apply_executed": 0,
-        "runtime_stage_receipt_written": 0,
-        "blockers": [],
-        "error": None,
-    }
-    blockers = _runtime_blockers()
-    if blockers:
-        stats["blockers"] = blockers
-        return stats
-    transport_blockers = await recent_ansible_transport_blockers(project_id=project_id)
-    if transport_blockers:
-        stats["blockers"] = transport_blockers
-        return stats
-
-    try:
     async with get_db_context(project_id) as db:
         result = await db.execute(
             text("""
@@ -2223,10 +2202,36 @@ async def run_failed_apply_check_mode_replay_once(
             {"window_hours": max(1, window_hours)},
         )
         row = result.mappings().one_or_none()
+        return dict(row) if row is not None else None
+
+
+async def run_failed_apply_check_mode_replay_once(
+    *,
+    project_id: str = "awoooi",
+    window_hours: int = 24,
+    timeout_seconds: int | None = None,
+) -> dict[str, Any]:
+    """Replay one failed apply in no-write check mode and persist its terminal receipt."""
+
+    stats: dict[str, Any] = {
+        "scanned": 0,
+        "replayed": 0,
+        "check_mode_passed": 0,
+        "check_mode_failed": 0,
+        "runtime_apply_executed": 0,
+        "runtime_stage_receipt_written": 0,
+        "blockers": [],
+        "error": None,
+    }
+    try:
+        row = await _load_open_failed_apply_retry_row(
+            project_id=project_id,
+            window_hours=window_hours,
+        )
         if row is None:
             return stats
         stats["scanned"] = 1
-            reconstructed = _claim_from_apply_operation_row(dict(row))
+        reconstructed = _claim_from_apply_operation_row(row)
         if reconstructed is None:
             stats["error"] = "failed_apply_receipt_could_not_be_reconstructed"
             return stats
@@ -2236,6 +2241,18 @@ async def run_failed_apply_check_mode_replay_once(
             claim.input_payload.get("automation_run_id")
             or claim.source_candidate_op_id
         )
+
+        blockers = _runtime_blockers()
+        if blockers:
+            stats["blockers"] = blockers
+            return stats
+        transport_blockers = await recent_ansible_transport_blockers(
+            project_id=project_id
+        )
+        if transport_blockers:
+            stats["blockers"] = transport_blockers
+            return stats
+
         input_payload = {
             **claim.input_payload,
             "automation_run_id": automation_run_id,
@@ -2253,13 +2270,15 @@ async def run_failed_apply_check_mode_replay_once(
                 "queue_ai_playbook_or_transport_repair_candidate"
             ),
         }
+        async with get_db_context(project_id) as db:
             claimed = await db.execute(
                 text("""
                     INSERT INTO automation_operation_log (
                         operation_type, actor, status, incident_id,
                         input, output, dry_run_result,
                         parent_op_id, tags
-                    ) VALUES (
+                    )
+                    SELECT
                         'ansible_execution_skipped',
                         'ansible_controlled_retry_worker',
                         'pending',
@@ -2269,6 +2288,13 @@ async def run_failed_apply_check_mode_replay_once(
                         CAST(:dry_run_result AS jsonb),
                         CAST(:parent_op_id AS uuid),
                         :tags
+                    WHERE NOT EXISTS (
+                        SELECT 1
+                        FROM automation_operation_log replay
+                        WHERE replay.operation_type = 'ansible_execution_skipped'
+                          AND replay.parent_op_id = CAST(:parent_op_id AS uuid)
+                          AND replay.input ->> 'execution_mode'
+                              = 'controlled_retry_check_mode_replay'
                     )
                     RETURNING op_id
                 """),
@@ -2299,7 +2325,11 @@ async def run_failed_apply_check_mode_replay_once(
                     ],
                 },
             )
-            replay_op_id = str(claimed.scalar_one())
+            replay_op_id_value = claimed.scalar_one_or_none()
+            if replay_op_id_value is None:
+                stats["blockers"] = ["failed_apply_retry_already_claimed"]
+                return stats
+            replay_op_id = str(replay_op_id_value)
         effective_timeout = (
             timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS
         )
@@ -3164,6 +3194,69 @@ async def run_pending_check_modes_once(
     expired_capability_count = await _expire_stale_ansible_execution_capabilities(
         project_id=project_id,
     )
+    effective_timeout_seconds = (
+        timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS
+    )
+    retry_stats = await run_failed_apply_check_mode_replay_once(
+        project_id=project_id,
+        window_hours=24,
+        timeout_seconds=effective_timeout_seconds,
+    )
+    failed_apply_retry_blockers = [
+        str(item)[:200]
+        for item in retry_stats.get("blockers") or []
+    ]
+    failed_apply_retry_error = (
+        str(retry_stats.get("error") or "")[:500] or None
+    )
+    failed_apply_retry_summary = {
+        "failed_apply_retry_scanned": int(retry_stats.get("scanned") or 0),
+        "failed_apply_retry_replayed": int(retry_stats.get("replayed") or 0),
+        "failed_apply_retry_check_mode_passed": int(
+            retry_stats.get("check_mode_passed") or 0
+        ),
+        "failed_apply_retry_check_mode_failed": int(
+            retry_stats.get("check_mode_failed") or 0
+        ),
+        "failed_apply_retry_stage_receipt_written": int(
+            retry_stats.get("runtime_stage_receipt_written") or 0
+        ),
+        "failed_apply_retry_blockers": failed_apply_retry_blockers,
+        "failed_apply_retry_error": failed_apply_retry_error,
+    }
+    failed_apply_retry_priority_tick = bool(
+        failed_apply_retry_summary["failed_apply_retry_scanned"]
+        or failed_apply_retry_summary["failed_apply_retry_replayed"]
+        or failed_apply_retry_blockers
+        or failed_apply_retry_error
+    )
+    failed_apply_retry_summary["failed_apply_retry_priority_tick"] = (
+        failed_apply_retry_priority_tick
+    )
+    if failed_apply_retry_priority_tick:
+        logger.info(
+            "ansible_execution_broker_failed_apply_retry_priority_tick",
+            project_id=project_id,
+            **failed_apply_retry_summary,
+        )
+        return {
+            "claimed": 0,
+            "reclaimed": 0,
+            "catalog_replayed": 0,
+            "completed": 0,
+            "failed": 0,
+            "apply_completed": 0,
+            "apply_failed": 0,
+            "apply_blocked": 0,
+            "capability_issued": 0,
+            "capability_revoked": 0,
+            "capability_expired": expired_capability_count,
+            "capability_revoke_failed": 0,
+            "catalog_replay_error": None,
+            "error": failed_apply_retry_error,
+            "blockers": failed_apply_retry_blockers,
+            **failed_apply_retry_summary,
+        }
     blockers = _runtime_blockers()
     if blockers:
         logger.warning("ansible_check_mode_runtime_blocked", blockers=blockers)
@@ -3173,6 +3266,7 @@ async def run_pending_check_modes_once(
             "failed": 0,
             "capability_expired": expired_capability_count,
             "blockers": blockers,
+            **failed_apply_retry_summary,
         }
     transport_blockers = await recent_ansible_transport_blockers(project_id=project_id)
     if transport_blockers:
@@ -3183,11 +3277,8 @@ async def run_pending_check_modes_once(
             "failed": 0,
             "capability_expired": expired_capability_count,
             "blockers": transport_blockers,
+            **failed_apply_retry_summary,
         }
-
-    effective_timeout_seconds = (
-        timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS
-    )
     reclaimed_claims = await claim_stale_pending_check_modes(
         project_id=project_id,
         limit=limit,
@@ -3340,5 +3431,6 @@ async def run_pending_check_modes_once(
         "capability_expired": expired_capability_count,
         "capability_revoke_failed": capability_revoke_failed,
         "catalog_replay_error": catalog_replay_error,
+        **failed_apply_retry_summary,
         "blockers": [],
     }

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 "verifier.incident_id = coalesce(" in preflight_source
-    assert "apply.input ->> 'incident_id'" in 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)