From 0903bc26dde8dcbc5b9ad7c74b20a2671d54b186 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 23:29:51 +0800 Subject: [PATCH] fix(agents): reuse consumer context in alert coverage --- ...gram_alert_monitoring_coverage_readback.py | 239 ++++++++++++++++-- ..._alert_monitoring_coverage_readback_api.py | 178 ++++++++++--- 2 files changed, 364 insertions(+), 53 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 c89ad0c1d..1bf5007dd 100644 --- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py +++ b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py @@ -21,6 +21,9 @@ from sqlalchemy import text from src.core.logging import get_logger from src.db.base import get_db_context +from src.services.ai_agent_log_controlled_writeback_consumer_readback import ( + load_latest_ai_agent_log_controlled_writeback_consumer_readback, +) from src.services.platform_operator_service import list_ai_alert_card_delivery_readback from src.services.snapshot_paths import default_security_dir, resolve_repo_root from src.services.telegram_alert_ai_automation_matrix import ( @@ -98,12 +101,20 @@ async def load_latest_telegram_alert_monitoring_coverage_readback( ) _require_safe_boundaries(monitoring_inventory, monitoring_readback_plan) - matrix, verifier, ai_alert_readback, runtime_log_readback, source_contract = ( + ( + matrix, + verifier, + ai_alert_readback, + runtime_log_readback, + consumer_readback, + source_contract, + ) = ( await asyncio.gather( _load_telegram_matrix_readback(project_id=project_id), _load_post_apply_verifier_readback(project_id=project_id), _load_ai_alert_card_delivery(project_id=project_id), _load_runtime_log_readback(project_id=project_id), + _load_log_controlled_writeback_consumer_readback(project_id=project_id), asyncio.to_thread(_inspect_source_contract, repo_root), ) ) @@ -116,6 +127,7 @@ async def load_latest_telegram_alert_monitoring_coverage_readback( post_apply_verifier=verifier, ai_alert_card_delivery_readback=ai_alert_readback, runtime_log_readback=runtime_log_readback, + log_controlled_writeback_consumer=consumer_readback, source_contract=source_contract, ) @@ -175,6 +187,47 @@ async def _load_post_apply_verifier_readback(*, project_id: str) -> dict[str, An } +async def _load_log_controlled_writeback_consumer_readback( + *, + project_id: str, +) -> dict[str, Any]: + try: + return await asyncio.wait_for( + load_latest_ai_agent_log_controlled_writeback_consumer_readback( + project_id=project_id + ), + timeout=_LIVE_READBACK_TIMEOUT_SECONDS, + ) + except Exception as exc: # pragma: no cover - live runtime pressure + logger.warning( + "telegram_alert_monitoring_consumer_readback_unavailable", + project_id=project_id, + error_type=type(exc).__name__, + ) + return { + "schema_version": "ai_agent_log_controlled_writeback_consumer_readback_v1", + "status": "blocked_waiting_controlled_writeback_consumer_db_readback", + "rollups": { + "controlled_consumer_readback_ready": False, + "ready_target_count": 0, + "telegram_alert_learning_context_readback_ready": False, + "telegram_alert_learning_context_receipt_count": 0, + "telegram_alert_learning_ai_agent_context_receipt_count": 0, + "error_type": type(exc).__name__, + }, + "telegram_alert_learning_context": { + "status": "blocked_telegram_alert_learning_registry_readback_unavailable", + "source_registry_status": "unavailable", + "active_blockers": [ + f"log_controlled_writeback_consumer_readback_unavailable:{type(exc).__name__}" + ], + }, + "active_blockers": [ + "log_controlled_writeback_consumer_readback_unavailable" + ], + } + + def build_telegram_alert_monitoring_coverage_readback( *, project_id: str, @@ -184,6 +237,7 @@ def build_telegram_alert_monitoring_coverage_readback( post_apply_verifier: Mapping[str, Any], ai_alert_card_delivery_readback: Mapping[str, Any], runtime_log_readback: Mapping[str, Any], + log_controlled_writeback_consumer: Mapping[str, Any] | None = None, source_contract: Mapping[str, Any], ) -> dict[str, Any]: """Build deterministic coverage from static inventory and live readbacks.""" @@ -194,6 +248,9 @@ def build_telegram_alert_monitoring_coverage_readback( verifier_rollups = _dict(post_apply_verifier.get("rollups")) ai_alert_summary = _dict(ai_alert_card_delivery_readback.get("summary")) runtime_summary = _dict(runtime_log_readback.get("summary")) + consumer_readback = _dict(log_controlled_writeback_consumer) + consumer_rollups = _dict(consumer_readback.get("rollups")) + consumer_context = _dict(consumer_readback.get("telegram_alert_learning_context")) monitoring_surface_count = _int(inventory_summary.get("surface_count")) live_evidence_received_count = _int( @@ -216,6 +273,25 @@ def build_telegram_alert_monitoring_coverage_readback( ai_alert_ready_total = _int( ai_alert_summary.get("learning_writeback_ready_total") ) + ai_loop_context = _effective_ai_loop_context( + verifier_ready=verifier_ready, + verifier_rollups=verifier_rollups, + ai_alert_total=ai_alert_total, + ai_alert_ready_total=ai_alert_ready_total, + consumer_rollups=consumer_rollups, + consumer_context=consumer_context, + ) + effective_verifier_ready = bool(ai_loop_context["ready"]) + effective_ai_alert_total = max(ai_alert_total, _int(ai_loop_context["receipt_count"])) + effective_ai_alert_ready_total = max( + ai_alert_ready_total, + _int(ai_loop_context["receipt_count"]), + ) + effective_ai_alert_summary = { + **ai_alert_summary, + "total": effective_ai_alert_total, + "learning_writeback_ready_total": effective_ai_alert_ready_total, + } runtime_event_type_counts = { str(key): _int(value) for key, value in _dict( @@ -232,22 +308,23 @@ def build_telegram_alert_monitoring_coverage_readback( readback_summary=readback_summary, matrix_summary=matrix_summary, verifier_rollups=verifier_rollups, - ai_alert_summary=ai_alert_summary, + ai_alert_summary=effective_ai_alert_summary, runtime_summary=runtime_summary, source_contract=source_contract, runtime_db_ok=runtime_db_ok, ai_alert_db_ok=ai_alert_db_ok, - verifier_ready=verifier_ready, + verifier_ready=effective_verifier_ready, + ai_loop_context=ai_loop_context, ) active_blockers = _active_blockers( monitoring_live_gap_count=monitoring_live_gap_count, direct_gap_count=direct_gap_count, runtime_db_ok=runtime_db_ok, ai_alert_db_ok=ai_alert_db_ok, - verifier_ready=verifier_ready, + verifier_ready=effective_verifier_ready, source_contract=source_contract, - ai_alert_total=ai_alert_total, - ai_alert_ready_total=ai_alert_ready_total, + ai_alert_total=effective_ai_alert_total, + ai_alert_ready_total=effective_ai_alert_ready_total, ) ready = not active_blockers monitoring_asset_rollups = _monitoring_asset_rollups( @@ -262,9 +339,9 @@ def build_telegram_alert_monitoring_coverage_readback( runtime_db_ok=runtime_db_ok, runtime_summary=runtime_summary, ai_alert_db_ok=ai_alert_db_ok, - ai_alert_total=ai_alert_total, - ai_alert_ready_total=ai_alert_ready_total, - verifier_ready=verifier_ready, + ai_alert_total=effective_ai_alert_total, + ai_alert_ready_total=effective_ai_alert_ready_total, + verifier_ready=effective_verifier_ready, source_ready=source_ready, ) work_item_progress = _work_item_progress( @@ -273,9 +350,9 @@ def build_telegram_alert_monitoring_coverage_readback( runtime_db_ok=runtime_db_ok, runtime_summary=runtime_summary, ai_alert_db_ok=ai_alert_db_ok, - ai_alert_total=ai_alert_total, - ai_alert_ready_total=ai_alert_ready_total, - verifier_ready=verifier_ready, + ai_alert_total=effective_ai_alert_total, + ai_alert_ready_total=effective_ai_alert_ready_total, + verifier_ready=effective_verifier_ready, source_ready=source_ready, ) ai_controlled_gap_queue = _ai_controlled_gap_queue( @@ -300,14 +377,18 @@ def build_telegram_alert_monitoring_coverage_readback( monitoring_live_gap_count == 0 and runtime_db_ok and ai_alert_db_ok ), "all_telegram_monitoring_alerts_ai_agent_automation_ready": ( - ready and ai_alert_total > 0 and ai_alert_ready_total >= ai_alert_total + ready + and effective_ai_alert_total > 0 + and effective_ai_alert_ready_total >= effective_ai_alert_total ), "ai_controlled_gap_queue_present": bool(ai_controlled_gap_queue), "all_known_telegram_egress_routes_controlled": direct_gap_count == 0, "all_known_telegram_egress_routes_have_db_or_log_receipt": ( direct_gap_count == 0 and ai_alert_db_ok and runtime_db_ok ), - "all_verified_ai_alert_context_receipts_reusable_by_ai_agent": verifier_ready, + "all_verified_ai_alert_context_receipts_reusable_by_ai_agent": ( + effective_verifier_ready + ), "all_monitoring_inventory_surfaces_have_accepted_live_receipt": ( monitoring_live_gap_count == 0 ), @@ -351,6 +432,8 @@ def build_telegram_alert_monitoring_coverage_readback( ), "ai_alert_card_delivery_total": ai_alert_total, "ai_alert_card_learning_writeback_ready_total": ai_alert_ready_total, + "effective_ai_alert_context_receipt_total": effective_ai_alert_total, + "effective_ai_alert_context_ready_total": effective_ai_alert_ready_total, "runtime_alert_operation_log_total_7d": _int( runtime_summary.get("alert_operation_log_total_7d") ), @@ -370,6 +453,19 @@ def build_telegram_alert_monitoring_coverage_readback( "verified_ai_agent_context_receipt_count": _int( verifier_rollups.get("verified_ai_agent_context_receipt_count") ), + "ai_loop_context_ready": effective_verifier_ready, + "ai_loop_context_source": str(ai_loop_context["source"]), + "ai_loop_context_fallback_used": bool(ai_loop_context["fallback_used"]), + "ai_loop_context_receipt_count": _int(ai_loop_context["receipt_count"]), + "ai_loop_ai_agent_context_receipt_count": _int( + ai_loop_context["ai_agent_context_receipt_count"] + ), + "log_controlled_consumer_readback_ready": bool( + consumer_rollups.get("controlled_consumer_readback_ready") + ), + "log_controlled_consumer_ready_target_count": _int( + consumer_rollups.get("ready_target_count") + ), "source_contract_ready": source_ready, "runtime_db_readback_ok": runtime_db_ok, "ai_alert_card_db_readback_ok": ai_alert_db_ok, @@ -418,12 +514,16 @@ def build_telegram_alert_monitoring_coverage_readback( "post_apply_verifier": ( "/api/v1/agents/telegram-alert-learning-context-post-apply-verifier" ), + "log_controlled_writeback_consumer": ( + "/api/v1/agents/agent-log-controlled-writeback-consumer-readback" + ), "ai_alert_card_delivery": "/api/v1/platform/runs/ai-alert-cards", "alert_operation_log": "/api/v1/alert-operation-logs", }, "operation_boundaries": { "metadata_read_performed": True, "runtime_db_read_performed": runtime_db_ok, + "consumer_context_read_performed": bool(consumer_readback), "telegram_send_performed": False, "bot_api_call_performed": False, "gateway_queue_write_performed": False, @@ -486,6 +586,95 @@ async def _load_ai_alert_card_delivery(*, project_id: str) -> dict[str, Any]: } +def _effective_ai_loop_context( + *, + verifier_ready: bool, + verifier_rollups: Mapping[str, Any], + ai_alert_total: int, + ai_alert_ready_total: int, + consumer_rollups: Mapping[str, Any], + consumer_context: Mapping[str, Any], +) -> dict[str, Any]: + verifier_receipt_count = _int(verifier_rollups.get("verified_context_receipt_count")) + verifier_ai_agent_count = _int( + verifier_rollups.get("verified_ai_agent_context_receipt_count") + ) + consumer_context_ready = ( + bool(consumer_rollups.get("telegram_alert_learning_context_readback_ready")) + or consumer_context.get("status") == "ai_loop_agent_context_receipt_ready" + ) + consumer_receipt_count = _int( + consumer_rollups.get("telegram_alert_learning_context_receipt_count") + or consumer_context.get("context_receipt_count") + ) + consumer_ai_agent_count = _int( + consumer_rollups.get("telegram_alert_learning_ai_agent_context_receipt_count") + or consumer_context.get("ai_agent_context_receipt_count") + ) + consumer_ready_target_count = _int( + consumer_rollups.get("telegram_alert_learning_ready_target_count") + or consumer_rollups.get("ready_target_count") + or consumer_context.get("ready_target_count") + ) + consumer_target_count = _int( + consumer_rollups.get("telegram_alert_learning_target_count") + or consumer_context.get("target_count") + ) + consumer_targets_ready = ( + consumer_context_ready + and consumer_receipt_count > 0 + and consumer_ai_agent_count > 0 + and consumer_target_count > 0 + and consumer_ready_target_count >= consumer_target_count + ) + + if verifier_ready: + return { + "ready": True, + "source": "post_apply_verifier", + "fallback_used": False, + "receipt_count": max( + verifier_receipt_count, + ai_alert_ready_total, + consumer_receipt_count, + ), + "ai_agent_context_receipt_count": max( + verifier_ai_agent_count, + consumer_ai_agent_count, + ), + "ready_target_count": max( + _int(verifier_rollups.get("verified_target_count")), + consumer_ready_target_count, + ), + "target_count": max(consumer_target_count, 0), + } + + if consumer_targets_ready: + return { + "ready": True, + "source": "log_controlled_writeback_consumer", + "fallback_used": True, + "receipt_count": max(consumer_receipt_count, ai_alert_ready_total), + "ai_agent_context_receipt_count": consumer_ai_agent_count, + "ready_target_count": consumer_ready_target_count, + "target_count": consumer_target_count, + } + + return { + "ready": False, + "source": "unavailable", + "fallback_used": False, + "receipt_count": max(verifier_receipt_count, ai_alert_ready_total), + "ai_agent_context_receipt_count": max( + verifier_ai_agent_count, + consumer_ai_agent_count, + ), + "ready_target_count": consumer_ready_target_count, + "target_count": consumer_target_count, + "ai_alert_total": ai_alert_total, + } + + async def _load_runtime_log_readback(*, project_id: str) -> dict[str, Any]: try: return await asyncio.wait_for( @@ -1048,10 +1237,14 @@ def _coverage_matrix( runtime_db_ok: bool, ai_alert_db_ok: bool, verifier_ready: bool, + ai_loop_context: Mapping[str, Any], ) -> list[dict[str, Any]]: direct_gap_count = _int(matrix_summary.get("known_direct_send_gap_count")) ai_alert_total = _int(ai_alert_summary.get("total")) ai_alert_ready_total = _int(ai_alert_summary.get("learning_writeback_ready_total")) + ai_loop_source = str(ai_loop_context.get("source") or "unavailable") + ai_loop_receipt_count = _int(ai_loop_context.get("receipt_count")) + ai_loop_fallback_used = bool(ai_loop_context.get("fallback_used")) return [ { "surface_id": "monitoring_inventory_static_scope", @@ -1152,19 +1345,27 @@ def _coverage_matrix( }, { "surface_id": "telegram_alert_ai_loop_context_verifier", - "status": "verified_context_ready" if verifier_ready else "verifier_not_ready", - "known_item_count": _int( - verifier_rollups.get("verified_context_receipt_count") + "status": ( + "consumer_context_fallback_ready" + if ai_loop_fallback_used and verifier_ready + else "verified_context_ready" + if verifier_ready + else "verifier_not_ready" ), + "known_item_count": ai_loop_receipt_count, "db_or_log_receipt": "ready_context_receipt_source", "ai_route": "ready_ai_agent_reusable_context", "controlled_queue": "ready_metadata_only_post_apply_verifier", - "post_verifier": "ready", + "post_verifier": "ready" + if not ai_loop_fallback_used + else "consumer_context_fallback", "learning_writeback": "ready_km_rag_mcp_playbook_ai_agent_targets", "gap_count": 0 if verifier_ready else 1, "evidence_refs": [ - "/api/v1/agents/telegram-alert-learning-context-post-apply-verifier" + "/api/v1/agents/telegram-alert-learning-context-post-apply-verifier", + "/api/v1/agents/agent-log-controlled-writeback-consumer-readback", ], + "ai_loop_context_source": ai_loop_source, }, { "surface_id": "awooop_operator_surfaces", 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 21d550106..379d994b1 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 @@ -31,7 +31,46 @@ def _source_contract() -> dict[str, bool]: } -def _payload() -> dict: +def _payload( + *, + post_apply_verifier: dict | None = None, + ai_alert_card_delivery_readback: dict | None = None, + runtime_log_readback: dict | None = None, + log_controlled_writeback_consumer: dict | None = None, +) -> dict: + default_post_apply_verifier = { + "schema_version": "telegram_alert_learning_context_post_apply_verifier_v1", + "status": "telegram_alert_learning_context_post_apply_verified", + "rollups": { + "telegram_alert_learning_context_post_apply_verifier_ready": True, + "verified_context_receipt_count": 36, + "verified_target_count": 6, + "verified_ai_agent_context_receipt_count": 6, + }, + } + default_ai_alert_card_delivery_readback = { + "status": "ok", + "summary": { + "total": 36, + "learning_writeback_ready_total": 36, + }, + } + default_runtime_log_readback = { + "status": "ok", + "summary": { + "alert_operation_log_total_7d": 41, + "telegram_sent_event_count_7d": 8, + "km_converted_event_count_7d": 4, + "playbook_draft_event_count_7d": 3, + "outbound_telegram_total_7d": 8, + }, + "event_type_counts_7d": { + "ALERT_RECEIVED": 15, + "TELEGRAM_SENT": 8, + "KM_CONVERTED": 4, + "PLAYBOOK_DRAFT_CREATED": 3, + }, + } return build_telegram_alert_monitoring_coverage_readback( project_id="awoooi", monitoring_inventory={ @@ -112,43 +151,48 @@ def _payload() -> dict: "learning_writeback_ready_surface_count": 9, }, }, - post_apply_verifier={ - "schema_version": "telegram_alert_learning_context_post_apply_verifier_v1", - "status": "telegram_alert_learning_context_post_apply_verified", - "rollups": { - "telegram_alert_learning_context_post_apply_verifier_ready": True, - "verified_context_receipt_count": 36, - "verified_target_count": 6, - "verified_ai_agent_context_receipt_count": 6, - }, - }, - ai_alert_card_delivery_readback={ - "status": "ok", - "summary": { - "total": 36, - "learning_writeback_ready_total": 36, - }, - }, - runtime_log_readback={ - "status": "ok", - "summary": { - "alert_operation_log_total_7d": 41, - "telegram_sent_event_count_7d": 8, - "km_converted_event_count_7d": 4, - "playbook_draft_event_count_7d": 3, - "outbound_telegram_total_7d": 8, - }, - "event_type_counts_7d": { - "ALERT_RECEIVED": 15, - "TELEGRAM_SENT": 8, - "KM_CONVERTED": 4, - "PLAYBOOK_DRAFT_CREATED": 3, - }, - }, + post_apply_verifier=post_apply_verifier or default_post_apply_verifier, + ai_alert_card_delivery_readback=( + ai_alert_card_delivery_readback + or default_ai_alert_card_delivery_readback + ), + runtime_log_readback=runtime_log_readback or default_runtime_log_readback, + log_controlled_writeback_consumer=log_controlled_writeback_consumer, source_contract=_source_contract(), ) +def _consumer_context_readback() -> dict: + return { + "schema_version": "ai_agent_log_controlled_writeback_consumer_readback_v1", + "status": "controlled_writeback_consumer_readback_ready", + "rollups": { + "controlled_consumer_readback_ready": True, + "ready_target_count": 6, + "telegram_alert_learning_context_readback_ready": True, + "telegram_alert_learning_context_receipt_count": 6, + "telegram_alert_learning_ai_agent_context_receipt_count": 1, + "telegram_alert_learning_ready_target_count": 6, + "telegram_alert_learning_target_count": 6, + }, + "telegram_alert_learning_context": { + "schema_version": "telegram_alert_ai_loop_consumer_context_readback_v1", + "status": "ai_loop_agent_context_receipt_ready", + "source_registry_status": ( + "fallback_log_controlled_writeback_consumer_context" + ), + "fallback_source_endpoint": ( + "/api/v1/agents/agent-log-controlled-writeback-consumer-readback" + ), + "target_count": 6, + "ready_target_count": 6, + "context_receipt_count": 6, + "ai_agent_context_receipt_count": 1, + "active_blockers": [], + }, + } + + def test_telegram_alert_monitoring_coverage_readback_surfaces_live_gaps(): payload = _payload() @@ -277,6 +321,72 @@ def test_telegram_alert_monitoring_coverage_readback_surfaces_live_gaps(): assert marker not in serialized +def test_telegram_alert_monitoring_coverage_uses_consumer_context_fallback(): + payload = _payload( + post_apply_verifier={ + "schema_version": "telegram_alert_learning_context_post_apply_verifier_v1", + "status": "telegram_alert_learning_context_post_apply_verifier_unavailable", + "rollups": { + "telegram_alert_learning_context_post_apply_verifier_ready": False, + "verified_context_receipt_count": 0, + "verified_target_count": 0, + "verified_ai_agent_context_receipt_count": 0, + }, + }, + ai_alert_card_delivery_readback={ + "status": "unavailable", + "summary": { + "total": 0, + "learning_writeback_ready_total": 0, + "db_read_status": "unavailable", + }, + }, + log_controlled_writeback_consumer=_consumer_context_readback(), + ) + + summary = payload["summary"] + assert summary["ai_alert_card_db_readback_ok"] is False + assert summary["ai_loop_context_ready"] is True + assert summary["ai_loop_context_source"] == "log_controlled_writeback_consumer" + assert summary["ai_loop_context_fallback_used"] is True + assert summary["ai_loop_context_receipt_count"] == 6 + assert summary["ai_loop_ai_agent_context_receipt_count"] == 1 + assert summary["effective_ai_alert_context_receipt_total"] == 6 + assert summary["effective_ai_alert_context_ready_total"] == 6 + assert summary["log_controlled_consumer_readback_ready"] is True + assert summary["telegram_monitoring_audit_completion_percent"] == 77.8 + + assert ( + payload["operator_answer"][ + "all_verified_ai_alert_context_receipts_reusable_by_ai_agent" + ] + is True + ) + assert ( + "awooop_ai_alert_card_delivery_db_readback_unavailable" + in payload["active_blockers"] + ) + assert "awooop_ai_alert_card_delivery_receipts_missing" not in payload[ + "active_blockers" + ] + assert "awooop_ai_alert_card_learning_writeback_refs_missing" not in payload[ + "active_blockers" + ] + assert "telegram_alert_ai_loop_post_apply_verifier_not_ready" not in payload[ + "active_blockers" + ] + + pipeline = {stage["stage_id"]: stage for stage in payload["alert_receipt_pipeline"]} + assert pipeline["km_rag_mcp_playbook_ai_agent_learning_refs"]["ready"] is True + matrix = {item["surface_id"]: item for item in payload["coverage_matrix"]} + assert matrix["telegram_alert_ai_loop_context_verifier"]["status"] == ( + "consumer_context_fallback_ready" + ) + assert matrix["telegram_alert_ai_loop_context_verifier"][ + "ai_loop_context_source" + ] == "log_controlled_writeback_consumer" + + def test_telegram_alert_monitoring_coverage_endpoint(monkeypatch): async def fake_loader(): return _payload()