Merge remote-tracking branch 'origin/main' into codex/agent99-alert-control-20260710

apps/api/src/services/awooop_ansible_check_mode_service.py

@@ -585,6 +585,69 @@ def _claim_from_apply_operation_row(row: dict[str, Any]) -> tuple[AnsibleCheckMo
     return claim, result
 
 
+def _claim_from_stale_check_mode_row(
+    row: dict[str, Any],
+) -> AnsibleCheckModeClaim | None:
+    input_payload = _json_loads(row.get("input"))
+    source_candidate_op_id = str(
+        input_payload.get("source_candidate_op_id")
+        or row.get("parent_op_id")
+        or ""
+    )
+    incident_id = str(
+        row.get("incident_id")
+        or input_payload.get("incident_id")
+        or ""
+    )
+    catalog_id = str(input_payload.get("catalog_id") or "")
+    inventory_hosts = input_payload.get("inventory_hosts")
+    source_playbook_path = str(
+        input_payload.get("source_candidate_playbook_path")
+        or input_payload.get("catalog_playbook_path")
+        or input_payload.get("apply_playbook_path")
+        or input_payload.get("playbook_path")
+        or ""
+    )
+    if (
+        not source_candidate_op_id
+        or not incident_id
+        or not catalog_id
+        or not source_playbook_path
+        or not isinstance(inventory_hosts, list)
+        or not inventory_hosts
+    ):
+        return None
+    try:
+        claim_input = build_ansible_check_mode_claim_input(
+            source_candidate_op_id=source_candidate_op_id,
+            candidate_input={
+                "incident_id": incident_id,
+                "executor": "ansible",
+                "executor_candidates": [
+                    {
+                        "catalog_id": catalog_id,
+                        "playbook_path": source_playbook_path,
+                        "inventory_hosts": inventory_hosts,
+                        "risk_level": input_payload.get("risk_level"),
+                    }
+                ],
+            },
+        )
+    except ValueError:
+        return None
+    return AnsibleCheckModeClaim(
+        op_id=str(row.get("op_id") or ""),
+        source_candidate_op_id=source_candidate_op_id,
+        incident_id=incident_id,
+        catalog_id=str(claim_input["catalog_id"]),
+        playbook_path=str(claim_input["playbook_path"]),
+        apply_playbook_path=str(claim_input["apply_playbook_path"]),
+        inventory_hosts=tuple(str(host) for host in claim_input["inventory_hosts"]),
+        risk_level=str(claim_input.get("risk_level") or ""),
+        input_payload=claim_input,
+    )
+
+
 async def _record_auto_repair_execution_receipt(
     claim: AnsibleCheckModeClaim,
     result: AnsibleRunResult,
@@ -1227,6 +1290,94 @@ async def claim_pending_check_modes(
     return claims
 
 
+async def claim_stale_pending_check_modes(
+    *,
+    project_id: str = "awoooi",
+    limit: int = 1,
+    stale_after_seconds: int,
+) -> list[AnsibleCheckModeClaim]:
+    """Lease and revalidate stale pending rows before replaying check-mode."""
+
+    claims: list[AnsibleCheckModeClaim] = []
+    async with get_db_context(project_id) as db:
+        result = await db.execute(
+            text("""
+                SELECT
+                    check_mode.op_id,
+                    check_mode.parent_op_id,
+                    coalesce(
+                        check_mode.incident_id::text,
+                        check_mode.input ->> 'incident_id'
+                    ) AS incident_id,
+                    check_mode.input
+                FROM automation_operation_log check_mode
+                WHERE check_mode.operation_type = 'ansible_check_mode_executed'
+                  AND check_mode.status = 'pending'
+                  AND coalesce(
+                        nullif(
+                            check_mode.dry_run_result ->> 'reclaimed_at',
+                            ''
+                        )::timestamptz,
+                        check_mode.created_at
+                      ) <= NOW() - (:stale_after_seconds * INTERVAL '1 second')
+                ORDER BY check_mode.created_at DESC
+                LIMIT :limit
+                FOR UPDATE SKIP LOCKED
+            """),
+            {
+                "limit": max(1, limit),
+                "stale_after_seconds": max(300, stale_after_seconds),
+            },
+        )
+        for row_value in result.mappings().all():
+            row = dict(row_value)
+            claim = _claim_from_stale_check_mode_row(row)
+            if claim is None:
+                await db.execute(
+                    text("""
+                        UPDATE automation_operation_log
+                        SET status = 'failed',
+                            error = 'stale_pending_reclaim_rejected_by_current_policy',
+                            dry_run_result = coalesce(
+                                dry_run_result,
+                                '{}'::jsonb
+                            ) || jsonb_build_object(
+                                'claim_state', 'stale_reclaim_rejected',
+                                'reclaimed_at', NOW(),
+                                'check_mode_executed', false
+                            )
+                        WHERE op_id = CAST(:op_id AS uuid)
+                          AND status = 'pending'
+                    """),
+                    {"op_id": str(row.get("op_id") or "")},
+                )
+                continue
+            await db.execute(
+                text("""
+                    UPDATE automation_operation_log
+                    SET input = CAST(:input AS jsonb),
+                        error = NULL,
+                        dry_run_result = coalesce(
+                            dry_run_result,
+                            '{}'::jsonb
+                        ) || jsonb_build_object(
+                            'claim_state', 'stale_reclaimed',
+                            'reclaimed_at', NOW(),
+                            'check_mode_executed', false,
+                            'apply_executed', false
+                        )
+                    WHERE op_id = CAST(:op_id AS uuid)
+                      AND status = 'pending'
+                """),
+                {
+                    "input": json.dumps(claim.input_payload, ensure_ascii=False),
+                    "op_id": claim.op_id,
+                },
+            )
+            claims.append(claim)
+    return claims
+
+
 async def recent_ansible_transport_blockers(
     *,
     project_id: str = "awoooi",
@@ -1355,6 +1506,12 @@ async def finalize_check_mode_claim(
             text("""
                 UPDATE automation_operation_log
                 SET status = :status,
+                    input = jsonb_set(
+                        coalesce(input, '{}'::jsonb),
+                        '{automation_run_id}',
+                        to_jsonb(CAST(:automation_run_id AS text)),
+                        true
+                    ),
                     output = CAST(:output AS jsonb),
                     dry_run_result = CAST(:dry_run_result AS jsonb),
                     error = :error,
@@ -1364,6 +1521,10 @@ async def finalize_check_mode_claim(
             """),
             {
                 "status": status,
+                "automation_run_id": str(
+                    claim.input_payload.get("automation_run_id")
+                    or claim.source_candidate_op_id
+                ),
                 "output": json.dumps(output, ensure_ascii=False),
                 "dry_run_result": json.dumps(dry_run_result, ensure_ascii=False),
                 "error": _tail(error or "", 2000) or None,
@@ -1450,6 +1611,10 @@ async def run_controlled_apply_for_claim(
                 "incident_db_id": _automation_operation_log_incident_id(claim.incident_id),
                 "input": json.dumps({
                     **claim.input_payload,
+                    "automation_run_id": str(
+                        claim.input_payload.get("automation_run_id")
+                        or claim.source_candidate_op_id
+                    ),
                     "execution_mode": "controlled_apply",
                     "check_mode_op_id": claim.op_id,
                     "playbook_path": claim.apply_playbook_path,
@@ -1566,11 +1731,27 @@ async def run_pending_check_modes_once(
         logger.warning("ansible_check_mode_transport_blocked", blockers=transport_blockers)
         return {"claimed": 0, "completed": 0, "failed": 0, "blockers": transport_blockers}
 
-    claims = await claim_pending_check_modes(
+    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,
-        candidate_max_age_hours=settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS,
+        stale_after_seconds=max(300, effective_timeout_seconds + 120),
     )
+    remaining_limit = max(0, limit - len(reclaimed_claims))
+    fresh_claims = (
+        await claim_pending_check_modes(
+            project_id=project_id,
+            limit=remaining_limit,
+            candidate_max_age_hours=(
+                settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS
+            ),
+        )
+        if remaining_limit
+        else []
+    )
+    claims = [*reclaimed_claims, *fresh_claims]
     completed = 0
     failed = 0
     apply_completed = 0
@@ -1579,7 +1760,7 @@ async def run_pending_check_modes_once(
     for claim in claims:
         result = await run_claimed_check_mode(
             claim,
-            timeout_seconds=timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS,
+            timeout_seconds=effective_timeout_seconds,
             project_id=project_id,
         )
         completed += 1
@@ -1599,6 +1780,7 @@ async def run_pending_check_modes_once(
                 apply_failed += 1
     return {
         "claimed": len(claims),
+        "reclaimed": len(reclaimed_claims),
         "completed": completed,
         "failed": failed,
         "apply_completed": apply_completed,