fix(operator): fail closed on missing backup evidence

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
+    )