diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index 7a553089a..d330746ae 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -1515,8 +1515,9 @@ 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. + # Resolve at most one Agent99 row per bounded page candidate. Materializing + # every Agent99 row for the project still grows with ledger history and can + # exhaust the readback timeout even when the requested page is small. list_sql = text(f""" WITH candidate_messages AS MATERIALIZED ( SELECT m.* @@ -1525,18 +1526,56 @@ async def list_ai_alert_card_delivery_readback( ORDER BY COALESCE(m.sent_at, m.queued_at) DESC, m.message_id DESC LIMIT :limit OFFSET :offset ), + candidate_refs AS MATERIALIZED ( + SELECT + m.message_id, + m.project_id, + NULLIF( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{agent99_dispatch_receipt,run_id}}', + '' + ) AS receipt_run_id, + NULLIF( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{agent99_dispatch_receipt,incident_id}}', + '' + ) AS receipt_incident_id + FROM candidate_messages m + ), 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' + cr.message_id, + matched.error_detail, + matched.state, + matched.trace_id + FROM candidate_refs cr + LEFT JOIN LATERAL ( + SELECT + ars.error_detail, + ars.state, + ars.trace_id, + ars.run_id, + ars.created_at + FROM awooop_run_state ars + WHERE ars.project_id = cr.project_id + AND ars.agent_id = 'agent99_controlled_dispatch' + AND ( + ( + cr.receipt_run_id IS NOT NULL + AND ars.run_id::text = cr.receipt_run_id + ) + OR ( + cr.receipt_incident_id IS NOT NULL + AND ars.trigger_ref LIKE ( + 'agent99:' || cr.receipt_incident_id || ':%' + ) + ) + ) + ORDER BY ( + ars.run_id::text = COALESCE(cr.receipt_run_id, '') + ) DESC, ars.created_at DESC + LIMIT 1 + ) matched ON TRUE ) SELECT m.message_id, @@ -1566,46 +1605,8 @@ async def list_ai_alert_card_delivery_readback( LEFT JOIN awooop_run_state r ON r.project_id = m.project_id AND r.run_id = m.run_id - LEFT JOIN LATERAL ( - SELECT - ars.error_detail, - ars.state, - ars.trace_id - FROM agent99_runs ars - WHERE ars.project_id = m.project_id - AND ( - ars.run_id::text = COALESCE( - COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> - '{{agent99_dispatch_receipt,run_id}}', - '' - ) - OR ( - COALESCE( - COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> - '{{agent99_dispatch_receipt,incident_id}}', - '' - ) <> '' - AND ars.trigger_ref LIKE ( - 'agent99:' || - COALESCE( - COALESCE( - m.source_envelope::jsonb, - '{{}}'::jsonb - ) #>> '{{agent99_dispatch_receipt,incident_id}}', - '' - ) || ':%' - ) - ) - ) - ORDER BY ( - ars.run_id::text = COALESCE( - COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> - '{{agent99_dispatch_receipt,run_id}}', - '' - ) - ) DESC, ars.created_at DESC - LIMIT 1 - ) ar ON TRUE + LEFT JOIN agent99_runs ar + ON ar.message_id = m.message_id ORDER BY COALESCE(m.sent_at, m.queued_at) DESC, m.message_id DESC """) @@ -1907,8 +1908,8 @@ 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. + # Mirror the pooled-session plan: the materialized Agent99 set + # is hard-bounded to one exact match per page candidate. list_rows = await asyncio.wait_for( conn.fetch(f""" WITH candidate_messages AS MATERIALIZED ( @@ -1920,18 +1921,65 @@ async def _load_ai_alert_card_delivery_readback_direct( m.message_id DESC LIMIT ${limit_param} OFFSET ${offset_param} ), + candidate_refs AS MATERIALIZED ( + SELECT + m.message_id, + m.project_id, + NULLIF( + COALESCE( + m.source_envelope::jsonb, + '{{}}'::jsonb + ) #>> '{{agent99_dispatch_receipt,run_id}}', + '' + ) AS receipt_run_id, + NULLIF( + COALESCE( + m.source_envelope::jsonb, + '{{}}'::jsonb + ) #>> '{{agent99_dispatch_receipt,incident_id}}', + '' + ) AS receipt_incident_id + FROM candidate_messages m + ), 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' + cr.message_id, + matched.error_detail, + matched.state, + matched.trace_id + FROM candidate_refs cr + LEFT JOIN LATERAL ( + SELECT + ars.error_detail, + ars.state, + ars.trace_id, + ars.run_id, + ars.created_at + FROM awooop_run_state ars + WHERE ars.project_id = cr.project_id + AND ars.agent_id = 'agent99_controlled_dispatch' + AND ( + ( + cr.receipt_run_id IS NOT NULL + AND ars.run_id::text = cr.receipt_run_id + ) + OR ( + cr.receipt_incident_id IS NOT NULL + AND ars.trigger_ref LIKE ( + 'agent99:' || + cr.receipt_incident_id || + ':%' + ) + ) + ) + ORDER BY ( + ars.run_id::text = COALESCE( + cr.receipt_run_id, + '' + ) + ) DESC, ars.created_at DESC + LIMIT 1 + ) matched ON TRUE ) SELECT m.message_id, @@ -1967,52 +2015,8 @@ async def _load_ai_alert_card_delivery_readback_direct( LEFT JOIN awooop_run_state r ON r.project_id = m.project_id AND r.run_id = m.run_id - LEFT JOIN LATERAL ( - SELECT - ars.error_detail, - ars.state, - ars.trace_id - FROM agent99_runs ars - WHERE ars.project_id = m.project_id - AND ( - ars.run_id::text = COALESCE( - COALESCE( - m.source_envelope::jsonb, - '{{}}'::jsonb - ) #>> '{{agent99_dispatch_receipt,run_id}}', - '' - ) - OR ( - COALESCE( - COALESCE( - m.source_envelope::jsonb, - '{{}}'::jsonb - ) #>> '{{agent99_dispatch_receipt,incident_id}}', - '' - ) <> '' - AND ars.trigger_ref LIKE ( - 'agent99:' || - COALESCE( - COALESCE( - m.source_envelope::jsonb, - '{{}}'::jsonb - ) #>> '{{agent99_dispatch_receipt,incident_id}}', - '' - ) || ':%' - ) - ) - ) - ORDER BY ( - ars.run_id::text = COALESCE( - COALESCE( - m.source_envelope::jsonb, - '{{}}'::jsonb - ) #>> '{{agent99_dispatch_receipt,run_id}}', - '' - ) - ) DESC, ars.created_at DESC - LIMIT 1 - ) ar ON TRUE + LEFT JOIN agent99_runs ar + ON ar.message_id = m.message_id ORDER BY COALESCE(m.sent_at, m.queued_at) DESC, m.message_id DESC """, *args, per_page, offset), timeout=_AI_ALERT_CARD_DIRECT_QUERY_TIMEOUT_SECONDS, @@ -2198,8 +2202,13 @@ def _ai_alert_card_delivery_source_unavailable_response( target["status"] = "blocked_by_ai_alert_card_delivery_readback_unavailable" target["next_action"] = "repair_awooop_ai_alert_card_delivery_readback" + backup_restore_filtered = event_type == BACKUP_RESTORE_EVENT_TYPE summary.update({ - "status": "source_unavailable_ai_controlled_repair_required", + "status": ( + "blocked_backup_restore_evidence_missing_source_unavailable" + if backup_restore_filtered + else "source_unavailable_ai_controlled_repair_required" + ), "db_read_status": "unavailable", "source_unavailable_reason": error_type, "source_unavailable_next_action": ( @@ -2218,6 +2227,15 @@ def _ai_alert_card_delivery_source_unavailable_response( "repair_awooop_ai_alert_card_delivery_readback_then_consume_learning_registry" ), }) + if backup_restore_filtered: + summary.update({ + "backup_restore_readback_required": True, + "no_false_green_enforced": True, + "backup_restore_next_action": ( + "repair_awooop_ai_alert_card_delivery_readback_then_collect_" + "backup_status_freshness_offsite_escrow_restore_and_resolution_receipts" + ), + }) learning_registry.update({ "status": "source_unavailable_ai_controlled_repair_required", "ready_target_count": 0, @@ -2281,7 +2299,8 @@ def _ai_alert_card_delivery_summary_from_row( 0, ) backup_restore_policy_applied = bool( - backup_restore_total + backup_restore_filtered + or backup_restore_total or backup_runtime_gate_excluded_total or backup_learning_writeback_excluded_total ) @@ -2291,14 +2310,18 @@ def _ai_alert_card_delivery_summary_from_row( status_value = "delivery_failure_observed" elif pending_total > 0: status_value = "delivery_pending_observed" - elif backup_restore_filtered and total > 0: - status_value = "blocked_backup_restore_evidence_or_verifier" + elif backup_restore_filtered: + status_value = ( + "blocked_backup_restore_evidence_missing" + if total == 0 + else "blocked_backup_restore_evidence_or_verifier" + ) elif backup_restore_policy_applied and total > 0: status_value = "observed_effective_backup_policy_projection_applied" learning_next_action = ( "collect_backup_status_freshness_offsite_escrow_restore_and_resolution_receipts" - if backup_restore_policy_applied and total > 0 + if backup_restore_policy_applied else "none" if total > 0 and learning_writeback_missing_total == 0 else "ensure_ai_alert_cards_include_event_type_lane_and_delivery_receipt" diff --git a/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py b/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py index 715b72a54..48a1a0d46 100644 --- a/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py +++ b/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import os +import time from decimal import Decimal from types import SimpleNamespace from urllib.parse import parse_qsl, urlsplit @@ -480,3 +481,153 @@ async def test_direct_readback_rewrites_ssl_and_keeps_rls_context_in_transaction assert len(rows) == 1 assert rows[0]["project_id"] == "awoooi" assert rows[0]["alert_card"]["event_type"] == ("backup_restore_escrow_signal") + + +@pytest.mark.asyncio +async def test_direct_readback_large_agent99_history_stays_within_bounded_budget( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A one-card page must not materialize unrelated Agent99 history.""" + + sqlalchemy_url = _explicit_local_test_database_url() + direct_url = platform_operator_service._asyncpg_direct_database_url(sqlalchemy_url) + connection = await asyncpg.connect(direct_url) + matching_run_id = UUID("c5ac50ab-9d55-5c4c-a23f-f16f3a772c39") + await connection.execute(""" + CREATE TEMP TABLE awooop_outbound_message ( + message_id UUID PRIMARY KEY, + project_id TEXT NOT NULL, + run_id UUID, + channel_type TEXT NOT NULL, + message_type TEXT NOT NULL, + provider_message_id TEXT, + send_status TEXT NOT NULL, + send_error TEXT, + queued_at TIMESTAMPTZ NOT NULL, + sent_at TIMESTAMPTZ, + triggered_by_state TEXT, + source_envelope JSONB NOT NULL + ) + """) + await connection.execute(""" + CREATE TEMP TABLE awooop_run_state ( + run_id UUID PRIMARY KEY, + project_id TEXT NOT NULL, + agent_id TEXT NOT NULL, + state TEXT NOT NULL, + trace_id TEXT, + trigger_ref TEXT, + error_detail JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ) + """) + await connection.execute( + """ + INSERT INTO awooop_run_state ( + run_id, project_id, agent_id, state, trace_id, trigger_ref, + error_detail, created_at + ) + SELECT + md5('unrelated-agent99-history-' || value::text)::uuid, + 'awoooi', + 'agent99_controlled_dispatch', + 'completed', + 'history-trace-' || value::text, + 'agent99:INC-UNRELATED-' || value::text || ':history', + '{}'::jsonb, + NOW() - (value * INTERVAL '1 second') + FROM generate_series(1, 25000) AS value + """ + ) + await connection.execute( + """ + INSERT INTO awooop_run_state ( + run_id, project_id, agent_id, state, trace_id, trigger_ref, + error_detail + ) VALUES ($1, 'awoooi', 'agent99_controlled_dispatch', 'completed', + 'matching-trace', 'agent99:INC-BENCHMARK:dispatch', $2::jsonb) + """, + matching_run_id, + json.dumps( + { + "schema_version": "agent99_controlled_dispatch_receipt_v1", + "identity": { + "incident_id": "INC-BENCHMARK", + "run_id": str(matching_run_id), + "trace_id": "matching-trace", + "work_item_id": "benchmark-work-item", + }, + } + ), + ) + await connection.execute( + """ + INSERT INTO awooop_outbound_message ( + message_id, project_id, channel_type, message_type, + provider_message_id, send_status, queued_at, sent_at, + triggered_by_state, source_envelope + ) VALUES ($1, 'awoooi', 'telegram', 'ai_alert_card', 'benchmark-1', + 'sent', NOW(), NOW(), 'controlled_apply', $2::jsonb) + """, + UUID("1dd3130f-e030-40b3-8d27-1f74818b4f87"), + json.dumps( + { + "ai_automation_alert_card": { + "event_type": "backup_restore_escrow_signal", + "lane": "backup_restore_escrow_triage", + "delivery_receipt_readback_required": True, + }, + "agent99_dispatch_receipt": { + "incident_id": "INC-BENCHMARK", + "run_id": str(matching_run_id), + }, + } + ), + ) + await connection.execute(""" + CREATE INDEX ON awooop_run_state (project_id, agent_id, run_id) + """) + await connection.execute(""" + CREATE INDEX ON awooop_run_state ( + project_id, agent_id, trigger_ref text_pattern_ops + ) + """) + + original_connect = asyncpg.connect + + async def reuse_connection(dsn: str, *args: object, **kwargs: object): + assert dsn == direct_url + assert not args + assert not kwargs + return connection + + monkeypatch.setattr(asyncpg, "connect", reuse_connection) + monkeypatch.setattr( + platform_operator_service, + "get_settings", + lambda: SimpleNamespace(DATABASE_URL=sqlalchemy_url), + ) + started_at = time.perf_counter() + try: + result = await platform_operator_service._load_ai_alert_card_delivery_readback_direct( + project_id="awoooi", + event_type="backup_restore_escrow_signal", + lane="backup_restore_escrow_triage", + page=1, + per_page=1, + ) + finally: + elapsed = time.perf_counter() - started_at + monkeypatch.setattr(asyncpg, "connect", original_connect) + if not connection.is_closed(): + await connection.close() + + assert result is not None + summary, rows = result + assert summary["total"] == 1 + assert len(rows) == 1 + assert rows[0]["agent99_current_run_state"] == "completed" + assert rows[0]["agent99_current_trace_id"] == "matching-trace" + assert elapsed < ( + platform_operator_service._AI_ALERT_CARD_DIRECT_QUERY_TIMEOUT_SECONDS + 0.25 + ) diff --git a/apps/api/tests/test_awooop_operator_timeline_labels.py b/apps/api/tests/test_awooop_operator_timeline_labels.py index a01e7f66f..6812b8a1c 100644 --- a/apps/api/tests/test_awooop_operator_timeline_labels.py +++ b/apps/api/tests/test_awooop_operator_timeline_labels.py @@ -1061,6 +1061,25 @@ def test_ai_alert_card_delivery_summary_keeps_no_false_green_status() -> None: assert summary["production_write_count"] == 0 +def test_backup_restore_filtered_empty_summary_blocks_missing_evidence() -> None: + summary = _ai_alert_card_delivery_summary_from_row( + {}, + project_id="awoooi", + event_type="backup_restore_escrow_signal", + lane="backup_restore_escrow_triage", + ) + + assert summary["status"] == "blocked_backup_restore_evidence_missing" + assert summary["total"] == 0 + assert summary["backup_restore_total"] == 0 + assert summary["backup_restore_readback_required"] is True + assert summary["no_false_green_enforced"] is True + assert summary["runtime_write_allowed"] is False + assert summary["learning_writeback_next_action"] == ( + "collect_backup_status_freshness_offsite_escrow_restore_and_resolution_receipts" + ) + + def test_backup_restore_delivery_readback_is_durable_but_never_false_green() -> None: run_id = UUID("0b25cafa-00a4-520b-8946-a3ea467b13fc") message_id = UUID("97ff9f41-7cdc-4893-8aab-634286fb128c") @@ -1691,12 +1710,17 @@ async def test_unfiltered_query_selects_declared_effective_and_dispatch_receipts 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 "candidate_refs AS MATERIALIZED" in list_query assert "agent99_runs AS MATERIALIZED" in list_query - assert "FROM agent99_runs ars" in list_query + assert "FROM candidate_refs cr" in list_query + assert "LEFT JOIN agent99_runs ar" in list_query + assert "ON ar.message_id = m.message_id" in list_query + assert "FROM agent99_runs ars" not in list_query assert list_query.index("LIMIT :limit OFFSET :offset") < list_query.index( - "LEFT JOIN LATERAL" + "candidate_refs AS MATERIALIZED" ) assert "LEFT JOIN LATERAL" in list_query + assert list_query.count("LIMIT 1") == 1 assert "agent99_current_receipt" in list_query assert payload["summary"]["declared_runtime_write_gate_open_count"] == 97 assert payload["summary"]["backup_restore_total"] == 98 @@ -1948,6 +1972,32 @@ def test_ai_alert_card_source_unavailable_response_exposes_ai_repair_queue() -> assert dumped["learning_registry"]["operation_boundaries"]["github_api_used"] is False +def test_backup_restore_source_unavailable_blocks_missing_evidence() -> None: + response = ListAiAlertCardsResponse.model_validate( + platform_operator_service._ai_alert_card_delivery_source_unavailable_response( + project_id="awoooi", + event_type="backup_restore_escrow_signal", + lane="backup_restore_escrow_triage", + page=1, + per_page=6, + error_type="DBAPIError", + ) + ) + + summary = response.model_dump(mode="json")["summary"] + assert summary["status"] == ( + "blocked_backup_restore_evidence_missing_source_unavailable" + ) + assert summary["db_read_status"] == "unavailable" + assert summary["controlled_repair_queue_present"] is True + assert summary["backup_restore_readback_required"] is True + assert summary["no_false_green_enforced"] is True + assert summary["runtime_write_allowed"] is False + assert summary["backup_restore_next_action"].startswith( + "repair_awooop_ai_alert_card_delivery_readback_then_collect_" + ) + + def test_ai_alert_card_summary_timeout_budget_is_bounded_and_receipted() -> None: class QueryCanceled(Exception): sqlstate = "57014" @@ -1979,11 +2029,16 @@ def test_ai_alert_card_direct_readback_bounds_lateral_join_candidates() -> None: ) assert "WITH candidate_messages AS MATERIALIZED" in source + assert "candidate_refs AS MATERIALIZED" in source assert "agent99_runs AS MATERIALIZED" in source - assert "FROM agent99_runs ars" in source + assert "FROM candidate_refs cr" in source + assert "LEFT JOIN agent99_runs ar" in source + assert "ON ar.message_id = m.message_id" in source + assert "FROM agent99_runs ars" not in source assert source.index("LIMIT ${limit_param} OFFSET ${offset_param}") < source.index( - "LEFT JOIN LATERAL" + "candidate_refs AS MATERIALIZED" ) + assert source.count("LIMIT 1") == 1 @pytest.mark.asyncio