From 155ea47958930be852f9bbd889d474064a8bad20 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 19:56:24 +0800 Subject: [PATCH] fix(agent): reconcile terminal incident read models --- .../src/jobs/awooop_ansible_check_mode_job.py | 37 +++++- .../awooop_ansible_check_mode_service.py | 121 +++++++++++++++++- .../src/workers/ansible_executor_broker.py | 4 + .../tests/test_ansible_verified_closure.py | 102 +++++++++++++++ 4 files changed, 261 insertions(+), 3 deletions(-) diff --git a/apps/api/src/jobs/awooop_ansible_check_mode_job.py b/apps/api/src/jobs/awooop_ansible_check_mode_job.py index b4236fb84..95d5509b8 100644 --- a/apps/api/src/jobs/awooop_ansible_check_mode_job.py +++ b/apps/api/src/jobs/awooop_ansible_check_mode_job.py @@ -14,7 +14,10 @@ import random import structlog from src.core.config import settings -from src.services.awooop_ansible_check_mode_service import run_pending_check_modes_once +from src.services.awooop_ansible_check_mode_service import ( + reconcile_verified_incident_working_memory_once, + run_pending_check_modes_once, +) logger = structlog.get_logger(__name__) _ERROR_BACKOFF_MIN_SECONDS = 15.0 @@ -45,9 +48,39 @@ async def run_awooop_ansible_check_mode_loop() -> None: limit=settings.AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT, timeout_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS, ) + projection = await reconcile_verified_incident_working_memory_once( + limit=max( + 1, + min(settings.AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT * 2, 10), + ), + ) + result.update( + { + "working_memory_projection_scanned": int( + projection.get("scanned") or 0 + ), + "working_memory_projection_updated": int( + projection.get("updated") or 0 + ), + "working_memory_projection_missing": int( + projection.get("missing") or 0 + ), + "working_memory_projection_failed": int( + projection.get("failed") or 0 + ), + "working_memory_projection_error": ( + str(projection.get("error") or "")[:500] or None + ), + } + ) if result.get("error") or result.get("catalog_replay_error"): sleep_seconds = _error_backoff_seconds() - if result.get("claimed") or result.get("blockers"): + if ( + result.get("claimed") + or result.get("blockers") + or result.get("working_memory_projection_updated") + or result.get("working_memory_projection_error") + ): logger.info("awooop_ansible_check_mode_worker_tick", **result) except Exception as exc: logger.warning("awooop_ansible_check_mode_worker_failed", error=str(exc)) diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 59144aa09..57a2ab6ea 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -2259,6 +2259,99 @@ async def _read_verified_apply_closure_prerequisites( } +async def _project_incident_terminal_to_working_memory( + *, + incident_id: str, + project_id: str, + incident_status: str, + updated_at: datetime, + resolved_at: datetime | None, +) -> str: + from src.models.incident import IncidentStatus + from src.services.incident_service import get_incident_service + + normalized_status = str(incident_status or "").strip().lower() + if normalized_status not in { + IncidentStatus.RESOLVED.value, + IncidentStatus.CLOSED.value, + }: + return "failed" + return await get_incident_service().project_terminal_status_to_working_memory( + incident_id, + project_id=project_id, + status=IncidentStatus(normalized_status), + updated_at=updated_at, + resolved_at=resolved_at, + ) + + +async def reconcile_verified_incident_working_memory_once( + *, + project_id: str = "awoooi", + window_hours: int = 24, + limit: int = 10, +) -> dict[str, Any]: + """Repair terminal DB truth that is still active in Redis working memory.""" + + stats: dict[str, Any] = { + "scanned": 0, + "updated": 0, + "current": 0, + "missing": 0, + "failed": 0, + "error": None, + } + try: + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + SELECT + incident.incident_id, + incident.status::text AS incident_status, + incident.updated_at, + incident.resolved_at + FROM incidents incident + WHERE incident.project_id = :project_id + AND upper(incident.status::text) IN ('RESOLVED', 'CLOSED') + AND incident.resolved_at IS NOT NULL + AND cast(incident.outcome AS jsonb) #>> + '{automation_terminal,incident_resolved}' = 'true' + AND incident.updated_at >= + NOW() - (:window_hours * INTERVAL '1 hour') + ORDER BY incident.updated_at DESC + LIMIT :limit + """), + { + "project_id": project_id, + "window_hours": max(1, int(window_hours)), + "limit": max(1, int(limit)), + }, + ) + rows = [dict(row) for row in result.mappings().all()] + + stats["scanned"] = len(rows) + for row in rows: + projection_status = await _project_incident_terminal_to_working_memory( + incident_id=str(row.get("incident_id") or ""), + project_id=project_id, + incident_status=str(row.get("incident_status") or ""), + updated_at=row["updated_at"], + resolved_at=row.get("resolved_at"), + ) + if projection_status in stats: + stats[projection_status] += 1 + else: + stats["failed"] += 1 + except Exception as exc: + stats["error"] = f"{type(exc).__name__}: {exc}"[:500] + logger.warning( + "ansible_incident_working_memory_reconcile_failed", + project_id=project_id, + error_type=type(exc).__name__, + ) + return stats + + async def _read_incident_closure_readback( claim: AnsibleCheckModeClaim, *, @@ -2633,7 +2726,7 @@ async def _commit_verified_incident_closure( and row.get("resolved_at") is not None ): raise RuntimeError("verified_incident_closure_atomic_readback_failed") - return { + closure_result = { **terminal, "incident_status": str(row.get("incident_status") or ""), "incident_row_version": ( @@ -2644,6 +2737,26 @@ async def _commit_verified_incident_closure( "closure_receipt_written": True, "resolved_lifecycle_written": True, } + projection_status = await _project_incident_terminal_to_working_memory( + incident_id=claim.incident_id, + project_id=project_id, + incident_status=closure_result["incident_status"], + updated_at=row["updated_at"], + resolved_at=row.get("resolved_at"), + ) + closure_result["working_memory_projection_status"] = projection_status + closure_result["working_memory_projection_ready"] = ( + projection_status in {"updated", "current"} + ) + if closure_result["working_memory_projection_ready"] is not True: + logger.warning( + "ansible_verified_incident_closure_working_memory_pending", + automation_run_id=automation_run_id, + incident_id=claim.incident_id, + apply_op_id=apply_op_id, + projection_status=projection_status, + ) + return closure_result except Exception as exc: logger.warning( "ansible_verified_incident_closure_atomic_write_failed", @@ -2689,6 +2802,12 @@ async def _finalize_verified_apply_closure( "resolved_lifecycle", ], } + if terminal.get("working_memory_projection_ready") is not True: + return { + "status": "working_memory_projection_pending", + "closed": False, + "missing": ["working_memory_terminal_projection"], + } readback = await _read_incident_closure_readback( claim, diff --git a/apps/api/src/workers/ansible_executor_broker.py b/apps/api/src/workers/ansible_executor_broker.py index 34b781c80..21bc2812d 100644 --- a/apps/api/src/workers/ansible_executor_broker.py +++ b/apps/api/src/workers/ansible_executor_broker.py @@ -10,6 +10,7 @@ from typing import Any import structlog from src.core.config import settings +from src.core.redis_client import close_redis_pool, init_redis_pool from src.db.base import close_db from src.jobs.awooop_ansible_check_mode_job import ( run_awooop_ansible_check_mode_loop, @@ -39,6 +40,8 @@ async def _main() -> None: if not ssh_key_path.is_file() or not known_hosts_path.is_file(): raise RuntimeError("ansible_execution_broker_transport_not_mounted") + await init_redis_pool() + shutdown_event = asyncio.Event() def _shutdown_handler(signum: int, _frame: Any) -> None: @@ -90,6 +93,7 @@ async def _main() -> None: await task except asyncio.CancelledError: pass + await close_redis_pool() await close_db() _HEALTH_PATH.unlink(missing_ok=True) _READY_PATH.unlink(missing_ok=True) diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index eb506ccef..c75c6858d 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -3,6 +3,7 @@ from __future__ import annotations import inspect import json from contextlib import asynccontextmanager +from datetime import UTC, datetime from unittest.mock import AsyncMock import pytest @@ -42,6 +43,17 @@ class _SequenceDB: return self._results.pop(0) if self._results else _MappingResult() +class _RowsResult: + def __init__(self, rows: list[dict]) -> None: + self._rows = rows + + def mappings(self) -> _RowsResult: + return self + + def all(self) -> list[dict]: + return self._rows + + def _claim() -> service.AnsibleCheckModeClaim: return service.AnsibleCheckModeClaim( op_id="00000000-0000-0000-0000-000000000101", @@ -224,6 +236,7 @@ async def test_closure_resolves_only_after_durable_readback( "incident_resolved": True, "closure_receipt_written": True, "resolved_lifecycle_written": True, + "working_memory_projection_ready": True, } ) monkeypatch.setattr( @@ -255,6 +268,95 @@ async def test_closure_resolves_only_after_durable_readback( assert atomic_writer.await_args.kwargs["closure_prerequisite_count"] == 1 +@pytest.mark.asyncio +async def test_closure_waits_for_working_memory_projection( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + service, + "_read_verified_apply_closure_prerequisites", + AsyncMock(return_value={"ready": True, "receipts": {"all": True}}), + ) + monkeypatch.setattr( + service, + "_commit_verified_incident_closure", + AsyncMock(return_value={"working_memory_projection_ready": False}), + ) + readback = AsyncMock() + monkeypatch.setattr(service, "_read_incident_closure_readback", readback) + + result = await service._finalize_verified_apply_closure( + _claim(), + apply_op_id="00000000-0000-0000-0000-000000000104", + project_id="awoooi", + ) + + assert result == { + "status": "working_memory_projection_pending", + "closed": False, + "missing": ["working_memory_terminal_projection"], + } + readback.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_terminal_working_memory_reconciler_projects_durable_rows( + monkeypatch: pytest.MonkeyPatch, +) -> None: + updated_at = datetime(2026, 7, 11, 11, 45, tzinfo=UTC) + db = _SequenceDB( + _RowsResult( + [ + { + "incident_id": "INC-20260711-C50677", + "incident_status": "RESOLVED", + "updated_at": updated_at, + "resolved_at": updated_at, + }, + { + "incident_id": "INC-20260711-8A28BC", + "incident_status": "RESOLVED", + "updated_at": updated_at, + "resolved_at": updated_at, + }, + ] + ) + ) + + @asynccontextmanager + async def fake_db_context(_project_id: str): + yield db + + project = AsyncMock(side_effect=["updated", "current"]) + monkeypatch.setattr(service, "get_db_context", fake_db_context) + monkeypatch.setattr( + service, + "_project_incident_terminal_to_working_memory", + project, + ) + + result = await service.reconcile_verified_incident_working_memory_once( + project_id="awoooi", + window_hours=24, + limit=6, + ) + + assert result == { + "scanned": 2, + "updated": 1, + "current": 1, + "missing": 0, + "failed": 0, + "error": None, + } + assert project.await_count == 2 + assert db.parameters[0] == { + "project_id": "awoooi", + "window_hours": 24, + "limit": 6, + } + + @pytest.mark.asyncio async def test_atomic_closure_rolls_back_when_bundle_readback_is_missing( monkeypatch: pytest.MonkeyPatch,