fix(awooop): fallback AI alert card readback to direct DB

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],

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)