fix(agent): fail closed on incident read drift

apps/api/tests/test_ansible_verified_closure.py

@@ -8,9 +8,11 @@ from unittest.mock import AsyncMock
 
 import pytest
 
+from src.core import redis_client as redis_client_module
 from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job
 from src.services import awooop_ansible_check_mode_service as service
 from src.services.telegram_gateway import TelegramGateway
+from src.workers import ansible_executor_broker as broker
 
 
 class _MappingResult:
@@ -269,7 +271,7 @@ async def test_closure_resolves_only_after_durable_readback(
 
 
 @pytest.mark.asyncio
-async def test_closure_waits_for_working_memory_projection(
+async def test_closure_preserves_durable_terminal_when_projection_is_degraded(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     monkeypatch.setattr(
@@ -282,7 +284,13 @@ async def test_closure_waits_for_working_memory_projection(
         "_commit_verified_incident_closure",
         AsyncMock(return_value={"working_memory_projection_ready": False}),
     )
-    readback = AsyncMock()
+    readback = AsyncMock(
+        return_value={
+            "closed": True,
+            "missing": [],
+            "incident_status": "RESOLVED",
+        }
+    )
     monkeypatch.setattr(service, "_read_incident_closure_readback", readback)
 
     result = await service._finalize_verified_apply_closure(
@@ -292,17 +300,21 @@ async def test_closure_waits_for_working_memory_projection(
     )
 
     assert result == {
-        "status": "working_memory_projection_pending",
-        "closed": False,
+        "status": "controlled_apply_closed_projection_degraded",
+        "closed": True,
         "missing": ["working_memory_terminal_projection"],
+        "incident_status": "RESOLVED",
+        "working_memory_projection_ready": False,
+        "read_model_degraded": True,
     }
-    readback.assert_not_awaited()
+    readback.assert_awaited_once()
 
 
 @pytest.mark.asyncio
 async def test_terminal_working_memory_reconciler_projects_durable_rows(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    service._WORKING_MEMORY_PROJECTION_CURSOR.clear()
     updated_at = datetime(2026, 7, 11, 11, 45, tzinfo=UTC)
     db = _SequenceDB(
         _RowsResult(
@@ -354,7 +366,167 @@ async def test_terminal_working_memory_reconciler_projects_durable_rows(
         "project_id": "awoooi",
         "window_hours": 24,
         "limit": 6,
+        "cursor_updated_at": None,
+        "cursor_incident_id": "",
     }
+    query = db.statements[0]
+    assert "working_memory_projection,ready" in query
+    assert "source_updated_at_epoch_us" in query
+    assert "cursor_updated_at" in query
+    assert "ORDER BY incident.updated_at ASC" in query
+    assert service._WORKING_MEMORY_PROJECTION_WINDOW_HOURS >= 7 * 24
+    service._WORKING_MEMORY_PROJECTION_CURSOR.clear()
+
+
+@pytest.mark.asyncio
+async def test_terminal_reconciler_advances_cursor_after_one_row_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    service._WORKING_MEMORY_PROJECTION_CURSOR.clear()
+    first_at = datetime(2026, 7, 11, 11, 40, tzinfo=UTC)
+    second_at = datetime(2026, 7, 11, 11, 41, tzinfo=UTC)
+    db = _SequenceDB(
+        _RowsResult(
+            [
+                {
+                    "incident_id": "INC-20260711-FAIL01",
+                    "incident_status": "RESOLVED",
+                    "updated_at": first_at,
+                    "resolved_at": first_at,
+                },
+                {
+                    "incident_id": "INC-20260711-NEXT01",
+                    "incident_status": "RESOLVED",
+                    "updated_at": second_at,
+                    "resolved_at": second_at,
+                },
+            ]
+        )
+    )
+
+    @asynccontextmanager
+    async def fake_db_context(_project_id: str):
+        yield db
+
+    project = AsyncMock(
+        side_effect=[RuntimeError("redis unavailable"), "updated"]
+    )
+    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",
+        limit=2,
+    )
+
+    assert result["scanned"] == 2
+    assert result["failed"] == 1
+    assert result["updated"] == 1
+    assert project.await_count == 2
+    assert service._WORKING_MEMORY_PROJECTION_CURSOR["awoooi"] == (
+        second_at,
+        "INC-20260711-NEXT01",
+    )
+    service._WORKING_MEMORY_PROJECTION_CURSOR.clear()
+
+
+@pytest.mark.asyncio
+async def test_projection_exception_does_not_erase_durable_closure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    terminal = {
+        "automation_run_id": "00000000-0000-0000-0000-000000000102",
+        "apply_op_id": "00000000-0000-0000-0000-000000000104",
+    }
+    updated_at = datetime(2026, 7, 11, 11, 45, tzinfo=UTC)
+    db = _SequenceDB(
+        _MappingResult(),
+        _MappingResult(
+            {
+                "incident_status": "RESOLVED",
+                "updated_at": updated_at,
+                "resolved_at": updated_at,
+                "outcome": {
+                    "automation_terminal": terminal,
+                    "proposal_executed": True,
+                    "execution_success": True,
+                },
+                "closure_receipt_written": True,
+                "resolved_lifecycle_written": True,
+            }
+        ),
+    )
+    transaction_completed = False
+
+    @asynccontextmanager
+    async def fake_db_context(_project_id: str):
+        nonlocal transaction_completed
+        yield db
+        transaction_completed = True
+
+    monkeypatch.setattr(service, "get_db_context", fake_db_context)
+    monkeypatch.setattr(
+        service,
+        "_project_incident_terminal_to_working_memory",
+        AsyncMock(side_effect=RuntimeError("redis unavailable")),
+    )
+
+    result = await service._commit_verified_incident_closure(
+        _claim(),
+        apply_op_id="00000000-0000-0000-0000-000000000104",
+        project_id="awoooi",
+        closure_prerequisite_count=12,
+    )
+
+    assert transaction_completed is True
+    assert result is not None
+    assert result["incident_resolved"] is True
+    assert result["closure_receipt_written"] is True
+    assert result["working_memory_projection_status"] == "failed"
+    assert result["working_memory_projection_ready"] is False
+
+
+@pytest.mark.asyncio
+async def test_broker_redis_projection_init_is_fail_soft(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        broker,
+        "init_redis_pool",
+        AsyncMock(side_effect=RuntimeError("redis unavailable")),
+    )
+
+    assert await broker._init_optional_projection_redis() is False
+
+
+@pytest.mark.asyncio
+async def test_failed_redis_init_does_not_poison_projection_retry(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    failed_pool = type(
+        "_FailedPool",
+        (),
+        {
+            "ping": AsyncMock(side_effect=RuntimeError("redis unavailable")),
+            "close": AsyncMock(),
+        },
+    )()
+    monkeypatch.setattr(redis_client_module, "_redis_pool", None)
+    monkeypatch.setattr(
+        redis_client_module.redis,
+        "from_url",
+        lambda *_args, **_kwargs: failed_pool,
+    )
+
+    with pytest.raises(RuntimeError, match="redis unavailable"):
+        await redis_client_module.init_redis_pool()
+
+    assert redis_client_module._redis_pool is None
+    failed_pool.close.assert_awaited_once_with()
 
 
 @pytest.mark.asyncio

apps/api/tests/test_incident_readback_source_truth.py

@@ -4,11 +4,15 @@ from datetime import UTC, datetime
 from unittest.mock import AsyncMock
 
 import pytest
+from fastapi import HTTPException
 
 from src.api.v1 import incidents as incidents_api
 from src.models.incident import Incident, IncidentStatus, Severity
 from src.repositories import incident_repository as incident_repository_module
-from src.services.incident_service import IncidentService
+from src.services.incident_service import (
+    IncidentDurableReadError,
+    IncidentService,
+)
 
 
 def _incident(status: IncidentStatus, *, minute: int) -> Incident:
@@ -20,7 +24,12 @@ def _incident(status: IncidentStatus, *, minute: int) -> Incident:
         affected_services=["awoooi-api"],
         created_at=datetime(2026, 7, 11, 10, 0, tzinfo=UTC),
         updated_at=timestamp,
-        resolved_at=timestamp if status is IncidentStatus.RESOLVED else None,
+        resolved_at=(
+            timestamp
+            if status in (IncidentStatus.RESOLVED, IncidentStatus.CLOSED)
+            else None
+        ),
+        closed_at=timestamp if status is IncidentStatus.CLOSED else None,
     )
 
 
@@ -31,7 +40,7 @@ async def test_readback_prefers_durable_resolved_incident_over_stale_cache(monke
     cached = _incident(IncidentStatus.INVESTIGATING, minute=11)
     monkeypatch.setattr(
         service,
-        "get_from_episodic_memory",
+        "_get_from_episodic_memory_strict",
         AsyncMock(return_value=durable),
     )
     working_read = AsyncMock(return_value=cached)
@@ -45,27 +54,28 @@ async def test_readback_prefers_durable_resolved_incident_over_stale_cache(monke
     assert result is durable
     assert result.status is IncidentStatus.RESOLVED
     working_read.assert_not_awaited()
-    service.get_from_episodic_memory.assert_awaited_once_with(
+    service._get_from_episodic_memory_strict.assert_awaited_once_with(
         durable.incident_id,
         project_id="awoooi",
     )
 
 
 @pytest.mark.asyncio
-async def test_readback_falls_back_to_working_memory_when_durable_record_unavailable(
+async def test_readback_returns_missing_without_trusting_working_memory(
     monkeypatch,
 ):
     service = IncidentService()
     cached = _incident(IncidentStatus.INVESTIGATING, minute=11)
     monkeypatch.setattr(
         service,
-        "get_from_episodic_memory",
+        "_get_from_episodic_memory_strict",
         AsyncMock(return_value=None),
     )
+    working_read = AsyncMock(return_value=cached)
     monkeypatch.setattr(
         service,
         "get_from_working_memory",
-        AsyncMock(return_value=cached),
+        working_read,
     )
 
     result = await service.get_for_readback(
@@ -73,7 +83,74 @@ async def test_readback_falls_back_to_working_memory_when_durable_record_unavail
         project_id="awoooi",
     )
 
-    assert result is cached
+    assert result is None
+    working_read.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_readback_db_failure_is_degraded_not_cache_not_found(monkeypatch):
+    service = IncidentService()
+    working_read = AsyncMock(
+        return_value=_incident(IncidentStatus.INVESTIGATING, minute=11)
+    )
+    monkeypatch.setattr(
+        service,
+        "_get_from_episodic_memory_strict",
+        AsyncMock(side_effect=RuntimeError("postgres unavailable")),
+    )
+    monkeypatch.setattr(service, "get_from_working_memory", working_read)
+
+    with pytest.raises(
+        IncidentDurableReadError,
+        match="incident_durable_readback_unavailable",
+    ):
+        await service.get_for_readback(
+            "INC-20260711-SOURCE-TRUTH",
+            project_id="awoooi",
+        )
+
+    working_read.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_terminal_projection_replaces_newer_wrong_cache_terminal(
+    monkeypatch,
+):
+    service = IncidentService()
+    durable = _incident(IncidentStatus.CLOSED, minute=38)
+    cached = _incident(IncidentStatus.RESOLVED, minute=59)
+    save = AsyncMock(return_value=True)
+    marker = AsyncMock(return_value=True)
+    monkeypatch.setattr(
+        service,
+        "_get_from_episodic_memory_strict",
+        AsyncMock(return_value=durable),
+    )
+    monkeypatch.setattr(
+        service,
+        "get_from_working_memory",
+        AsyncMock(return_value=cached),
+    )
+    monkeypatch.setattr(service, "save_to_working_memory", save)
+    monkeypatch.setattr(service, "_mark_terminal_projection_ready", marker)
+
+    result = await service.project_terminal_status_to_working_memory(
+        durable.incident_id,
+        project_id="awoooi",
+        status=IncidentStatus.CLOSED,
+        updated_at=durable.updated_at,
+        resolved_at=durable.resolved_at,
+    )
+
+    assert result == "updated"
+    save.assert_awaited_once_with(durable)
+    assert save.await_args.args[0].status is IncidentStatus.CLOSED
+    marker.assert_awaited_once_with(
+        incident_id=durable.incident_id,
+        project_id="awoooi",
+        status=IncidentStatus.CLOSED,
+        updated_at=durable.updated_at,
+    )
 
 
 @pytest.mark.asyncio
@@ -118,3 +195,55 @@ async def test_active_incidents_come_from_durable_repository(monkeypatch):
 
     assert incidents == [durable_active]
     repository.get_active.assert_awaited_once_with(project_id="awoooi")
+
+
+@pytest.mark.asyncio
+async def test_active_incidents_fail_closed_when_postgres_is_unavailable(
+    monkeypatch,
+):
+    service = IncidentService()
+
+    class _Repository:
+        get_active = AsyncMock(side_effect=RuntimeError("postgres unavailable"))
+
+    monkeypatch.setattr(
+        incident_repository_module,
+        "get_incident_repository",
+        lambda: _Repository(),
+    )
+
+    with pytest.raises(
+        IncidentDurableReadError,
+        match="active_incidents_durable_read_unavailable",
+    ):
+        await service.get_active_incidents(project_id="awoooi")
+
+
+@pytest.mark.asyncio
+async def test_incident_detail_reports_durable_readback_degraded(monkeypatch):
+    service = type(
+        "_Service",
+        (),
+        {
+            "get_for_readback": AsyncMock(
+                side_effect=IncidentDurableReadError(
+                    "incident_durable_readback_unavailable"
+                )
+            )
+        },
+    )()
+    monkeypatch.setattr(incidents_api, "get_incident_service", lambda: service)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await incidents_api.get_incident(
+            "INC-20260711-SOURCE-TRUTH",
+            project_id="awoooi",
+        )
+
+    assert exc_info.value.status_code == 503
+    assert exc_info.value.detail == {
+        "status": "degraded",
+        "reason_code": "incident_durable_readback_unavailable",
+        "source_of_truth": "postgresql",
+        "redis_fallback_used": False,
+    }