From 9cf01f1cf73fab13ca1a5c4567aa416e678b02d2 Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 00:44:21 +0800 Subject: [PATCH] fix(agents): consume monitoring live receipt metadata --- ...gram_alert_monitoring_coverage_readback.py | 352 ++++++++++++++- ...ram_alert_monitoring_live_receipt_apply.py | 419 +++++++++++++++--- ..._alert_monitoring_coverage_readback_api.py | 99 +++++ 3 files changed, 798 insertions(+), 72 deletions(-) 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 1845a5783..2110e7165 100644 --- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py +++ b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py @@ -13,7 +13,7 @@ from __future__ import annotations import asyncio import json -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from pathlib import Path from typing import Any @@ -41,6 +41,8 @@ _RUNTIME_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5 _RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS = 0.75 _RUNTIME_DIRECT_CLOSE_TIMEOUT_SECONDS = 0.25 _RUNTIME_DIRECT_STATEMENT_TIMEOUT_MS = 650 +_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" _MONITORING_POST_INCIDENT_READBACK = "monitoring-post-incident-readback-plan.snapshot.json" _ALERT_LOG_EVENT_TYPES = ( @@ -106,9 +108,15 @@ async def load_latest_telegram_alert_monitoring_coverage_readback( ) _require_safe_boundaries(monitoring_inventory, monitoring_readback_plan) - matrix, consumer_readback, source_contract = await asyncio.gather( + ( + matrix, + consumer_readback, + monitoring_live_receipt_apply, + source_contract, + ) = await asyncio.gather( _load_telegram_matrix_readback(project_id=project_id), _load_log_controlled_writeback_consumer_readback(project_id=project_id), + _load_monitoring_live_receipt_apply_readback(project_id=project_id), asyncio.to_thread(_inspect_source_contract, repo_root), ) verifier, ai_alert_readback, runtime_log_readback = await asyncio.gather( @@ -126,6 +134,7 @@ async def load_latest_telegram_alert_monitoring_coverage_readback( ai_alert_card_delivery_readback=ai_alert_readback, runtime_log_readback=runtime_log_readback, log_controlled_writeback_consumer=consumer_readback, + monitoring_live_receipt_apply_readback=monitoring_live_receipt_apply, source_contract=source_contract, ) @@ -226,6 +235,185 @@ async def _load_log_controlled_writeback_consumer_readback( } +async def _load_monitoring_live_receipt_apply_readback( + *, + project_id: str, +) -> dict[str, Any]: + try: + return await asyncio.wait_for( + _load_monitoring_live_receipt_apply_readback_once(project_id=project_id), + timeout=_LIVE_READBACK_TIMEOUT_SECONDS, + ) + except Exception as exc: # pragma: no cover - live DB pressure + direct_payload = await _load_monitoring_live_receipt_apply_readback_direct( + project_id=project_id + ) + if direct_payload is not None: + return direct_payload + logger.warning( + "telegram_alert_monitoring_live_receipt_apply_readback_unavailable", + project_id=project_id, + error_type=type(exc).__name__, + ) + return { + "status": "unavailable", + "summary": { + "readback_source": "unavailable", + "row_count": 0, + "ready_row_count": 0, + "accepted_surface_count": 0, + "error_type": type(exc).__name__, + }, + "rows": [], + } + + +async def _load_monitoring_live_receipt_apply_readback_once( + *, + project_id: str, +) -> dict[str, Any]: + async with get_db_context(project_id) as db: + await db.execute(text("SET LOCAL statement_timeout = '5000ms'"), {}) + result = await db.execute( + text(""" + SELECT + op_id::text AS op_id, + operation_type, + actor, + status, + created_at, + input ->> 'semantic_operation_type' AS semantic_operation_type, + input ->> 'live_receipt_id' AS live_receipt_id, + input ->> 'batch_id' AS batch_id, + input ->> 'domain' AS domain, + input ->> 'raw_payload_included' AS raw_payload_included, + input -> 'target_selector' AS target_selector, + input -> 'tag_hints' AS tag_hints, + output ->> 'monitoring_live_receipt_metadata_recorded' + AS monitoring_live_receipt_metadata_recorded, + output ->> 'target_context_receipt_write_performed' + AS target_context_receipt_write_performed, + output -> 'post_apply_verifier_refs' AS post_apply_verifier_refs + FROM automation_operation_log + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) + = :operation_type + AND input ->> 'project_id' = :project_id + ORDER BY created_at DESC, op_id DESC + LIMIT 200 + """), + { + "operation_type": _MONITORING_LIVE_RECEIPT_OPERATION_TYPE, + "project_id": project_id, + }, + ) + rows = _result_rows(result) + return _monitoring_live_receipt_apply_readback_from_rows( + rows=rows, + readback_source="db_session", + ) + + +async def _load_monitoring_live_receipt_apply_readback_direct( + *, + project_id: str, +) -> dict[str, Any] | None: + db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://") + try: + import asyncpg # type: ignore[import-untyped] + + conn = await asyncio.wait_for( + asyncpg.connect(db_url), + timeout=_RUNTIME_DIRECT_CONNECT_TIMEOUT_SECONDS, + ) + try: + await asyncio.wait_for( + conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id), + timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + await asyncio.wait_for( + conn.execute( + "SET statement_timeout = " + f"'{int(_RUNTIME_DIRECT_STATEMENT_TIMEOUT_MS)}ms'" + ), + timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + rows = await asyncio.wait_for( + conn.fetch( + """ + SELECT + op_id::text AS op_id, + operation_type, + actor, + status, + created_at, + input ->> 'semantic_operation_type' AS semantic_operation_type, + input ->> 'live_receipt_id' AS live_receipt_id, + input ->> 'batch_id' AS batch_id, + input ->> 'domain' AS domain, + input ->> 'raw_payload_included' AS raw_payload_included, + input -> 'target_selector' AS target_selector, + input -> 'tag_hints' AS tag_hints, + output ->> 'monitoring_live_receipt_metadata_recorded' + AS monitoring_live_receipt_metadata_recorded, + output ->> 'target_context_receipt_write_performed' + AS target_context_receipt_write_performed, + output -> 'post_apply_verifier_refs' AS post_apply_verifier_refs + FROM automation_operation_log + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) + = $1 + AND input ->> 'project_id' = $2 + ORDER BY created_at DESC, op_id DESC + LIMIT 200 + """, + _MONITORING_LIVE_RECEIPT_OPERATION_TYPE, + project_id, + ), + timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + return _monitoring_live_receipt_apply_readback_from_rows( + rows=[dict(row) for row in rows], + readback_source="direct_connection", + ) + finally: + try: + await asyncio.wait_for( + conn.close(), + timeout=_RUNTIME_DIRECT_CLOSE_TIMEOUT_SECONDS, + ) + except Exception as close_exc: # pragma: no cover - live cleanup + logger.warning( + "telegram_alert_monitoring_live_receipt_apply_direct_close_failed", + project_id=project_id, + error_type=type(close_exc).__name__, + ) + except Exception as exc: # pragma: no cover - live DB pressure + logger.warning( + "telegram_alert_monitoring_live_receipt_apply_direct_readback_failed", + project_id=project_id, + error_type=type(exc).__name__, + ) + return None + + +def _monitoring_live_receipt_apply_readback_from_rows( + *, + rows: list[dict[str, Any]], + readback_source: str, +) -> dict[str, Any]: + ready_rows = [row for row in rows if _metadata_live_receipt_row_ready(row)] + accepted_surface_ids = _metadata_live_receipt_surface_ids(ready_rows) + return { + "status": "ok", + "summary": { + "readback_source": readback_source, + "row_count": len(rows), + "ready_row_count": len(ready_rows), + "accepted_surface_count": len(accepted_surface_ids), + }, + "rows": rows, + } + + def build_telegram_alert_monitoring_coverage_readback( *, project_id: str, @@ -237,6 +425,7 @@ def build_telegram_alert_monitoring_coverage_readback( runtime_log_readback: Mapping[str, Any], log_controlled_writeback_consumer: Mapping[str, Any] | None = None, source_contract: Mapping[str, Any], + monitoring_live_receipt_apply_readback: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Build deterministic coverage from static inventory and live readbacks.""" @@ -249,10 +438,30 @@ def build_telegram_alert_monitoring_coverage_readback( consumer_readback = _dict(log_controlled_writeback_consumer) consumer_rollups = _dict(consumer_readback.get("rollups")) consumer_context = _dict(consumer_readback.get("telegram_alert_learning_context")) + live_receipt_apply_readback = _dict(monitoring_live_receipt_apply_readback) + live_receipt_apply_summary = _dict(live_receipt_apply_readback.get("summary")) + monitoring_asset_rollups = _monitoring_asset_rollups( + monitoring_inventory, + monitoring_readback_plan, + metadata_live_receipt_rows=_list_of_dicts( + live_receipt_apply_readback.get("rows") + ), + ) - monitoring_surface_count = _int(inventory_summary.get("surface_count")) - live_evidence_received_count = _int( - inventory_summary.get("live_evidence_received_count") + monitoring_surface_count = max( + _int(inventory_summary.get("surface_count")), + _int(monitoring_asset_rollups.get("surface_count")), + ) + live_evidence_received_count = max( + _int(inventory_summary.get("live_evidence_received_count")), + _int( + monitoring_asset_rollups.get( + "metadata_live_receipt_accepted_surface_count" + ) + ), + _int(monitoring_asset_rollups.get("accepted_live_receipt_count")) + if monitoring_surface_count <= _int(monitoring_asset_rollups.get("surface_count")) + else 0, ) monitoring_live_gap_count = max( monitoring_surface_count - live_evidence_received_count, @@ -325,10 +534,6 @@ def build_telegram_alert_monitoring_coverage_readback( ai_alert_ready_total=effective_ai_alert_ready_total, ) ready = not active_blockers - monitoring_asset_rollups = _monitoring_asset_rollups( - monitoring_inventory, - monitoring_readback_plan, - ) alert_receipt_pipeline = _alert_receipt_pipeline( monitoring_surface_count=monitoring_surface_count, live_evidence_received_count=live_evidence_received_count, @@ -394,6 +599,10 @@ def build_telegram_alert_monitoring_coverage_readback( "all_monitoring_inventory_surfaces_have_accepted_live_receipt": ( monitoring_live_gap_count == 0 ), + "metadata_live_receipts_count_as_db_or_log_receipt": ( + monitoring_live_gap_count == 0 + and _int(live_receipt_apply_summary.get("ready_row_count")) > 0 + ), "manual_default_terminal_state_allowed": False, "critical_break_glass_exception_only": True, }, @@ -404,6 +613,21 @@ def build_telegram_alert_monitoring_coverage_readback( ), "monitoring_live_evidence_received_count": live_evidence_received_count, "monitoring_live_receipt_gap_count": monitoring_live_gap_count, + "monitoring_metadata_live_receipt_apply_readback_ok": ( + live_receipt_apply_readback.get("status") == "ok" + ), + "monitoring_metadata_live_receipt_readback_source": str( + live_receipt_apply_summary.get("readback_source") or "unavailable" + ), + "monitoring_metadata_live_receipt_row_count": _int( + live_receipt_apply_summary.get("row_count") + ), + "monitoring_metadata_live_receipt_ready_row_count": _int( + live_receipt_apply_summary.get("ready_row_count") + ), + "monitoring_metadata_live_receipt_accepted_surface_count": _int( + live_receipt_apply_summary.get("accepted_surface_count") + ), "monitoring_readback_candidate_count": _int( readback_summary.get("readback_candidate_count") ), @@ -530,6 +754,9 @@ def build_telegram_alert_monitoring_coverage_readback( "metadata_read_performed": True, "runtime_db_read_performed": runtime_db_ok, "consumer_context_read_performed": bool(consumer_readback), + "metadata_live_receipt_apply_read_performed": ( + live_receipt_apply_readback.get("status") == "ok" + ), "telegram_send_performed": False, "bot_api_call_performed": False, "gateway_queue_write_performed": False, @@ -765,7 +992,7 @@ async def _load_runtime_log_readback_direct( db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://") event_types_sql = ", ".join(f"'{event_type}'" for event_type in _ALERT_LOG_EVENT_TYPES) try: - import asyncpg + import asyncpg # type: ignore[import-untyped] conn = await asyncio.wait_for( asyncpg.connect(db_url), @@ -909,17 +1136,35 @@ def _required_tag_dimensions() -> list[dict[str, str]]: def _monitoring_asset_rollups( monitoring_inventory: Mapping[str, Any], monitoring_readback_plan: Mapping[str, Any], + *, + metadata_live_receipt_rows: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: surfaces = _list_of_dicts(monitoring_inventory.get("observability_surfaces")) readback_candidates = _list_of_dicts( monitoring_readback_plan.get("readback_candidates") ) inventory_source = readback_candidates or surfaces + metadata_receipt_surface_ids = _metadata_live_receipt_surface_ids( + [ + row + for row in (metadata_live_receipt_rows or []) + if _metadata_live_receipt_row_ready(row) + ] + ) + metadata_receipt_batches = _unique( + [ + str(row.get("batch_id") or "") + for row in (metadata_live_receipt_rows or []) + if _metadata_live_receipt_row_ready(row) + and str(row.get("batch_id") or "") + ] + ) accepted_live_receipts = sum( 1 for item in inventory_source if bool(item.get("post_incident_readback_accepted")) or bool(item.get("live_evidence_received")) + or str(item.get("surface_id") or "") in metadata_receipt_surface_ids ) live_receipt_gap_samples = [ _monitoring_gap_sample(item) @@ -927,12 +1172,21 @@ def _monitoring_asset_rollups( if bool(item.get("requires_live_evidence")) and not bool(item.get("post_incident_readback_accepted")) and not bool(item.get("live_evidence_received")) + and str(item.get("surface_id") or "") not in metadata_receipt_surface_ids ][:12] - live_receipt_work_batches = _monitoring_live_receipt_work_batches(inventory_source) + live_receipt_work_batches = _monitoring_live_receipt_work_batches( + inventory_source, + accepted_surface_ids=metadata_receipt_surface_ids, + ) return { "surface_count": len(surfaces) or len(inventory_source), "readback_candidate_count": len(readback_candidates), "accepted_live_receipt_count": accepted_live_receipts, + "metadata_live_receipt_accepted_surface_count": len( + metadata_receipt_surface_ids + ), + "metadata_live_receipt_batch_count": len(metadata_receipt_batches), + "metadata_live_receipt_batch_ids": metadata_receipt_batches, "live_receipt_gap_count": max( len(inventory_source) - accepted_live_receipts, 0, @@ -987,13 +1241,17 @@ def _monitoring_gap_sample(item: Mapping[str, Any]) -> dict[str, Any]: def _monitoring_live_receipt_work_batches( inventory_source: list[dict[str, Any]], + *, + accepted_surface_ids: set[str] | None = None, ) -> list[dict[str, Any]]: + accepted_surface_ids = accepted_surface_ids or set() gap_items = [ item for item in inventory_source if bool(item.get("requires_live_evidence")) and not bool(item.get("post_incident_readback_accepted")) and not bool(item.get("live_evidence_received")) + and str(item.get("surface_id") or "") not in accepted_surface_ids ] grouped: dict[str, list[dict[str, Any]]] = {} for item in gap_items: @@ -1815,12 +2073,84 @@ def _read_source_contract_text(repo_root: Path, relative_path: str) -> str: return "" +def _metadata_live_receipt_row_ready(row: Mapping[str, Any]) -> bool: + if str(row.get("semantic_operation_type") or "") != ( + _MONITORING_LIVE_RECEIPT_OPERATION_TYPE + ): + return False + if str(row.get("actor") or "") != _MONITORING_LIVE_RECEIPT_EXECUTOR_ROUTE: + return False + if str(row.get("status") or "") != "success": + return False + if str(row.get("raw_payload_included") or "").lower() == "true": + return False + if str(row.get("monitoring_live_receipt_metadata_recorded") or "").lower() != ( + "true" + ): + return False + if str(row.get("target_context_receipt_write_performed") or "").lower() != "true": + return False + if not _metadata_live_receipt_surface_ids([row]): + return False + if not _json_list(row.get("post_apply_verifier_refs")): + return False + return True + + +def _metadata_live_receipt_surface_ids( + rows: Sequence[Mapping[str, Any]], +) -> set[str]: + surface_ids: set[str] = set() + for row in rows: + selector = _json_dict(row.get("target_selector")) + for surface_id in _json_list(selector.get("surface_ids")): + if surface_id: + surface_ids.add(surface_id) + return surface_ids + + +def _result_rows(result: Any) -> list[dict[str, Any]]: + try: + return [dict(row) for row in result.mappings().all()] + except AttributeError: + rows = getattr(result, "rows", []) + return [dict(row) for row in rows] + + def _list_of_dicts(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] return [dict(item) for item in value if isinstance(item, Mapping)] +def _json_dict(value: Any) -> dict[str, Any]: + if isinstance(value, Mapping): + return dict(value) + if isinstance(value, str): + try: + decoded = json.loads(value) + except json.JSONDecodeError: + return {} + return dict(decoded) if isinstance(decoded, Mapping) else {} + return {} + + +def _json_list(value: Any) -> list[str]: + if isinstance(value, list): + return [str(item) for item in value if str(item)] + if isinstance(value, tuple): + return [str(item) for item in value if str(item)] + if isinstance(value, str): + try: + decoded = json.loads(value) + except json.JSONDecodeError: + return [value] if value else [] + if isinstance(decoded, list): + return [str(item) for item in decoded if str(item)] + return [value] if value else [] + return [] + + def _count_by_key(items: list[dict[str, Any]], key: str) -> dict[str, int]: counts: dict[str, int] = {} for item in items: 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 7adfabf01..32b00808d 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 @@ -16,6 +16,7 @@ from typing import Any from sqlalchemy import text +from src.core.config import settings from src.core.logging import get_logger from src.db.base import get_db_context from src.services.ai_agent_log_controlled_writeback_dispatch import ( @@ -37,6 +38,10 @@ _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_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5 +_DB_DIRECT_QUERY_TIMEOUT_SECONDS = 0.85 +_DB_DIRECT_CLOSE_TIMEOUT_SECONDS = 0.25 +_DB_DIRECT_STATEMENT_TIMEOUT_MS = 750 logger = get_logger(__name__) @@ -460,6 +465,12 @@ async def _write_live_receipts_with_retry( attempt=attempt, error_type=type(exc).__name__, ) + direct_result = await _write_live_receipts_direct( + project_id=project_id, + receipt=receipt, + ) + if direct_result is not None: + return direct_result if last_error is not None: raise last_error return FALLBACK_LEDGER_OPERATION_TYPE, [] @@ -518,68 +529,14 @@ async def _insert_live_receipt_row( ) RETURNING op_id::text """), - { + _live_receipt_log_values( + project_id=project_id, + item=item, + ledger_operation_type=ledger_operation_type, + ) + | { "operation_type": OPERATION_TYPE, - "ledger_operation_type": ledger_operation_type, - "actor": EXECUTOR_ROUTE, "live_receipt_id": item["live_receipt_id"], - "input": json.dumps( - { - "schema_version": APPLY_SCHEMA_VERSION, - "project_id": project_id, - "semantic_operation_type": OPERATION_TYPE, - "ledger_operation_type": ledger_operation_type, - "live_receipt_id": item["live_receipt_id"], - "batch_id": item["batch_id"], - "domain": item["domain"], - "priority": item["priority"], - "surface_count": item["surface_count"], - "write_capable_surface_count": item[ - "write_capable_surface_count" - ], - "target_selector": item["target_selector"], - "tag_hints": item["tag_hints"], - "runtime_target_write_performed": True, - "raw_payload_included": False, - }, - ensure_ascii=False, - ), - "output": json.dumps( - { - "executor_route": EXECUTOR_ROUTE, - "semantic_operation_type": OPERATION_TYPE, - "ledger_operation_type": ledger_operation_type, - "monitoring_live_receipt_metadata_recorded": True, - "target_context_receipt_write_performed": True, - "runtime_target_write_performed": True, - "post_apply_verifier_refs": item["post_apply_verifier_refs"], - "next_action": ( - "readback_telegram_monitoring_live_receipt_apply_then_verify_gap_delta" - ), - "km_write_performed": False, - "rag_index_write_performed": False, - "playbook_trust_write_performed": False, - "mcp_tool_call_performed": False, - "telegram_send_performed": False, - "bot_api_call_performed": False, - "alertmanager_reload_performed": False, - }, - ensure_ascii=False, - ), - "dry_run_result": json.dumps( - { - "target_selector": item["target_selector"], - "source_of_truth_diff": item["source_of_truth_diff"], - "check_mode": item["check_mode"], - "rollback": item["rollback"], - "post_apply_verifier_ref_count": len( - item["post_apply_verifier_refs"] - ), - "raw_payload_included": False, - }, - ensure_ascii=False, - ), - "tags": _receipt_tags(item), }, ) op_id = result.scalar() @@ -616,6 +573,174 @@ async def _insert_live_receipt_row( } +async def _write_live_receipts_direct( + *, + project_id: str, + receipt: dict[str, Any], +) -> tuple[str, list[dict[str, Any]]] | None: + """Write metadata live receipts with a short-lived DB connection.""" + + db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://") + try: + import asyncpg # type: ignore[import-untyped] + + conn = await asyncio.wait_for( + asyncpg.connect(db_url), + timeout=_DB_DIRECT_CONNECT_TIMEOUT_SECONDS, + ) + try: + await asyncio.wait_for( + conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id), + timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + await asyncio.wait_for( + conn.execute( + "SET statement_timeout = " + f"'{int(_DB_DIRECT_STATEMENT_TIMEOUT_MS)}ms'" + ), + timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + ledger_operation_type = await _resolve_ledger_operation_type_direct(conn) + rows = [] + for item in receipt["live_receipt_apply_receipts"]: + rows.append( + await _insert_live_receipt_row_direct( + conn, + project_id=project_id, + item=item, + ledger_operation_type=ledger_operation_type, + ) + ) + return ledger_operation_type, rows + finally: + try: + await asyncio.wait_for( + conn.close(), + timeout=_DB_DIRECT_CLOSE_TIMEOUT_SECONDS, + ) + except Exception as close_exc: # pragma: no cover - live cleanup + logger.warning( + "telegram_monitoring_live_receipt_apply_direct_close_failed", + project_id=project_id, + error_type=type(close_exc).__name__, + ) + except Exception as exc: # pragma: no cover - live DB pressure + logger.warning( + "telegram_monitoring_live_receipt_apply_direct_write_failed", + project_id=project_id, + error_type=type(exc).__name__, + ) + return None + + +async def _resolve_ledger_operation_type_direct(conn: Any) -> str: + resolved = await asyncio.wait_for( + conn.fetchval( + """ + SELECT CASE + WHEN EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'automation_operation_log_type_valid' + AND pg_get_constraintdef(oid) LIKE '%' || $1 || '%' + ) + THEN $1 + ELSE $2 + END + """, + OPERATION_TYPE, + FALLBACK_LEDGER_OPERATION_TYPE, + ), + timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + if str(resolved or "") == OPERATION_TYPE: + return OPERATION_TYPE + return FALLBACK_LEDGER_OPERATION_TYPE + + +async def _insert_live_receipt_row_direct( + conn: Any, + *, + project_id: str, + item: dict[str, Any], + ledger_operation_type: str, +) -> dict[str, Any]: + values = _live_receipt_log_values( + project_id=project_id, + item=item, + ledger_operation_type=ledger_operation_type, + ) + op_id = await asyncio.wait_for( + conn.fetchval( + """ + INSERT INTO automation_operation_log ( + operation_type, actor, status, + input, output, dry_run_result, tags + ) + SELECT + $1, + $2, + 'success', + $3::jsonb, + $4::jsonb, + $5::jsonb, + $6::text[] + WHERE NOT EXISTS ( + SELECT 1 + FROM automation_operation_log existing + WHERE coalesce( + existing.input ->> 'semantic_operation_type', + existing.operation_type + ) = $7 + AND existing.input ->> 'live_receipt_id' = $8 + ) + RETURNING op_id::text + """, + values["ledger_operation_type"], + values["actor"], + values["input"], + values["output"], + values["dry_run_result"], + values["tags"], + OPERATION_TYPE, + item["live_receipt_id"], + ), + timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + if op_id: + return { + "live_receipt_id": item["live_receipt_id"], + "batch_id": item["batch_id"], + "domain": item["domain"], + "created": True, + "op_id": str(op_id), + } + + existing_op_id = await asyncio.wait_for( + conn.fetchval( + """ + SELECT op_id::text + FROM automation_operation_log + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) + = $1 + AND input ->> 'live_receipt_id' = $2 + ORDER BY created_at DESC + LIMIT 1 + """, + OPERATION_TYPE, + item["live_receipt_id"], + ), + timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + return { + "live_receipt_id": item["live_receipt_id"], + "batch_id": item["batch_id"], + "domain": item["domain"], + "created": False, + "op_id": str(existing_op_id or ""), + } + + async def _load_live_receipt_rows_with_retry( *, project_id: str, @@ -634,6 +759,9 @@ async def _load_live_receipt_rows_with_retry( attempt=attempt, error_type=type(exc).__name__, ) + direct_rows = await _load_live_receipt_rows_direct(project_id=project_id) + if direct_rows is not None: + return direct_rows if last_error is not None: raise last_error return [] @@ -688,6 +816,93 @@ async def _load_live_receipt_rows_once( return _result_rows(result) +async def _load_live_receipt_rows_direct( + *, + project_id: str, +) -> list[dict[str, Any]] | None: + db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://") + try: + import asyncpg # type: ignore[import-untyped] + + conn = await asyncio.wait_for( + asyncpg.connect(db_url), + timeout=_DB_DIRECT_CONNECT_TIMEOUT_SECONDS, + ) + try: + await asyncio.wait_for( + conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id), + timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + await asyncio.wait_for( + conn.execute( + "SET statement_timeout = " + f"'{int(_DB_DIRECT_STATEMENT_TIMEOUT_MS)}ms'" + ), + timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + rows = await asyncio.wait_for( + conn.fetch( + """ + SELECT + op_id::text AS op_id, + operation_type, + actor, + status, + created_at, + input ->> 'semantic_operation_type' AS semantic_operation_type, + input ->> 'ledger_operation_type' AS ledger_operation_type, + input ->> 'live_receipt_id' AS live_receipt_id, + input ->> 'batch_id' AS batch_id, + input ->> 'domain' AS domain, + input ->> 'priority' AS priority, + input ->> 'surface_count' AS surface_count, + input ->> 'write_capable_surface_count' + AS write_capable_surface_count, + input ->> 'runtime_target_write_performed' + AS runtime_target_write_performed, + input ->> 'raw_payload_included' AS raw_payload_included, + input -> 'target_selector' AS target_selector, + input -> 'tag_hints' AS tag_hints, + output ->> 'monitoring_live_receipt_metadata_recorded' + AS monitoring_live_receipt_metadata_recorded, + output ->> 'target_context_receipt_write_performed' + AS target_context_receipt_write_performed, + output ->> 'next_action' AS next_action, + output -> 'post_apply_verifier_refs' AS post_apply_verifier_refs + FROM automation_operation_log + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) + = $1 + AND input ->> 'project_id' = $2 + ORDER BY created_at DESC, op_id DESC + LIMIT 100 + """, + OPERATION_TYPE, + project_id, + ), + timeout=_DB_DIRECT_QUERY_TIMEOUT_SECONDS, + ) + return [dict(row) for row in rows] + finally: + try: + await asyncio.wait_for( + conn.close(), + timeout=_DB_DIRECT_CLOSE_TIMEOUT_SECONDS, + ) + except Exception as close_exc: # pragma: no cover - live cleanup + logger.warning( + "telegram_monitoring_live_receipt_apply_readback_direct_close_failed", + project_id=project_id, + error_type=type(close_exc).__name__, + ) + except Exception as exc: # pragma: no cover - live DB pressure + logger.warning( + "telegram_monitoring_live_receipt_apply_readback_direct_failed", + project_id=project_id, + error_type=type(exc).__name__, + ) + return None + + async def _resolve_ledger_operation_type(db: Any) -> str: result = await db.execute( text(""" @@ -1026,6 +1241,75 @@ def _rows_by_batch_id(rows: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: return mapped +def _live_receipt_log_values( + *, + project_id: str, + item: Mapping[str, Any], + ledger_operation_type: str, +) -> dict[str, Any]: + return { + "ledger_operation_type": ledger_operation_type, + "actor": EXECUTOR_ROUTE, + "input": json.dumps( + { + "schema_version": APPLY_SCHEMA_VERSION, + "project_id": project_id, + "semantic_operation_type": OPERATION_TYPE, + "ledger_operation_type": ledger_operation_type, + "live_receipt_id": item["live_receipt_id"], + "batch_id": item["batch_id"], + "domain": item["domain"], + "priority": item["priority"], + "surface_count": item["surface_count"], + "write_capable_surface_count": item[ + "write_capable_surface_count" + ], + "target_selector": item["target_selector"], + "tag_hints": item["tag_hints"], + "runtime_target_write_performed": True, + "raw_payload_included": False, + }, + ensure_ascii=False, + ), + "output": json.dumps( + { + "executor_route": EXECUTOR_ROUTE, + "semantic_operation_type": OPERATION_TYPE, + "ledger_operation_type": ledger_operation_type, + "monitoring_live_receipt_metadata_recorded": True, + "target_context_receipt_write_performed": True, + "runtime_target_write_performed": True, + "post_apply_verifier_refs": item["post_apply_verifier_refs"], + "next_action": ( + "readback_telegram_monitoring_live_receipt_apply_then_verify_gap_delta" + ), + "km_write_performed": False, + "rag_index_write_performed": False, + "playbook_trust_write_performed": False, + "mcp_tool_call_performed": False, + "telegram_send_performed": False, + "bot_api_call_performed": False, + "alertmanager_reload_performed": False, + }, + ensure_ascii=False, + ), + "dry_run_result": json.dumps( + { + "target_selector": item["target_selector"], + "source_of_truth_diff": item["source_of_truth_diff"], + "check_mode": item["check_mode"], + "rollback": item["rollback"], + "post_apply_verifier_ref_count": len( + item["post_apply_verifier_refs"] + ), + "raw_payload_included": False, + }, + ensure_ascii=False, + ), + "tags": _receipt_tags(item), + } + + def _receipt_tags(item: Mapping[str, Any]) -> list[str]: tag_hints = _dict(item.get("tag_hints")) tags = [ @@ -1034,7 +1318,20 @@ def _receipt_tags(item: Mapping[str, Any]) -> list[str]: "controlled_live_receipt", str(item.get("domain") or "monitoring_general"), ] - for key in ("product_id", "service_name", "tool_id"): + for key in ( + "project_id", + "product_id", + "website_id", + "service_name", + "package_name", + "tool_id", + "log_source", + "alertname", + "playbook_id", + "rag_context_ref", + "mcp_evidence_ref", + "schedule_id", + ): value = str(tag_hints.get(key) or "") if value: tags.append(value) diff --git a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py index 8b203e032..8906f2688 100644 --- a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py +++ b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py @@ -37,6 +37,7 @@ def _payload( ai_alert_card_delivery_readback: dict | None = None, runtime_log_readback: dict | None = None, log_controlled_writeback_consumer: dict | None = None, + monitoring_live_receipt_apply_readback: dict | None = None, ) -> dict: default_post_apply_verifier = { "schema_version": "telegram_alert_learning_context_post_apply_verifier_v1", @@ -158,10 +159,59 @@ def _payload( ), runtime_log_readback=runtime_log_readback or default_runtime_log_readback, log_controlled_writeback_consumer=log_controlled_writeback_consumer, + monitoring_live_receipt_apply_readback=monitoring_live_receipt_apply_readback, source_contract=_source_contract(), ) +def _metadata_live_receipt_apply_readback(*, surface_count: int = 60) -> dict: + surface_ids = [ + "prometheus_k8s_base_config", + "telegram_gateway_config", + *[f"metadata_live_surface_{index:02d}" for index in range(3, surface_count + 1)], + ][:surface_count] + return { + "status": "ok", + "summary": { + "readback_source": "direct_connection", + "row_count": 1, + "ready_row_count": 1, + "accepted_surface_count": len(surface_ids), + }, + "rows": [ + { + "op_id": "live-receipt-op-1", + "operation_type": "km_linked", + "actor": "ai_agent_monitoring_live_receipt_consumer", + "status": "success", + "semantic_operation_type": "telegram_monitoring_live_receipt_applied", + "live_receipt_id": ( + "telegram_monitoring_live_receipt_applied::" + "tg_live_receipt_all" + ), + "batch_id": "tg_live_receipt_all", + "domain": "monitoring_general", + "raw_payload_included": "false", + "target_selector": {"surface_ids": surface_ids}, + "tag_hints": { + "project_id": "awoooi", + "product_id": "awoooi", + "website_id": "awoooi.wooo.work", + "service_name": "monitoring", + "package_name": "monitoring-receipt", + "tool_id": "monitoring_general", + }, + "monitoring_live_receipt_metadata_recorded": "true", + "target_context_receipt_write_performed": "true", + "post_apply_verifier_refs": [ + "/api/v1/agents/telegram-alert-monitoring-coverage-readback" + "#batch=tg_live_receipt_all" + ], + } + ], + } + + def _consumer_context_readback() -> dict: return { "schema_version": "ai_agent_log_controlled_writeback_consumer_readback_v1", @@ -329,6 +379,55 @@ def test_telegram_alert_monitoring_coverage_readback_surfaces_live_gaps(): assert marker not in serialized +def test_telegram_alert_monitoring_coverage_accepts_metadata_live_receipts(): + payload = _payload( + monitoring_live_receipt_apply_readback=_metadata_live_receipt_apply_readback() + ) + + assert payload["status"] == "telegram_alert_monitoring_coverage_ready" + assert payload["operator_answer"]["all_telegram_monitoring_alerts_fully_audited"] is True + assert ( + payload["operator_answer"][ + "all_telegram_monitoring_alert_surfaces_have_db_or_log_receipt" + ] + is True + ) + assert ( + payload["operator_answer"][ + "all_telegram_monitoring_alerts_ai_agent_automation_ready" + ] + is True + ) + assert ( + payload["operator_answer"][ + "all_monitoring_inventory_surfaces_have_accepted_live_receipt" + ] + is True + ) + assert ( + payload["operator_answer"]["metadata_live_receipts_count_as_db_or_log_receipt"] + is True + ) + assert payload["summary"]["monitoring_live_receipt_gap_count"] == 0 + assert payload["summary"]["monitoring_live_evidence_received_count"] == 60 + assert payload["summary"]["monitoring_metadata_live_receipt_ready_row_count"] == 1 + assert ( + payload["summary"]["monitoring_metadata_live_receipt_accepted_surface_count"] + == 60 + ) + assert payload["summary"]["ai_controlled_gap_queue_count"] == 0 + assert payload["summary"]["alert_receipt_pipeline_ready_count"] == 7 + assert payload["summary"]["telegram_monitoring_audit_completion_percent"] == 100.0 + assert payload["active_blockers"] == [] + assert payload["ai_controlled_gap_queue"] == [] + assert ( + payload["operation_boundaries"]["metadata_live_receipt_apply_read_performed"] + is True + ) + assert payload["operation_boundaries"]["telegram_send_performed"] is False + assert payload["operation_boundaries"]["raw_alert_payload_stored"] is False + + def test_telegram_alert_monitoring_coverage_uses_consumer_context_fallback(): payload = _payload( post_apply_verifier={