fix(incidents): prefer durable closure readback

apps/api/src/api/v1/incidents.py

@@ -306,7 +306,7 @@ async def get_incident(incident_id: str) -> IncidentResponse:
     incident_service = get_incident_service()
 
     try:
-        incident = await incident_service.get_from_working_memory(incident_id)
+        incident = await incident_service.get_for_readback(incident_id)
         if incident is None:
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,

apps/api/src/services/incident_service.py

@@ -664,15 +664,110 @@ class IncidentService:
             )
             return None
 
+    async def project_terminal_status_to_working_memory(
+        self,
+        incident_id: str,
+        *,
+        project_id: str,
+        status: IncidentStatus,
+        updated_at: datetime,
+        resolved_at: datetime | None,
+    ) -> Literal["updated", "current", "missing", "failed"]:
+        """Project a durable terminal DB state into the Redis read model."""
+
+        if status not in (IncidentStatus.RESOLVED, IncidentStatus.CLOSED):
+            raise ValueError("working_memory_terminal_projection_requires_terminal_status")
+
+        incident = await self.get_from_working_memory(incident_id)
+        hydrated_from_episodic = incident is None
+        if incident is None:
+            incident = await self.get_from_episodic_memory(
+                incident_id,
+                project_id=project_id,
+            )
+        if incident is None:
+            logger.warning(
+                "working_memory_terminal_projection_missing_incident",
+                incident_id=incident_id,
+                project_id=project_id,
+            )
+            return "missing"
+
+        current_updated_at = incident.updated_at
+        if current_updated_at.tzinfo is None:
+            current_updated_at = current_updated_at.replace(tzinfo=UTC)
+        durable_updated_at = updated_at
+        if durable_updated_at.tzinfo is None:
+            durable_updated_at = durable_updated_at.replace(tzinfo=UTC)
+
+        already_current = bool(
+            not hydrated_from_episodic
+            and incident.status in (IncidentStatus.RESOLVED, IncidentStatus.CLOSED)
+            and current_updated_at >= durable_updated_at
+            and (resolved_at is None or incident.resolved_at is not None)
+        )
+        if already_current:
+            return "current"
+
+        if incident.status != IncidentStatus.CLOSED:
+            incident.status = status
+        incident.updated_at = updated_at
+        incident.resolved_at = resolved_at or incident.resolved_at
+        if not await self.save_to_working_memory(incident):
+            logger.warning(
+                "working_memory_terminal_projection_failed",
+                incident_id=incident_id,
+                project_id=project_id,
+            )
+            return "failed"
+
+        logger.info(
+            "working_memory_terminal_projection_updated",
+            incident_id=incident_id,
+            project_id=project_id,
+            status=incident.status.value,
+            hydrated_from_episodic=hydrated_from_episodic,
+        )
+        return "updated"
+
+    async def get_for_readback(self, incident_id: str) -> Incident | None:
+        """Return durable incident truth, with Redis as an availability fallback."""
+        incident = await self.get_from_episodic_memory(incident_id)
+        if incident is not None:
+            return incident
+
+        logger.warning(
+            "incident_readback_falling_back_to_working_memory",
+            incident_id=incident_id,
+        )
+        return await self.get_from_working_memory(incident_id)
+
     async def get_active_incidents(self) -> list[Incident]:
         """
-        列出所有活躍的 Incidents (從 Working Memory)
+        列出所有活躍的 Incidents。
 
-        方案 C: 解析時正規化舊格式 Enum 值
+        PostgreSQL 是 durable source of truth；只有 repository 暫時不可用時，
+        才退回 Redis working memory 維持讀取可用性。
 
         Returns:
             list[Incident]: 活躍事件列表 (investigating 或 mitigating)
         """
+        try:
+            from src.repositories.incident_repository import get_incident_repository
+
+            incidents = await get_incident_repository().get_active()
+            logger.info(
+                "get_active_incidents_from_episodic_memory",
+                count=len(incidents),
+            )
+            return incidents
+        except Exception as e:
+            logger.warning(
+                "active_incidents_episodic_read_failed",
+                error=str(e),
+                fallback="working_memory",
+            )
+
         redis_client = get_redis()
         incidents: list[Incident] = []
 
@@ -805,7 +900,12 @@ class IncidentService:
             )
             return False
 
-    async def get_from_episodic_memory(self, incident_id: str) -> Incident | None:
+    async def get_from_episodic_memory(
+        self,
+        incident_id: str,
+        *,
+        project_id: str | None = None,
+    ) -> Incident | None:
         """
         從 Episodic Memory 讀取 Incident
 
@@ -813,7 +913,10 @@ class IncidentService:
             Incident | None: 事件資料，若不存在則返回 None
         """
         try:
-            async with get_db_context() as db:
+            db_context = (
+                get_db_context(project_id) if project_id else get_db_context()
+            )
+            async with db_context as db:
                 from sqlalchemy import select
 
                 stmt = select(IncidentRecord).where(

apps/api/tests/test_incident_readback_source_truth.py

@@ -0,0 +1,104 @@
+"""Incident readback must not regress durable closure to stale cache state."""
+
+from datetime import UTC, datetime
+from unittest.mock import AsyncMock
+
+import pytest
+
+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
+
+
+def _incident(status: IncidentStatus, *, minute: int) -> Incident:
+    timestamp = datetime(2026, 7, 11, 11, minute, tzinfo=UTC)
+    return Incident(
+        incident_id="INC-20260711-SOURCE-TRUTH",
+        status=status,
+        severity=Severity.P2,
+        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,
+    )
+
+
+@pytest.mark.asyncio
+async def test_readback_prefers_durable_resolved_incident_over_stale_cache(monkeypatch):
+    service = IncidentService()
+    durable = _incident(IncidentStatus.RESOLVED, minute=38)
+    cached = _incident(IncidentStatus.INVESTIGATING, minute=11)
+    monkeypatch.setattr(
+        service,
+        "get_from_episodic_memory",
+        AsyncMock(return_value=durable),
+    )
+    working_read = AsyncMock(return_value=cached)
+    monkeypatch.setattr(service, "get_from_working_memory", working_read)
+
+    result = await service.get_for_readback(durable.incident_id)
+
+    assert result is durable
+    assert result.status is IncidentStatus.RESOLVED
+    working_read.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_readback_falls_back_to_working_memory_when_durable_record_unavailable(
+    monkeypatch,
+):
+    service = IncidentService()
+    cached = _incident(IncidentStatus.INVESTIGATING, minute=11)
+    monkeypatch.setattr(
+        service,
+        "get_from_episodic_memory",
+        AsyncMock(return_value=None),
+    )
+    monkeypatch.setattr(
+        service,
+        "get_from_working_memory",
+        AsyncMock(return_value=cached),
+    )
+
+    result = await service.get_for_readback(cached.incident_id)
+
+    assert result is cached
+
+
+@pytest.mark.asyncio
+async def test_incident_detail_endpoint_uses_durable_readback(monkeypatch):
+    durable = _incident(IncidentStatus.RESOLVED, minute=38)
+
+    class _Service:
+        get_for_readback = AsyncMock(return_value=durable)
+
+    service = _Service()
+    monkeypatch.setattr(incidents_api, "get_incident_service", lambda: service)
+
+    result = await incidents_api.get_incident(durable.incident_id)
+
+    assert result.incident_id == durable.incident_id
+    assert result.status == IncidentStatus.RESOLVED.value
+    service.get_for_readback.assert_awaited_once_with(durable.incident_id)
+
+
+@pytest.mark.asyncio
+async def test_active_incidents_come_from_durable_repository(monkeypatch):
+    service = IncidentService()
+    durable_active = _incident(IncidentStatus.INVESTIGATING, minute=38)
+
+    class _Repository:
+        get_active = AsyncMock(return_value=[durable_active])
+
+    repository = _Repository()
+    monkeypatch.setattr(
+        incident_repository_module,
+        "get_incident_repository",
+        lambda: repository,
+    )
+
+    incidents = await service.get_active_incidents()
+
+    assert incidents == [durable_active]
+    repository.get_active.assert_awaited_once_with()