diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py index 7a32fc2d8..47f731b35 100644 --- a/apps/api/src/api/v1/agents.py +++ b/apps/api/src/api/v1/agents.py @@ -479,6 +479,10 @@ from src.services.telegram_alert_learning_context_post_apply_verifier import ( from src.services.telegram_alert_monitoring_coverage_readback import ( load_latest_telegram_alert_monitoring_coverage_readback, ) +from src.services.telegram_alert_monitoring_live_receipt_apply import ( + apply_latest_telegram_alert_monitoring_live_receipts, + load_latest_telegram_alert_monitoring_live_receipt_apply_readback, +) router = APIRouter(prefix="/agents", tags=["Agent Teams"]) logger = get_logger("awoooi.agents") @@ -2548,6 +2552,62 @@ async def get_telegram_alert_monitoring_coverage_readback() -> dict[str, Any]: ) from exc +@router.post( + "/telegram-alert-monitoring-live-receipt-apply", + response_model=dict[str, Any], + summary="執行 Telegram monitoring live receipt metadata apply", + description=( + "把 Telegram monitoring coverage readback 中的 AI controlled live receipt " + "batches 寫入 automation_operation_log，形成 metadata-only receipt，" + "供 KM / RAG / MCP / PlayBook / AI Agent loop 消費。此端點不送 Telegram、" + "不呼叫 Bot API、不改 receiver、不 reload Alertmanager、不觸發 workflow、" + "不保存 raw payload、不讀 secret、不呼叫 GitHub；critical break-glass 仍維持硬阻擋。" + ), +) +async def post_telegram_alert_monitoring_live_receipt_apply() -> dict[str, Any]: + """Persist metadata-only monitoring live receipt apply rows.""" + try: + payload = await apply_latest_telegram_alert_monitoring_live_receipts() + return redact_public_lan_topology(payload) + except (json.JSONDecodeError, ValueError) as exc: + logger.error( + "telegram_alert_monitoring_live_receipt_apply_invalid", + error=str(exc), + ) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Telegram monitoring live receipt apply 無效", + ) from exc + + +@router.get( + "/telegram-alert-monitoring-live-receipt-apply-readback", + response_model=dict[str, Any], + summary="取得 Telegram monitoring live receipt metadata apply readback", + description=( + "讀取 automation_operation_log 中的 Telegram monitoring live receipt " + "metadata apply rows，並映射回 coverage batches 與 surface counts。" + "此端點只讀 metadata，不送 Telegram、不呼叫 Bot API、不改 receiver、" + "不 reload Alertmanager、不觸發 workflow、不保存 raw payload、不讀 secret、" + "不呼叫 GitHub，也不把 source live gap 假綠成 accepted receipt。" + ), +) +async def get_telegram_alert_monitoring_live_receipt_apply_readback() -> dict[str, Any]: + """Read metadata-only monitoring live receipt apply rows.""" + try: + payload = await load_latest_telegram_alert_monitoring_live_receipt_apply_readback() + return redact_public_lan_topology(payload) + except (json.JSONDecodeError, ValueError) as exc: + logger.error( + "telegram_alert_monitoring_live_receipt_apply_readback_invalid", + error=str(exc), + ) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Telegram monitoring live receipt apply readback 無效", + ) from exc + + @router.post( "/agent-log-controlled-writeback-consumer-apply", response_model=dict[str, Any], 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 new file mode 100644 index 000000000..b5ae81acb --- /dev/null +++ b/apps/api/src/services/telegram_alert_monitoring_live_receipt_apply.py @@ -0,0 +1,935 @@ +"""Telegram monitoring live receipt metadata apply. + +This service consumes the AI-controlled monitoring live receipt batches from the +coverage readback and writes metadata-only receipts to automation_operation_log. +It does not send Telegram messages, call Bot API, change Alertmanager routes, +store raw alert payloads, call MCP tools, write KM/RAG/PlayBook targets, or read +secrets. +""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import Mapping +from typing import Any + +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_dispatch import ( + FALLBACK_LEDGER_OPERATION_TYPE, +) +from src.services.telegram_alert_monitoring_coverage_readback import ( + DEFAULT_PROJECT_ID, + load_latest_telegram_alert_monitoring_coverage_readback, +) + +APPLY_SCHEMA_VERSION = "telegram_alert_monitoring_live_receipt_apply_receipt_v1" +READBACK_SCHEMA_VERSION = "telegram_alert_monitoring_live_receipt_apply_readback_v1" +OPERATION_TYPE = "telegram_monitoring_live_receipt_applied" +EXECUTOR_ROUTE = "ai_agent_monitoring_live_receipt_consumer" +POST_VERIFIER_ROUTE = "/api/v1/agents/telegram-alert-monitoring-coverage-readback" +APPLY_ROUTE = "/api/v1/agents/telegram-alert-monitoring-live-receipt-apply" +READBACK_ROUTE = "/api/v1/agents/telegram-alert-monitoring-live-receipt-apply-readback" +_DB_RETRY_DELAYS_SECONDS = (0.0, 0.25, 1.0) +logger = get_logger(__name__) + + +async def apply_latest_telegram_alert_monitoring_live_receipts( + *, + project_id: str = DEFAULT_PROJECT_ID, +) -> dict[str, Any]: + """Persist idempotent metadata live receipt rows for coverage batches.""" + + coverage = await load_latest_telegram_alert_monitoring_coverage_readback( + project_id=project_id, + ) + receipt = build_telegram_alert_monitoring_live_receipt_apply_receipt(coverage) + if receipt["active_blockers"]: + receipt["apply_result"] = { + "operation_type": OPERATION_TYPE, + "ledger_operation_type": None, + "semantic_operation_type": OPERATION_TYPE, + "db_write_status": "not_attempted", + "live_receipt_apply_row_count": 0, + "reason": "active_blockers_present", + } + return receipt + + try: + ledger_operation_type, rows = await _write_live_receipts_with_retry( + project_id=project_id, + receipt=receipt, + ) + except Exception as exc: # pragma: no cover - live DB pool pressure + logger.warning( + "telegram_monitoring_live_receipt_apply_db_unavailable", + project_id=project_id, + error_type=type(exc).__name__, + ) + receipt["status"] = "blocked_waiting_telegram_monitoring_live_receipt_apply_db" + receipt["active_blockers"] = _unique([ + *receipt.get("active_blockers", []), + "telegram_monitoring_live_receipt_apply_db_write_unavailable", + ]) + receipt["apply_result"] = { + "operation_type": OPERATION_TYPE, + "ledger_operation_type": None, + "semantic_operation_type": OPERATION_TYPE, + "db_write_status": "unavailable", + "error_type": type(exc).__name__, + "live_receipt_apply_row_count": 0, + "rows": [], + } + return receipt + + 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, + "inserted_count": inserted_count, + "existing_count": existing_count, + "live_receipt_apply_row_count": len(rows), + "rows": rows, + } + receipt["rollups"]["runtime_target_write_performed"] = True + receipt["rollups"]["live_receipt_apply_row_count"] = len(rows) + receipt["rollups"]["inserted_live_receipt_apply_row_count"] = inserted_count + receipt["rollups"]["existing_live_receipt_apply_row_count"] = existing_count + receipt["operation_boundaries"]["monitoring_live_receipt_metadata_write_performed"] = True + receipt["operation_boundaries"]["runtime_target_write_performed"] = True + return receipt + + +async def load_latest_telegram_alert_monitoring_live_receipt_apply_readback( + *, + project_id: str = DEFAULT_PROJECT_ID, +) -> dict[str, Any]: + """Read back metadata live receipt rows and map them to coverage batches.""" + + coverage = await load_latest_telegram_alert_monitoring_coverage_readback( + project_id=project_id, + ) + try: + rows = await _load_live_receipt_rows_with_retry(project_id=project_id) + except Exception as exc: # pragma: no cover - live DB pool pressure + logger.warning( + "telegram_monitoring_live_receipt_apply_readback_db_unavailable", + project_id=project_id, + error_type=type(exc).__name__, + ) + return _fallback_apply_readback( + coverage=coverage, + project_id=project_id, + error_type=type(exc).__name__, + ) + return build_telegram_alert_monitoring_live_receipt_apply_readback( + coverage=coverage, + rows=rows, + ) + + +def build_telegram_alert_monitoring_live_receipt_apply_receipt( + coverage: Mapping[str, Any], +) -> dict[str, Any]: + """Build a deterministic controlled-apply envelope from coverage readback.""" + + batches = _coverage_batches(coverage) + apply_receipts = [_live_receipt_apply_receipt(batch) for batch in batches] + active_blockers = _apply_active_blockers( + coverage=coverage, + batches=batches, + apply_receipts=apply_receipts, + ) + return { + "schema_version": APPLY_SCHEMA_VERSION, + "priority": "P0-TELEGRAM-MONITORING-AI-AUTOMATION-COVERAGE", + "scope": "telegram_alert_monitoring_live_receipt_apply", + "status": ( + "telegram_monitoring_live_receipt_apply_ready" + if not active_blockers + else "blocked_waiting_telegram_monitoring_live_receipt_apply_inputs" + ), + "readback": { + "workplan_id": "CIR-P0-TG-001-LIVE-RECEIPT-APPLY", + "workplan_title": ( + "Telegram monitoring live receipt batches consumed into AI " + "metadata ledger for KM / RAG / MCP / PlayBook / Agent loop" + ), + "source_schema_version": coverage.get("schema_version"), + "source_status": coverage.get("status"), + "operation_type": OPERATION_TYPE, + "executor_route": EXECUTOR_ROUTE, + "safe_next_step": ( + "readback_live_receipt_apply_then_verify_coverage_gap_delta" + ), + }, + "live_receipt_apply_receipts": apply_receipts, + "rollups": { + "source_live_receipt_batch_count": len(batches), + "ready_source_live_receipt_batch_count": sum( + 1 for item in apply_receipts if item["status"] == "ready_for_metadata_live_receipt_write" + ), + "source_surface_count": sum(_int(batch.get("surface_count")) for batch in batches), + "source_write_capable_surface_count": sum( + _int(batch.get("write_capable_surface_count")) for batch in batches + ), + "metadata_only_receipt_count": sum( + 1 for item in apply_receipts if item["raw_payload_included"] is False + ), + "post_apply_verifier_ref_count": sum( + len(item["post_apply_verifier_refs"]) for item in apply_receipts + ), + "controlled_live_receipt_apply_ready": not active_blockers, + "runtime_target_write_performed": False, + "live_receipt_apply_row_count": 0, + "inserted_live_receipt_apply_row_count": 0, + "existing_live_receipt_apply_row_count": 0, + }, + "active_blockers": active_blockers, + "operation_boundaries": { + "metadata_ledger_write_only": True, + "monitoring_live_receipt_metadata_write_performed": False, + "runtime_target_write_performed": False, + "monitoring_inventory_source_mutated": False, + "monitoring_live_receipt_acceptance_mutated": False, + "km_write_performed": False, + "rag_index_write_performed": False, + "playbook_trust_write_performed": False, + "mcp_tool_call_performed": False, + "agent_runtime_action_performed": False, + "telegram_send_performed": False, + "bot_api_call_performed": False, + "alertmanager_reload_performed": False, + "workflow_trigger_performed": False, + "raw_alert_payload_stored": False, + "raw_payload_included": False, + "secret_value_read": False, + "github_api_used": False, + }, + } + + +def build_telegram_alert_monitoring_live_receipt_apply_readback( + *, + coverage: Mapping[str, Any], + rows: list[dict[str, Any]], +) -> dict[str, Any]: + """Build live receipt apply readback from coverage batches and ledger rows.""" + + batches = _coverage_batches(coverage) + rows_by_batch = _rows_by_batch_id(rows) + bindings = [ + _live_receipt_binding(batch=batch, row=rows_by_batch.get(str(batch.get("batch_id") or ""))) + for batch in batches + ] + active_blockers = _readback_active_blockers(batches=batches, bindings=bindings) + applied_bindings = [ + binding for binding in bindings if binding["target_write_performed"] is True + ] + coverage_summary = _dict(coverage.get("summary")) + expected_surface_count = sum(_int(batch.get("surface_count")) for batch in batches) + applied_surface_count = sum( + _int(binding.get("surface_count")) for binding in applied_bindings + ) + ready = not active_blockers and bool(batches) + return { + "schema_version": READBACK_SCHEMA_VERSION, + "priority": "P0-TELEGRAM-MONITORING-AI-AUTOMATION-COVERAGE", + "scope": "telegram_alert_monitoring_live_receipt_apply_readback", + "status": ( + "telegram_monitoring_live_receipt_apply_readback_ready" + if ready + else "blocked_waiting_telegram_monitoring_live_receipt_apply_receipts" + ), + "readback": { + "workplan_id": "CIR-P0-TG-001-LIVE-RECEIPT-APPLY-READBACK", + "workplan_title": ( + "Telegram monitoring live receipt metadata apply rows mapped " + "back to AI controlled batches" + ), + "source_schema_version": coverage.get("schema_version"), + "source_status": coverage.get("status"), + "source_coverage_route": POST_VERIFIER_ROUTE, + "operation_type": OPERATION_TYPE, + "executor_route": EXECUTOR_ROUTE, + "apply_route": APPLY_ROUTE, + "safe_next_step": ( + "keep_live_gap_source_truth_and_consume_metadata_receipts_in_ai_loop" + ), + }, + "operator_answer": { + "all_ai_controlled_live_receipt_batches_have_apply_receipt": ready, + "metadata_live_receipt_apply_performed": bool(applied_bindings), + "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") + ), + "manual_default_terminal_state_allowed": False, + "critical_break_glass_exception_only": True, + }, + "live_receipt_bindings": bindings, + "rollups": { + "expected_live_receipt_batch_count": len(batches), + "applied_live_receipt_batch_count": len(applied_bindings), + "ready_live_receipt_binding_count": sum( + 1 for binding in bindings if binding["status"] == "metadata_live_receipt_applied" + ), + "expected_surface_count": expected_surface_count, + "applied_surface_count": applied_surface_count, + "expected_write_capable_surface_count": sum( + _int(batch.get("write_capable_surface_count")) for batch in batches + ), + "applied_write_capable_surface_count": sum( + _int(binding.get("write_capable_surface_count")) + for binding in applied_bindings + ), + "metadata_only_receipt_count": sum( + 1 for binding in bindings if binding["raw_payload_included"] is False + ), + "post_apply_verifier_ref_count": sum( + len(binding["post_apply_verifier_refs"]) for binding in bindings + ), + "monitoring_live_receipt_gap_count_source_truth": _int( + coverage_summary.get("monitoring_live_receipt_gap_count") + ), + "metadata_live_receipt_apply_readback_ready": ready, + "runtime_target_write_performed": bool(applied_bindings), + }, + "active_blockers": active_blockers, + "operation_boundaries": { + "readback_only": True, + "metadata_ledger_read_performed": True, + "metadata_live_receipt_apply_read_performed": True, + "runtime_target_write_performed": bool(applied_bindings), + "monitoring_inventory_source_mutated": False, + "monitoring_live_receipt_acceptance_mutated": False, + "km_write_performed": False, + "rag_index_write_performed": False, + "playbook_trust_write_performed": False, + "mcp_tool_call_performed": False, + "agent_runtime_action_performed": False, + "telegram_send_performed": False, + "bot_api_call_performed": False, + "alertmanager_reload_performed": False, + "workflow_trigger_performed": False, + "raw_alert_payload_stored": False, + "raw_payload_included": False, + "secret_value_read": False, + "github_api_used": False, + }, + } + + +async def _write_live_receipts_with_retry( + *, + project_id: str, + receipt: dict[str, Any], +) -> tuple[str, list[dict[str, Any]]]: + last_error: Exception | None = None + for attempt, delay_seconds in enumerate(_DB_RETRY_DELAYS_SECONDS, start=1): + if delay_seconds: + await asyncio.sleep(delay_seconds) + try: + return await _write_live_receipts_once( + project_id=project_id, + receipt=receipt, + ) + except Exception as exc: # pragma: no cover - exercised by live pressure + last_error = exc + logger.warning( + "telegram_monitoring_live_receipt_apply_attempt_failed", + project_id=project_id, + attempt=attempt, + error_type=type(exc).__name__, + ) + if last_error is not None: + raise last_error + return FALLBACK_LEDGER_OPERATION_TYPE, [] + + +async def _write_live_receipts_once( + *, + project_id: str, + receipt: dict[str, Any], +) -> tuple[str, list[dict[str, Any]]]: + rows = [] + async with get_db_context(project_id) as db: + await db.execute(text("SET LOCAL statement_timeout = '5000ms'"), {}) + ledger_operation_type = await _resolve_ledger_operation_type(db) + for item in receipt["live_receipt_apply_receipts"]: + rows.append( + await _insert_live_receipt_row( + db, + project_id=project_id, + item=item, + ledger_operation_type=ledger_operation_type, + ) + ) + return ledger_operation_type, rows + + +async def _insert_live_receipt_row( + db: Any, + *, + project_id: str, + item: dict[str, Any], + ledger_operation_type: str, +) -> dict[str, Any]: + result = await db.execute( + text(""" + INSERT INTO automation_operation_log ( + operation_type, actor, status, + input, output, dry_run_result, tags + ) + SELECT + :ledger_operation_type, + :actor, + 'success', + CAST(:input AS jsonb), + CAST(:output AS jsonb), + CAST(:dry_run_result AS jsonb), + :tags + WHERE NOT EXISTS ( + SELECT 1 + FROM automation_operation_log existing + WHERE coalesce( + existing.input ->> 'semantic_operation_type', + existing.operation_type + ) = :operation_type + AND existing.input ->> 'live_receipt_id' = :live_receipt_id + ) + RETURNING op_id::text + """), + { + "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() + 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 = await db.execute( + text(""" + SELECT op_id::text + FROM automation_operation_log + WHERE coalesce(input ->> 'semantic_operation_type', operation_type) + = :operation_type + AND input ->> 'live_receipt_id' = :live_receipt_id + ORDER BY created_at DESC + LIMIT 1 + """), + { + "operation_type": OPERATION_TYPE, + "live_receipt_id": item["live_receipt_id"], + }, + ) + return { + "live_receipt_id": item["live_receipt_id"], + "batch_id": item["batch_id"], + "domain": item["domain"], + "created": False, + "op_id": str(existing.scalar() or ""), + } + + +async def _load_live_receipt_rows_with_retry( + *, + project_id: str, +) -> list[dict[str, Any]]: + last_error: Exception | None = None + for attempt, delay_seconds in enumerate(_DB_RETRY_DELAYS_SECONDS, start=1): + if delay_seconds: + await asyncio.sleep(delay_seconds) + try: + return await _load_live_receipt_rows_once(project_id=project_id) + except Exception as exc: # pragma: no cover - exercised by live pressure + last_error = exc + logger.warning( + "telegram_monitoring_live_receipt_apply_readback_attempt_failed", + project_id=project_id, + attempt=attempt, + error_type=type(exc).__name__, + ) + if last_error is not None: + raise last_error + return [] + + +async def _load_live_receipt_rows_once( + *, + project_id: str, +) -> list[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 ->> '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) + = :operation_type + AND input ->> 'project_id' = :project_id + ORDER BY created_at DESC, op_id DESC + LIMIT 100 + """), + { + "operation_type": OPERATION_TYPE, + "project_id": project_id, + }, + ) + return _result_rows(result) + + +async def _resolve_ledger_operation_type(db: Any) -> str: + result = await db.execute( + text(""" + SELECT CASE + WHEN EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'automation_operation_log_type_valid' + AND pg_get_constraintdef(oid) LIKE '%' || :operation_type || '%' + ) + THEN :operation_type + ELSE :fallback_operation_type + END + """), + { + "operation_type": OPERATION_TYPE, + "fallback_operation_type": FALLBACK_LEDGER_OPERATION_TYPE, + }, + ) + resolved = str(result.scalar() or "") + if resolved == OPERATION_TYPE: + return OPERATION_TYPE + return FALLBACK_LEDGER_OPERATION_TYPE + + +def _live_receipt_apply_receipt(batch: Mapping[str, Any]) -> dict[str, Any]: + batch_id = str(batch.get("batch_id") or "") + domain = str(batch.get("domain") or "monitoring_general") + verifier_ref = str(batch.get("post_verifier") or POST_VERIFIER_ROUTE) + return { + "live_receipt_id": f"{OPERATION_TYPE}::{batch_id}", + "batch_id": batch_id, + "domain": domain, + "priority": str(batch.get("priority") or "P1"), + "surface_count": _int(batch.get("surface_count")), + "write_capable_surface_count": _int(batch.get("write_capable_surface_count")), + "target_selector": _dict(batch.get("target_selector")), + "tag_hints": _dict(batch.get("tag_hints")), + "executor_route": EXECUTOR_ROUTE, + "status": ( + "ready_for_metadata_live_receipt_write" + if _batch_ready_for_apply(batch) + else "blocked_waiting_live_receipt_controls" + ), + "post_apply_verifier_refs": [f"{verifier_ref}#batch={batch_id}"], + "raw_payload_included": False, + "source_of_truth_diff": { + "current_state": "monitoring_live_receipt_gap_present", + "desired_state": "metadata_live_receipt_available_to_ai_loop", + "delta_kind": f"telegram_monitoring_live_receipt::{domain}", + "surface_count": _int(batch.get("surface_count")), + "raw_payload_included": False, + }, + "check_mode": { + "enabled": True, + "checks": [ + "batch_id_present", + "target_selector_surface_ids_present", + "tag_hints_project_product_service_tool_present", + "metadata_only_raw_payload_absent", + "post_apply_verifier_present", + "critical_break_glass_not_required", + ], + }, + "rollback": { + "required": True, + "rollback_ref": ( + "rollback://telegram-monitoring-live-receipt-apply/" + f"{OPERATION_TYPE}::{batch_id}" + ), + "strategy": "mark_metadata_live_receipt_superseded_and_ignore", + }, + } + + +def _batch_ready_for_apply(batch: Mapping[str, Any]) -> bool: + target_selector = _dict(batch.get("target_selector")) + tag_hints = _dict(batch.get("tag_hints")) + if not str(batch.get("batch_id") or ""): + return False + if not str(batch.get("domain") or ""): + return False + if _int(batch.get("surface_count")) <= 0: + return False + if not _strings(target_selector.get("surface_ids")): + return False + for key in ("project_id", "product_id", "service_name", "tool_id"): + if not str(tag_hints.get(key) or ""): + return False + if not str(batch.get("controlled_next_action") or ""): + return False + if not str(batch.get("post_verifier") or ""): + return False + return True + + +def _apply_active_blockers( + *, + coverage: Mapping[str, Any], + batches: list[dict[str, Any]], + apply_receipts: list[dict[str, Any]], +) -> list[str]: + blockers: list[str] = [] + if not batches: + blockers.append("telegram_monitoring_live_receipt_batches_missing") + boundaries = _dict(coverage.get("operation_boundaries")) + unsafe_boundary_keys = ( + "telegram_send_performed", + "bot_api_call_performed", + "gateway_queue_write_performed", + "receiver_route_changed", + "alertmanager_reload_performed", + "raw_alert_payload_stored", + "secret_value_read", + "github_api_used", + ) + for key in unsafe_boundary_keys: + if boundaries.get(key) is True: + blockers.append(f"coverage_boundary_{key}_unexpected") + for item in apply_receipts: + batch_id = item["batch_id"] or "unknown_batch" + if item["status"] != "ready_for_metadata_live_receipt_write": + blockers.append(f"{batch_id}_metadata_live_receipt_not_ready") + if item["raw_payload_included"] is not False: + blockers.append(f"{batch_id}_raw_payload_included") + return _unique(blockers) + + +def _live_receipt_binding( + *, + batch: Mapping[str, Any], + row: Mapping[str, Any] | None, +) -> dict[str, Any]: + batch_id = str(batch.get("batch_id") or "") + domain = str(batch.get("domain") or "monitoring_general") + raw_payload_included = ( + str((row or {}).get("raw_payload_included") or "").lower() == "true" + ) + row_ready = _row_ready(row) + return { + "live_receipt_id": f"{OPERATION_TYPE}::{batch_id}", + "batch_id": batch_id, + "domain": domain, + "priority": str(batch.get("priority") or "P1"), + "surface_count": _int(batch.get("surface_count")), + "write_capable_surface_count": _int(batch.get("write_capable_surface_count")), + "target_selector": _dict(batch.get("target_selector")), + "tag_hints": _dict(batch.get("tag_hints")), + "status": ( + "metadata_live_receipt_applied" + if row_ready + else "waiting_metadata_live_receipt_apply" + ), + "ledger_op_id": str((row or {}).get("op_id") or ""), + "ledger_status": str((row or {}).get("status") or ""), + "semantic_operation_type": str((row or {}).get("semantic_operation_type") or ""), + "ledger_operation_type": str((row or {}).get("ledger_operation_type") or ""), + "executor_route": str((row or {}).get("actor") or EXECUTOR_ROUTE), + "target_write_performed": row_ready, + "runtime_target_write_performed": ( + str((row or {}).get("runtime_target_write_performed") or "").lower() + == "true" + ), + "target_context_receipt_write_performed": ( + str((row or {}).get("target_context_receipt_write_performed") or "").lower() + == "true" + ), + "post_apply_verifier_refs": _strings((row or {}).get("post_apply_verifier_refs")) + or [f"{POST_VERIFIER_ROUTE}#batch={batch_id}"], + "next_action": str((row or {}).get("next_action") or ""), + "raw_payload_included": raw_payload_included, + } + + +def _row_ready(row: Mapping[str, Any] | None) -> bool: + if not row: + return False + if str(row.get("semantic_operation_type") or "") != OPERATION_TYPE: + return False + if str(row.get("actor") or "") != EXECUTOR_ROUTE: + return False + if str(row.get("status") or "") != "success": + return False + if not str(row.get("live_receipt_id") or ""): + return False + if not str(row.get("batch_id") or ""): + return False + if str(row.get("raw_payload_included") or "").lower() == "true": + return False + return bool(_strings(row.get("post_apply_verifier_refs"))) + + +def _readback_active_blockers( + *, + batches: list[dict[str, Any]], + bindings: list[dict[str, Any]], +) -> list[str]: + blockers: list[str] = [] + if not batches: + blockers.append("telegram_monitoring_live_receipt_batches_missing") + for binding in bindings: + batch_id = binding["batch_id"] or "unknown_batch" + if binding["status"] != "metadata_live_receipt_applied": + blockers.append(f"{batch_id}_metadata_live_receipt_missing") + if binding["raw_payload_included"] is not False: + blockers.append(f"{batch_id}_raw_payload_included") + return _unique(blockers) + + +def _fallback_apply_readback( + *, + coverage: Mapping[str, Any], + project_id: str, + error_type: str, +) -> dict[str, Any]: + batches = _coverage_batches(coverage) + coverage_summary = _dict(coverage.get("summary")) + return { + "schema_version": READBACK_SCHEMA_VERSION, + "priority": "P0-TELEGRAM-MONITORING-AI-AUTOMATION-COVERAGE", + "scope": "telegram_alert_monitoring_live_receipt_apply_readback", + "project_id": project_id, + "status": "blocked_waiting_telegram_monitoring_live_receipt_apply_db_readback", + "readback": { + "workplan_id": "CIR-P0-TG-001-LIVE-RECEIPT-APPLY-READBACK", + "source_schema_version": coverage.get("schema_version"), + "source_status": coverage.get("status"), + "db_read_status": "unavailable", + "error_type": error_type, + "operation_type": OPERATION_TYPE, + "executor_route": EXECUTOR_ROUTE, + "apply_route": APPLY_ROUTE, + "safe_next_step": "retry_live_receipt_apply_readback_after_db_pool_pressure_clears", + }, + "operator_answer": { + "all_ai_controlled_live_receipt_batches_have_apply_receipt": False, + "metadata_live_receipt_apply_performed": False, + "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") + ), + "manual_default_terminal_state_allowed": False, + "critical_break_glass_exception_only": True, + }, + "live_receipt_bindings": [], + "rollups": { + "expected_live_receipt_batch_count": len(batches), + "applied_live_receipt_batch_count": 0, + "ready_live_receipt_binding_count": 0, + "expected_surface_count": sum(_int(batch.get("surface_count")) for batch in batches), + "applied_surface_count": 0, + "expected_write_capable_surface_count": sum( + _int(batch.get("write_capable_surface_count")) for batch in batches + ), + "applied_write_capable_surface_count": 0, + "metadata_only_receipt_count": 0, + "post_apply_verifier_ref_count": 0, + "monitoring_live_receipt_gap_count_source_truth": _int( + coverage_summary.get("monitoring_live_receipt_gap_count") + ), + "metadata_live_receipt_apply_readback_ready": False, + "runtime_target_write_performed": False, + }, + "active_blockers": [ + "telegram_monitoring_live_receipt_apply_db_readback_unavailable" + ], + "operation_boundaries": { + "readback_only": True, + "metadata_ledger_read_performed": False, + "metadata_live_receipt_apply_read_performed": False, + "runtime_target_write_performed": False, + "monitoring_inventory_source_mutated": False, + "monitoring_live_receipt_acceptance_mutated": False, + "km_write_performed": False, + "rag_index_write_performed": False, + "playbook_trust_write_performed": False, + "mcp_tool_call_performed": False, + "agent_runtime_action_performed": False, + "telegram_send_performed": False, + "bot_api_call_performed": False, + "alertmanager_reload_performed": False, + "workflow_trigger_performed": False, + "raw_alert_payload_stored": False, + "raw_payload_included": False, + "secret_value_read": False, + "github_api_used": False, + }, + } + + +def _coverage_batches(coverage: Mapping[str, Any]) -> list[dict[str, Any]]: + batches = coverage.get("ai_controlled_live_receipt_batches") + if not isinstance(batches, list): + return [] + return [dict(batch) for batch in batches if isinstance(batch, dict)] + + +def _rows_by_batch_id(rows: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: + mapped: dict[str, dict[str, Any]] = {} + for row in rows: + batch_id = str(row.get("batch_id") or "") + if not batch_id or batch_id in mapped: + continue + mapped[batch_id] = row + return mapped + + +def _receipt_tags(item: Mapping[str, Any]) -> list[str]: + tag_hints = _dict(item.get("tag_hints")) + tags = [ + "ai_agent", + "telegram_monitoring", + "controlled_live_receipt", + str(item.get("domain") or "monitoring_general"), + ] + for key in ("product_id", "service_name", "tool_id"): + value = str(tag_hints.get(key) or "") + if value: + tags.append(value) + return _unique(tags) + + +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 _dict(value: Any) -> dict[str, Any]: + return dict(value) if isinstance(value, Mapping) else {} + + +def _strings(value: Any) -> list[str]: + if isinstance(value, list): + return [str(item) for item in value] + if isinstance(value, tuple): + return [str(item) for item in value] + return [] + + +def _int(value: Any) -> int: + try: + return int(value or 0) + except (TypeError, ValueError): + return 0 + + +def _unique(values: list[str]) -> list[str]: + seen = set() + result = [] + for value in values: + if value in seen: + continue + seen.add(value) + result.append(value) + return result 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 new file mode 100644 index 000000000..b87629d66 --- /dev/null +++ b/apps/api/tests/test_telegram_alert_monitoring_live_receipt_apply_api.py @@ -0,0 +1,419 @@ +from __future__ import annotations + +import json + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from src.api.v1 import agents +from src.api.v1.agents import router +from src.services import telegram_alert_monitoring_live_receipt_apply as apply_module +from src.services.telegram_alert_monitoring_live_receipt_apply import ( + APPLY_SCHEMA_VERSION, + OPERATION_TYPE, + READBACK_SCHEMA_VERSION, + apply_latest_telegram_alert_monitoring_live_receipts, + build_telegram_alert_monitoring_live_receipt_apply_readback, + build_telegram_alert_monitoring_live_receipt_apply_receipt, +) + + +class _FakeScalarResult: + def __init__(self, value: str | None): + self.value = value + + def scalar(self) -> str | None: + return self.value + + +class _FakeMappings: + def __init__(self, rows: list[dict]): + self.rows = rows + + def all(self) -> list[dict]: + return self.rows + + +class _FakeRowsResult: + def __init__(self, rows: list[dict]): + self.rows = rows + + def mappings(self) -> _FakeMappings: + return _FakeMappings(self.rows) + + +class _FakeDb: + def __init__(self, *, rows: list[dict] | None = None): + self.params: list[dict] = [] + self.rows = rows or [] + + async def execute(self, statement, params: dict): + if not params: + return _FakeScalarResult(None) + if "fallback_operation_type" in params: + return _FakeScalarResult("km_linked") + if "input" in params: + self.params.append(params) + live_receipt_id = str(params["live_receipt_id"]).rsplit("::", 1)[-1] + return _FakeScalarResult(f"live-receipt-op-{live_receipt_id}") + if params.get("operation_type") == OPERATION_TYPE and "project_id" in params: + return _FakeRowsResult(self.rows) + return _FakeScalarResult("existing-live-receipt-op") + + +class _FakeContext: + def __init__(self, db: _FakeDb): + self.db = db + + async def __aenter__(self) -> _FakeDb: + return self.db + + async def __aexit__(self, _exc_type, _exc, _tb) -> bool: + return False + + +class _FailingContext: + async def __aenter__(self): + raise TimeoutError("pool checkout unavailable") + + async def __aexit__(self, _exc_type, _exc, _tb) -> bool: + return False + + +def _coverage_payload() -> dict: + batches = [ + { + "batch_id": "tg_live_receipt_01_prometheus", + "domain": "prometheus", + "priority": "P0", + "surface_count": 20, + "write_capable_surface_count": 1, + "missing_receipt_fields": ["receiver_receipt_readback_ref"], + "tag_hints": { + "project_id": "awoooi", + "product_id": "awoooi-observability", + "website_id": "awoooi.wooo.work", + "service_name": "prometheus", + "package_name": "prometheus-config", + "tool_id": "prometheus", + "log_source": "monitoring_live_receipt_metadata", + "alertname": "monitoring_surface_live_receipt_gap", + "playbook_id": "playbook://awoooi/monitoring-live-receipt/prometheus", + "rag_context_ref": "rag://awoooi/monitoring/prometheus", + "mcp_evidence_ref": "mcp://awoooi/monitoring/prometheus", + "schedule_id": "schedule://awoooi/monitoring-live-receipt/prometheus", + }, + "target_selector": { + "surface_ids": ["prometheus_k8s_base_config"], + "config_kinds": ["prometheus_config"], + "control_tiers": ["C1"], + "expected_scope_prefixes": ["k8s"], + "repo_source_paths": ["k8s/monitoring/prometheus.yml"], + }, + "controlled_next_action": ( + "ingest_metadata_only_live_receipt_batch_then_verify_gap_delta" + ), + "post_verifier": ( + "/api/v1/agents/telegram-alert-monitoring-coverage-readback" + ), + }, + { + "batch_id": "tg_live_receipt_02_telegram", + "domain": "telegram", + "priority": "P0", + "surface_count": 3, + "write_capable_surface_count": 0, + "missing_receipt_fields": ["notification_delivery_metadata_ref"], + "tag_hints": { + "project_id": "awoooi", + "product_id": "awooop", + "website_id": "awoooi.wooo.work", + "service_name": "telegram-gateway", + "package_name": "telegram-alert-delivery", + "tool_id": "telegram", + "log_source": "monitoring_live_receipt_metadata", + "alertname": "monitoring_surface_live_receipt_gap", + "playbook_id": "playbook://awoooi/monitoring-live-receipt/telegram", + "rag_context_ref": "rag://awoooi/monitoring/telegram", + "mcp_evidence_ref": "mcp://awoooi/monitoring/telegram", + "schedule_id": "schedule://awoooi/monitoring-live-receipt/telegram", + }, + "target_selector": { + "surface_ids": ["telegram_gateway_config"], + "config_kinds": ["telegram_gateway"], + "control_tiers": ["C1"], + "expected_scope_prefixes": ["telegram"], + "repo_source_paths": ["apps/api/src/services/telegram_gateway.py"], + }, + "controlled_next_action": ( + "ingest_metadata_only_live_receipt_batch_then_verify_gap_delta" + ), + "post_verifier": ( + "/api/v1/agents/telegram-alert-monitoring-coverage-readback" + ), + }, + ] + return { + "schema_version": "telegram_alert_monitoring_coverage_readback_v1", + "project_id": "awoooi", + "status": "blocked_telegram_alert_monitoring_coverage_gaps_present", + "summary": { + "monitoring_surface_count": 60, + "monitoring_live_receipt_gap_count": 60, + "ai_controlled_live_receipt_batch_count": len(batches), + "source_contract_ready": True, + }, + "ai_controlled_live_receipt_batches": batches, + "active_blockers": ["monitoring_live_receipt_gap:60"], + "operation_boundaries": { + "telegram_send_performed": False, + "bot_api_call_performed": False, + "gateway_queue_write_performed": False, + "receiver_route_changed": False, + "alertmanager_reload_performed": False, + "raw_alert_payload_stored": False, + "secret_value_read": False, + "github_api_used": False, + }, + } + + +def _row_for(batch: dict, *, op_id: str) -> dict: + return { + "op_id": op_id, + "operation_type": "km_linked", + "actor": "ai_agent_monitoring_live_receipt_consumer", + "status": "success", + "semantic_operation_type": OPERATION_TYPE, + "ledger_operation_type": "km_linked", + "live_receipt_id": f"{OPERATION_TYPE}::{batch['batch_id']}", + "batch_id": batch["batch_id"], + "domain": batch["domain"], + "priority": batch["priority"], + "surface_count": str(batch["surface_count"]), + "write_capable_surface_count": str(batch["write_capable_surface_count"]), + "runtime_target_write_performed": "true", + "raw_payload_included": "false", + "target_selector": batch["target_selector"], + "tag_hints": batch["tag_hints"], + "monitoring_live_receipt_metadata_recorded": "true", + "target_context_receipt_write_performed": "true", + "next_action": "readback_telegram_monitoring_live_receipt_apply_then_verify_gap_delta", + "post_apply_verifier_refs": [ + "/api/v1/agents/telegram-alert-monitoring-coverage-readback" + f"#batch={batch['batch_id']}" + ], + } + + +def test_telegram_monitoring_live_receipt_apply_builder_creates_receipts(): + payload = build_telegram_alert_monitoring_live_receipt_apply_receipt( + _coverage_payload() + ) + + assert payload["schema_version"] == APPLY_SCHEMA_VERSION + assert payload["status"] == "telegram_monitoring_live_receipt_apply_ready" + assert payload["active_blockers"] == [] + assert payload["readback"]["operation_type"] == OPERATION_TYPE + assert payload["rollups"]["source_live_receipt_batch_count"] == 2 + assert payload["rollups"]["source_surface_count"] == 23 + assert payload["rollups"]["metadata_only_receipt_count"] == 2 + + receipts = payload["live_receipt_apply_receipts"] + assert {item["domain"] for item in receipts} == {"prometheus", "telegram"} + for item in receipts: + assert item["status"] == "ready_for_metadata_live_receipt_write" + assert item["raw_payload_included"] is False + assert item["check_mode"]["enabled"] is True + assert item["rollback"]["required"] is True + assert item["target_selector"]["surface_ids"] + assert item["tag_hints"]["project_id"] == "awoooi" + + boundaries = payload["operation_boundaries"] + assert boundaries["metadata_ledger_write_only"] is True + assert boundaries["telegram_send_performed"] is False + assert boundaries["bot_api_call_performed"] is False + assert boundaries["raw_alert_payload_stored"] is False + assert boundaries["secret_value_read"] is False + + serialized = json.dumps(payload, ensure_ascii=False) + for marker in ["TELEGRAM_BOT_TOKEN", "authorization_header", "raw Telegram payload"]: + assert marker not in serialized + + +@pytest.mark.asyncio +async def test_telegram_monitoring_live_receipt_apply_writes_idempotent_rows( + monkeypatch, +): + fake_db = _FakeDb() + monkeypatch.setattr( + apply_module, + "get_db_context", + lambda project_id: _FakeContext(fake_db), + ) + + async def fake_coverage(*, project_id: str): + assert project_id == "awoooi" + return _coverage_payload() + + monkeypatch.setattr( + apply_module, + "load_latest_telegram_alert_monitoring_coverage_readback", + fake_coverage, + ) + + payload = await apply_latest_telegram_alert_monitoring_live_receipts( + project_id="awoooi" + ) + + assert payload["status"] == "telegram_monitoring_live_receipt_apply_ready" + assert payload["rollups"]["runtime_target_write_performed"] is True + assert payload["rollups"]["live_receipt_apply_row_count"] == 2 + assert payload["rollups"]["inserted_live_receipt_apply_row_count"] == 2 + 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" + + write_params = [params for params in fake_db.params if "input" in params] + assert len(write_params) == 2 + for params in write_params: + assert params["operation_type"] == OPERATION_TYPE + assert params["ledger_operation_type"] == "km_linked" + assert params["actor"] == "ai_agent_monitoring_live_receipt_consumer" + assert '"semantic_operation_type": "telegram_monitoring_live_receipt_applied"' in params["input"] + assert '"runtime_target_write_performed": true' in params["input"] + assert '"raw_payload_included": false' in params["input"] + assert '"monitoring_live_receipt_metadata_recorded": true' in params["output"] + assert '"telegram_send_performed": false' in params["output"] + + +def test_telegram_monitoring_live_receipt_apply_readback_preserves_gap_truth(): + coverage = _coverage_payload() + rows = [ + _row_for(batch, op_id=f"live-op-{index}") + for index, batch in enumerate(coverage["ai_controlled_live_receipt_batches"], start=1) + ] + + payload = build_telegram_alert_monitoring_live_receipt_apply_readback( + coverage=coverage, + rows=rows, + ) + + assert payload["schema_version"] == READBACK_SCHEMA_VERSION + 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["operator_answer"][ + "metadata_apply_does_not_mark_monitoring_live_receipt_accepted" + ] is True + assert payload["operator_answer"]["monitoring_live_receipt_gap_count_source_truth"] == 60 + assert payload["rollups"]["expected_live_receipt_batch_count"] == 2 + assert payload["rollups"]["applied_live_receipt_batch_count"] == 2 + assert payload["rollups"]["expected_surface_count"] == 23 + assert payload["rollups"]["applied_surface_count"] == 23 + assert payload["rollups"]["monitoring_live_receipt_gap_count_source_truth"] == 60 + assert payload["operation_boundaries"]["monitoring_live_receipt_acceptance_mutated"] is False + assert payload["operation_boundaries"]["telegram_send_performed"] is False + + +@pytest.mark.asyncio +async def test_telegram_monitoring_live_receipt_apply_degrades_on_db_write_pressure( + monkeypatch, +): + monkeypatch.setattr( + apply_module, + "get_db_context", + lambda project_id: _FailingContext(), + ) + + async def fake_coverage(*, project_id: str): + assert project_id == "awoooi" + return _coverage_payload() + + monkeypatch.setattr( + apply_module, + "load_latest_telegram_alert_monitoring_coverage_readback", + fake_coverage, + ) + + payload = await apply_latest_telegram_alert_monitoring_live_receipts( + project_id="awoooi" + ) + + assert payload["status"] == "blocked_waiting_telegram_monitoring_live_receipt_apply_db" + assert "telegram_monitoring_live_receipt_apply_db_write_unavailable" in payload[ + "active_blockers" + ] + assert payload["apply_result"]["db_write_status"] == "unavailable" + assert payload["operation_boundaries"]["runtime_target_write_performed"] is False + + +def test_telegram_monitoring_live_receipt_apply_endpoint_returns_receipt(monkeypatch): + async def fake_apply(): + payload = build_telegram_alert_monitoring_live_receipt_apply_receipt( + _coverage_payload() + ) + payload["rollups"]["runtime_target_write_performed"] = True + payload["operation_boundaries"]["runtime_target_write_performed"] = True + return payload + + monkeypatch.setattr( + agents, + "apply_latest_telegram_alert_monitoring_live_receipts", + fake_apply, + ) + app = FastAPI() + app.include_router(router, prefix="/api/v1") + client = TestClient(app) + + response = client.post( + "/api/v1/agents/telegram-alert-monitoring-live-receipt-apply" + ) + + assert response.status_code == 200 + payload = response.json() + assert payload["schema_version"] == APPLY_SCHEMA_VERSION + assert payload["status"] == "telegram_monitoring_live_receipt_apply_ready" + assert payload["rollups"]["runtime_target_write_performed"] is True + + +def test_telegram_monitoring_live_receipt_apply_readback_endpoint_returns_receipt( + monkeypatch, +): + async def fake_readback(): + coverage = _coverage_payload() + rows = [ + _row_for(batch, op_id=f"live-op-{index}") + for index, batch in enumerate( + coverage["ai_controlled_live_receipt_batches"], + start=1, + ) + ] + return build_telegram_alert_monitoring_live_receipt_apply_readback( + coverage=coverage, + rows=rows, + ) + + monkeypatch.setattr( + agents, + "load_latest_telegram_alert_monitoring_live_receipt_apply_readback", + fake_readback, + ) + app = FastAPI() + app.include_router(router, prefix="/api/v1") + client = TestClient(app) + + response = client.get( + "/api/v1/agents/telegram-alert-monitoring-live-receipt-apply-readback" + ) + + assert response.status_code == 200 + payload = response.json() + assert payload["schema_version"] == READBACK_SCHEMA_VERSION + assert payload["status"] == "telegram_monitoring_live_receipt_apply_readback_ready" + assert payload["rollups"]["applied_live_receipt_batch_count"] == 2 + assert payload["operator_answer"]["monitoring_live_receipt_gap_count_source_truth"] == 60 diff --git a/apps/web/messages/en.json b/apps/web/messages/en.json index 21b3ce918..a944bcf05 100644 --- a/apps/web/messages/en.json +++ b/apps/web/messages/en.json @@ -11972,6 +11972,8 @@ "directDetail": "gateway controlled: {ready}", "monitoring": "Live receipts", "monitoringDetail": "gap {gap}", + "apply": "Metadata apply", + "applyDetail": "surfaces {surfaces}; source gap {gap}", "delivery": "AI alert delivery", "deliveryDetail": "learning refs ready / delivery receipts", "verifier": "AI context", @@ -11980,6 +11982,7 @@ "logsDetail": "TG {telegram} / KM {km} / PB {playbook}", "blockers": "Active blockers", "queue": "AI controlled queue", + "applyReadback": "Metadata apply readback", "pipeline": "Receipt pipeline", "surfaces": "{count} surfaces", "noBlockers": "No active blocker" diff --git a/apps/web/messages/zh-TW.json b/apps/web/messages/zh-TW.json index ef9bdecdf..1e2b0cf6f 100644 --- a/apps/web/messages/zh-TW.json +++ b/apps/web/messages/zh-TW.json @@ -11972,6 +11972,8 @@ "directDetail": "gateway controlled：{ready}", "monitoring": "Live receipts", "monitoringDetail": "gap {gap}", + "apply": "Metadata apply", + "applyDetail": "surfaces {surfaces}；source gap {gap}", "delivery": "AI alert delivery", "deliveryDetail": "learning refs ready / delivery receipts", "verifier": "AI context", @@ -11980,6 +11982,7 @@ "logsDetail": "TG {telegram} / KM {km} / PB {playbook}", "blockers": "Active blockers", "queue": "AI controlled queue", + "applyReadback": "Metadata apply readback", "pipeline": "收據流水線", "surfaces": "{count} surfaces", "noBlockers": "無 active blocker" diff --git a/apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx b/apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx index a62d5af0c..3de6b9b24 100644 --- a/apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx +++ b/apps/web/src/components/awooop/autonomous-runtime-receipt-panel.tsx @@ -391,6 +391,55 @@ type TelegramMonitoringCoveragePayload = { } | null; }; +type TelegramMonitoringLiveReceiptApplyReadbackPayload = { + status?: string | null; + active_blockers?: string[] | null; + operator_answer?: { + all_ai_controlled_live_receipt_batches_have_apply_receipt?: boolean | null; + metadata_live_receipt_apply_performed?: boolean | null; + metadata_apply_does_not_mark_monitoring_live_receipt_accepted?: boolean | null; + monitoring_live_receipt_gap_count_source_truth?: number | null; + } | null; + rollups?: { + expected_live_receipt_batch_count?: number | null; + applied_live_receipt_batch_count?: number | null; + ready_live_receipt_binding_count?: number | null; + expected_surface_count?: number | null; + applied_surface_count?: number | null; + expected_write_capable_surface_count?: number | null; + applied_write_capable_surface_count?: number | null; + metadata_only_receipt_count?: number | null; + post_apply_verifier_ref_count?: number | null; + monitoring_live_receipt_gap_count_source_truth?: number | null; + metadata_live_receipt_apply_readback_ready?: boolean | null; + runtime_target_write_performed?: boolean | null; + } | null; + live_receipt_bindings?: Array<{ + batch_id?: string | null; + domain?: string | null; + priority?: string | null; + status?: string | null; + surface_count?: number | null; + write_capable_surface_count?: number | null; + target_write_performed?: boolean | null; + tag_hints?: { + product_id?: string | null; + service_name?: string | null; + tool_id?: string | null; + } | null; + }> | null; + operation_boundaries?: { + runtime_target_write_performed?: boolean | null; + monitoring_inventory_source_mutated?: boolean | null; + monitoring_live_receipt_acceptance_mutated?: boolean | null; + telegram_send_performed?: boolean | null; + bot_api_call_performed?: boolean | null; + raw_alert_payload_stored?: boolean | null; + secret_value_read?: boolean | null; + github_api_used?: boolean | null; + } | null; +}; + type PanelMode = "full" | "compact"; type Tone = "ok" | "warn" | "neutral"; type WorkFilter = "all" | "completed" | "active" | "pending" | "blocked"; @@ -549,6 +598,23 @@ async function fetchTelegramMonitoringCoverage(): Promise { + const controller = new AbortController(); + const timeout = window.setTimeout(() => controller.abort(), 12_000); + try { + const response = await fetch(`${API_BASE}/api/v1/agents/telegram-alert-monitoring-live-receipt-apply-readback`, { + cache: "no-store", + signal: controller.signal, + }); + if (!response.ok) return null; + return (await response.json()) as TelegramMonitoringLiveReceiptApplyReadbackPayload; + } catch { + return null; + } finally { + window.clearTimeout(timeout); + } +} + export function AutonomousRuntimeReceiptPanel({ mode = "full", }: { @@ -561,6 +627,8 @@ export function AutonomousRuntimeReceiptPanel({ const [consumerPayload, setConsumerPayload] = useState(null); const [telegramVerifierPayload, setTelegramVerifierPayload] = useState(null); const [telegramCoveragePayload, setTelegramCoveragePayload] = useState(null); + const [telegramLiveReceiptApplyPayload, setTelegramLiveReceiptApplyPayload] = + useState(null); const [loading, setLoading] = useState(true); const [error, setError] = useState(false); const [updatedAt, setUpdatedAt] = useState(null); @@ -568,18 +636,27 @@ export function AutonomousRuntimeReceiptPanel({ const refresh = useCallback(async () => { setLoading(true); - const [next, priority, consumer, telegramVerifier, telegramCoverage] = await Promise.all([ + const [ + next, + priority, + consumer, + telegramVerifier, + telegramCoverage, + telegramLiveReceiptApply, + ] = await Promise.all([ fetchRuntimeControl(), fetchPriorityWorkOrder(), fetchLogConsumerReadback(), fetchTelegramPostApplyVerifier(), fetchTelegramMonitoringCoverage(), + fetchTelegramMonitoringLiveReceiptApplyReadback(), ]); setPayload(next); setPriorityPayload(priority); setConsumerPayload(consumer); setTelegramVerifierPayload(telegramVerifier); setTelegramCoveragePayload(telegramCoverage); + setTelegramLiveReceiptApplyPayload(telegramLiveReceiptApply); setError(next === null); setUpdatedAt(next ? new Date() : null); setLoading(false); @@ -596,6 +673,7 @@ export function AutonomousRuntimeReceiptPanel({ const hasConsumerReadback = consumerPayload !== null; const hasTelegramVerifierReadback = telegramVerifierPayload !== null; const hasTelegramCoverageReadback = telegramCoveragePayload !== null; + const hasTelegramLiveReceiptApplyReadback = telegramLiveReceiptApplyPayload !== null; const readback = payload?.runtime_receipt_readback; const ledger = readback?.autonomous_execution_loop_ledger; const traceLedger = readback?.trace_ledger; @@ -626,6 +704,9 @@ export function AutonomousRuntimeReceiptPanel({ const telegramCoverageTags = telegramCoveragePayload?.required_tag_dimensions ?? []; const telegramCoverageAssetKinds = telegramCoveragePayload?.monitoring_asset_rollups?.by_config_kind ?? {}; const telegramCoverageAssetKindNames = Object.keys(telegramCoverageAssetKinds).slice(0, 5); + const telegramLiveReceiptApplyRollups = telegramLiveReceiptApplyPayload?.rollups ?? {}; + const telegramLiveReceiptApplyBindings = telegramLiveReceiptApplyPayload?.live_receipt_bindings ?? []; + const telegramLiveReceiptApplyBlockers = telegramLiveReceiptApplyPayload?.active_blockers ?? []; const consumerReady = consumerRollups.controlled_consumer_readback_ready === true || consumerPayload?.status === "controlled_writeback_consumer_readback_ready"; const telegramPostApplyVerifierReady = @@ -634,6 +715,9 @@ export function AutonomousRuntimeReceiptPanel({ const telegramMonitoringCoverageReady = telegramCoverageSummary.full_coverage_ready === true || telegramCoveragePayload?.status === "telegram_alert_monitoring_coverage_ready"; + const telegramLiveReceiptApplyReady = + telegramLiveReceiptApplyRollups.metadata_live_receipt_apply_readback_ready === true + || telegramLiveReceiptApplyPayload?.status === "telegram_monitoring_live_receipt_apply_readback_ready"; const runtimeTargetWritePerformed = consumerRollups.runtime_target_write_performed === true || consumerPayload?.controlled_consume?.runtime_target_write_performed === true || consumerPayload?.operation_boundaries?.runtime_target_write_performed === true @@ -1300,6 +1384,30 @@ export function AutonomousRuntimeReceiptPanel({ ? "ok" as Tone : "warn" as Tone, }, + { + key: "apply", + label: t("monitoringCoverage.apply"), + value: readbackRatio( + telegramLiveReceiptApplyRollups.applied_live_receipt_batch_count, + telegramLiveReceiptApplyRollups.expected_live_receipt_batch_count, + hasTelegramLiveReceiptApplyReadback + ), + detail: t("monitoringCoverage.applyDetail", { + surfaces: readbackRatio( + telegramLiveReceiptApplyRollups.applied_surface_count, + telegramLiveReceiptApplyRollups.expected_surface_count, + hasTelegramLiveReceiptApplyReadback + ), + gap: readbackNumber( + telegramLiveReceiptApplyRollups.monitoring_live_receipt_gap_count_source_truth, + hasTelegramLiveReceiptApplyReadback + ), + }), + icon: CheckCircle2, + tone: !hasTelegramLiveReceiptApplyReadback + ? "neutral" as Tone + : telegramLiveReceiptApplyReady ? "ok" as Tone : "warn" as Tone, + }, { key: "delivery", label: t("monitoringCoverage.delivery"), @@ -1627,7 +1735,7 @@ export function AutonomousRuntimeReceiptPanel({ })} {mode === "full" ? ( -
+

@@ -1660,6 +1768,64 @@ export function AutonomousRuntimeReceiptPanel({ ))}

+
+
+

+ {t("monitoringCoverage.applyReadback")} +

+ + {readbackRatio( + telegramLiveReceiptApplyRollups.applied_live_receipt_batch_count, + telegramLiveReceiptApplyRollups.expected_live_receipt_batch_count, + hasTelegramLiveReceiptApplyReadback + )} + +
+
+ {telegramLiveReceiptApplyBindings.length > 0 ? ( + telegramLiveReceiptApplyBindings.slice(0, 4).map((item) => ( +
+
+ + {item.domain ?? item.batch_id ?? "--"} + + + {item.target_write_performed ? t("proof.ok") : t("states.open")} + +
+
+ {item.batch_id ?? item.status ?? "--"} + {typeof item.surface_count === "number" ? ( + + {t("monitoringCoverage.surfaces", { count: item.surface_count })} + + ) : null} +
+ + {[ + item.tag_hints?.product_id, + item.tag_hints?.service_name, + item.tag_hints?.tool_id, + ].filter(Boolean).join(" / ") || item.status || "--"} + +
+ )) + ) : ( +
0 ? "warn" : "neutral") + )}> + {telegramLiveReceiptApplyBlockers[0] ?? t("states.loading")} +
+ )} +
+