1517 lines
55 KiB
Python
1517 lines
55 KiB
Python
"""AwoooP read-only truth chain aggregation.
|
|
|
|
T0 only: this service does not mutate incident, approval, execution, or channel
|
|
state. It stitches existing durable records into one operator-facing status so
|
|
Telegram cards can be audited without guessing which subsystem owns the truth.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from datetime import UTC, date, datetime, timedelta
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import structlog
|
|
from sqlalchemy import text
|
|
|
|
from src.db.base import get_db_context
|
|
from src.services.awooop_ansible_audit_service import build_ansible_truth
|
|
from src.services.drift_repeat_state import build_drift_repeat_state
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
_MAX_ROWS = 100
|
|
_JSON_TEXT_FIELDS = {"gate_result", "source_envelope"}
|
|
_QUALITY_SUMMARY_CONCURRENCY = 8
|
|
|
|
|
|
def _clean(value: Any) -> Any:
|
|
"""Convert DB values into JSON-friendly primitives."""
|
|
if isinstance(value, UUID):
|
|
return str(value)
|
|
if isinstance(value, (datetime, date)):
|
|
return value.isoformat()
|
|
if isinstance(value, Decimal):
|
|
return float(value)
|
|
if isinstance(value, dict):
|
|
return {str(k): _clean(v) for k, v in value.items()}
|
|
if isinstance(value, list):
|
|
return [_clean(v) for v in value]
|
|
return value
|
|
|
|
|
|
def _clean_row(row: Any) -> dict[str, Any]:
|
|
cleaned: dict[str, Any] = {}
|
|
for key, value in dict(row).items():
|
|
if key in _JSON_TEXT_FIELDS and isinstance(value, str):
|
|
try:
|
|
value = json.loads(value)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
cleaned[key] = _clean(value)
|
|
return cleaned
|
|
|
|
|
|
async def _fetch_all(db: Any, sql: str, params: dict[str, Any]) -> list[dict[str, Any]]:
|
|
result = await db.execute(text(sql), params)
|
|
return [_clean_row(row) for row in result.mappings().all()]
|
|
|
|
|
|
async def _fetch_one(
|
|
db: Any,
|
|
sql: str,
|
|
params: dict[str, Any],
|
|
) -> dict[str, Any] | None:
|
|
result = await db.execute(text(sql), params)
|
|
row = result.mappings().first()
|
|
return _clean_row(row) if row else None
|
|
|
|
|
|
def _source_type(source_id: str, incident: dict[str, Any] | None, drift: dict[str, Any] | None) -> str:
|
|
if incident is not None:
|
|
return "incident"
|
|
if drift is not None:
|
|
return "drift_report"
|
|
try:
|
|
UUID(source_id)
|
|
except ValueError:
|
|
return "unknown"
|
|
return "run"
|
|
|
|
|
|
def _failed_mcp_tools(evidence_rows: list[dict[str, Any]]) -> list[str]:
|
|
failed: set[str] = set()
|
|
for evidence in evidence_rows:
|
|
mcp_health = evidence.get("mcp_health")
|
|
if isinstance(mcp_health, dict):
|
|
failed.update(str(tool) for tool, ok in mcp_health.items() if ok is False)
|
|
return sorted(failed)
|
|
|
|
|
|
def _approval_ids(approvals: list[dict[str, Any]]) -> list[str]:
|
|
return [str(row["id"]) for row in approvals if row.get("id")]
|
|
|
|
|
|
def _operation_ids(automation_ops: list[dict[str, Any]]) -> list[str]:
|
|
return [str(row["op_id"]) for row in automation_ops if row.get("op_id")]
|
|
|
|
|
|
def _auto_repair_ids(auto_repair_executions: list[dict[str, Any]]) -> list[str]:
|
|
return [str(row["id"]) for row in auto_repair_executions if row.get("id")]
|
|
|
|
|
|
def _incident_fingerprints(incident: dict[str, Any] | None) -> list[str]:
|
|
"""Extract durable alert fingerprints from incident signals, when present."""
|
|
if not incident:
|
|
return []
|
|
signals = incident.get("signals")
|
|
if not isinstance(signals, list):
|
|
return []
|
|
|
|
fingerprints: set[str] = set()
|
|
for signal in signals:
|
|
if not isinstance(signal, dict):
|
|
continue
|
|
direct = signal.get("fingerprint")
|
|
if direct:
|
|
fingerprints.add(str(direct))
|
|
labels = signal.get("labels")
|
|
if isinstance(labels, dict):
|
|
value = labels.get("fingerprint")
|
|
if value:
|
|
fingerprints.add(str(value))
|
|
return sorted(fingerprints)
|
|
|
|
|
|
def _looks_like_no_action(value: Any) -> bool:
|
|
text = str(value or "").upper()
|
|
return (
|
|
"NO_ACTION" in text
|
|
or "NO-ACTION" in text
|
|
or "NOACTION" in text
|
|
or text.startswith("OBSERVE")
|
|
or text.startswith("INVESTIGATE")
|
|
)
|
|
|
|
|
|
def _approval_has_no_action(approvals: list[dict[str, Any]]) -> bool:
|
|
return any(_looks_like_no_action(row.get("action")) for row in approvals)
|
|
|
|
|
|
def _is_no_action_operation(row: dict[str, Any]) -> bool:
|
|
"""Return true for durable audit rows that represent observation, not repair."""
|
|
if str(row.get("operation_type") or "") != "playbook_executed":
|
|
return False
|
|
return any(
|
|
_looks_like_no_action(row.get(key))
|
|
for key in (
|
|
"input_action",
|
|
"output_action",
|
|
"output_reason",
|
|
"output_not_used_reason",
|
|
)
|
|
)
|
|
|
|
|
|
def _is_audit_only_operation(row: dict[str, Any]) -> bool:
|
|
operation_type = str(row.get("operation_type") or "")
|
|
status = str(row.get("status") or "").lower()
|
|
if status == "dry_run":
|
|
return True
|
|
return operation_type in {
|
|
"ansible_candidate_matched",
|
|
"ansible_execution_skipped",
|
|
}
|
|
|
|
|
|
def _effective_execution_ops(automation_ops: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return [
|
|
row
|
|
for row in automation_ops
|
|
if not _is_no_action_operation(row) and not _is_audit_only_operation(row)
|
|
]
|
|
|
|
|
|
def build_incident_reconciliation(
|
|
*,
|
|
incident: dict[str, Any] | None,
|
|
approvals: list[dict[str, Any]],
|
|
evidence_rows: list[dict[str, Any]],
|
|
automation_ops: list[dict[str, Any]],
|
|
auto_repair_executions: list[dict[str, Any]] | None = None,
|
|
timeline_events: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
"""Build a read-only consistency report across incident lifecycle tables."""
|
|
if incident is None:
|
|
return {
|
|
"schema_version": "incident_reconciliation_v1",
|
|
"applicable": False,
|
|
"consistency_status": "not_applicable",
|
|
"operator_next_state": "not_applicable",
|
|
"facts": {},
|
|
"mismatches": [],
|
|
}
|
|
|
|
incident_status = str(incident.get("status") or "unknown").upper()
|
|
incident_closed = incident_status in {"RESOLVED", "CLOSED"}
|
|
latest_approval = approvals[0] if approvals else None
|
|
approval_status = str((latest_approval or {}).get("status") or "none").upper()
|
|
approval_action = str((latest_approval or {}).get("action") or "")
|
|
approval_resolved = bool((latest_approval or {}).get("resolved_at"))
|
|
attempted = sum(int(row.get("sensors_attempted") or 0) for row in evidence_rows)
|
|
succeeded = sum(int(row.get("sensors_succeeded") or 0) for row in evidence_rows)
|
|
repair_rows = auto_repair_executions or []
|
|
executed_ops = [
|
|
row
|
|
for row in automation_ops
|
|
if str(row.get("status") or "").lower()
|
|
in {"success", "completed", "executed"}
|
|
]
|
|
successful_repairs = [row for row in repair_rows if row.get("success") is True]
|
|
effective_execution_count = len(executed_ops) + len(successful_repairs)
|
|
latest_verification = _latest_verification_result(incident, evidence_rows)
|
|
mismatches: list[dict[str, Any]] = []
|
|
|
|
def add(code: str, severity: str, message: str) -> None:
|
|
mismatches.append({
|
|
"code": code,
|
|
"severity": severity,
|
|
"message": message,
|
|
})
|
|
|
|
if (
|
|
latest_approval
|
|
and not incident_closed
|
|
and (approval_resolved or approval_status in {"APPROVED", "REJECTED"})
|
|
):
|
|
if effective_execution_count:
|
|
add(
|
|
"incident_open_after_successful_execution",
|
|
"high",
|
|
"Execution evidence exists while the incident is still open.",
|
|
)
|
|
else:
|
|
add(
|
|
"incident_open_after_approval_resolved",
|
|
"high",
|
|
"Approval reached a terminal state while the incident is still open.",
|
|
)
|
|
|
|
if approval_status == "APPROVED" and not effective_execution_count:
|
|
add(
|
|
"approval_approved_without_execution_record",
|
|
"high",
|
|
"Approval is approved but automation_operation_log has no linked execution record.",
|
|
)
|
|
|
|
if (
|
|
approval_status == "APPROVED"
|
|
and "NO_ACTION" in approval_action.upper()
|
|
and not effective_execution_count
|
|
):
|
|
add(
|
|
"approval_no_action_without_execution",
|
|
"high",
|
|
"Approval resolved to NO_ACTION and no executor produced a successful operation.",
|
|
)
|
|
|
|
if successful_repairs and str(latest_verification or "").lower() == "degraded":
|
|
add(
|
|
"verification_degraded_after_auto_repair",
|
|
"medium",
|
|
"Auto-repair succeeded, but post-execution verification reported degraded evidence.",
|
|
)
|
|
|
|
if attempted > 0 and succeeded == 0:
|
|
add(
|
|
"evidence_all_sensors_failed",
|
|
"medium",
|
|
"Evidence collection attempted sensors but none succeeded.",
|
|
)
|
|
|
|
if latest_approval and not timeline_events:
|
|
add(
|
|
"timeline_missing_for_approval",
|
|
"medium",
|
|
"Approval exists but timeline_events has no linked lifecycle entries.",
|
|
)
|
|
|
|
high_count = sum(1 for row in mismatches if row["severity"] == "high")
|
|
medium_count = sum(1 for row in mismatches if row["severity"] == "medium")
|
|
if high_count:
|
|
consistency_status = "blocked"
|
|
operator_next_state = "manual_required"
|
|
elif medium_count:
|
|
consistency_status = "degraded"
|
|
operator_next_state = "investigate"
|
|
else:
|
|
consistency_status = "consistent"
|
|
operator_next_state = "continue"
|
|
|
|
return {
|
|
"schema_version": "incident_reconciliation_v1",
|
|
"applicable": True,
|
|
"consistency_status": consistency_status,
|
|
"operator_next_state": operator_next_state,
|
|
"facts": {
|
|
"incident_id": incident.get("incident_id"),
|
|
"incident_status": incident_status,
|
|
"incident_closed": incident_closed,
|
|
"latest_approval_id": (latest_approval or {}).get("id"),
|
|
"latest_approval_status": approval_status,
|
|
"latest_approval_action": approval_action,
|
|
"approval_resolved": approval_resolved,
|
|
"evidence_records": len(evidence_rows),
|
|
"sensors_attempted": attempted,
|
|
"sensors_succeeded": succeeded,
|
|
"automation_operation_records": len(automation_ops),
|
|
"executed_operation_records": len(executed_ops),
|
|
"auto_repair_execution_records": len(repair_rows),
|
|
"successful_auto_repair_records": len(successful_repairs),
|
|
"effective_execution_records": effective_execution_count,
|
|
"latest_verification_result": latest_verification,
|
|
"timeline_events": len(timeline_events),
|
|
},
|
|
"mismatches": mismatches,
|
|
}
|
|
|
|
|
|
def _truth_status(
|
|
*,
|
|
incident: dict[str, Any] | None,
|
|
approvals: list[dict[str, Any]],
|
|
evidence_rows: list[dict[str, Any]],
|
|
automation_ops: list[dict[str, Any]],
|
|
drift: dict[str, Any] | None,
|
|
drift_repeat_count: int,
|
|
gateway_mcp_total: int,
|
|
legacy_mcp_total: int,
|
|
outbound_visible_total: int,
|
|
inbound_visible_total: int = 0,
|
|
auto_repair_executions: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Derive the current operator-visible truth-chain stage."""
|
|
blockers: list[str] = []
|
|
needs_human = False
|
|
stage = "not_found"
|
|
stage_status = "missing"
|
|
|
|
if drift is not None:
|
|
stage = "dedup_or_repeat_updated" if drift_repeat_count > 1 else "received"
|
|
stage_status = str(drift.get("status") or "unknown")
|
|
interpretation = drift.get("interpretation") or {}
|
|
confidence = interpretation.get("confidence") if isinstance(interpretation, dict) else None
|
|
if stage_status == "pending":
|
|
needs_human = True
|
|
blockers.append("drift_report_pending_without_resolution")
|
|
if confidence in (0, 0.0):
|
|
blockers.append("drift_ai_confidence_zero")
|
|
|
|
if incident is None and drift is None and inbound_visible_total > 0:
|
|
stage = "inbound_received"
|
|
stage_status = "observed"
|
|
|
|
if incident is not None:
|
|
incident_status = str(incident.get("status") or "unknown")
|
|
repair_rows = auto_repair_executions or []
|
|
effective_ops = _effective_execution_ops(automation_ops)
|
|
has_execution_records = bool(effective_ops or repair_rows)
|
|
stage = "received"
|
|
stage_status = incident_status.lower()
|
|
if incident_status in {"RESOLVED", "CLOSED"}:
|
|
stage = "resolved"
|
|
stage_status = "success"
|
|
elif evidence_rows:
|
|
attempted = sum(int(row.get("sensors_attempted") or 0) for row in evidence_rows)
|
|
succeeded = sum(int(row.get("sensors_succeeded") or 0) for row in evidence_rows)
|
|
if attempted > 0 and succeeded == 0:
|
|
stage = "evidence_degraded"
|
|
stage_status = "warning"
|
|
needs_human = True
|
|
blockers.append("all_evidence_sensors_failed")
|
|
else:
|
|
stage = "evidence_collected"
|
|
stage_status = "completed"
|
|
|
|
approval_statuses = {str(row.get("status") or "").upper() for row in approvals}
|
|
approval_actions = " ".join(str(row.get("action") or "") for row in approvals).upper()
|
|
approval_no_action = _approval_has_no_action(approvals)
|
|
if any(status in {"PENDING", "WAITING_APPROVAL"} for status in approval_statuses):
|
|
stage = "approval_required"
|
|
stage_status = "waiting"
|
|
needs_human = True
|
|
elif not has_execution_records and (approval_no_action or "NO_ACTION" in approval_actions):
|
|
if approval_statuses:
|
|
stage = "manual_required"
|
|
stage_status = "blocked"
|
|
needs_human = True
|
|
blockers.append("approval_resolved_no_action_without_execution")
|
|
elif "APPROVED" in approval_statuses and not has_execution_records:
|
|
stage = "execution_missing"
|
|
stage_status = "blocked"
|
|
needs_human = True
|
|
blockers.append("approved_without_execution_record")
|
|
|
|
op_statuses = {str(row.get("status") or "").lower() for row in effective_ops}
|
|
repair_successes = {row.get("success") for row in repair_rows}
|
|
execution_succeeded = (op_statuses & {"success", "completed"}) or True in repair_successes
|
|
execution_failed = (op_statuses & {"failed", "error"}) or False in repair_successes
|
|
if op_statuses or repair_successes:
|
|
if execution_succeeded:
|
|
stage = "execution_succeeded"
|
|
stage_status = "success"
|
|
elif execution_failed:
|
|
stage = "execution_failed"
|
|
stage_status = "error"
|
|
needs_human = True
|
|
else:
|
|
stage = "execution_started"
|
|
stage_status = "running"
|
|
|
|
if incident_status == "INVESTIGATING" and approvals:
|
|
if execution_succeeded:
|
|
blockers.append("incident_open_after_successful_execution")
|
|
needs_human = True
|
|
elif not has_execution_records:
|
|
blockers.append("incident_still_investigating_after_approval")
|
|
|
|
if gateway_mcp_total == 0:
|
|
blockers.append("awooop_mcp_gateway_audit_empty")
|
|
if legacy_mcp_total == 0 and incident is not None:
|
|
blockers.append("legacy_mcp_audit_missing")
|
|
if outbound_visible_total == 0:
|
|
blockers.append("outbound_mirror_not_visible_for_source")
|
|
|
|
return {
|
|
"current_stage": stage,
|
|
"stage_status": stage_status,
|
|
"needs_human": needs_human,
|
|
"blockers": blockers,
|
|
}
|
|
|
|
|
|
def _latest_verification_result(
|
|
incident: dict[str, Any] | None,
|
|
evidence_rows: list[dict[str, Any]],
|
|
) -> str | None:
|
|
if incident and incident.get("verification_result"):
|
|
return str(incident["verification_result"])
|
|
for row in evidence_rows:
|
|
if row.get("verification_result"):
|
|
return str(row["verification_result"])
|
|
return None
|
|
|
|
|
|
def build_automation_quality(
|
|
*,
|
|
incident: dict[str, Any] | None,
|
|
approvals: list[dict[str, Any]],
|
|
evidence_rows: list[dict[str, Any]],
|
|
automation_ops: list[dict[str, Any]],
|
|
auto_repair_executions: list[dict[str, Any]],
|
|
gateway_mcp_summary: dict[str, Any],
|
|
legacy_mcp_summary: dict[str, Any],
|
|
outbound_rows: list[dict[str, Any]],
|
|
km_entries: list[dict[str, Any]],
|
|
timeline_events: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
"""Summarize whether a card reached the real automation flywheel."""
|
|
if incident is None:
|
|
return {
|
|
"schema_version": "automation_quality_v1",
|
|
"applicable": False,
|
|
"verdict": "not_applicable",
|
|
"score": 0,
|
|
"gates": [],
|
|
"facts": {},
|
|
"blockers": [],
|
|
}
|
|
|
|
blockers: list[str] = []
|
|
gates: list[dict[str, Any]] = []
|
|
|
|
def gate(name: str, status: str, detail: str | None = None) -> None:
|
|
gates.append({"name": name, "status": status, "detail": detail})
|
|
if status in {"failed", "missing"}:
|
|
blockers.append(name)
|
|
|
|
evidence_attempted = sum(int(row.get("sensors_attempted") or 0) for row in evidence_rows)
|
|
evidence_succeeded = sum(int(row.get("sensors_succeeded") or 0) for row in evidence_rows)
|
|
gateway_total = int(gateway_mcp_summary.get("total") or 0)
|
|
legacy_total = int(legacy_mcp_summary.get("total") or 0)
|
|
effective_ops = _effective_execution_ops(automation_ops)
|
|
noop_ops = [row for row in automation_ops if _is_no_action_operation(row)]
|
|
audit_only_ops = [row for row in automation_ops if _is_audit_only_operation(row)]
|
|
automation_statuses = {str(row.get("status") or "").lower() for row in effective_ops}
|
|
auto_repair_successes = {row.get("success") for row in auto_repair_executions}
|
|
has_execution = bool(effective_ops or auto_repair_executions)
|
|
verification_result = _latest_verification_result(incident, evidence_rows)
|
|
approval_statuses = {str(row.get("status") or "").upper() for row in approvals}
|
|
approval_actions = " ".join(str(row.get("action") or "") for row in approvals).upper()
|
|
approval_no_action = _approval_has_no_action(approvals)
|
|
|
|
gate("source_persisted", "passed", str(incident.get("incident_id")))
|
|
gate("outbound_recorded", "passed" if outbound_rows else "missing", str(len(outbound_rows)))
|
|
if not evidence_rows:
|
|
gate("evidence_collected", "missing", "no incident_evidence rows")
|
|
elif evidence_attempted > 0 and evidence_succeeded == 0:
|
|
gate("evidence_collected", "failed", f"{evidence_succeeded}/{evidence_attempted}")
|
|
elif evidence_succeeded > 0:
|
|
gate("evidence_collected", "passed", f"{evidence_succeeded}/{evidence_attempted}")
|
|
else:
|
|
gate("evidence_collected", "warning", f"{evidence_succeeded}/{evidence_attempted}")
|
|
|
|
if gateway_total > 0:
|
|
gate("mcp_gateway_observed", "passed", str(gateway_total))
|
|
elif legacy_total > 0:
|
|
gate("mcp_gateway_observed", "warning", f"legacy_only={legacy_total}")
|
|
else:
|
|
gate("mcp_gateway_observed", "missing", "no mcp audit")
|
|
|
|
if any(status in {"PENDING", "WAITING_APPROVAL"} for status in approval_statuses):
|
|
gate("approval_state", "warning", "waiting_approval")
|
|
elif approval_statuses and (approval_no_action or "NO_ACTION" in approval_actions) and not has_execution:
|
|
gate("approval_state", "failed", "approved_no_action_without_execution")
|
|
elif approvals:
|
|
gate("approval_state", "passed", ",".join(sorted(approval_statuses)))
|
|
else:
|
|
gate("approval_state", "not_applicable", "no approval")
|
|
|
|
gate("execution_recorded", "passed" if has_execution else "missing", str(len(effective_ops) + len(auto_repair_executions)))
|
|
gate("auto_repair_recorded", "passed" if auto_repair_executions else "missing", str(len(auto_repair_executions)))
|
|
|
|
verification_status = str(verification_result or "").lower()
|
|
|
|
if not has_execution:
|
|
gate("verification_recorded", "not_applicable", "no execution")
|
|
elif verification_status == "success":
|
|
gate("verification_recorded", "passed", verification_result)
|
|
elif verification_result:
|
|
gate("verification_recorded", "warning", verification_result)
|
|
else:
|
|
gate("verification_recorded", "missing", "execution without verification_result")
|
|
|
|
if not has_execution:
|
|
gate("learning_recorded", "not_applicable", "no execution")
|
|
elif km_entries:
|
|
gate("learning_recorded", "passed", str(len(km_entries)))
|
|
else:
|
|
gate("learning_recorded", "missing", "execution without KM entry")
|
|
|
|
gate("timeline_recorded", "passed" if timeline_events else "missing", str(len(timeline_events)))
|
|
|
|
if has_execution and (
|
|
False in auto_repair_successes or automation_statuses & {"failed", "error"}
|
|
):
|
|
verdict = "execution_failed"
|
|
elif has_execution and verification_status == "success":
|
|
verdict = "auto_repaired_verified"
|
|
elif has_execution and verification_result:
|
|
verdict = "auto_repaired_verification_degraded"
|
|
elif has_execution:
|
|
verdict = "execution_unverified"
|
|
elif approval_statuses and (approval_no_action or "NO_ACTION" in approval_actions):
|
|
verdict = "manual_required_no_action"
|
|
elif any(status in {"PENDING", "WAITING_APPROVAL"} for status in approval_statuses):
|
|
verdict = "approval_required"
|
|
elif evidence_rows or gateway_total or legacy_total:
|
|
verdict = "observed_not_executed"
|
|
else:
|
|
verdict = "received_only"
|
|
|
|
score_weights = {
|
|
"source_persisted": 10,
|
|
"outbound_recorded": 10,
|
|
"evidence_collected": 15,
|
|
"mcp_gateway_observed": 15,
|
|
"approval_state": 10,
|
|
"execution_recorded": 15,
|
|
"auto_repair_recorded": 10,
|
|
"verification_recorded": 10,
|
|
"learning_recorded": 3,
|
|
"timeline_recorded": 2,
|
|
}
|
|
score = 0
|
|
for row in gates:
|
|
weight = score_weights.get(str(row["name"]), 0)
|
|
if row["status"] == "passed":
|
|
score += weight
|
|
elif row["status"] == "not_applicable" and row["name"] == "approval_state":
|
|
score += weight
|
|
elif row["status"] == "warning":
|
|
score += weight // 2
|
|
|
|
return {
|
|
"schema_version": "automation_quality_v1",
|
|
"applicable": True,
|
|
"verdict": verdict,
|
|
"score": score,
|
|
"gates": gates,
|
|
"facts": {
|
|
"incident_id": incident.get("incident_id"),
|
|
"evidence_records": len(evidence_rows),
|
|
"sensors_attempted": evidence_attempted,
|
|
"sensors_succeeded": evidence_succeeded,
|
|
"mcp_gateway_total": gateway_total,
|
|
"legacy_mcp_total": legacy_total,
|
|
"approvals": len(approvals),
|
|
"automation_operation_records": len(automation_ops),
|
|
"effective_execution_records": len(effective_ops),
|
|
"noop_operation_records": len(noop_ops),
|
|
"audit_only_operation_records": len(audit_only_ops),
|
|
"auto_repair_execution_records": len(auto_repair_executions),
|
|
"verification_result": verification_result,
|
|
"knowledge_entries": len(km_entries),
|
|
"timeline_events": len(timeline_events),
|
|
"outbound_messages": len(outbound_rows),
|
|
},
|
|
"blockers": blockers,
|
|
}
|
|
|
|
|
|
def _automation_quality_score_bucket(score: int) -> str:
|
|
if score >= 85:
|
|
return "green"
|
|
if score >= 60:
|
|
return "yellow"
|
|
return "red"
|
|
|
|
|
|
def summarize_automation_quality_records(
|
|
*,
|
|
project_id: str,
|
|
window_hours: int,
|
|
records: list[dict[str, Any]],
|
|
limit: int,
|
|
) -> dict[str, Any]:
|
|
"""Aggregate per-incident automation quality into an operator summary."""
|
|
verdicts: dict[str, dict[str, Any]] = {}
|
|
gate_failures: dict[str, dict[str, Any]] = {}
|
|
score_buckets: dict[str, int] = {"green": 0, "yellow": 0, "red": 0}
|
|
examples: list[dict[str, Any]] = []
|
|
total_score = 0
|
|
evaluated_total = 0
|
|
verified_total = 0
|
|
|
|
for record in records:
|
|
incident = record.get("incident") if isinstance(record.get("incident"), dict) else {}
|
|
truth_status = record.get("truth_status") if isinstance(record.get("truth_status"), dict) else {}
|
|
quality = record.get("automation_quality") if isinstance(record.get("automation_quality"), dict) else {}
|
|
if quality.get("applicable") is not True:
|
|
continue
|
|
|
|
evaluated_total += 1
|
|
score = int(quality.get("score") or 0)
|
|
total_score += score
|
|
bucket = _automation_quality_score_bucket(score)
|
|
score_buckets[bucket] += 1
|
|
|
|
verdict = str(quality.get("verdict") or "unknown")
|
|
if verdict == "auto_repaired_verified":
|
|
verified_total += 1
|
|
verdict_row = verdicts.setdefault(
|
|
verdict,
|
|
{
|
|
"verdict": verdict,
|
|
"total": 0,
|
|
"score_sum": 0,
|
|
"min_score": score,
|
|
"max_score": score,
|
|
"needs_human": False,
|
|
},
|
|
)
|
|
verdict_row["total"] += 1
|
|
verdict_row["score_sum"] += score
|
|
verdict_row["min_score"] = min(int(verdict_row["min_score"]), score)
|
|
verdict_row["max_score"] = max(int(verdict_row["max_score"]), score)
|
|
verdict_row["needs_human"] = bool(
|
|
verdict_row["needs_human"] or truth_status.get("needs_human")
|
|
)
|
|
|
|
for gate in quality.get("gates") or []:
|
|
if not isinstance(gate, dict):
|
|
continue
|
|
gate_status = str(gate.get("status") or "")
|
|
if gate_status not in {"failed", "missing"}:
|
|
continue
|
|
gate_name = str(gate.get("name") or "unknown")
|
|
gate_row = gate_failures.setdefault(
|
|
gate_name,
|
|
{"gate": gate_name, "total": 0, "statuses": {}},
|
|
)
|
|
gate_row["total"] += 1
|
|
gate_row["statuses"][gate_status] = int(gate_row["statuses"].get(gate_status, 0)) + 1
|
|
|
|
examples.append({
|
|
"incident_id": incident.get("incident_id"),
|
|
"alertname": incident.get("alertname"),
|
|
"severity": incident.get("severity"),
|
|
"status": incident.get("status"),
|
|
"created_at": incident.get("created_at"),
|
|
"truth_stage": truth_status.get("current_stage"),
|
|
"truth_stage_status": truth_status.get("stage_status"),
|
|
"needs_human": bool(truth_status.get("needs_human")),
|
|
"verdict": verdict,
|
|
"score": score,
|
|
"score_bucket": bucket,
|
|
"blockers": list(quality.get("blockers") or [])[:8],
|
|
})
|
|
|
|
by_verdict = []
|
|
for row in verdicts.values():
|
|
total = int(row["total"])
|
|
row["avg_score"] = round(float(row.pop("score_sum")) / total, 1) if total else 0.0
|
|
by_verdict.append(row)
|
|
by_verdict.sort(key=lambda row: (-int(row["total"]), str(row["verdict"])))
|
|
|
|
failing_gates = sorted(
|
|
gate_failures.values(),
|
|
key=lambda row: (-int(row["total"]), str(row["gate"])),
|
|
)
|
|
|
|
return {
|
|
"schema_version": "automation_quality_summary_v1",
|
|
"project_id": project_id,
|
|
"window_hours": window_hours,
|
|
"limit": limit,
|
|
"incident_total": len(records),
|
|
"evaluated_total": evaluated_total,
|
|
"verified_auto_repair_total": verified_total,
|
|
"average_score": round(total_score / evaluated_total, 1) if evaluated_total else 0.0,
|
|
"score_buckets": score_buckets,
|
|
"by_verdict": by_verdict,
|
|
"gate_failures": failing_gates,
|
|
"examples": examples[:25],
|
|
"production_claim": {
|
|
"can_claim_full_auto_repair": evaluated_total > 0 and verified_total == evaluated_total,
|
|
"reason": (
|
|
"all_evaluated_incidents_auto_repaired_verified"
|
|
if evaluated_total > 0 and verified_total == evaluated_total
|
|
else "some_incidents_are_not_auto_repaired_verified"
|
|
),
|
|
},
|
|
}
|
|
|
|
|
|
def _summarize_mcp(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
|
by_tool: dict[str, dict[str, Any]] = {}
|
|
success_count = 0
|
|
failure_count = 0
|
|
for row in rows:
|
|
key = f"{row.get('mcp_server') or 'unknown'}:{row.get('tool_name') or 'unknown'}"
|
|
item = by_tool.setdefault(
|
|
key,
|
|
{
|
|
"mcp_server": row.get("mcp_server"),
|
|
"tool_name": row.get("tool_name"),
|
|
"success": 0,
|
|
"failed": 0,
|
|
"last_error": None,
|
|
},
|
|
)
|
|
if row.get("success") is True:
|
|
item["success"] += 1
|
|
success_count += 1
|
|
else:
|
|
item["failed"] += 1
|
|
failure_count += 1
|
|
item["last_error"] = row.get("error_message")
|
|
return {
|
|
"total": len(rows),
|
|
"success": success_count,
|
|
"failed": failure_count,
|
|
"by_tool": list(by_tool.values()),
|
|
}
|
|
|
|
|
|
def _as_bool(value: Any) -> bool:
|
|
if isinstance(value, bool):
|
|
return value
|
|
return str(value).lower() == "true"
|
|
|
|
|
|
def _counter_bucket(
|
|
buckets: dict[str, dict[str, Any]],
|
|
key: str,
|
|
*,
|
|
label: str,
|
|
) -> dict[str, Any]:
|
|
return buckets.setdefault(
|
|
key,
|
|
{
|
|
label: key,
|
|
"total": 0,
|
|
"success": 0,
|
|
"failed": 0,
|
|
"blocked": 0,
|
|
},
|
|
)
|
|
|
|
|
|
def _summarize_gateway_mcp(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
|
by_agent: dict[str, dict[str, Any]] = {}
|
|
by_tool: dict[str, dict[str, Any]] = {}
|
|
by_scope: dict[str, dict[str, Any]] = {}
|
|
success_count = 0
|
|
failed_count = 0
|
|
blocked_count = 0
|
|
first_class_count = 0
|
|
bridge_count = 0
|
|
policy_enforced_count = 0
|
|
approval_executor_count = 0
|
|
|
|
for row in rows:
|
|
gate_result = row.get("gate_result") if isinstance(row.get("gate_result"), dict) else {}
|
|
gateway_path = str(gate_result.get("gateway_path") or "")
|
|
schema_version = str(gate_result.get("schema_version") or "")
|
|
policy_enforced = _as_bool(gate_result.get("policy_enforced"))
|
|
required_scope = str(gate_result.get("required_scope") or "unknown")
|
|
status = str(row.get("result_status") or "unknown").lower()
|
|
is_blocked = status == "blocked" or row.get("block_gate") is not None
|
|
is_success = status == "success"
|
|
is_failed = status == "failed"
|
|
agent_id = str(row.get("agent_id") or "unknown")
|
|
tool_name = str(row.get("tool_name") or "unknown")
|
|
|
|
if gateway_path == "awooop_mcp_gateway" and policy_enforced:
|
|
first_class_count += 1
|
|
if schema_version == "legacy_mcp_bridge_v1" or policy_enforced is False:
|
|
bridge_count += 1
|
|
if policy_enforced:
|
|
policy_enforced_count += 1
|
|
if agent_id == "approval_executor":
|
|
approval_executor_count += 1
|
|
if is_success:
|
|
success_count += 1
|
|
if is_failed:
|
|
failed_count += 1
|
|
if is_blocked:
|
|
blocked_count += 1
|
|
|
|
for bucket in (
|
|
_counter_bucket(by_agent, agent_id, label="agent_id"),
|
|
_counter_bucket(by_tool, tool_name, label="tool_name"),
|
|
_counter_bucket(by_scope, required_scope, label="required_scope"),
|
|
):
|
|
bucket["total"] += 1
|
|
if is_success:
|
|
bucket["success"] += 1
|
|
if is_failed:
|
|
bucket["failed"] += 1
|
|
if is_blocked:
|
|
bucket["blocked"] += 1
|
|
|
|
blockers: list[str] = []
|
|
if blocked_count:
|
|
stage = "gateway_blocked"
|
|
stage_status = "blocked"
|
|
blockers.append("mcp_gateway_blocked")
|
|
elif first_class_count and failed_count:
|
|
stage = "provider_failed_after_gateway"
|
|
stage_status = "failed"
|
|
blockers.append("provider_failed_after_gateway")
|
|
elif success_count:
|
|
stage = "gateway_execution_succeeded"
|
|
stage_status = "success"
|
|
elif bridge_count and not first_class_count:
|
|
stage = "legacy_bridge_only"
|
|
stage_status = "observed"
|
|
blockers.append("legacy_bridge_not_policy_enforced")
|
|
else:
|
|
stage = "no_gateway_records"
|
|
stage_status = "missing"
|
|
|
|
return {
|
|
"total": len(rows),
|
|
"success": success_count,
|
|
"failed": failed_count,
|
|
"blocked": blocked_count,
|
|
"first_class_total": first_class_count,
|
|
"legacy_bridge_total": bridge_count,
|
|
"policy_enforced_total": policy_enforced_count,
|
|
"approval_executor_total": approval_executor_count,
|
|
"stage": stage,
|
|
"stage_status": stage_status,
|
|
"needs_human": bool(blocked_count or (first_class_count and failed_count)),
|
|
"blockers": blockers,
|
|
"by_agent": list(by_agent.values()),
|
|
"by_tool": list(by_tool.values()),
|
|
"by_scope": list(by_scope.values()),
|
|
}
|
|
|
|
|
|
async def fetch_truth_chain(source_id: str, project_id: str = "awoooi") -> dict[str, Any]:
|
|
"""Return a read-only truth chain for an incident, drift report, or run id."""
|
|
async with get_db_context(project_id) as db:
|
|
incident = await _fetch_one(
|
|
db,
|
|
"""
|
|
SELECT
|
|
incident_id,
|
|
project_id,
|
|
status::text AS status,
|
|
severity::text AS severity,
|
|
alertname,
|
|
alert_category,
|
|
notification_type,
|
|
created_at,
|
|
updated_at,
|
|
resolved_at,
|
|
verification_result,
|
|
frequency_snapshot,
|
|
signals,
|
|
decision_chain
|
|
FROM incidents
|
|
WHERE incident_id = :source_id
|
|
AND (project_id = :project_id OR project_id IS NULL)
|
|
""",
|
|
{"source_id": source_id, "project_id": project_id},
|
|
)
|
|
|
|
drift = await _fetch_one(
|
|
db,
|
|
"""
|
|
SELECT
|
|
report_id,
|
|
namespace,
|
|
status,
|
|
triggered_by,
|
|
high_count,
|
|
medium_count,
|
|
info_count,
|
|
scanned_at,
|
|
created_at,
|
|
resolved_at,
|
|
interpretation,
|
|
items,
|
|
narrative_text
|
|
FROM drift_reports
|
|
WHERE report_id = :source_id
|
|
""",
|
|
{"source_id": source_id},
|
|
)
|
|
|
|
runs = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
run_id,
|
|
project_id,
|
|
agent_id,
|
|
state,
|
|
trigger_type,
|
|
trigger_ref,
|
|
is_shadow,
|
|
step_count,
|
|
created_at,
|
|
started_at,
|
|
completed_at,
|
|
error_code,
|
|
error_detail
|
|
FROM awooop_run_state
|
|
WHERE project_id = :project_id
|
|
AND (run_id::text = :source_id OR trigger_ref = :source_id)
|
|
ORDER BY created_at DESC
|
|
LIMIT :limit
|
|
""",
|
|
{"source_id": source_id, "project_id": project_id, "limit": _MAX_ROWS},
|
|
)
|
|
|
|
approvals: list[dict[str, Any]] = []
|
|
evidence_rows: list[dict[str, Any]] = []
|
|
timeline_events: list[dict[str, Any]] = []
|
|
legacy_mcp_rows: list[dict[str, Any]] = []
|
|
automation_ops: list[dict[str, Any]] = []
|
|
auto_repair_executions: list[dict[str, Any]] = []
|
|
km_entries: list[dict[str, Any]] = []
|
|
inbound_rows: list[dict[str, Any]] = []
|
|
gateway_mcp_rows: list[dict[str, Any]] = []
|
|
outbound_rows: list[dict[str, Any]] = []
|
|
|
|
if incident is not None:
|
|
incident_id = str(incident["incident_id"])
|
|
approvals = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
id,
|
|
incident_id,
|
|
status::text AS status,
|
|
risk_level::text AS risk_level,
|
|
action,
|
|
description,
|
|
hit_count,
|
|
created_at,
|
|
updated_at,
|
|
resolved_at,
|
|
matched_playbook_id,
|
|
extra_metadata,
|
|
decision_fusion_details
|
|
FROM approval_records
|
|
WHERE incident_id = :incident_id
|
|
ORDER BY created_at DESC
|
|
LIMIT :limit
|
|
""",
|
|
{"incident_id": incident_id, "limit": _MAX_ROWS},
|
|
)
|
|
approval_ids = _approval_ids(approvals)
|
|
evidence_rows = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
id,
|
|
incident_id,
|
|
matched_playbook_id,
|
|
collected_at,
|
|
collection_duration_ms,
|
|
sensors_attempted,
|
|
sensors_succeeded,
|
|
evidence_summary,
|
|
metrics_snapshot,
|
|
mcp_health,
|
|
verification_result,
|
|
pre_execution_state,
|
|
post_execution_state,
|
|
self_healing_score,
|
|
self_healing_detail
|
|
FROM incident_evidence
|
|
WHERE incident_id = :incident_id
|
|
ORDER BY collected_at DESC
|
|
LIMIT :limit
|
|
""",
|
|
{"incident_id": incident_id, "limit": _MAX_ROWS},
|
|
)
|
|
timeline_events = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
id,
|
|
event_type,
|
|
status,
|
|
title,
|
|
description,
|
|
actor,
|
|
actor_role,
|
|
risk_level,
|
|
approval_id,
|
|
incident_id,
|
|
created_at
|
|
FROM timeline_events
|
|
WHERE incident_id = :incident_id
|
|
OR approval_id = ANY(:approval_ids)
|
|
ORDER BY created_at ASC
|
|
LIMIT :limit
|
|
""",
|
|
{
|
|
"incident_id": incident_id,
|
|
"approval_ids": approval_ids or ["__none__"],
|
|
"limit": _MAX_ROWS,
|
|
},
|
|
)
|
|
legacy_mcp_rows = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
id,
|
|
session_id,
|
|
flywheel_node,
|
|
mcp_server,
|
|
tool_name,
|
|
duration_ms,
|
|
success,
|
|
error_message,
|
|
incident_id,
|
|
agent_role,
|
|
created_at
|
|
FROM mcp_audit_log
|
|
WHERE incident_id = :incident_id
|
|
ORDER BY created_at ASC
|
|
LIMIT :limit
|
|
""",
|
|
{"incident_id": incident_id, "limit": _MAX_ROWS},
|
|
)
|
|
automation_ops = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
op_id,
|
|
operation_type,
|
|
status,
|
|
incident_id,
|
|
run_id,
|
|
actor,
|
|
dry_run_result,
|
|
error,
|
|
duration_ms,
|
|
tags,
|
|
input ->> 'action' AS input_action,
|
|
input ->> 'executor' AS input_executor,
|
|
input ->> 'execution_backend' AS input_execution_backend,
|
|
input ->> 'playbook_id' AS input_playbook_id,
|
|
input ->> 'playbook_path' AS input_playbook_path,
|
|
input ->> 'ansible_playbook_path' AS input_ansible_playbook_path,
|
|
input ->> 'check_mode' AS input_check_mode,
|
|
input ->> 'not_used_reason' AS input_not_used_reason,
|
|
output ->> 'action' AS output_action,
|
|
output ->> 'reason' AS output_reason,
|
|
output ->> 'executor' AS output_executor,
|
|
output ->> 'execution_backend' AS output_execution_backend,
|
|
output ->> 'playbook_id' AS output_playbook_id,
|
|
output ->> 'playbook_path' AS output_playbook_path,
|
|
output ->> 'ansible_playbook_path' AS output_ansible_playbook_path,
|
|
output ->> 'check_mode' AS output_check_mode,
|
|
output ->> 'not_used_reason' AS output_not_used_reason,
|
|
created_at
|
|
FROM automation_operation_log
|
|
WHERE incident_id::text = :incident_id
|
|
OR coalesce(input::text, '') LIKE :needle
|
|
OR coalesce(output::text, '') LIKE :needle
|
|
OR coalesce(array_to_string(tags, ','), '') LIKE :needle
|
|
ORDER BY created_at DESC
|
|
LIMIT :limit
|
|
""",
|
|
{"incident_id": incident_id, "needle": f"%{incident_id}%", "limit": _MAX_ROWS},
|
|
)
|
|
auto_repair_executions = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
id,
|
|
incident_id,
|
|
playbook_id,
|
|
playbook_name,
|
|
success,
|
|
executed_steps,
|
|
error_message,
|
|
triggered_by,
|
|
similarity_score,
|
|
risk_level,
|
|
execution_time_ms,
|
|
created_at
|
|
FROM auto_repair_executions
|
|
WHERE incident_id = :incident_id
|
|
ORDER BY created_at DESC
|
|
LIMIT :limit
|
|
""",
|
|
{"incident_id": incident_id, "limit": _MAX_ROWS},
|
|
)
|
|
km_entries = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
id,
|
|
title,
|
|
entry_type::text AS entry_type,
|
|
status::text AS status,
|
|
related_incident_id,
|
|
related_approval_id,
|
|
created_at
|
|
FROM knowledge_entries
|
|
WHERE related_incident_id = :incident_id
|
|
ORDER BY created_at DESC
|
|
LIMIT :limit
|
|
""",
|
|
{"incident_id": incident_id, "limit": _MAX_ROWS},
|
|
)
|
|
|
|
drift_repeats: dict[str, Any] = {
|
|
"occurrences_12h": 0,
|
|
"first_scanned_at": None,
|
|
"last_scanned_at": None,
|
|
"reports": [],
|
|
}
|
|
if drift is not None:
|
|
recent_drift_reports = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
report_id,
|
|
namespace,
|
|
status,
|
|
scanned_at,
|
|
created_at,
|
|
items,
|
|
interpretation,
|
|
narrative_text
|
|
FROM drift_reports
|
|
WHERE created_at > now() - interval '24 hours'
|
|
AND namespace = :namespace
|
|
ORDER BY scanned_at DESC
|
|
LIMIT 200
|
|
""",
|
|
{"namespace": drift["namespace"]},
|
|
)
|
|
drift_repeats = build_drift_repeat_state(drift, recent_drift_reports)
|
|
|
|
gateway_mcp_rows = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
call_id,
|
|
project_id,
|
|
run_id,
|
|
trace_id,
|
|
agent_id,
|
|
tool_name,
|
|
gate_result,
|
|
result_status,
|
|
block_gate,
|
|
block_reason,
|
|
latency_ms,
|
|
created_at
|
|
FROM awooop_mcp_gateway_audit
|
|
WHERE project_id = :project_id
|
|
AND (
|
|
trace_id = :source_id
|
|
OR run_id::text = :source_id
|
|
)
|
|
ORDER BY created_at ASC
|
|
LIMIT :limit
|
|
""",
|
|
{"source_id": source_id, "project_id": project_id, "limit": _MAX_ROWS},
|
|
)
|
|
incident_fingerprints = _incident_fingerprints(incident)
|
|
fingerprint_needle = (
|
|
f"%{incident_fingerprints[0]}%"
|
|
if incident_fingerprints
|
|
else ""
|
|
)
|
|
fingerprint_value = incident_fingerprints[0] if incident_fingerprints else ""
|
|
inbound_rows = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
event_id,
|
|
project_id,
|
|
channel_type,
|
|
provider_event_id,
|
|
platform_subject_id,
|
|
channel_user_id,
|
|
channel_chat_id,
|
|
run_id,
|
|
content_type,
|
|
content_hash,
|
|
content_preview,
|
|
content_redacted,
|
|
redaction_version,
|
|
source_envelope,
|
|
attachment_sha256,
|
|
is_duplicate,
|
|
provider_ts,
|
|
received_at
|
|
FROM awooop_conversation_event
|
|
WHERE project_id = :project_id
|
|
AND (
|
|
run_id::text = :source_id
|
|
OR provider_event_id = :source_id
|
|
OR content_preview ILIKE :source_needle
|
|
OR coalesce(source_envelope #> '{source_refs,event_ids}', '[]'::jsonb) ? :source_id
|
|
OR coalesce(source_envelope #> '{source_refs,incident_ids}', '[]'::jsonb) ? :source_id
|
|
OR coalesce(source_envelope #> '{source_refs,approval_ids}', '[]'::jsonb) ? :source_id
|
|
OR coalesce(source_envelope #> '{source_refs,alert_ids}', '[]'::jsonb) ? :source_id
|
|
OR coalesce(source_envelope #> '{source_refs,sentry_issue_ids}', '[]'::jsonb) ? :source_id
|
|
OR coalesce(source_envelope #> '{source_refs,signoz_alerts}', '[]'::jsonb) ? :source_id
|
|
OR (
|
|
:fingerprint_needle != ''
|
|
AND (
|
|
provider_event_id ILIKE :fingerprint_needle
|
|
OR content_preview ILIKE :fingerprint_needle
|
|
OR coalesce(source_envelope #> '{source_refs,fingerprints}', '[]'::jsonb) ? :fingerprint_value
|
|
)
|
|
)
|
|
)
|
|
ORDER BY received_at DESC
|
|
LIMIT :limit
|
|
""",
|
|
{
|
|
"source_id": source_id,
|
|
"project_id": project_id,
|
|
"source_needle": f"%{source_id}%",
|
|
"fingerprint_needle": fingerprint_needle,
|
|
"fingerprint_value": fingerprint_value,
|
|
"limit": _MAX_ROWS,
|
|
},
|
|
)
|
|
outbound_rows = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
message_id,
|
|
project_id,
|
|
run_id,
|
|
channel_type,
|
|
message_type,
|
|
content_hash,
|
|
content_preview,
|
|
content_redacted,
|
|
redaction_version,
|
|
source_envelope,
|
|
provider_message_id,
|
|
send_status,
|
|
queued_at,
|
|
sent_at,
|
|
triggered_by_state
|
|
FROM awooop_outbound_message
|
|
WHERE project_id = :project_id
|
|
AND (
|
|
run_id::text = :source_id
|
|
OR content_preview ILIKE :needle
|
|
OR coalesce(source_envelope #> '{source_refs,incident_ids}', '[]'::jsonb) ? :source_id
|
|
OR coalesce(source_envelope #> '{source_refs,code_refs}', '[]'::jsonb) ? :source_id
|
|
)
|
|
ORDER BY queued_at DESC
|
|
LIMIT :limit
|
|
""",
|
|
{
|
|
"source_id": source_id,
|
|
"project_id": project_id,
|
|
"needle": f"%{source_id}%",
|
|
"limit": _MAX_ROWS,
|
|
},
|
|
)
|
|
|
|
source_type = _source_type(source_id, incident, drift)
|
|
legacy_mcp_summary = _summarize_mcp(legacy_mcp_rows)
|
|
gateway_mcp_summary = _summarize_gateway_mcp(gateway_mcp_rows)
|
|
truth_status = _truth_status(
|
|
incident=incident,
|
|
approvals=approvals,
|
|
evidence_rows=evidence_rows,
|
|
automation_ops=automation_ops,
|
|
drift=drift,
|
|
drift_repeat_count=int(drift_repeats.get("occurrences_12h") or 0),
|
|
gateway_mcp_total=len(gateway_mcp_rows),
|
|
legacy_mcp_total=legacy_mcp_summary["total"],
|
|
outbound_visible_total=len(outbound_rows),
|
|
inbound_visible_total=len(inbound_rows),
|
|
auto_repair_executions=auto_repair_executions,
|
|
)
|
|
if incident is None and drift is None and not runs and gateway_mcp_rows:
|
|
truth_status = {
|
|
"current_stage": gateway_mcp_summary["stage"],
|
|
"stage_status": gateway_mcp_summary["stage_status"],
|
|
"needs_human": gateway_mcp_summary["needs_human"],
|
|
"blockers": gateway_mcp_summary["blockers"],
|
|
}
|
|
reconciliation = build_incident_reconciliation(
|
|
incident=incident,
|
|
approvals=approvals,
|
|
evidence_rows=evidence_rows,
|
|
automation_ops=automation_ops,
|
|
auto_repair_executions=auto_repair_executions,
|
|
timeline_events=timeline_events,
|
|
)
|
|
|
|
evidence_totals = {
|
|
"records": len(evidence_rows),
|
|
"sensors_attempted": sum(int(row.get("sensors_attempted") or 0) for row in evidence_rows),
|
|
"sensors_succeeded": sum(int(row.get("sensors_succeeded") or 0) for row in evidence_rows),
|
|
"failed_tools": _failed_mcp_tools(evidence_rows),
|
|
}
|
|
automation_quality = build_automation_quality(
|
|
incident=incident,
|
|
approvals=approvals,
|
|
evidence_rows=evidence_rows,
|
|
automation_ops=automation_ops,
|
|
auto_repair_executions=auto_repair_executions,
|
|
gateway_mcp_summary=gateway_mcp_summary,
|
|
legacy_mcp_summary=legacy_mcp_summary,
|
|
outbound_rows=outbound_rows,
|
|
km_entries=km_entries,
|
|
timeline_events=timeline_events,
|
|
)
|
|
|
|
result = {
|
|
"project_id": project_id,
|
|
"source_id": source_id,
|
|
"source_type": source_type,
|
|
"found": (
|
|
incident is not None
|
|
or drift is not None
|
|
or bool(runs)
|
|
or bool(gateway_mcp_rows)
|
|
or bool(inbound_rows)
|
|
or bool(outbound_rows)
|
|
),
|
|
"truth_status": truth_status,
|
|
"linked_ids": {
|
|
"incident_id": incident.get("incident_id") if incident else None,
|
|
"approval_ids": _approval_ids(approvals),
|
|
"run_ids": [row["run_id"] for row in runs],
|
|
"drift_report_id": drift.get("report_id") if drift else None,
|
|
"operation_ids": _operation_ids(automation_ops),
|
|
"auto_repair_execution_ids": _auto_repair_ids(auto_repair_executions),
|
|
"conversation_event_ids": [row["event_id"] for row in inbound_rows],
|
|
},
|
|
"incident": incident,
|
|
"drift": {
|
|
"report": drift,
|
|
"repeat_state": drift_repeats,
|
|
},
|
|
"approvals": approvals,
|
|
"evidence": {
|
|
"summary": evidence_totals,
|
|
"records": evidence_rows,
|
|
},
|
|
"mcp": {
|
|
"awooop_gateway": {
|
|
**gateway_mcp_summary,
|
|
"records": gateway_mcp_rows,
|
|
},
|
|
"legacy": {
|
|
**legacy_mcp_summary,
|
|
"records": legacy_mcp_rows,
|
|
},
|
|
},
|
|
"execution": {
|
|
"automation_operation_log": automation_ops,
|
|
"auto_repair_executions": auto_repair_executions,
|
|
"ansible": build_ansible_truth(automation_ops, incident=incident, drift=drift),
|
|
},
|
|
"automation_quality": automation_quality,
|
|
"reconciliation": reconciliation,
|
|
"learning": {
|
|
"knowledge_entries": km_entries,
|
|
},
|
|
"timeline_events": timeline_events,
|
|
"channel": {
|
|
"inbound_events_visible": len(inbound_rows),
|
|
"inbound_events": inbound_rows,
|
|
"outbound_messages_visible": len(outbound_rows),
|
|
"outbound_messages": outbound_rows,
|
|
"visibility_note": (
|
|
"If inbound is zero while Alertmanager fired, or outbound is zero while "
|
|
"Telegram delivered a card, channel mirrors or RLS project context are "
|
|
"not reliable sources of truth yet."
|
|
),
|
|
},
|
|
}
|
|
logger.info(
|
|
"awooop_truth_chain_fetched",
|
|
project_id=project_id,
|
|
source_id=source_id,
|
|
source_type=source_type,
|
|
current_stage=truth_status["current_stage"],
|
|
)
|
|
return result
|
|
|
|
|
|
async def fetch_automation_quality_summary(
|
|
*,
|
|
project_id: str = "awoooi",
|
|
hours: int = 24,
|
|
limit: int = 200,
|
|
) -> dict[str, Any]:
|
|
"""Return a recent incident-level quality summary for the automation flywheel."""
|
|
bounded_hours = max(1, min(int(hours), 168))
|
|
bounded_limit = max(1, min(int(limit), 500))
|
|
cutoff = datetime.now(UTC) - timedelta(hours=bounded_hours)
|
|
|
|
async with get_db_context(project_id) as db:
|
|
incidents = await _fetch_all(
|
|
db,
|
|
"""
|
|
SELECT
|
|
incident_id,
|
|
project_id,
|
|
status::text AS status,
|
|
severity::text AS severity,
|
|
alertname,
|
|
alert_category,
|
|
notification_type,
|
|
created_at,
|
|
updated_at,
|
|
resolved_at,
|
|
verification_result
|
|
FROM incidents
|
|
WHERE (project_id = :project_id OR project_id IS NULL)
|
|
AND created_at >= :cutoff
|
|
ORDER BY created_at DESC
|
|
LIMIT :limit
|
|
""",
|
|
{
|
|
"project_id": project_id,
|
|
"cutoff": cutoff,
|
|
"limit": bounded_limit,
|
|
},
|
|
)
|
|
|
|
semaphore = asyncio.Semaphore(_QUALITY_SUMMARY_CONCURRENCY)
|
|
|
|
async def _quality_record(incident: dict[str, Any]) -> dict[str, Any] | None:
|
|
incident_id = str(incident.get("incident_id") or "")
|
|
if not incident_id:
|
|
return None
|
|
async with semaphore:
|
|
truth_chain = await fetch_truth_chain(source_id=incident_id, project_id=project_id)
|
|
return {
|
|
"incident": truth_chain.get("incident") or incident,
|
|
"truth_status": truth_chain.get("truth_status") or {},
|
|
"automation_quality": truth_chain.get("automation_quality") or {},
|
|
}
|
|
|
|
records = [
|
|
record
|
|
for record in await asyncio.gather(*(_quality_record(incident) for incident in incidents))
|
|
if record is not None
|
|
]
|
|
|
|
summary = summarize_automation_quality_records(
|
|
project_id=project_id,
|
|
window_hours=bounded_hours,
|
|
records=records,
|
|
limit=bounded_limit,
|
|
)
|
|
logger.info(
|
|
"awooop_automation_quality_summary_fetched",
|
|
project_id=project_id,
|
|
window_hours=bounded_hours,
|
|
incident_total=summary["incident_total"],
|
|
evaluated_total=summary["evaluated_total"],
|
|
can_claim_full_auto_repair=summary["production_claim"]["can_claim_full_auto_repair"],
|
|
)
|
|
return summary
|