fix(agent): run failed apply replay on broker

apps/api/src/services/awooop_ansible_check_mode_service.py

@@ -2146,6 +2146,65 @@ async def backfill_missing_auto_repair_execution_receipts_once(
     return stats
 
 
+async def _load_open_failed_apply_retry_row(
+    *,
+    project_id: str,
+    window_hours: int,
+) -> dict[str, Any] | None:
+    """Read one durable failed-apply retry candidate before transport checks."""
+
+    async with get_db_context(project_id) as db:
+        result = await db.execute(
+            text("""
+                SELECT
+                    apply.op_id::text AS op_id,
+                    apply.parent_op_id::text AS parent_op_id,
+                    coalesce(
+                        apply.incident_id::text,
+                        apply.input ->> 'incident_id'
+                    ) AS incident_id,
+                    apply.input,
+                    apply.output,
+                    apply.dry_run_result,
+                    apply.error,
+                    apply.duration_ms,
+                    apply.status,
+                    apply.created_at
+                FROM automation_operation_log apply
+                WHERE apply.operation_type = 'ansible_apply_executed'
+                  AND apply.status = 'failed'
+                  AND apply.created_at >= NOW() - (
+                      :window_hours * INTERVAL '1 hour'
+                  )
+                  AND EXISTS (
+                      SELECT 1
+                      FROM incident_evidence verifier
+                      WHERE verifier.incident_id = coalesce(
+                                apply.incident_id::text,
+                                apply.input ->> 'incident_id'
+                            )
+                        AND verifier.post_execution_state ->> 'apply_op_id'
+                            = apply.op_id::text
+                        AND verifier.verification_result = 'failed'
+                  )
+                  AND NOT EXISTS (
+                      SELECT 1
+                      FROM automation_operation_log replay
+                      WHERE replay.operation_type = 'ansible_execution_skipped'
+                        AND replay.parent_op_id = apply.op_id
+                        AND replay.input ->> 'execution_mode'
+                            = 'controlled_retry_check_mode_replay'
+                  )
+                ORDER BY apply.created_at DESC
+                LIMIT 1
+                FOR UPDATE SKIP LOCKED
+            """),
+            {"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",
@@ -2164,102 +2223,62 @@ async def run_failed_apply_check_mode_replay_once(
         "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:
+        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(row)
+        if reconstructed is None:
+            stats["error"] = "failed_apply_receipt_could_not_be_reconstructed"
+            return stats
+        claim, _failed_apply_result = reconstructed
+        apply_op_id = str(row.get("op_id") or "")
+        automation_run_id = str(
+            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,
+            "execution_mode": "controlled_retry_check_mode_replay",
+            "retry_of_apply_op_id": apply_op_id,
+            "check_mode": True,
+            "diff": True,
+            "apply_enabled": False,
+            "approval_required_before_apply": False,
+            "owner_review_required": False,
+            "controlled_apply_allowed": True,
+            "controlled_apply_blocker": None,
+            "bounded_execution_scope": "retry_check_mode_only",
+            "next_controlled_action": (
+                "queue_ai_playbook_or_transport_repair_candidate"
+            ),
+        }
         async with get_db_context(project_id) as db:
-            result = await db.execute(
-                text("""
-                    SELECT
-                        apply.op_id::text AS op_id,
-                        apply.parent_op_id::text AS parent_op_id,
-                        coalesce(
-                            apply.incident_id::text,
-                            apply.input ->> 'incident_id'
-                        ) AS incident_id,
-                        apply.input,
-                        apply.output,
-                        apply.dry_run_result,
-                        apply.error,
-                        apply.duration_ms,
-                        apply.status,
-                        apply.created_at
-                    FROM automation_operation_log apply
-                    WHERE apply.operation_type = 'ansible_apply_executed'
-                      AND apply.status = 'failed'
-                      AND apply.created_at >= NOW() - (
-                          :window_hours * INTERVAL '1 hour'
-                      )
-                      AND EXISTS (
-                          SELECT 1
-                          FROM incident_evidence verifier
-                          WHERE verifier.incident_id = coalesce(
-                                    apply.incident_id::text,
-                                    apply.input ->> 'incident_id'
-                                )
-                            AND verifier.post_execution_state ->> 'apply_op_id'
-                                = apply.op_id::text
-                            AND verifier.verification_result = 'failed'
-                      )
-                      AND NOT EXISTS (
-                          SELECT 1
-                          FROM automation_operation_log replay
-                          WHERE replay.operation_type = 'ansible_execution_skipped'
-                            AND replay.parent_op_id = apply.op_id
-                            AND replay.input ->> 'execution_mode'
-                                = 'controlled_retry_check_mode_replay'
-                      )
-                    ORDER BY apply.created_at DESC
-                    LIMIT 1
-                    FOR UPDATE SKIP LOCKED
-                """),
-                {"window_hours": max(1, window_hours)},
-            )
-            row = result.mappings().one_or_none()
-            if row is None:
-                return stats
-            stats["scanned"] = 1
-            reconstructed = _claim_from_apply_operation_row(dict(row))
-            if reconstructed is None:
-                stats["error"] = "failed_apply_receipt_could_not_be_reconstructed"
-                return stats
-            claim, _failed_apply_result = reconstructed
-            apply_op_id = str(row.get("op_id") or "")
-            automation_run_id = str(
-                claim.input_payload.get("automation_run_id")
-                or claim.source_candidate_op_id
-            )
-            input_payload = {
-                **claim.input_payload,
-                "automation_run_id": automation_run_id,
-                "execution_mode": "controlled_retry_check_mode_replay",
-                "retry_of_apply_op_id": apply_op_id,
-                "check_mode": True,
-                "diff": True,
-                "apply_enabled": False,
-                "approval_required_before_apply": False,
-                "owner_review_required": False,
-                "controlled_apply_allowed": True,
-                "controlled_apply_blocker": None,
-                "bounded_execution_scope": "retry_check_mode_only",
-                "next_controlled_action": (
-                    "queue_ai_playbook_or_transport_repair_candidate"
-                ),
-            }
             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)