fix(agents): reuse cached AI alert card readback

apps/api/tests/test_awooop_operator_timeline_labels.py

@@ -1373,6 +1373,84 @@ async def test_list_ai_alert_card_delivery_readback_failsofts_slow_db(
     )
 
 
+@pytest.mark.asyncio
+async def test_list_ai_alert_card_delivery_readback_uses_cache_after_live_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class SlowDb:
+        async def execute(self, *_args, **_kwargs):
+            await asyncio.sleep(1)
+
+    class SlowDbContext:
+        async def __aenter__(self):
+            return SlowDb()
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+    cached_response = (
+        platform_operator_service._ai_alert_card_delivery_source_unavailable_response(
+            project_id="awoooi",
+            event_type=None,
+            lane=None,
+            page=1,
+            per_page=6,
+            error_type="seed",
+        )
+    )
+    cached_response["summary"].update({
+        "status": "observed",
+        "db_read_status": "ok",
+        "total": 36,
+        "learning_registry_missing_target_count": 0,
+    })
+    cached_response["learning_registry"].update({
+        "status": "learning_registry_ready",
+        "ready_target_count": 6,
+        "missing_target_count": 0,
+    })
+
+    async def cached_summary(*_args, **_kwargs):
+        return cached_response
+
+    monkeypatch.setattr(
+        platform_operator_service,
+        "_AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS",
+        0.01,
+    )
+    monkeypatch.setattr(
+        platform_operator_service,
+        "_AI_ALERT_CARD_DB_CONTEXT_EXIT_TIMEOUT_SECONDS",
+        0.01,
+    )
+    monkeypatch.setattr(
+        platform_operator_service,
+        "get_db_context",
+        lambda _project_id: SlowDbContext(),
+    )
+    monkeypatch.setattr(
+        platform_operator_service,
+        "get_cached_operator_summary_async",
+        cached_summary,
+    )
+
+    payload = await platform_operator_service.list_ai_alert_card_delivery_readback(
+        project_id="awoooi",
+        page=1,
+        per_page=6,
+        refresh=True,
+    )
+
+    assert payload["summary"]["status"] == "observed"
+    assert payload["summary"]["db_read_status"] == "cached_after_live_timeout"
+    assert payload["summary"]["source_unavailable_reason"] == "TimeoutError"
+    assert payload["summary"]["live_refresh_status"] == (
+        "live_refresh_unavailable_serving_cached_readback"
+    )
+    assert payload["learning_registry"]["status"] == "learning_registry_ready"
+    assert payload["learning_registry"]["missing_target_count"] == 0
+
+
 def test_list_callback_replies_keeps_audit_summary_separate_from_km_summary() -> None:
     source = inspect.getsource(platform_operator_service.list_callback_replies)