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

fix(alerts): require runtime closure for coverage
All checks were successful
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 2m27s
Details
CD Pipeline / build-and-deploy (push) Successful in 6m46s
Details
CD Pipeline / post-deploy-checks (push) Successful in 1m41s
Details

Browse Source
This commit is contained in:
ogt
2026-07-11 10:43:33 +08:00
parent f64174106b
commit de53670cbd
2 changed files with 404 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -57,6 +57,56 @@ _ALERT_LOG_EVENT_TYPES = (
"EXECUTION_STARTED",
"EXECUTION_COMPLETED",
)
_RUNTIME_LIFECYCLE_SQL = """
WITH run_lifecycle AS (
SELECT
COALESCE(
NULLIF(context->>'automation_run_id', ''),
NULLIF(context->>'run_id', '')
) AS automation_run_id,
COUNT(*) FILTER (
WHERE event_type = 'AUTO_REPAIR_TRIGGERED'
) AS triggered_count,
COUNT(*) FILTER (
WHERE event_type = 'EXECUTION_STARTED'
) AS started_count,
COUNT(*) FILTER (
WHERE event_type = 'EXECUTION_COMPLETED'
AND success IS TRUE
) AS completed_success_count,
COUNT(*) FILTER (
WHERE event_type = 'TELEGRAM_RESULT_SENT'
) AS result_sent_count,
MAX(created_at) AS latest_at
FROM alert_operation_log
WHERE created_at >= NOW() - INTERVAL '7 days'
AND COALESCE(
NULLIF(context->>'automation_run_id', ''),
NULLIF(context->>'run_id', '')
) IS NOT NULL
GROUP BY automation_run_id
)
SELECT
COUNT(*) FILTER (
WHERE triggered_count > 0
) AS triggered_run_count,
COUNT(*) FILTER (
WHERE triggered_count > 0
AND started_count > 0
AND completed_success_count > 0
AND result_sent_count > 0
) AS closed_run_count,
MAX(latest_at) FILTER (
WHERE triggered_count > 0
) AS latest_triggered_at,
MAX(latest_at) FILTER (
WHERE triggered_count > 0
AND started_count > 0
AND completed_success_count > 0
AND result_sent_count > 0
) AS latest_closed_at
FROM run_lifecycle
"""
_REQUIRED_TAG_DIMENSIONS = (
(
"project",
@@ -468,12 +518,62 @@ def build_telegram_alert_monitoring_coverage_readback(
0,
)
direct_gap_count = _int(matrix_summary.get("known_direct_send_gap_count"))
matrix_surface_count = _int(
matrix_summary.get("telegram_alert_surface_count")
)
matrix_db_receipt_gap_count = max(
matrix_surface_count
- _int(matrix_summary.get("db_or_log_receipt_ready_surface_count")),
0,
)
matrix_ai_route_gap_count = max(
matrix_surface_count
- _int(matrix_summary.get("ai_route_ready_surface_count")),
0,
)
matrix_controlled_queue_gap_count = max(
matrix_surface_count
- _int(matrix_summary.get("controlled_queue_ready_surface_count")),
0,
)
matrix_post_verifier_gap_count = max(
matrix_surface_count
- _int(matrix_summary.get("post_verifier_ready_surface_count")),
0,
)
matrix_learning_writeback_gap_count = max(
matrix_surface_count
- _int(matrix_summary.get("learning_writeback_ready_surface_count")),
0,
)
matrix_control_gap_count = max(
matrix_db_receipt_gap_count,
matrix_ai_route_gap_count,
matrix_controlled_queue_gap_count,
matrix_post_verifier_gap_count,
matrix_learning_writeback_gap_count,
)
matrix_control_contract_ready = (
matrix_surface_count > 0 and matrix_control_gap_count == 0
)
verifier_ready = bool(
verifier_rollups.get(
"telegram_alert_learning_context_post_apply_verifier_ready"
)
)
runtime_db_ok = runtime_log_readback.get("status") == "ok"
runtime_triggered_run_count = _int(
runtime_summary.get("automation_triggered_run_count_7d")
)
runtime_closed_run_count = _int(
runtime_summary.get("automation_closed_run_count_7d")
)
runtime_open_run_count = _int(
runtime_summary.get("automation_open_run_count_7d")
)
runtime_closure_ready = bool(
runtime_summary.get("automation_runtime_closure_ready")
)
ai_alert_db_ok = ai_alert_card_delivery_readback.get("status") == "ok"
source_ready = not _source_contract_blockers(source_contract)
ai_alert_total = _int(ai_alert_summary.get("total"))
@@ -532,6 +632,14 @@ def build_telegram_alert_monitoring_coverage_readback(
source_contract=source_contract,
ai_alert_total=effective_ai_alert_total,
ai_alert_ready_total=effective_ai_alert_ready_total,
matrix_db_receipt_gap_count=matrix_db_receipt_gap_count,
matrix_ai_route_gap_count=matrix_ai_route_gap_count,
matrix_controlled_queue_gap_count=matrix_controlled_queue_gap_count,
matrix_post_verifier_gap_count=matrix_post_verifier_gap_count,
matrix_learning_writeback_gap_count=matrix_learning_writeback_gap_count,
runtime_triggered_run_count=runtime_triggered_run_count,
runtime_open_run_count=runtime_open_run_count,
runtime_closure_ready=runtime_closure_ready,
)
ready = not active_blockers
alert_receipt_pipeline = _alert_receipt_pipeline(
@@ -546,6 +654,12 @@ def build_telegram_alert_monitoring_coverage_readback(
ai_alert_ready_total=effective_ai_alert_ready_total,
verifier_ready=effective_verifier_ready,
source_ready=source_ready,
matrix_surface_count=matrix_surface_count,
matrix_control_gap_count=matrix_control_gap_count,
matrix_control_contract_ready=matrix_control_contract_ready,
runtime_closed_run_count=runtime_closed_run_count,
runtime_open_run_count=runtime_open_run_count,
runtime_closure_ready=runtime_closure_ready,
)
work_item_progress = _work_item_progress(
monitoring_live_gap_count=monitoring_live_gap_count,
@@ -557,6 +671,8 @@ def build_telegram_alert_monitoring_coverage_readback(
ai_alert_ready_total=effective_ai_alert_ready_total,
verifier_ready=effective_verifier_ready,
source_ready=source_ready,
matrix_control_contract_ready=matrix_control_contract_ready,
runtime_closure_ready=runtime_closure_ready,
)
ai_controlled_gap_queue = _ai_controlled_gap_queue(
active_blockers=active_blockers,
@@ -577,10 +693,15 @@ def build_telegram_alert_monitoring_coverage_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
monitoring_live_gap_count == 0
and matrix_db_receipt_gap_count == 0
and runtime_db_ok
and ai_alert_db_ok
),
"all_telegram_monitoring_alerts_ai_agent_automation_ready": (
ready
and matrix_control_contract_ready
and runtime_closure_ready
and effective_ai_alert_total > 0
and effective_ai_alert_ready_total >= effective_ai_alert_total
),
@@ -641,6 +762,20 @@ def build_telegram_alert_monitoring_coverage_readback(
matrix_summary.get("telegram_alert_surface_count")
),
"known_direct_send_gap_count": direct_gap_count,
"telegram_matrix_db_receipt_gap_count": matrix_db_receipt_gap_count,
"telegram_matrix_ai_route_gap_count": matrix_ai_route_gap_count,
"telegram_matrix_controlled_queue_gap_count": (
matrix_controlled_queue_gap_count
),
"telegram_matrix_post_verifier_gap_count": (
matrix_post_verifier_gap_count
),
"telegram_matrix_learning_writeback_gap_count": (
matrix_learning_writeback_gap_count
),
"telegram_matrix_control_contract_ready": (
matrix_control_contract_ready
),
"db_or_log_receipt_ready_surface_count": _int(
matrix_summary.get("db_or_log_receipt_ready_surface_count")
),
@@ -672,6 +807,19 @@ def build_telegram_alert_monitoring_coverage_readback(
"runtime_playbook_draft_event_count_7d": _int(
runtime_summary.get("playbook_draft_event_count_7d")
),
"runtime_automation_triggered_run_count_7d": (
runtime_triggered_run_count
),
"runtime_automation_closed_run_count_7d": (
runtime_closed_run_count
),
"runtime_automation_open_run_count_7d": (
runtime_open_run_count
),
"runtime_automation_closure_percent": float(
runtime_summary.get("automation_runtime_closure_percent") or 0.0
),
"runtime_automation_closure_ready": runtime_closure_ready,
"verified_context_receipt_count": _int(
verifier_rollups.get("verified_context_receipt_count")
),
@@ -941,6 +1089,11 @@ async def _load_runtime_log_readback(*, project_id: str) -> dict[str, Any]:
"telegram_sent_event_count_7d": 0,
"km_converted_event_count_7d": 0,
"playbook_draft_event_count_7d": 0,
"automation_triggered_run_count_7d": 0,
"automation_closed_run_count_7d": 0,
"automation_open_run_count_7d": 0,
"automation_runtime_closure_percent": 0.0,
"automation_runtime_closure_ready": False,
"error_type": type(exc).__name__,
},
"event_type_counts_7d": {},
@@ -963,6 +1116,7 @@ async def _load_runtime_log_readback_once(*, project_id: str) -> dict[str, Any]:
GROUP BY event_type
"""),
)
lifecycle_result = await db.execute(text(_RUNTIME_LIFECYCLE_SQL))
outbound_result = await db.execute(
text("""
SELECT
@@ -986,9 +1140,11 @@ async def _load_runtime_log_readback_once(*, project_id: str) -> dict[str, Any]:
)
event_rows = [_dict(row) for row in event_result.mappings().all()]
lifecycle_row = _dict(lifecycle_result.mappings().first() or {})
outbound_row = _dict(outbound_result.mappings().first() or {})
return _runtime_log_readback_from_rows(
event_rows=event_rows,
lifecycle_row=lifecycle_row,
outbound_row=outbound_row,
readback_source="db_session",
)
@@ -1034,6 +1190,10 @@ async def _load_runtime_log_readback_direct(
"""),
timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS,
)
lifecycle_row = await asyncio.wait_for(
conn.fetchrow(_RUNTIME_LIFECYCLE_SQL),
timeout=_RUNTIME_DIRECT_QUERY_TIMEOUT_SECONDS,
)
outbound_row = await asyncio.wait_for(
conn.fetchrow("""
SELECT
@@ -1057,6 +1217,9 @@ async def _load_runtime_log_readback_direct(
)
return _runtime_log_readback_from_rows(
event_rows=[dict(row) for row in event_rows],
lifecycle_row=(
dict(lifecycle_row) if lifecycle_row is not None else {}
),
outbound_row=dict(outbound_row) if outbound_row is not None else {},
readback_source="direct_connection",
)
@@ -1086,6 +1249,7 @@ def _runtime_log_readback_from_rows(
event_rows: list[dict[str, Any]],
outbound_row: dict[str, Any],
readback_source: str,
lifecycle_row: dict[str, Any] | None = None,
) -> dict[str, Any]:
event_counts = {str(row["event_type"]): _int(row["count"]) for row in event_rows}
latest_event_at = max(
@@ -1093,6 +1257,19 @@ def _runtime_log_readback_from_rows(
default=None,
)
total = sum(event_counts.values())
lifecycle = lifecycle_row or {}
triggered_run_count = _int(lifecycle.get("triggered_run_count"))
closed_run_count = min(
_int(lifecycle.get("closed_run_count")),
triggered_run_count,
)
open_run_count = max(
triggered_run_count - closed_run_count,
0,
)
runtime_closure_ready = (
triggered_run_count > 0 and open_run_count == 0
)
return {
"status": "ok",
"summary": {
@@ -1112,6 +1289,26 @@ def _runtime_log_readback_from_rows(
"NOTIFICATION_CLASSIFIED",
0,
),
"automation_triggered_run_count_7d": triggered_run_count,
"automation_closed_run_count_7d": closed_run_count,
"automation_open_run_count_7d": open_run_count,
"automation_runtime_closure_percent": round(
closed_run_count * 100 / triggered_run_count,
1,
)
if triggered_run_count
else 0.0,
"automation_runtime_closure_ready": runtime_closure_ready,
"latest_automation_triggered_at": (
str(lifecycle.get("latest_triggered_at"))
if lifecycle.get("latest_triggered_at")
else None
),
"latest_automation_closed_at": (
str(lifecycle.get("latest_closed_at"))
if lifecycle.get("latest_closed_at")
else None
),
"outbound_telegram_total_7d": _int(
outbound_row.get("outbound_telegram_total")
),
@@ -1431,6 +1628,12 @@ def _alert_receipt_pipeline(
ai_alert_ready_total: int,
verifier_ready: bool,
source_ready: bool,
matrix_surface_count: int,
matrix_control_gap_count: int,
matrix_control_contract_ready: bool,
runtime_closed_run_count: int,
runtime_open_run_count: int,
runtime_closure_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"))
@@ -1449,6 +1652,18 @@ def _alert_receipt_pipeline(
"gap_count": direct_gap_count,
"next_action": "keep_all_telegram_alerts_on_gateway_receipt_path",
},
{
"stage_id": "telegram_alert_surface_control_contracts",
"ready": matrix_control_contract_ready,
"receipt_count": max(
matrix_surface_count - matrix_control_gap_count,
0,
),
"gap_count": matrix_control_gap_count,
"next_action": (
"close_db_route_queue_verifier_learning_gaps_per_alert_surface"
),
},
{
"stage_id": "alert_operation_log_runtime_readback",
"ready": runtime_db_ok and runtime_total > 0,
@@ -1456,6 +1671,19 @@ def _alert_receipt_pipeline(
"gap_count": 0 if runtime_db_ok and runtime_total > 0 else 1,
"next_action": "append_alert_lifecycle_events_with_metadata_tags",
},
{
"stage_id": "runtime_ai_automation_lifecycle_closure",
"ready": runtime_closure_ready,
"receipt_count": runtime_closed_run_count,
"gap_count": (
0
if runtime_closure_ready
else max(runtime_open_run_count, 1)
),
"next_action": (
"complete_detect_check_apply_verify_result_writeback_same_run"
),
},
{
"stage_id": "awooop_outbound_telegram_ai_alert_card_mirror",
"ready": ai_alert_db_ok and ai_alert_total > 0 and outbound_total >= 0,
@@ -1500,13 +1728,20 @@ def _work_item_progress(
ai_alert_ready_total: int,
verifier_ready: bool,
source_ready: bool,
matrix_control_contract_ready: bool,
runtime_closure_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),
(
"telegram_alert_surface_control_contracts_complete",
matrix_control_contract_ready,
),
("runtime_alert_operation_log_readback_ready", runtime_db_ok),
("runtime_alert_operation_log_has_recent_events", runtime_total > 0),
("runtime_ai_automation_lifecycle_closed", runtime_closure_ready),
("ai_alert_card_delivery_readback_ready", ai_alert_db_ok),
("ai_alert_card_delivery_has_receipts", ai_alert_total > 0),
(
@@ -1780,12 +2015,32 @@ def _active_blockers(
source_contract: Mapping[str, Any],
ai_alert_total: int,
ai_alert_ready_total: int,
matrix_db_receipt_gap_count: int,
matrix_ai_route_gap_count: int,
matrix_controlled_queue_gap_count: int,
matrix_post_verifier_gap_count: int,
matrix_learning_writeback_gap_count: int,
runtime_triggered_run_count: int,
runtime_open_run_count: int,
runtime_closure_ready: bool,
) -> list[str]:
blockers = []
if monitoring_live_gap_count:
blockers.append(f"monitoring_live_receipt_gap:{monitoring_live_gap_count}")
if direct_gap_count:
blockers.append(f"telegram_direct_send_gap:{direct_gap_count}")
matrix_gaps = (
("db_receipt", matrix_db_receipt_gap_count),
("ai_route", matrix_ai_route_gap_count),
("controlled_queue", matrix_controlled_queue_gap_count),
("post_verifier", matrix_post_verifier_gap_count),
("learning_writeback", matrix_learning_writeback_gap_count),
)
blockers.extend(
f"telegram_alert_matrix_{gap_name}_gap:{gap_count}"
for gap_name, gap_count in matrix_gaps
if gap_count
)
if not runtime_db_ok:
blockers.append("runtime_alert_operation_log_db_readback_unavailable")
if not ai_alert_db_ok:
@@ -1794,6 +2049,13 @@ def _active_blockers(
blockers.append("awooop_ai_alert_card_delivery_receipts_missing")
if ai_alert_ready_total == 0:
blockers.append("awooop_ai_alert_card_learning_writeback_refs_missing")
if runtime_triggered_run_count == 0:
blockers.append("runtime_ai_automation_trigger_receipt_missing")
elif not runtime_closure_ready:
blockers.append(
"runtime_ai_automation_lifecycle_open:"
f"{max(runtime_open_run_count, 1)}"
)
if not verifier_ready:
blockers.append("telegram_alert_ai_loop_post_apply_verifier_not_ready")
blockers.extend(_source_contract_blockers(source_contract))
@@ -1979,6 +2241,28 @@ def _gap_target_selector(blocker: str) -> dict[str, Any]:
"target_tables": ["awooop_outbound_message"],
"target_contexts": ["telegram_gateway"],
}
if blocker.startswith("telegram_alert_matrix_"):
return {
"target_surface": "telegram_alert_ai_automation_matrix",
"target_tables": ["alert_operation_log", "awooop_outbound_message"],
"target_contexts": [
"ai_route",
"controlled_queue",
"post_verifier",
"learning_writeback",
],
}
if blocker.startswith("runtime_ai_automation_"):
return {
"target_surface": "alert_runtime_lifecycle",
"target_tables": ["alert_operation_log"],
"target_contexts": [
"check_mode",
"controlled_apply",
"post_verifier",
"telegram_result",
],
}
if blocker.startswith("awooop_ai_alert_card"):
return {
"target_surface": "awooop_ai_alert_card_delivery",
@@ -2009,6 +2293,12 @@ def _gap_next_action(blocker: str) -> str:
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.startswith("telegram_alert_matrix_"):
return "close_per_surface_ai_route_queue_verifier_learning_contract"
if blocker == "runtime_ai_automation_trigger_receipt_missing":
return "dispatch_bounded_alert_canary_and_record_runtime_lifecycle"
if blocker.startswith("runtime_ai_automation_lifecycle_open"):
return "resume_open_alert_lifecycle_to_verify_or_rollback_terminal"
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":