wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(awooop): surface telegram alert automation coverage
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 58s
Details
CD Pipeline / build-and-deploy (push) Successful in 5m42s
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Your Name
2026-07-03 09:31:05 +08:00
parent e28ebd5b3e
commit cc5868b657
6 changed files with 751 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

402
apps/api/src/services/telegram_alert_monitoring_coverage_readback.py
View File

@@ -46,6 +46,28 @@ _ALERT_LOG_EVENT_TYPES = (
"EXECUTION_STARTED",
"EXECUTION_COMPLETED",
)
_REQUIRED_TAG_DIMENSIONS = (
(
"project",
"project_id",
"bind every alert/log receipt to the project source of truth",
),
(
"product",
"product_id",
"separate AWOOOI/AwoooP/IwoooS and adjacent product automation lanes",
),
("website", "website_id", "cluster public website and route evidence"),
("service", "service_name", "cluster runtime service, pod, job, or daemon signals"),
("package", "package_name", "cluster package, dependency, and version-radar signals"),
("tool", "tool_id", "cluster MCP, script, exporter, and connector evidence"),
("log", "log_source", "preserve metadata-only log source identity"),
("alert", "alertname", "bind alert fingerprints, severity, and recurrence state"),
("playbook", "playbook_id", "route candidates into controlled apply playbooks"),
("rag", "rag_context_ref", "make receipts retrievable by RAG without raw payloads"),
("mcp", "mcp_evidence_ref", "bind tool evidence without bypassing the MCP registry"),
("schedule", "schedule_id", "cluster cron, worker, and maintenance-window signals"),
)
logger = get_logger(__name__)
@@ -137,6 +159,12 @@ def build_telegram_alert_monitoring_coverage_readback(
ai_alert_ready_total = _int(
ai_alert_summary.get("learning_writeback_ready_total")
)
runtime_event_type_counts = {
str(key): _int(value)
for key, value in _dict(
runtime_log_readback.get("event_type_counts_7d")
).items()
}
coverage_matrix = _coverage_matrix(
project_id=project_id,
@@ -165,6 +193,38 @@ def build_telegram_alert_monitoring_coverage_readback(
ai_alert_ready_total=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,
monitoring_live_gap_count=monitoring_live_gap_count,
direct_gap_count=direct_gap_count,
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,
source_ready=source_ready,
)
work_item_progress = _work_item_progress(
monitoring_live_gap_count=monitoring_live_gap_count,
direct_gap_count=direct_gap_count,
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,
source_ready=source_ready,
)
ai_controlled_gap_queue = _ai_controlled_gap_queue(
active_blockers=active_blockers,
monitoring_asset_rollups=monitoring_asset_rollups,
)
return {
"schema_version": SCHEMA_VERSION,
@@ -178,6 +238,14 @@ def build_telegram_alert_monitoring_coverage_readback(
),
"mode": "metadata_only_db_log_ai_context_readback",
"operator_answer": {
"all_telegram_monitoring_alerts_fully_audited": ready,
"all_telegram_monitoring_alert_surfaces_have_db_or_log_receipt": (
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
),
"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
@@ -248,8 +316,27 @@ def build_telegram_alert_monitoring_coverage_readback(
"source_contract_ready": source_ready,
"runtime_db_readback_ok": runtime_db_ok,
"ai_alert_card_db_readback_ok": ai_alert_db_ok,
"runtime_event_type_with_receipts_count": len(runtime_event_type_counts),
"alert_receipt_pipeline_stage_count": len(alert_receipt_pipeline),
"alert_receipt_pipeline_ready_count": sum(
1 for stage in alert_receipt_pipeline if stage["ready"]
),
"ai_controlled_gap_queue_count": len(ai_controlled_gap_queue),
"required_tag_dimension_count": len(_REQUIRED_TAG_DIMENSIONS),
"monitoring_asset_config_kind_count": len(
monitoring_asset_rollups["by_config_kind"]
),
"telegram_monitoring_audit_completion_percent": work_item_progress[
"completion_percent"
],
"full_coverage_ready": ready,
},
"required_tag_dimensions": _required_tag_dimensions(),
"monitoring_asset_rollups": monitoring_asset_rollups,
"runtime_event_type_counts_7d": runtime_event_type_counts,
"alert_receipt_pipeline": alert_receipt_pipeline,
"work_item_progress": work_item_progress,
"ai_controlled_gap_queue": ai_controlled_gap_queue,
"coverage_matrix": coverage_matrix,
"active_blockers": active_blockers,
"next_controlled_actions": _next_controlled_actions(active_blockers),
@@ -435,6 +522,240 @@ async def _load_runtime_log_readback_once(*, project_id: str) -> dict[str, Any]:
}
def _required_tag_dimensions() -> list[dict[str, str]]:
return [
{
"dimension": dimension,
"canonical_field": canonical_field,
"why": why,
}
for dimension, canonical_field, why in _REQUIRED_TAG_DIMENSIONS
]
def _monitoring_asset_rollups(
monitoring_inventory: Mapping[str, Any],
monitoring_readback_plan: Mapping[str, Any],
) -> 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
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"))
)
live_receipt_gap_samples = [
_monitoring_gap_sample(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"))
][:12]
return {
"surface_count": len(surfaces) or len(inventory_source),
"readback_candidate_count": len(readback_candidates),
"accepted_live_receipt_count": accepted_live_receipts,
"live_receipt_gap_count": max(
len(inventory_source) - accepted_live_receipts,
0,
),
"write_capable_surface_count": sum(
1 for item in inventory_source if bool(item.get("write_capable_surface"))
),
"requires_live_evidence_count": sum(
1 for item in inventory_source if bool(item.get("requires_live_evidence"))
),
"by_config_kind": _count_by_key(inventory_source, "config_kind"),
"by_control_tier": _count_by_key(inventory_source, "control_tier"),
"by_status": _count_by_key(inventory_source, "status"),
"by_reviewer_outcome": _count_by_key(inventory_source, "reviewer_outcome"),
"by_expected_scope_prefix": _count_by_expected_scope_prefix(inventory_source),
"live_receipt_gap_samples": live_receipt_gap_samples,
}
def _monitoring_gap_sample(item: Mapping[str, Any]) -> dict[str, Any]:
return {
"surface_id": str(item.get("surface_id") or ""),
"label": str(item.get("label") or ""),
"config_kind": str(item.get("config_kind") or "unknown"),
"control_tier": str(item.get("control_tier") or "unknown"),
"expected_scope": str(item.get("expected_scope") or ""),
"repo_source_path": str(
item.get("repo_source_path") or item.get("source_path") or ""
),
"status": str(item.get("status") or "waiting_live_receipt"),
"reviewer_outcome": str(
item.get("reviewer_outcome") or "waiting_post_incident_readback"
),
"requires_live_evidence": bool(item.get("requires_live_evidence")),
"write_capable_surface": bool(item.get("write_capable_surface")),
"missing_receipt_fields": [
key
for key in (
"monitoring_incident_or_change_ref",
"receiver_receipt_readback_ref",
"notification_delivery_metadata_ref",
"alert_chain_health_readback_ref",
"postcheck_readback_ref",
"recurrence_guard_ref",
)
if not item.get(key)
],
}
def _alert_receipt_pipeline(
*,
monitoring_surface_count: int,
live_evidence_received_count: int,
monitoring_live_gap_count: int,
direct_gap_count: int,
runtime_db_ok: bool,
runtime_summary: Mapping[str, Any],
ai_alert_db_ok: bool,
ai_alert_total: int,
ai_alert_ready_total: int,
verifier_ready: bool,
source_ready: bool,
) -> list[dict[str, Any]]:
runtime_total = _int(runtime_summary.get("alert_operation_log_total_7d"))
outbound_total = _int(runtime_summary.get("outbound_telegram_total_7d"))
return [
{
"stage_id": "monitoring_inventory_live_receipts",
"ready": monitoring_live_gap_count == 0,
"receipt_count": min(live_evidence_received_count, monitoring_surface_count),
"gap_count": monitoring_live_gap_count,
"next_action": "ingest_monitoring_surface_live_receipts_to_db_log",
},
{
"stage_id": "telegram_egress_gateway_control",
"ready": direct_gap_count == 0,
"receipt_count": 0,
"gap_count": direct_gap_count,
"next_action": "keep_all_telegram_alerts_on_gateway_receipt_path",
},
{
"stage_id": "alert_operation_log_runtime_readback",
"ready": runtime_db_ok and runtime_total > 0,
"receipt_count": runtime_total,
"gap_count": 0 if runtime_db_ok and runtime_total > 0 else 1,
"next_action": "append_alert_lifecycle_events_with_metadata_tags",
},
{
"stage_id": "awooop_outbound_telegram_ai_alert_card_mirror",
"ready": ai_alert_db_ok and ai_alert_total > 0 and outbound_total >= 0,
"receipt_count": ai_alert_total,
"gap_count": 0 if ai_alert_db_ok and ai_alert_total > 0 else 1,
"next_action": "mirror_telegram_cards_with_ai_automation_metadata",
},
{
"stage_id": "km_rag_mcp_playbook_ai_agent_learning_refs",
"ready": ai_alert_total > 0 and ai_alert_ready_total >= ai_alert_total,
"receipt_count": ai_alert_ready_total,
"gap_count": max(ai_alert_total - ai_alert_ready_total, 0)
if ai_alert_total > 0
else 1,
"next_action": "write_metadata_receipts_to_km_rag_mcp_playbook_context",
},
{
"stage_id": "ai_loop_post_apply_verifier",
"ready": verifier_ready,
"receipt_count": 1 if verifier_ready else 0,
"gap_count": 0 if verifier_ready else 1,
"next_action": "verify_context_receipts_before_calling_ai_automation_ready",
},
{
"stage_id": "source_contract_and_operator_surface",
"ready": source_ready,
"receipt_count": 1 if source_ready else 0,
"gap_count": 0 if source_ready else 1,
"next_action": "surface_receipts_in_runs_work_items_alerts",
},
]
def _work_item_progress(
*,
monitoring_live_gap_count: int,
direct_gap_count: int,
runtime_db_ok: bool,
runtime_summary: Mapping[str, Any],
ai_alert_db_ok: bool,
ai_alert_total: int,
ai_alert_ready_total: int,
verifier_ready: bool,
source_ready: bool,
) -> dict[str, Any]:
runtime_total = _int(runtime_summary.get("alert_operation_log_total_7d"))
checks = [
("source_contract_ready", source_ready),
("direct_telegram_send_gap_closed", direct_gap_count == 0),
("runtime_alert_operation_log_readback_ready", runtime_db_ok),
("runtime_alert_operation_log_has_recent_events", runtime_total > 0),
("ai_alert_card_delivery_readback_ready", ai_alert_db_ok),
("ai_alert_card_delivery_has_receipts", ai_alert_total > 0),
(
"ai_alert_card_learning_writeback_refs_ready",
ai_alert_total > 0 and ai_alert_ready_total >= ai_alert_total,
),
("ai_loop_post_apply_verifier_ready", verifier_ready),
("monitoring_live_receipt_coverage_complete", monitoring_live_gap_count == 0),
]
completed = sum(1 for _, passed in checks if passed)
return {
"work_item_id": "CIR-P0-TG-001",
"priority": "P0",
"completed_check_count": completed,
"total_check_count": len(checks),
"completion_percent": round(completed * 100 / len(checks), 1),
"checks": [
{
"check_id": check_id,
"status": "completed" if passed else "in_progress",
}
for check_id, passed in checks
],
}
def _ai_controlled_gap_queue(
*,
active_blockers: list[str],
monitoring_asset_rollups: Mapping[str, Any],
) -> list[dict[str, Any]]:
gap_samples = _list_of_dicts(monitoring_asset_rollups.get("live_receipt_gap_samples"))
queue: list[dict[str, Any]] = []
for index, blocker in enumerate(active_blockers, start=1):
queue.append({
"work_item_id": f"CIR-P0-TG-001-{index:02d}",
"parent_work_item_id": "CIR-P0-TG-001",
"priority": "P0" if index <= 4 else "P1",
"blocker": blocker,
"status": "queued_ai_controlled_apply",
"owner_review_required_for_low_medium_high": False,
"critical_break_glass_required": True,
"target_selector": _gap_target_selector(blocker),
"controlled_next_action": _gap_next_action(blocker),
"post_verifier": (
"/api/v1/agents/telegram-alert-monitoring-coverage-readback"
),
"sample_surfaces": gap_samples[:3] if blocker.startswith("monitoring_") else [],
"operation_boundaries": {
"requires_secret": False,
"requires_runtime_send": False,
"stores_raw_payload": False,
"destructive_action": False,
},
})
return queue
def _coverage_matrix(
*,
project_id: str,
@@ -781,6 +1102,64 @@ def _source_contract_blockers(source_contract: Mapping[str, Any]) -> list[str]:
]
def _gap_target_selector(blocker: str) -> dict[str, Any]:
if blocker.startswith("monitoring_live_receipt_gap"):
return {
"target_surface": "monitoring_inventory",
"target_tables": ["alert_operation_log", "awooop_outbound_message"],
"target_contexts": ["km", "rag", "playbook", "mcp", "ai_agent"],
}
if blocker.startswith("telegram_direct_send_gap"):
return {
"target_surface": "telegram_egress",
"target_tables": ["awooop_outbound_message"],
"target_contexts": ["telegram_gateway"],
}
if blocker.startswith("awooop_ai_alert_card"):
return {
"target_surface": "awooop_ai_alert_card_delivery",
"target_tables": ["awooop_outbound_message"],
"target_contexts": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
}
if blocker.startswith("runtime_alert_operation_log"):
return {
"target_surface": "alert_operation_log",
"target_tables": ["alert_operation_log"],
"target_contexts": ["alert_lifecycle", "ai_lane"],
}
if blocker.startswith("source_contract_missing"):
return {
"target_surface": "source_contract",
"target_tables": [],
"target_contexts": ["runs", "work_items", "alerts"],
}
return {
"target_surface": "telegram_monitoring_ai_automation",
"target_tables": ["alert_operation_log", "awooop_outbound_message"],
"target_contexts": ["km", "rag", "playbook", "mcp", "verifier", "ai_agent"],
}
def _gap_next_action(blocker: str) -> str:
if blocker.startswith("monitoring_live_receipt_gap"):
return "classify_monitoring_surfaces_and_ingest_metadata_live_receipts"
if blocker.startswith("telegram_direct_send_gap"):
return "migrate_direct_sender_to_telegram_gateway_receipt_path"
if blocker == "runtime_alert_operation_log_db_readback_unavailable":
return "retry_runtime_alert_operation_log_readback_with_bounded_timeout"
if blocker == "awooop_ai_alert_card_delivery_db_readback_unavailable":
return "retry_awooop_ai_alert_card_delivery_readback"
if blocker == "awooop_ai_alert_card_delivery_receipts_missing":
return "bind_telegram_alert_cards_to_awooop_outbound_message_mirror"
if blocker == "awooop_ai_alert_card_learning_writeback_refs_missing":
return "populate_km_rag_mcp_playbook_ai_agent_learning_refs"
if blocker == "telegram_alert_ai_loop_post_apply_verifier_not_ready":
return "run_ai_loop_context_post_apply_verifier_readback"
if blocker.startswith("source_contract_missing"):
return "restore_operator_visible_source_contract_marker"
return "continue_ai_controlled_gap_closure"
def _load_required_json(path: Path) -> dict[str, Any]:
if not path.exists():
raise FileNotFoundError(f"required monitoring coverage source missing: {path}")
@@ -841,6 +1220,29 @@ def _read_source_contract_text(repo_root: Path, relative_path: str) -> str:
return ""
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 _count_by_key(items: list[dict[str, Any]], key: str) -> dict[str, int]:
counts: dict[str, int] = {}
for item in items:
value = str(item.get(key) or "unknown")
counts[value] = counts.get(value, 0) + 1
return dict(sorted(counts.items()))
def _count_by_expected_scope_prefix(items: list[dict[str, Any]]) -> dict[str, int]:
counts: dict[str, int] = {}
for item in items:
expected_scope = str(item.get("expected_scope") or "unknown")
prefix = expected_scope.split("_", 1)[0] or "unknown"
counts[prefix] = counts.get(prefix, 0) + 1
return dict(sorted(counts.items()))
def _dict(value: Any) -> dict[str, Any]:
return dict(value) if isinstance(value, Mapping) else {}