From e2cfbaabc8dd50cbd2605caf39c7d518b633f361 Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 00:28:41 +0800 Subject: [PATCH] fix(awooop): fallback AI alert card readback to direct DB --- .../src/services/platform_operator_service.py | 242 +++++++++++++++++- .../test_awooop_operator_timeline_labels.py | 122 +++++++++ 2 files changed, 351 insertions(+), 13 deletions(-) diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index 94bc6036b..e9871a746 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -111,6 +111,18 @@ _AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS = float( _AI_ALERT_CARD_STATEMENT_TIMEOUT_MS = int( os.getenv("AWOOOP_AI_ALERT_CARD_STATEMENT_TIMEOUT_MS", "200") ) +_AI_ALERT_CARD_DIRECT_CONNECT_TIMEOUT_SECONDS = float( + os.getenv("AWOOOP_AI_ALERT_CARD_DIRECT_CONNECT_TIMEOUT_SECONDS", "1.5") +) +_AI_ALERT_CARD_DIRECT_QUERY_TIMEOUT_SECONDS = float( + os.getenv("AWOOOP_AI_ALERT_CARD_DIRECT_QUERY_TIMEOUT_SECONDS", "1.25") +) +_AI_ALERT_CARD_DIRECT_CLOSE_TIMEOUT_SECONDS = float( + os.getenv("AWOOOP_AI_ALERT_CARD_DIRECT_CLOSE_TIMEOUT_SECONDS", "0.25") +) +_AI_ALERT_CARD_DIRECT_STATEMENT_TIMEOUT_MS = int( + os.getenv("AWOOOP_AI_ALERT_CARD_DIRECT_STATEMENT_TIMEOUT_MS", "1000") +) _AI_ALERT_CARD_LEARNING_REGISTRY_TARGETS: tuple[dict[str, str], ...] = ( { "target": "km", @@ -1406,6 +1418,7 @@ async def list_ai_alert_card_delivery_readback( params["lane"] = normalized_lane where_sql = " AND ".join(where_clauses) + direct_fallback_used = False summary_sql = text(f""" SELECT COUNT(*) AS total, @@ -1527,27 +1540,38 @@ 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, + direct_readback = await _load_ai_alert_card_delivery_readback_direct( 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, + if direct_readback is not None: + summary_row, rows = direct_readback + direct_fallback_used = True + else: + 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, + lane=normalized_lane or None, + page=normalized_page, + per_page=normalized_per_page, 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, - lane=normalized_lane or None, - page=normalized_page, - per_page=normalized_per_page, - error_type=type(exc).__name__, - ) summary = _ai_alert_card_delivery_summary_from_row( summary_row, @@ -1555,6 +1579,14 @@ async def list_ai_alert_card_delivery_readback( event_type=normalized_event_type or None, lane=normalized_lane or None, ) + if direct_fallback_used: + summary.update({ + "db_read_status": "direct_connection_after_session_timeout", + "live_refresh_status": "live_refresh_served_by_direct_connection", + "source_unavailable_reason": "session_timeout_recovered", + "source_unavailable_next_action": "keep_direct_fallback_and_retry_session_pool", + "controlled_repair_queue_present": False, + }) items = [_ai_alert_card_delivery_item(row) for row in rows] learning_registry = _ai_alert_card_learning_registry_readback( items=items, @@ -1608,6 +1640,190 @@ async def list_ai_alert_card_delivery_readback( return response +async def _load_ai_alert_card_delivery_readback_direct( + *, + project_id: str, + event_type: str, + lane: str, + page: int, + per_page: int, +) -> tuple[dict[str, Any], list[dict[str, Any]]] | None: + """Read AI alert card rows through a short-lived DB connection under pool pressure.""" + + db_url = get_settings().DATABASE_URL.replace( + "postgresql+asyncpg://", + "postgresql://", + ) + args: list[Any] = [project_id] + where_clauses = [ + "m.project_id = $1", + "m.channel_type = 'telegram'", + "COALESCE(m.source_envelope::jsonb, '{}'::jsonb) ? 'ai_automation_alert_card'", + ] + if event_type: + args.append(event_type) + where_clauses.append( + "COALESCE(m.source_envelope::jsonb, '{}'::jsonb) #>> " + f"'{{ai_automation_alert_card,event_type}}' = ${len(args)}" + ) + if lane: + args.append(lane) + where_clauses.append( + "COALESCE(m.source_envelope::jsonb, '{}'::jsonb) #>> " + f"'{{ai_automation_alert_card,lane}}' = ${len(args)}" + ) + where_sql = " AND ".join(where_clauses) + limit_param = len(args) + 1 + offset_param = len(args) + 2 + offset = (page - 1) * per_page + try: + import asyncpg + + conn = await asyncio.wait_for( + asyncpg.connect(db_url), + timeout=_AI_ALERT_CARD_DIRECT_CONNECT_TIMEOUT_SECONDS, + ) + try: + await asyncio.wait_for( + conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id), + timeout=_AI_ALERT_CARD_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + await asyncio.wait_for( + conn.execute( + "SET statement_timeout = " + f"'{int(_AI_ALERT_CARD_DIRECT_STATEMENT_TIMEOUT_MS)}ms'" + ), + timeout=_AI_ALERT_CARD_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + summary_row = await asyncio.wait_for( + conn.fetchrow(f""" + SELECT + COUNT(*) AS total, + COUNT(*) FILTER (WHERE m.send_status = 'sent') AS sent_total, + COUNT(*) FILTER (WHERE m.send_status = 'failed') AS failed_total, + COUNT(*) FILTER (WHERE m.send_status = 'pending') AS pending_total, + COUNT(*) FILTER (WHERE m.send_status = 'shadow') AS shadow_total, + COUNT(*) FILTER ( + WHERE COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{ai_automation_alert_card,delivery_receipt_readback_required}}', + '' + ) = 'true' + ) AS delivery_receipt_required_total, + COUNT(*) FILTER ( + WHERE COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{ai_automation_alert_card,runtime_write_gate_count}}', + '0' + ) <> '0' + ) AS runtime_write_gate_open_count, + COUNT(*) FILTER ( + WHERE COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{ai_automation_alert_card,delivery_receipt_readback_required}}', + '' + ) = 'true' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{ai_automation_alert_card,event_type}}', + '' + ) <> '' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{ai_automation_alert_card,lane}}', + '' + ) <> '' + ) AS learning_writeback_ready_total, + MAX(m.sent_at) AS latest_sent_at, + MAX(m.queued_at) AS latest_queued_at + FROM awooop_outbound_message m + WHERE {where_sql} + """, *args), + timeout=_AI_ALERT_CARD_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + list_rows = await asyncio.wait_for( + conn.fetch(f""" + SELECT + m.message_id, + m.project_id, + m.run_id, + m.channel_type, + m.message_type, + m.provider_message_id, + m.send_status, + m.send_error, + m.queued_at, + m.sent_at, + m.triggered_by_state, + ( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) + -> 'ai_automation_alert_card' + )::text AS alert_card_json, + ( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) + -> 'source_refs' + )::text AS source_refs_json, + r.agent_id, + r.state AS run_state, + r.created_at AS run_created_at + FROM awooop_outbound_message m + LEFT JOIN awooop_run_state r + ON r.project_id = m.project_id + AND r.run_id = m.run_id + WHERE {where_sql} + ORDER BY COALESCE(m.sent_at, m.queued_at) DESC, m.message_id DESC + LIMIT ${limit_param} OFFSET ${offset_param} + """, *args, per_page, offset), + timeout=_AI_ALERT_CARD_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + return ( + dict(summary_row) if summary_row is not None else {}, + [_ai_alert_card_direct_row(row) for row in list_rows], + ) + finally: + try: + await asyncio.wait_for( + conn.close(), + timeout=_AI_ALERT_CARD_DIRECT_CLOSE_TIMEOUT_SECONDS, + ) + except Exception as close_exc: # pragma: no cover - live cleanup + logger.warning( + "operator_ai_alert_card_delivery_direct_close_unavailable", + project_id=project_id, + error_type=type(close_exc).__name__, + ) + except Exception as exc: # pragma: no cover - live DB pressure + logger.warning( + "operator_ai_alert_card_delivery_direct_readback_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_direct_row(row: Mapping[str, Any]) -> dict[str, Any]: + payload = dict(row) + payload["alert_card"] = _json_object(payload.pop("alert_card_json", None)) + payload["source_refs"] = _json_object(payload.pop("source_refs_json", None)) + return payload + + +def _json_object(value: Any) -> dict[str, Any]: + if isinstance(value, Mapping): + return dict(value) + if isinstance(value, str) and value.strip(): + try: + payload = json.loads(value) + except json.JSONDecodeError: + return {} + return dict(payload) if isinstance(payload, Mapping) else {} + return {} + + async def _get_cached_ai_alert_card_delivery_readback( *, cache_key: Mapping[str, Any], diff --git a/apps/api/tests/test_awooop_operator_timeline_labels.py b/apps/api/tests/test_awooop_operator_timeline_labels.py index f597fd5b9..987b0d29d 100644 --- a/apps/api/tests/test_awooop_operator_timeline_labels.py +++ b/apps/api/tests/test_awooop_operator_timeline_labels.py @@ -1314,6 +1314,11 @@ async def test_list_ai_alert_card_delivery_readback_failsofts_db_unavailable( "get_db_context", lambda _project_id: FailingDbContext(), ) + monkeypatch.setattr( + platform_operator_service, + "_load_ai_alert_card_delivery_readback_direct", + _no_ai_alert_card_direct_readback, + ) payload = await platform_operator_service.list_ai_alert_card_delivery_readback( project_id="awoooi", @@ -1358,6 +1363,11 @@ async def test_list_ai_alert_card_delivery_readback_failsofts_slow_db( "get_db_context", lambda _project_id: SlowDbContext(), ) + monkeypatch.setattr( + platform_operator_service, + "_load_ai_alert_card_delivery_readback_direct", + _no_ai_alert_card_direct_readback, + ) payload = await platform_operator_service.list_ai_alert_card_delivery_readback( project_id="awoooi", @@ -1428,6 +1438,11 @@ async def test_list_ai_alert_card_delivery_readback_uses_cache_after_live_timeou "get_db_context", lambda _project_id: SlowDbContext(), ) + monkeypatch.setattr( + platform_operator_service, + "_load_ai_alert_card_delivery_readback_direct", + _no_ai_alert_card_direct_readback, + ) monkeypatch.setattr( platform_operator_service, "get_cached_operator_summary_async", @@ -1451,6 +1466,113 @@ async def test_list_ai_alert_card_delivery_readback_uses_cache_after_live_timeou assert payload["learning_registry"]["missing_target_count"] == 0 +async def _no_ai_alert_card_direct_readback(**_kwargs): + return None + + +@pytest.mark.asyncio +async def test_list_ai_alert_card_delivery_readback_uses_direct_fallback( + 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 + + async def direct_readback(**kwargs): + assert kwargs["project_id"] == "awoooi" + return ( + { + "total": 1, + "sent_total": 1, + "failed_total": 0, + "pending_total": 0, + "shadow_total": 0, + "delivery_receipt_required_total": 1, + "runtime_write_gate_open_count": 0, + "learning_writeback_ready_total": 1, + "latest_sent_at": None, + "latest_queued_at": None, + }, + [ + { + "message_id": "message-1", + "project_id": "awoooi", + "run_id": "run-1", + "channel_type": "telegram", + "message_type": "ai_alert_card", + "provider_message_id": "provider-1", + "send_status": "sent", + "send_error": None, + "queued_at": None, + "sent_at": None, + "triggered_by_state": "controlled_apply", + "alert_card": { + "event_type": "provider_freshness_signal", + "lane": "source_provider_freshness_triage", + "target": "provider_freshness_signal", + "delivery_receipt_readback_required": True, + "runtime_write_gate_count": 0, + }, + "source_refs": { + "fingerprints": ["fp-1"], + "alert_ids": ["alert-1"], + }, + "agent_id": "agent", + "run_state": "completed", + "run_created_at": None, + } + ], + ) + + 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, + "_load_ai_alert_card_delivery_readback_direct", + direct_readback, + ) + + payload = await platform_operator_service.list_ai_alert_card_delivery_readback( + project_id="awoooi", + page=1, + per_page=6, + refresh=True, + ) + + assert payload["total"] == 1 + assert payload["summary"]["status"] == "observed" + assert payload["summary"]["db_read_status"] == ( + "direct_connection_after_session_timeout" + ) + assert payload["summary"]["live_refresh_status"] == ( + "live_refresh_served_by_direct_connection" + ) + assert payload["summary"]["controlled_repair_queue_present"] is False + assert payload["learning_registry"]["status"] == "learning_registry_ready" + assert payload["learning_registry"]["ready_target_count"] == 6 + assert payload["items"][0]["learning_writeback_ready"] is True + + def test_list_callback_replies_keeps_audit_summary_separate_from_km_summary() -> None: source = inspect.getsource(platform_operator_service.list_callback_replies)