771 lines
26 KiB
Python
771 lines
26 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 json
|
|
from datetime import date, datetime
|
|
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"}
|
|
|
|
|
|
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 _build_reconciliation(
|
|
*,
|
|
incident: dict[str, Any] | None,
|
|
approvals: list[dict[str, Any]],
|
|
evidence_rows: list[dict[str, Any]],
|
|
automation_ops: list[dict[str, Any]],
|
|
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)
|
|
executed_ops = [
|
|
row
|
|
for row in automation_ops
|
|
if str(row.get("status") or "").lower()
|
|
in {"success", "completed", "executed"}
|
|
]
|
|
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"})
|
|
):
|
|
add(
|
|
"incident_open_after_approval_resolved",
|
|
"high",
|
|
"Approval reached a terminal state while the incident is still open.",
|
|
)
|
|
|
|
if approval_status == "APPROVED" and not automation_ops:
|
|
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 executed_ops
|
|
):
|
|
add(
|
|
"approval_no_action_without_execution",
|
|
"high",
|
|
"Approval resolved to NO_ACTION and no executor produced a successful operation.",
|
|
)
|
|
|
|
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),
|
|
"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,
|
|
) -> 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 not None:
|
|
incident_status = str(incident.get("status") or "unknown")
|
|
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()
|
|
if any(status in {"PENDING", "WAITING_APPROVAL"} for status in approval_statuses):
|
|
stage = "approval_required"
|
|
stage_status = "waiting"
|
|
needs_human = True
|
|
elif "APPROVED" in approval_statuses and not automation_ops:
|
|
if "NO_ACTION" in approval_actions:
|
|
stage = "manual_required"
|
|
stage_status = "blocked"
|
|
needs_human = True
|
|
blockers.append("approval_resolved_no_action_without_execution")
|
|
else:
|
|
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 automation_ops}
|
|
if op_statuses:
|
|
if op_statuses & {"success", "completed"}:
|
|
stage = "execution_succeeded"
|
|
stage_status = "success"
|
|
elif op_statuses & {"failed", "error"}:
|
|
stage = "execution_failed"
|
|
stage_status = "error"
|
|
needs_human = True
|
|
else:
|
|
stage = "execution_started"
|
|
stage_status = "running"
|
|
|
|
if incident_status == "INVESTIGATING" and automation_ops == [] and approvals:
|
|
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 _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()),
|
|
}
|
|
|
|
|
|
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,
|
|
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]] = []
|
|
km_entries: 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 ->> '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 ->> '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},
|
|
)
|
|
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},
|
|
)
|
|
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
|
|
)
|
|
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)
|
|
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),
|
|
)
|
|
reconciliation = _build_reconciliation(
|
|
incident=incident,
|
|
approvals=approvals,
|
|
evidence_rows=evidence_rows,
|
|
automation_ops=automation_ops,
|
|
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),
|
|
}
|
|
|
|
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),
|
|
"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),
|
|
},
|
|
"incident": incident,
|
|
"drift": {
|
|
"report": drift,
|
|
"repeat_state": drift_repeats,
|
|
},
|
|
"approvals": approvals,
|
|
"evidence": {
|
|
"summary": evidence_totals,
|
|
"records": evidence_rows,
|
|
},
|
|
"mcp": {
|
|
"awooop_gateway": {
|
|
"total": len(gateway_mcp_rows),
|
|
"records": gateway_mcp_rows,
|
|
},
|
|
"legacy": {
|
|
**legacy_mcp_summary,
|
|
"records": legacy_mcp_rows,
|
|
},
|
|
},
|
|
"execution": {
|
|
"automation_operation_log": automation_ops,
|
|
"ansible": build_ansible_truth(automation_ops, incident=incident, drift=drift),
|
|
},
|
|
"reconciliation": reconciliation,
|
|
"learning": {
|
|
"knowledge_entries": km_entries,
|
|
},
|
|
"timeline_events": timeline_events,
|
|
"channel": {
|
|
"outbound_messages_visible": len(outbound_rows),
|
|
"outbound_messages": outbound_rows,
|
|
"visibility_note": (
|
|
"If this is zero while Telegram delivered a card, the outbound mirror "
|
|
"or RLS project context is not a reliable source 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
|