diff --git a/apps/api/src/api/v1/platform/operator_runs.py b/apps/api/src/api/v1/platform/operator_runs.py index a9991cdfb..606f4c010 100644 --- a/apps/api/src/api/v1/platform/operator_runs.py +++ b/apps/api/src/api/v1/platform/operator_runs.py @@ -160,6 +160,7 @@ class AiAlertCardDeliverySummary(BaseModel): source_unavailable_reason: str | None = None source_unavailable_next_action: str = "none" controlled_repair_queue_present: bool = False + live_refresh_status: str = "ok" total: int sent_total: int failed_total: int diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index 94a90bf6c..e498c601b 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -1362,23 +1362,14 @@ async def list_ai_alert_card_delivery_readback( "per_page": normalized_per_page, } if not refresh: - try: - cached_response = await get_cached_operator_summary_async( - "ai_alert_card_delivery_readback", - cache_key, - ttl_seconds=_AI_ALERT_CARD_CACHE_TTL_SECONDS, - ) - except Exception as exc: # pragma: no cover - cache pressure - logger.warning( - "operator_ai_alert_card_delivery_readback_cache_unavailable", - project_id=normalized_project_id, - event_type=normalized_event_type, - lane=normalized_lane, - page=normalized_page, - per_page=normalized_per_page, - error_type=type(exc).__name__, - ) - cached_response = None + cached_response = await _get_cached_ai_alert_card_delivery_readback( + cache_key=cache_key, + project_id=normalized_project_id, + event_type=normalized_event_type, + lane=normalized_lane, + page=normalized_page, + per_page=normalized_per_page, + ) if cached_response is not None and "learning_registry" in cached_response: logger.info( "operator_ai_alert_card_delivery_readback_cache_hit", @@ -1536,6 +1527,19 @@ async def list_ai_alert_card_delivery_readback( per_page=normalized_per_page, error_type=type(exc).__name__, ) + cached_fallback = await _get_cached_ai_alert_card_delivery_readback( + cache_key=cache_key, + project_id=normalized_project_id, + event_type=normalized_event_type, + lane=normalized_lane, + page=normalized_page, + per_page=normalized_per_page, + ) + if cached_fallback is not None and "learning_registry" in cached_fallback: + return _ai_alert_card_delivery_cached_live_timeout_response( + cached_response=cached_fallback, + error_type=type(exc).__name__, + ) return _ai_alert_card_delivery_source_unavailable_response( project_id=normalized_project_id, event_type=normalized_event_type or None, @@ -1604,6 +1608,54 @@ async def list_ai_alert_card_delivery_readback( return response +async def _get_cached_ai_alert_card_delivery_readback( + *, + cache_key: Mapping[str, Any], + project_id: str, + event_type: str, + lane: str, + page: int, + per_page: int, +) -> dict[str, Any] | None: + try: + return await get_cached_operator_summary_async( + "ai_alert_card_delivery_readback", + cache_key, + ttl_seconds=_AI_ALERT_CARD_CACHE_TTL_SECONDS, + ) + except Exception as exc: # pragma: no cover - cache pressure + logger.warning( + "operator_ai_alert_card_delivery_readback_cache_unavailable", + project_id=project_id, + event_type=event_type, + lane=lane, + page=page, + per_page=per_page, + error_type=type(exc).__name__, + ) + return None + + +def _ai_alert_card_delivery_cached_live_timeout_response( + *, + cached_response: Mapping[str, Any], + error_type: str, +) -> dict[str, Any]: + response = dict(cached_response) + summary = dict(_as_dict(response.get("summary"))) + summary.update({ + "db_read_status": "cached_after_live_timeout", + "source_unavailable_reason": error_type, + "source_unavailable_next_action": ( + "retry_live_refresh_keep_cached_learning_registry" + ), + "live_refresh_status": "live_refresh_unavailable_serving_cached_readback", + "controlled_repair_queue_present": False, + }) + response["summary"] = summary + return response + + def _ai_alert_card_delivery_source_unavailable_response( *, project_id: str, @@ -1635,6 +1687,7 @@ def _ai_alert_card_delivery_source_unavailable_response( "repair_awooop_ai_alert_card_delivery_readback" ), "controlled_repair_queue_present": True, + "live_refresh_status": "live_refresh_unavailable_no_cached_readback", "runtime_write_allowed": False, "learning_writeback_next_action": ( "repair_awooop_ai_alert_card_delivery_readback" diff --git a/apps/api/tests/test_awooop_operator_timeline_labels.py b/apps/api/tests/test_awooop_operator_timeline_labels.py index dcc398951..f597fd5b9 100644 --- a/apps/api/tests/test_awooop_operator_timeline_labels.py +++ b/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)