diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index 73897771b..7a553089a 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -1515,7 +1515,29 @@ async def list_ai_alert_card_delivery_readback( FROM awooop_outbound_message m WHERE {where_sql} """) + # Bound the page and the Agent99 subset before the lateral lookup. Without + # both materialized sets PostgreSQL can rescan the full run ledger per card. list_sql = text(f""" + WITH candidate_messages AS MATERIALIZED ( + SELECT m.* + FROM awooop_outbound_message m + WHERE {where_sql} + ORDER BY COALESCE(m.sent_at, m.queued_at) DESC, m.message_id DESC + LIMIT :limit OFFSET :offset + ), + agent99_runs AS MATERIALIZED ( + SELECT + ars.error_detail, + ars.state, + ars.trace_id, + ars.run_id, + ars.project_id, + ars.trigger_ref, + ars.created_at + FROM awooop_run_state ars + WHERE ars.project_id = :project_id + AND ars.agent_id = 'agent99_controlled_dispatch' + ) SELECT m.message_id, m.project_id, @@ -1540,7 +1562,7 @@ async def list_ai_alert_card_delivery_readback( r.agent_id, r.state AS run_state, r.created_at AS run_created_at - FROM awooop_outbound_message m + FROM candidate_messages m LEFT JOIN awooop_run_state r ON r.project_id = m.project_id AND r.run_id = m.run_id @@ -1549,9 +1571,8 @@ async def list_ai_alert_card_delivery_readback( ars.error_detail, ars.state, ars.trace_id - FROM awooop_run_state ars + FROM agent99_runs ars WHERE ars.project_id = m.project_id - AND ars.agent_id = 'agent99_controlled_dispatch' AND ( ars.run_id::text = COALESCE( COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> @@ -1585,9 +1606,7 @@ async def list_ai_alert_card_delivery_readback( ) DESC, ars.created_at DESC LIMIT 1 ) ar ON TRUE - WHERE {where_sql} ORDER BY COALESCE(m.sent_at, m.queued_at) DESC, m.message_id DESC - LIMIT :limit OFFSET :offset """) try: @@ -1888,8 +1907,32 @@ async def _load_ai_alert_card_delivery_readback_direct( """, *args), timeout=_AI_ALERT_CARD_DIRECT_QUERY_TIMEOUT_SECONDS, ) + # Mirror the pooled-session plan so the pressure fallback does + # not reintroduce a per-card full run-ledger scan. list_rows = await asyncio.wait_for( conn.fetch(f""" + WITH candidate_messages AS MATERIALIZED ( + SELECT m.* + FROM awooop_outbound_message m + WHERE {where_sql} + ORDER BY + COALESCE(m.sent_at, m.queued_at) DESC, + m.message_id DESC + LIMIT ${limit_param} OFFSET ${offset_param} + ), + agent99_runs AS MATERIALIZED ( + SELECT + ars.error_detail, + ars.state, + ars.trace_id, + ars.run_id, + ars.project_id, + ars.trigger_ref, + ars.created_at + FROM awooop_run_state ars + WHERE ars.project_id = $1 + AND ars.agent_id = 'agent99_controlled_dispatch' + ) SELECT m.message_id, m.project_id, @@ -1920,7 +1963,7 @@ async def _load_ai_alert_card_delivery_readback_direct( r.agent_id, r.state AS run_state, r.created_at AS run_created_at - FROM awooop_outbound_message m + FROM candidate_messages m LEFT JOIN awooop_run_state r ON r.project_id = m.project_id AND r.run_id = m.run_id @@ -1929,9 +1972,8 @@ async def _load_ai_alert_card_delivery_readback_direct( ars.error_detail, ars.state, ars.trace_id - FROM awooop_run_state ars + FROM agent99_runs ars WHERE ars.project_id = m.project_id - AND ars.agent_id = 'agent99_controlled_dispatch' AND ( ars.run_id::text = COALESCE( COALESCE( @@ -1971,9 +2013,7 @@ async def _load_ai_alert_card_delivery_readback_direct( ) DESC, ars.created_at DESC LIMIT 1 ) ar ON TRUE - 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, ) diff --git a/apps/api/tests/test_awooop_operator_timeline_labels.py b/apps/api/tests/test_awooop_operator_timeline_labels.py index d9f51f64c..a01e7f66f 100644 --- a/apps/api/tests/test_awooop_operator_timeline_labels.py +++ b/apps/api/tests/test_awooop_operator_timeline_labels.py @@ -1690,6 +1690,12 @@ async def test_unfiltered_query_selects_declared_effective_and_dispatch_receipts assert "AS declared_learning_writeback_ready_total" in summary_query assert summary_query.count("<> 'backup_restore_escrow_signal'") >= 2 assert "-> 'agent99_dispatch_receipt' AS agent99_dispatch_receipt" in list_query + assert "WITH candidate_messages AS MATERIALIZED" in list_query + assert "agent99_runs AS MATERIALIZED" in list_query + assert "FROM agent99_runs ars" in list_query + assert list_query.index("LIMIT :limit OFFSET :offset") < list_query.index( + "LEFT JOIN LATERAL" + ) assert "LEFT JOIN LATERAL" in list_query assert "agent99_current_receipt" in list_query assert payload["summary"]["declared_runtime_write_gate_open_count"] == 97 @@ -1967,6 +1973,19 @@ def test_ai_alert_card_summary_timeout_budget_is_bounded_and_receipted() -> None assert "query_timeout_seconds=_AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS" in source +def test_ai_alert_card_direct_readback_bounds_lateral_join_candidates() -> None: + source = inspect.getsource( + platform_operator_service._load_ai_alert_card_delivery_readback_direct + ) + + assert "WITH candidate_messages AS MATERIALIZED" in source + assert "agent99_runs AS MATERIALIZED" in source + assert "FROM agent99_runs ars" in source + assert source.index("LIMIT ${limit_param} OFFSET ${offset_param}") < source.index( + "LEFT JOIN LATERAL" + ) + + @pytest.mark.asyncio async def test_list_ai_alert_card_delivery_readback_failsofts_db_unavailable( monkeypatch: pytest.MonkeyPatch,