From 403c0a3d054886a25e957c615fdfccbbcccfefaf Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 01:17:21 +0800 Subject: [PATCH] fix(agents): stabilize alert automation readbacks --- ..._controlled_writeback_consumer_readback.py | 10 +- ...gram_alert_monitoring_coverage_readback.py | 6 +- ...ram_alert_monitoring_live_receipt_apply.py | 81 +++++++- ...alert_monitoring_live_receipt_apply_api.py | 186 ++++++++++++++++++ 4 files changed, 269 insertions(+), 14 deletions(-) diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py index ded81f716..040dd5ff2 100644 --- a/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py +++ b/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), diff --git a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py index 2110e7165..4812cba38 100644 --- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py +++ b/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" diff --git a/apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py b/apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py index 32b00808d..6ac18b19e 100644 --- a/apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py +++ b/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], diff --git a/apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py b/apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py index 974124f3a..5d211ba0e 100644 --- a/apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py +++ b/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,