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

feat(awooop): expose alert noise reduction readback
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) Failing after 16s
Details
CD Pipeline / build-and-deploy (push) Has been skipped
Details
CD Pipeline / post-deploy-checks (push) Has been skipped
Details

Browse Source
This commit is contained in:
Your Name
2026-06-29 17:38:06 +08:00
parent 876ba030d8
commit 81588f9a44
5 changed files with 503 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

410
apps/api/src/services/ai_agent_autonomous_runtime_control.py
View File

@@ -136,22 +136,48 @@ def _status_counts(
status_key: str,
) -> dict[str, Any]:
by_status: dict[str, int] = {}
recent_by_status: dict[str, int] = {}
total = 0
recent = 0
for row in rows:
item = _row_mapping(row)
status = str(item.get(status_key) or "unknown")
row_total = _int_value(item.get("total"))
row_recent = _int_value(item.get("recent"))
by_status[status] = by_status.get(status, 0) + row_total
recent_by_status[status] = recent_by_status.get(status, 0) + row_recent
total += row_total
recent += _int_value(item.get("recent"))
recent += row_recent
return {
"total": total,
"recent": recent,
"by_status": by_status,
"recent_by_status": recent_by_status,
}
def _status_total(summary: Mapping[str, Any] | None, *statuses: str) -> int:
if not isinstance(summary, Mapping):
return 0
if not statuses:
return _int_value(summary.get("total"))
by_status = summary.get("by_status")
if not isinstance(by_status, Mapping):
return 0
return sum(_int_value(by_status.get(status)) for status in statuses)
def _status_recent(summary: Mapping[str, Any] | None, *statuses: str) -> int:
if not isinstance(summary, Mapping):
return 0
if not statuses:
return _int_value(summary.get("recent"))
by_status = summary.get("recent_by_status")
if not isinstance(by_status, Mapping):
return 0
return sum(_int_value(by_status.get(status)) for status in statuses)
def _trace_stage(
*,
stage_id: str,
@@ -976,12 +1002,253 @@ def _build_learning_loop_readback(
}
def _alert_noise_stage(
*,
stage_id: str,
display_name: str,
evidence_sources: list[str],
total: int,
recent: int,
required_for_noise_reduction: bool,
feeds_controlled_queue: bool,
next_action_if_missing: str,
) -> dict[str, Any]:
present = total > 0
return {
"stage_id": stage_id,
"display_name": display_name,
"evidence_sources": evidence_sources,
"present": present,
"total": max(0, total),
"recent": max(0, recent),
"required_for_noise_reduction": required_for_noise_reduction,
"feeds_controlled_queue": feeds_controlled_queue,
"next_action_if_missing": None if present else next_action_if_missing,
}
def _build_alert_noise_reduction_readback(
*,
alert_operation_summary: Mapping[str, Any],
alertmanager_event_summary: Mapping[str, Any],
grouped_alert_summary: Mapping[str, Any],
operation_summary: Mapping[str, Any],
agent_decision_wiring: Mapping[str, Any],
learning_loop: Mapping[str, Any],
) -> dict[str, Any]:
"""Expose alert storm control and AI controlled routing receipts."""
alert_received_total = (
_status_total(alert_operation_summary, "ALERT_RECEIVED")
+ _status_total(alertmanager_event_summary, "received")
)
alert_received_recent = (
_status_recent(alert_operation_summary, "ALERT_RECEIVED")
+ _status_recent(alertmanager_event_summary, "received")
)
converged_duplicate_total = _status_total(alertmanager_event_summary, "converged")
converged_duplicate_recent = _status_recent(alertmanager_event_summary, "converged")
llm_inflight_suppressed_total = _status_total(
alertmanager_event_summary,
"llm_inflight_suppressed",
)
llm_inflight_suppressed_recent = _status_recent(
alertmanager_event_summary,
"llm_inflight_suppressed",
)
grouped_child_total = _status_total(grouped_alert_summary, "grouped_child_alert")
grouped_child_recent = _status_recent(grouped_alert_summary, "grouped_child_alert")
duplicate_convergence_total = (
converged_duplicate_total + llm_inflight_suppressed_total + grouped_child_total
)
duplicate_convergence_recent = (
converged_duplicate_recent + llm_inflight_suppressed_recent + grouped_child_recent
)
controlled_route_total = (
_trace_total(
operation_summary,
"ansible_candidate_matched",
"ansible_check_mode_executed",
"ansible_apply_executed",
)
+ _status_total(
alert_operation_summary,
"AUTO_REPAIR_TRIGGERED",
"EXECUTION_STARTED",
"EXECUTION_COMPLETED",
"NOTIFICATION_CLASSIFIED",
)
)
controlled_route_recent = (
_trace_recent(
operation_summary,
"ansible_candidate_matched",
"ansible_check_mode_executed",
"ansible_apply_executed",
)
+ _status_recent(
alert_operation_summary,
"AUTO_REPAIR_TRIGGERED",
"EXECUTION_STARTED",
"EXECUTION_COMPLETED",
"NOTIFICATION_CLASSIFIED",
)
)
guardrail_total = _status_total(
alert_operation_summary,
"GUARDRAIL_BLOCKED",
"STATE_GUARD_BLOCKED",
"ESCALATED",
"SILENCED",
)
guardrail_recent = _status_recent(
alert_operation_summary,
"GUARDRAIL_BLOCKED",
"STATE_GUARD_BLOCKED",
"ESCALATED",
"SILENCED",
)
decision_complete = agent_decision_wiring.get("status") == "completed"
learning_complete = learning_loop.get("status") == "completed"
stages = [
_alert_noise_stage(
stage_id="alert_intake_receipts",
display_name="Alertmanager receipts recorded",
evidence_sources=[
"alert_operation_log:ALERT_RECEIVED",
"awooop_conversation_event:received",
],
total=alert_received_total,
recent=alert_received_recent,
required_for_noise_reduction=True,
feeds_controlled_queue=True,
next_action_if_missing="record_alertmanager_received_events_before_any_notification_or_ai_route",
),
_alert_noise_stage(
stage_id="duplicate_convergence",
display_name="Duplicate and recurring alerts converge",
evidence_sources=[
"awooop_conversation_event:converged",
"awooop_conversation_event:llm_inflight_suppressed",
"awooop_conversation_event:alert-group",
],
total=duplicate_convergence_total,
recent=duplicate_convergence_recent,
required_for_noise_reduction=True,
feeds_controlled_queue=True,
next_action_if_missing="enable_converged_fingerprint_and_grouped_child_alert_receipts",
),
_alert_noise_stage(
stage_id="notification_suppression",
display_name="Telegram flood is suppressed into parent/digest receipts",
evidence_sources=[
"awooop_conversation_event:alert-group",
"telegram_gateway:grouped_alert_digest_dedup",
],
total=grouped_child_total + llm_inflight_suppressed_total,
recent=grouped_child_recent + llm_inflight_suppressed_recent,
required_for_noise_reduction=True,
feeds_controlled_queue=False,
next_action_if_missing="write_grouped_child_alert_event_or_inflight_suppression_receipt",
),
_alert_noise_stage(
stage_id="ai_controlled_routing",
display_name="Alerts route to AI controlled candidate/check/apply queue",
evidence_sources=[
"automation_operation_log:ansible_candidate_matched",
"automation_operation_log:ansible_check_mode_executed",
"automation_operation_log:ansible_apply_executed",
"alert_operation_log:AUTO_REPAIR_TRIGGERED",
],
total=controlled_route_total if decision_complete else 0,
recent=controlled_route_recent,
required_for_noise_reduction=True,
feeds_controlled_queue=True,
next_action_if_missing="route_repeated_non_critical_alerts_to_controlled_candidate_check_apply",
),
_alert_noise_stage(
stage_id="learning_feedback",
display_name="Suppressed alert patterns feed KM/RAG/PlayBook learning",
evidence_sources=[
"alert_noise_reduction",
"ai_agent_learning_loop_readback",
],
total=1 if learning_complete and duplicate_convergence_total > 0 else 0,
recent=1 if learning_complete and duplicate_convergence_recent > 0 else 0,
required_for_noise_reduction=True,
feeds_controlled_queue=True,
next_action_if_missing="keep_p1c_learning_loop_complete_before_closing_alert_noise_reduction",
),
_alert_noise_stage(
stage_id="break_glass_boundary",
display_name="Critical / guardrail cases remain isolated from default alert routing",
evidence_sources=[
"alert_operation_log:GUARDRAIL_BLOCKED",
"alert_operation_log:ESCALATED",
"current_policy:critical_break_glass_required",
],
total=guardrail_total,
recent=guardrail_recent,
required_for_noise_reduction=False,
feeds_controlled_queue=False,
next_action_if_missing="record_guardrail_or_break_glass_receipts_only_for_true_hard_blockers",
),
]
missing_required = [
str(stage["stage_id"])
for stage in stages
if stage["required_for_noise_reduction"] is True and stage["present"] is not True
]
present_required_count = sum(
1
for stage in stages
if stage["required_for_noise_reduction"] is True and stage["present"] is True
)
required_count = sum(1 for stage in stages if stage["required_for_noise_reduction"] is True)
return {
"schema_version": "ai_agent_alert_noise_reduction_readback_v1",
"status": "completed" if not missing_required else "in_progress",
"stages": stages,
"missing_required_stage_ids": missing_required,
"routing_policy": {
"manual_default_route_allowed": False,
"low_medium_high_alerts_route_to_ai_controlled_queue": True,
"critical_break_glass_still_required": True,
"telegram_child_alert_flood_allowed": False,
},
"public_safety": {
"stores_raw_alert_payload": False,
"stores_secret_values": False,
"executes_on_read": False,
"reads_raw_sessions": False,
},
"rollups": {
"stage_count": len(stages),
"required_stage_count": required_count,
"required_stage_present_count": present_required_count,
"required_stage_missing_count": len(missing_required),
"alert_received_total": alert_received_total,
"alert_received_recent": alert_received_recent,
"converged_duplicate_total": converged_duplicate_total,
"llm_inflight_suppressed_total": llm_inflight_suppressed_total,
"grouped_child_alert_total": grouped_child_total,
"suppressed_alert_total": duplicate_convergence_total,
"suppressed_alert_recent": duplicate_convergence_recent,
"controlled_route_total": controlled_route_total,
"controlled_route_recent": controlled_route_recent,
"break_glass_or_guardrail_total": guardrail_total,
},
}
def _build_work_item_progress(
*,
trace_ledger: Mapping[str, Any],
log_integration_taxonomy: Mapping[str, Any],
agent_decision_wiring: Mapping[str, Any],
learning_loop: Mapping[str, Any],
alert_noise_reduction: Mapping[str, Any],
db_read_status: str,
) -> dict[str, Any]:
"""Build ordered work items that the UI and agent can keep advancing."""
@@ -1015,6 +1282,16 @@ def _build_work_item_progress(
and learning_loop.get("schema_version") == "ai_agent_learning_loop_readback_v1"
and learning_loop_missing == 0
)
alert_noise_rollups = alert_noise_reduction.get("rollups")
if not isinstance(alert_noise_rollups, Mapping):
alert_noise_rollups = {}
alert_noise_missing = _int_value(alert_noise_rollups.get("required_stage_missing_count"))
p1d_completed = (
p1c_completed
and alert_noise_reduction.get("schema_version")
== "ai_agent_alert_noise_reduction_readback_v1"
and alert_noise_missing == 0
)
deployed_readback_complete = (
db_read_status == "ok"
and trace_ledger.get("schema_version") == "ai_agent_autonomous_trace_ledger_v1"
@@ -1086,8 +1363,9 @@ def _build_work_item_progress(
"work_item_id": "P1-D-alert-noise-reduction",
"priority": "P1-D",
"title": "Alert grouping and AI controlled workflow routing",
"status": "pending",
"status": "completed" if p1d_completed else "in_progress" if p1c_completed else "pending",
"exit_criteria": "repeated alerts are clustered, deduped, routed to controlled automation, and no longer default to manual handling",
"remaining_alert_noise_stage_count": alert_noise_missing,
},
{
"work_item_id": "P2-A-ui-ux-productization",
@@ -1869,6 +2147,9 @@ def build_runtime_receipt_readback_from_rows(
executor_log_count_rows: Iterable[Mapping[str, Any] | Any] = (),
timeline_count_rows: Iterable[Mapping[str, Any] | Any] = (),
playbook_trust_count_rows: Iterable[Mapping[str, Any] | Any] = (),
alert_operation_count_rows: Iterable[Mapping[str, Any] | Any] = (),
alertmanager_event_count_rows: Iterable[Mapping[str, Any] | Any] = (),
grouped_alert_event_count_rows: Iterable[Mapping[str, Any] | Any] = (),
error_type: str | None = None,
) -> dict[str, Any]:
"""Build the live executor receipt readback from already-fetched rows."""
@@ -1895,6 +2176,18 @@ def build_runtime_receipt_readback_from_rows(
executor_log_summary = _status_counts(executor_log_count_rows, status_key="status")
timeline_summary = _status_counts(timeline_count_rows, status_key="status")
playbook_trust_summary = _status_counts(playbook_trust_count_rows, status_key="status")
alert_operation_summary = _status_counts(
alert_operation_count_rows,
status_key="event_type",
)
alertmanager_event_summary = _status_counts(
alertmanager_event_count_rows,
status_key="stage",
)
grouped_alert_summary = _status_counts(
grouped_alert_event_count_rows,
status_key="status",
)
latest_closure = _latest_flow_closure(
operation_latest_rows=operation_latest,
verifier_latest_rows=verifier_latest,
@@ -1971,11 +2264,20 @@ def build_runtime_receipt_readback_from_rows(
controlled_retry_package=retry_package,
loop_ledger=loop_ledger,
)
alert_noise_reduction = _build_alert_noise_reduction_readback(
alert_operation_summary=alert_operation_summary,
alertmanager_event_summary=alertmanager_event_summary,
grouped_alert_summary=grouped_alert_summary,
operation_summary=operation_summary,
agent_decision_wiring=agent_decision_wiring,
learning_loop=learning_loop,
)
work_item_progress = _build_work_item_progress(
trace_ledger=trace_ledger,
log_integration_taxonomy=log_integration_taxonomy,
agent_decision_wiring=agent_decision_wiring,
learning_loop=learning_loop,
alert_noise_reduction=alert_noise_reduction,
db_read_status=db_read_status,
)
apply_summary = operation_summary.get("ansible_apply_executed") or {}
@@ -2099,6 +2401,7 @@ def build_runtime_receipt_readback_from_rows(
"log_integration_taxonomy": log_integration_taxonomy,
"agent_decision_wiring": agent_decision_wiring,
"learning_loop": learning_loop,
"alert_noise_reduction": alert_noise_reduction,
"work_item_progress": work_item_progress,
}
if error_type:
@@ -2247,6 +2550,36 @@ def _attach_runtime_receipt_readback(
"similar_case_source_total"
)
),
"live_alert_noise_stage_count": _int_value(
((readback.get("alert_noise_reduction") or {}).get("rollups") or {}).get(
"stage_count"
)
),
"live_alert_noise_required_present_count": _int_value(
((readback.get("alert_noise_reduction") or {}).get("rollups") or {}).get(
"required_stage_present_count"
)
),
"live_alert_noise_required_missing_count": _int_value(
((readback.get("alert_noise_reduction") or {}).get("rollups") or {}).get(
"required_stage_missing_count"
)
),
"live_alert_noise_complete_count": (
1
if (readback.get("alert_noise_reduction") or {}).get("status") == "completed"
else 0
),
"live_alert_noise_suppressed_count": _int_value(
((readback.get("alert_noise_reduction") or {}).get("rollups") or {}).get(
"suppressed_alert_total"
)
),
"live_alert_noise_controlled_route_count": _int_value(
((readback.get("alert_noise_reduction") or {}).get("rollups") or {}).get(
"controlled_route_total"
)
),
"live_work_item_count": _int_value(
((readback.get("work_item_progress") or {}).get("rollups") or {}).get(
"work_item_count"
@@ -2387,7 +2720,7 @@ def build_ai_agent_autonomous_runtime_control() -> dict[str, Any]:
"deploy_readback_marker": _DEPLOY_READBACK_MARKER,
"deploy_attempt_note": _DEPLOY_ATTEMPT_NOTE,
"legacy_no_send_no_live_rules_overridden": True,
"implementation_completion_percent": 88,
"implementation_completion_percent": 91,
"status_note": (
"目前有效規則：low / medium / high 風險由 AI Agent 在 allowlist、"
"Ansible check-mode、verifier、rollback、KM 與 Telegram receipt 下受控自動處理。"
@@ -2600,6 +2933,18 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
_RUNTIME_PLAYBOOK_TRUST_COUNTS_SQL,
_RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL,
)
alert_operation_counts = await _safe_aux_rows(
"alert_operation_counts",
_RUNTIME_ALERT_OPERATION_COUNTS_SQL,
)
alertmanager_event_counts = await _safe_aux_rows(
"alertmanager_event_counts",
_RUNTIME_ALERTMANAGER_EVENT_COUNTS_SQL,
)
grouped_alert_event_counts = await _safe_aux_rows(
"grouped_alert_event_counts",
_RUNTIME_GROUPED_ALERT_EVENT_COUNTS_SQL,
)
except Exception as exc:
logger.warning(
"ai_agent_autonomous_runtime_receipt_readback_failed",
@@ -2633,6 +2978,9 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
executor_log_count_rows=executor_log_counts,
timeline_count_rows=timeline_counts,
playbook_trust_count_rows=playbook_trust_counts,
alert_operation_count_rows=alert_operation_counts,
alertmanager_event_count_rows=alertmanager_event_counts,
grouped_alert_event_count_rows=grouped_alert_event_counts,
)
@@ -2984,6 +3332,62 @@ _RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL = """
"""
_RUNTIME_ALERT_OPERATION_COUNTS_SQL = """
SELECT
event_type,
count(*) AS total,
count(*) FILTER (
WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
) AS recent
FROM alert_operation_log
WHERE event_type IN (
'ALERT_RECEIVED',
'AUTO_REPAIR_TRIGGERED',
'EXECUTION_STARTED',
'EXECUTION_COMPLETED',
'NOTIFICATION_CLASSIFIED',
'GUARDRAIL_BLOCKED',
'STATE_GUARD_BLOCKED',
'SILENCED',
'ESCALATED'
)
GROUP BY event_type
ORDER BY event_type
"""
_RUNTIME_ALERTMANAGER_EVENT_COUNTS_SQL = """
SELECT
COALESCE(NULLIF(source_envelope ->> 'stage', ''), 'unknown') AS stage,
count(*) AS total,
count(*) FILTER (
WHERE received_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
) AS recent
FROM awooop_conversation_event
WHERE project_id = :project_id
AND COALESCE(
NULLIF(source_envelope ->> 'provider', ''),
platform_subject_id,
''
) = 'alertmanager'
GROUP BY COALESCE(NULLIF(source_envelope ->> 'stage', ''), 'unknown')
ORDER BY stage
"""
_RUNTIME_GROUPED_ALERT_EVENT_COUNTS_SQL = """
SELECT
'grouped_child_alert' AS status,
count(*) AS total,
count(*) FILTER (
WHERE received_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
) AS recent
FROM awooop_conversation_event
WHERE project_id = :project_id
AND channel_chat_id LIKE 'alert-group:%'
"""
def _validate_payload(payload: dict[str, Any]) -> None:
if payload.get("schema_version") != _SCHEMA_VERSION:
raise ValueError(f"schema_version must be {_SCHEMA_VERSION}")