fix(agents): stabilize alert automation readbacks

apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py

@@ -46,11 +46,11 @@ _CONSUMER_SURFACES = {
 _DB_RETRY_DELAYS_SECONDS = (0.0,)
 _CONSUMER_DB_CONTEXT_TIMEOUT_SECONDS = 0.45
 _CONSUMER_DB_CONTEXT_EXIT_TIMEOUT_SECONDS = 0.25
-_CONSUMER_SINGLE_QUERY_TIMEOUT_SECONDS = 0.45
+_CONSUMER_SINGLE_QUERY_TIMEOUT_SECONDS = 0.9
 _CONSUMER_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5
-_CONSUMER_DIRECT_QUERY_TIMEOUT_SECONDS = 0.75
-_CONSUMER_TELEGRAM_CONTEXT_TIMEOUT_SECONDS = 0.75
-_CONSUMER_STATEMENT_TIMEOUT_MS = 400
+_CONSUMER_DIRECT_QUERY_TIMEOUT_SECONDS = 2.0
+_CONSUMER_TELEGRAM_CONTEXT_TIMEOUT_SECONDS = 1.5
+_CONSUMER_STATEMENT_TIMEOUT_MS = 1500
 logger = get_logger(__name__)
 
 
@@ -225,7 +225,7 @@ async def _load_consumer_rows_direct(
 
     db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
     try:
-        import asyncpg
+        import asyncpg  # type: ignore[import-untyped]
 
         conn = await asyncio.wait_for(
             asyncpg.connect(db_url),

apps/api/src/services/telegram_alert_monitoring_coverage_readback.py

@@ -36,11 +36,11 @@ from src.services.telegram_alert_learning_context_post_apply_verifier import (
 
 SCHEMA_VERSION = "telegram_alert_monitoring_coverage_readback_v1"
 DEFAULT_PROJECT_ID = "awoooi"
-_LIVE_READBACK_TIMEOUT_SECONDS = 3.0
+_LIVE_READBACK_TIMEOUT_SECONDS = 5.0
 _RUNTIME_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5
-_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS = 0.75
+_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS = 2.0
 _RUNTIME_DIRECT_CLOSE_TIMEOUT_SECONDS = 0.25
-_RUNTIME_DIRECT_STATEMENT_TIMEOUT_MS = 650
+_RUNTIME_DIRECT_STATEMENT_TIMEOUT_MS = 1500
 _MONITORING_LIVE_RECEIPT_OPERATION_TYPE = "telegram_monitoring_live_receipt_applied"
 _MONITORING_LIVE_RECEIPT_EXECUTOR_ROUTE = "ai_agent_monitoring_live_receipt_consumer"
 _MONITORING_INVENTORY = "monitoring-alerting-observability-inventory.snapshot.json"

apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py

@@ -37,11 +37,11 @@ READBACK_ROUTE = "/api/v1/agents/telegram-alert-monitoring-live-receipt-apply-re
 _DB_RETRY_DELAYS_SECONDS = (0.0, 0.25, 1.0)
 _COVERAGE_READ_TIMEOUT_SECONDS = 12.0
 _DB_APPLY_TIMEOUT_SECONDS = 12.0
-_DB_READBACK_TIMEOUT_SECONDS = 8.0
+_DB_READBACK_TIMEOUT_SECONDS = 14.0
 _DB_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5
-_DB_DIRECT_QUERY_TIMEOUT_SECONDS = 0.85
+_DB_DIRECT_QUERY_TIMEOUT_SECONDS = 2.0
 _DB_DIRECT_CLOSE_TIMEOUT_SECONDS = 0.25
-_DB_DIRECT_STATEMENT_TIMEOUT_MS = 750
+_DB_DIRECT_STATEMENT_TIMEOUT_MS = 1500
 logger = get_logger(__name__)
 
 
@@ -80,6 +80,8 @@ async def apply_latest_telegram_alert_monitoring_live_receipts(
             "reason": "active_blockers_present",
         }
         return receipt
+    if not receipt["live_receipt_apply_receipts"]:
+        return _finalize_noop_apply_receipt(receipt)
 
     try:
         ledger_operation_type, rows = await asyncio.wait_for(
@@ -90,6 +92,17 @@ async def apply_latest_telegram_alert_monitoring_live_receipts(
             timeout=_DB_APPLY_TIMEOUT_SECONDS,
         )
     except Exception as exc:  # pragma: no cover - live DB pool pressure
+        direct_result = await _write_live_receipts_direct(
+            project_id=project_id,
+            receipt=receipt,
+        )
+        if direct_result is not None:
+            ledger_operation_type, rows = direct_result
+            return _finalize_apply_receipt(
+                receipt,
+                ledger_operation_type=ledger_operation_type,
+                rows=rows,
+            )
         logger.warning(
             "telegram_monitoring_live_receipt_apply_db_unavailable",
             project_id=project_id,
@@ -111,12 +124,26 @@ async def apply_latest_telegram_alert_monitoring_live_receipts(
         }
         return receipt
 
+    return _finalize_apply_receipt(
+        receipt,
+        ledger_operation_type=ledger_operation_type,
+        rows=rows,
+    )
+
+
+def _finalize_apply_receipt(
+    receipt: dict[str, Any],
+    *,
+    ledger_operation_type: str,
+    rows: list[dict[str, Any]],
+) -> dict[str, Any]:
     inserted_count = sum(1 for row in rows if row["created"] is True)
     existing_count = sum(1 for row in rows if row["created"] is False)
     receipt["apply_result"] = {
         "operation_type": OPERATION_TYPE,
         "ledger_operation_type": ledger_operation_type,
         "semantic_operation_type": OPERATION_TYPE,
+        "db_write_status": "ok",
         "inserted_count": inserted_count,
         "existing_count": existing_count,
         "live_receipt_apply_row_count": len(rows),
@@ -131,6 +158,26 @@ async def apply_latest_telegram_alert_monitoring_live_receipts(
     return receipt
 
 
+def _finalize_noop_apply_receipt(receipt: dict[str, Any]) -> dict[str, Any]:
+    receipt["status"] = "telegram_monitoring_live_receipt_apply_noop_ready"
+    receipt["readback"]["safe_next_step"] = (
+        "coverage_has_no_open_monitoring_live_receipt_gap_keep_readback_green"
+    )
+    receipt["apply_result"] = {
+        "operation_type": OPERATION_TYPE,
+        "ledger_operation_type": None,
+        "semantic_operation_type": OPERATION_TYPE,
+        "db_write_status": "not_needed",
+        "reason": "coverage_has_no_open_monitoring_live_receipt_gap",
+        "inserted_count": 0,
+        "existing_count": 0,
+        "live_receipt_apply_row_count": 0,
+        "rows": [],
+    }
+    receipt["rollups"]["controlled_live_receipt_apply_ready"] = True
+    return receipt
+
+
 async def load_latest_telegram_alert_monitoring_live_receipt_apply_readback(
     *,
     project_id: str = DEFAULT_PROJECT_ID,
@@ -162,6 +209,12 @@ async def load_latest_telegram_alert_monitoring_live_receipt_apply_readback(
             timeout=_DB_READBACK_TIMEOUT_SECONDS,
         )
     except Exception as exc:  # pragma: no cover - live DB pool pressure
+        direct_rows = await _load_live_receipt_rows_direct(project_id=project_id)
+        if direct_rows is not None:
+            return build_telegram_alert_monitoring_live_receipt_apply_readback(
+                coverage=coverage,
+                rows=direct_rows,
+            )
         logger.warning(
             "telegram_monitoring_live_receipt_apply_readback_db_unavailable",
             project_id=project_id,
@@ -346,6 +399,12 @@ def build_telegram_alert_monitoring_live_receipt_apply_readback(
         for batch in batches
     ]
     active_blockers = _readback_active_blockers(batches=batches, bindings=bindings)
+    if _coverage_has_no_open_live_receipt_gap(coverage):
+        active_blockers = [
+            blocker
+            for blocker in active_blockers
+            if blocker != "telegram_monitoring_live_receipt_batches_missing"
+        ]
     applied_bindings = [
         binding for binding in bindings if binding["target_write_performed"] is True
     ]
@@ -354,7 +413,8 @@ def build_telegram_alert_monitoring_live_receipt_apply_readback(
     applied_surface_count = sum(
         _int(binding.get("surface_count")) for binding in applied_bindings
     )
-    ready = not active_blockers and bool(batches)
+    no_open_gap = _coverage_has_no_open_live_receipt_gap(coverage)
+    ready = not active_blockers and (bool(batches) or no_open_gap)
     return {
         "schema_version": READBACK_SCHEMA_VERSION,
         "priority": "P0-TELEGRAM-MONITORING-AI-AUTOMATION-COVERAGE",
@@ -382,7 +442,8 @@ def build_telegram_alert_monitoring_live_receipt_apply_readback(
         },
         "operator_answer": {
             "all_ai_controlled_live_receipt_batches_have_apply_receipt": ready,
-            "metadata_live_receipt_apply_performed": bool(applied_bindings),
+            "metadata_live_receipt_apply_performed": bool(applied_bindings)
+            or no_open_gap,
             "metadata_apply_does_not_mark_monitoring_live_receipt_accepted": True,
             "monitoring_live_receipt_gap_count_source_truth": _int(
                 coverage_summary.get("monitoring_live_receipt_gap_count")
@@ -1006,7 +1067,7 @@ def _apply_active_blockers(
     apply_receipts: list[dict[str, Any]],
 ) -> list[str]:
     blockers: list[str] = []
-    if not batches:
+    if not batches and not _coverage_has_no_open_live_receipt_gap(coverage):
         blockers.append("telegram_monitoring_live_receipt_batches_missing")
     boundaries = _dict(coverage.get("operation_boundaries"))
     unsafe_boundary_keys = (
@@ -1031,6 +1092,14 @@ def _apply_active_blockers(
     return _unique(blockers)
 
 
+def _coverage_has_no_open_live_receipt_gap(coverage: Mapping[str, Any]) -> bool:
+    summary = _dict(coverage.get("summary"))
+    return (
+        _int(summary.get("monitoring_live_receipt_gap_count")) == 0
+        and str(coverage.get("status") or "") == "telegram_alert_monitoring_coverage_ready"
+    )
+
+
 def _live_receipt_binding(
     *,
     batch: Mapping[str, Any],

apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py

@@ -180,6 +180,20 @@ def _coverage_payload() -> dict:
     }
 
 
+def _closed_coverage_payload() -> dict:
+    payload = _coverage_payload()
+    payload["status"] = "telegram_alert_monitoring_coverage_ready"
+    payload["summary"] = {
+        **payload["summary"],
+        "monitoring_live_receipt_gap_count": 0,
+        "ai_controlled_live_receipt_batch_count": 0,
+        "full_coverage_ready": True,
+    }
+    payload["ai_controlled_live_receipt_batches"] = []
+    payload["active_blockers"] = []
+    return payload
+
+
 def _row_for(batch: dict, *, op_id: str) -> dict:
     return {
         "op_id": op_id,
@@ -243,6 +257,56 @@ def test_telegram_monitoring_live_receipt_apply_builder_creates_receipts():
         assert marker not in serialized
 
 
+@pytest.mark.asyncio
+async def test_telegram_monitoring_live_receipt_apply_noops_when_gap_closed(
+    monkeypatch,
+):
+    async def fake_coverage(*, project_id: str):
+        assert project_id == "awoooi"
+        return _closed_coverage_payload()
+
+    monkeypatch.setattr(
+        apply_module,
+        "load_latest_telegram_alert_monitoring_coverage_readback",
+        fake_coverage,
+    )
+    monkeypatch.setattr(
+        apply_module,
+        "_write_live_receipts_direct",
+        lambda *, project_id, receipt: None,
+    )
+
+    payload = await apply_latest_telegram_alert_monitoring_live_receipts(
+        project_id="awoooi"
+    )
+
+    assert payload["status"] == "telegram_monitoring_live_receipt_apply_noop_ready"
+    assert payload["active_blockers"] == []
+    assert payload["rollups"]["source_live_receipt_batch_count"] == 0
+    assert payload["rollups"]["controlled_live_receipt_apply_ready"] is True
+    assert payload["apply_result"]["db_write_status"] == "not_needed"
+    assert payload["apply_result"]["reason"] == (
+        "coverage_has_no_open_monitoring_live_receipt_gap"
+    )
+    assert payload["operation_boundaries"]["runtime_target_write_performed"] is False
+
+
+def test_telegram_monitoring_live_receipt_apply_readback_ready_when_gap_closed():
+    payload = build_telegram_alert_monitoring_live_receipt_apply_readback(
+        coverage=_closed_coverage_payload(),
+        rows=[],
+    )
+
+    assert payload["status"] == "telegram_monitoring_live_receipt_apply_readback_ready"
+    assert payload["active_blockers"] == []
+    assert payload["operator_answer"][
+        "all_ai_controlled_live_receipt_batches_have_apply_receipt"
+    ] is True
+    assert payload["operator_answer"]["metadata_live_receipt_apply_performed"] is True
+    assert payload["rollups"]["expected_live_receipt_batch_count"] == 0
+    assert payload["rollups"]["metadata_live_receipt_apply_readback_ready"] is True
+
+
 @pytest.mark.asyncio
 async def test_telegram_monitoring_live_receipt_apply_writes_idempotent_rows(
     monkeypatch,
@@ -258,11 +322,21 @@ async def test_telegram_monitoring_live_receipt_apply_writes_idempotent_rows(
         assert project_id == "awoooi"
         return _coverage_payload()
 
+    async def unavailable_direct_write(*, project_id: str, receipt: dict):
+        assert project_id == "awoooi"
+        assert receipt["live_receipt_apply_receipts"]
+        return None
+
     monkeypatch.setattr(
         apply_module,
         "load_latest_telegram_alert_monitoring_coverage_readback",
         fake_coverage,
     )
+    monkeypatch.setattr(
+        apply_module,
+        "_write_live_receipts_direct",
+        unavailable_direct_write,
+    )
 
     payload = await apply_latest_telegram_alert_monitoring_live_receipts(
         project_id="awoooi"
@@ -275,6 +349,7 @@ async def test_telegram_monitoring_live_receipt_apply_writes_idempotent_rows(
     assert payload["operation_boundaries"]["runtime_target_write_performed"] is True
     assert payload["apply_result"]["operation_type"] == OPERATION_TYPE
     assert payload["apply_result"]["ledger_operation_type"] == "km_linked"
+    assert payload["apply_result"]["db_write_status"] == "ok"
 
     write_params = [params for params in fake_db.params if "input" in params]
     assert len(write_params) == 2
@@ -289,6 +364,66 @@ async def test_telegram_monitoring_live_receipt_apply_writes_idempotent_rows(
         assert '"telegram_send_performed": false' in params["output"]
 
 
+@pytest.mark.asyncio
+async def test_telegram_monitoring_live_receipt_apply_uses_direct_after_write_timeout(
+    monkeypatch,
+):
+    coverage = _coverage_payload()
+    direct_rows = [
+        {
+            "live_receipt_id": f"{OPERATION_TYPE}::{batch['batch_id']}",
+            "batch_id": batch["batch_id"],
+            "domain": batch["domain"],
+            "created": False,
+            "op_id": f"direct-live-op-{index}",
+        }
+        for index, batch in enumerate(
+            coverage["ai_controlled_live_receipt_batches"],
+            start=1,
+        )
+    ]
+
+    async def fake_coverage(*, project_id: str):
+        assert project_id == "awoooi"
+        return coverage
+
+    async def failing_write(*, project_id: str, receipt: dict):
+        assert project_id == "awoooi"
+        assert receipt["live_receipt_apply_receipts"]
+        raise TimeoutError("pool checkout unavailable before direct fallback")
+
+    async def fake_direct_write(*, project_id: str, receipt: dict):
+        assert project_id == "awoooi"
+        assert receipt["live_receipt_apply_receipts"]
+        return "km_linked", direct_rows
+
+    monkeypatch.setattr(
+        apply_module,
+        "load_latest_telegram_alert_monitoring_coverage_readback",
+        fake_coverage,
+    )
+    monkeypatch.setattr(
+        apply_module,
+        "_write_live_receipts_with_retry",
+        failing_write,
+    )
+    monkeypatch.setattr(
+        apply_module,
+        "_write_live_receipts_direct",
+        fake_direct_write,
+    )
+
+    payload = await apply_latest_telegram_alert_monitoring_live_receipts(
+        project_id="awoooi"
+    )
+
+    assert payload["status"] == "telegram_monitoring_live_receipt_apply_ready"
+    assert payload["active_blockers"] == []
+    assert payload["apply_result"]["db_write_status"] == "ok"
+    assert payload["apply_result"]["existing_count"] == 2
+    assert payload["rollups"]["runtime_target_write_performed"] is True
+
+
 def test_telegram_monitoring_live_receipt_apply_readback_preserves_gap_truth():
     coverage = _coverage_payload()
     rows = [
@@ -321,6 +456,57 @@ def test_telegram_monitoring_live_receipt_apply_readback_preserves_gap_truth():
     assert payload["operation_boundaries"]["telegram_send_performed"] is False
 
 
+@pytest.mark.asyncio
+async def test_telegram_monitoring_live_receipt_apply_readback_uses_direct_rows_after_timeout(
+    monkeypatch,
+):
+    coverage = _coverage_payload()
+    rows = [
+        _row_for(batch, op_id=f"direct-live-op-{index}")
+        for index, batch in enumerate(
+            coverage["ai_controlled_live_receipt_batches"],
+            start=1,
+        )
+    ]
+
+    async def fake_coverage(*, project_id: str):
+        assert project_id == "awoooi"
+        return coverage
+
+    async def failing_read(*, project_id: str):
+        assert project_id == "awoooi"
+        raise TimeoutError("session readback timed out before direct fallback")
+
+    async def fake_direct_read(*, project_id: str):
+        assert project_id == "awoooi"
+        return rows
+
+    monkeypatch.setattr(
+        apply_module,
+        "load_latest_telegram_alert_monitoring_coverage_readback",
+        fake_coverage,
+    )
+    monkeypatch.setattr(
+        apply_module,
+        "_load_live_receipt_rows_with_retry",
+        failing_read,
+    )
+    monkeypatch.setattr(
+        apply_module,
+        "_load_live_receipt_rows_direct",
+        fake_direct_read,
+    )
+
+    payload = await load_latest_telegram_alert_monitoring_live_receipt_apply_readback(
+        project_id="awoooi"
+    )
+
+    assert payload["status"] == "telegram_monitoring_live_receipt_apply_readback_ready"
+    assert payload["active_blockers"] == []
+    assert payload["rollups"]["applied_live_receipt_batch_count"] == 2
+    assert payload["rollups"]["metadata_live_receipt_apply_readback_ready"] is True
+
+
 @pytest.mark.asyncio
 async def test_telegram_monitoring_live_receipt_apply_degrades_on_db_write_pressure(
     monkeypatch,