feat(awooop): expose ai automation log taxonomy

apps/api/src/services/ai_agent_autonomous_runtime_control.py

@@ -151,6 +151,451 @@ def _status_counts(
     }
 
 
+def _trace_stage(
+    *,
+    stage_id: str,
+    display_name: str,
+    source_tables: list[str],
+    total: int,
+    recent: int,
+    required_for_closed_loop: bool,
+    feeds_learning: bool,
+    public_safe: bool = True,
+    next_action_if_missing: str | None = None,
+) -> dict[str, Any]:
+    present = total > 0
+    return {
+        "stage_id": stage_id,
+        "display_name": display_name,
+        "source_tables": source_tables,
+        "recorded": present,
+        "record_quality": "recorded" if present else "missing",
+        "total": max(0, total),
+        "recent": max(0, recent),
+        "required_for_closed_loop": required_for_closed_loop,
+        "feeds_learning": feeds_learning,
+        "public_safe": public_safe,
+        "next_action_if_missing": None if present else next_action_if_missing,
+    }
+
+
+def _trace_total(summary: Mapping[str, Any] | None, *operation_types: str) -> int:
+    if not isinstance(summary, Mapping):
+        return 0
+    if not operation_types:
+        return _int_value(summary.get("total"))
+    return sum(
+        _int_value((summary.get(operation_type) or {}).get("total"))
+        for operation_type in operation_types
+    )
+
+
+def _trace_recent(summary: Mapping[str, Any] | None, *operation_types: str) -> int:
+    if not isinstance(summary, Mapping):
+        return 0
+    if not operation_types:
+        return _int_value(summary.get("recent"))
+    return sum(
+        _int_value((summary.get(operation_type) or {}).get("recent"))
+        for operation_type in operation_types
+    )
+
+
+def _build_trace_ledger(
+    *,
+    operation_summary: Mapping[str, Any],
+    auto_repair_summary: Mapping[str, Any],
+    verifier_summary: Mapping[str, Any],
+    km_summary: Mapping[str, Any],
+    telegram_summary: Mapping[str, Any],
+    mcp_gateway_summary: Mapping[str, Any],
+    legacy_mcp_summary: Mapping[str, Any],
+    service_log_summary: Mapping[str, Any],
+    executor_log_summary: Mapping[str, Any],
+    timeline_summary: Mapping[str, Any],
+    playbook_trust_summary: Mapping[str, Any],
+    latest_flow_closure: Mapping[str, Any],
+    loop_ledger: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Build the full public-safe AI automation trace ledger."""
+
+    mcp_total = _trace_total(mcp_gateway_summary) + _trace_total(legacy_mcp_summary)
+    mcp_recent = _trace_recent(mcp_gateway_summary) + _trace_recent(legacy_mcp_summary)
+    stages = [
+        _trace_stage(
+            stage_id="mcp_context",
+            display_name="MCP sensor / tool context",
+            source_tables=["awooop_mcp_gateway_audit", "mcp_audit_log"],
+            total=mcp_total,
+            recent=mcp_recent,
+            required_for_closed_loop=False,
+            feeds_learning=True,
+            next_action_if_missing="record_mcp_gateway_or_legacy_mcp_audit_for_every_ai_decision",
+        ),
+        _trace_stage(
+            stage_id="service_log_evidence",
+            display_name="Sanitized service / package log evidence",
+            source_tables=["incident_evidence.recent_logs", "incident_evidence.evidence_summary"],
+            total=_trace_total(service_log_summary),
+            recent=_trace_recent(service_log_summary),
+            required_for_closed_loop=False,
+            feeds_learning=True,
+            next_action_if_missing="collect_sanitized_service_log_evidence_before_ai_decision",
+        ),
+        _trace_stage(
+            stage_id="candidate",
+            display_name="AI candidate / playbook match",
+            source_tables=["automation_operation_log"],
+            total=_trace_total(operation_summary, "ansible_candidate_matched"),
+            recent=_trace_recent(operation_summary, "ansible_candidate_matched"),
+            required_for_closed_loop=True,
+            feeds_learning=True,
+            next_action_if_missing="candidate_backfill_worker_enqueue_allowlisted_playbook",
+        ),
+        _trace_stage(
+            stage_id="check_mode",
+            display_name="No-write check-mode / dry-run",
+            source_tables=["automation_operation_log"],
+            total=_trace_total(operation_summary, "ansible_check_mode_executed"),
+            recent=_trace_recent(operation_summary, "ansible_check_mode_executed"),
+            required_for_closed_loop=True,
+            feeds_learning=True,
+            next_action_if_missing="ansible_check_mode_worker_claims_candidate",
+        ),
+        _trace_stage(
+            stage_id="executor_log_projection",
+            display_name="Executor stdout / stderr / dry-run projection",
+            source_tables=[
+                "automation_operation_log.output",
+                "automation_operation_log.error",
+                "automation_operation_log.stderr_feed_back",
+                "automation_operation_log.dry_run_result",
+            ],
+            total=_trace_total(executor_log_summary),
+            recent=_trace_recent(executor_log_summary),
+            required_for_closed_loop=False,
+            feeds_learning=True,
+            next_action_if_missing="persist_sanitized_executor_log_projection_for_failed_or_applied_actions",
+        ),
+        _trace_stage(
+            stage_id="controlled_apply",
+            display_name="Controlled apply execution",
+            source_tables=["automation_operation_log"],
+            total=_trace_total(operation_summary, "ansible_apply_executed"),
+            recent=_trace_recent(operation_summary, "ansible_apply_executed"),
+            required_for_closed_loop=True,
+            feeds_learning=True,
+            next_action_if_missing="controlled_apply_worker_waits_for_check_mode_success",
+        ),
+        _trace_stage(
+            stage_id="auto_repair_execution_receipt",
+            display_name="Auto-repair execution receipt",
+            source_tables=["auto_repair_executions"],
+            total=_trace_total(auto_repair_summary),
+            recent=_trace_recent(auto_repair_summary),
+            required_for_closed_loop=True,
+            feeds_learning=True,
+            next_action_if_missing="receipt_backfill_records_auto_repair_execution",
+        ),
+        _trace_stage(
+            stage_id="post_apply_verifier",
+            display_name="Post-apply verifier evidence",
+            source_tables=["incident_evidence"],
+            total=_trace_total(verifier_summary),
+            recent=_trace_recent(verifier_summary),
+            required_for_closed_loop=True,
+            feeds_learning=True,
+            next_action_if_missing="post_apply_verifier_writes_incident_evidence",
+        ),
+        _trace_stage(
+            stage_id="rag_km_learning",
+            display_name="RAG / KM / PlayBook learning writeback",
+            source_tables=["knowledge_entries"],
+            total=_trace_total(km_summary),
+            recent=_trace_recent(km_summary),
+            required_for_closed_loop=True,
+            feeds_learning=True,
+            next_action_if_missing="hermes_writes_km_playbook_trust_candidate",
+        ),
+        _trace_stage(
+            stage_id="playbook_trust",
+            display_name="PlayBook trust / success-failure learning",
+            source_tables=[
+                "playbooks.trust_score",
+                "playbooks.success_count",
+                "playbooks.failure_count",
+                "playbooks.review_required",
+            ],
+            total=_trace_total(playbook_trust_summary),
+            recent=_trace_recent(playbook_trust_summary),
+            required_for_closed_loop=False,
+            feeds_learning=True,
+            next_action_if_missing="write_playbook_trust_delta_after_verified_execution",
+        ),
+        _trace_stage(
+            stage_id="timeline_projection",
+            display_name="Operator timeline projection",
+            source_tables=["timeline_events"],
+            total=_trace_total(timeline_summary),
+            recent=_trace_recent(timeline_summary),
+            required_for_closed_loop=False,
+            feeds_learning=True,
+            next_action_if_missing="project_ai_runtime_stage_to_timeline_events",
+        ),
+        _trace_stage(
+            stage_id="telegram_receipt",
+            display_name="Telegram Gateway receipt",
+            source_tables=["awooop_outbound_message"],
+            total=_trace_total(telegram_summary),
+            recent=_trace_recent(telegram_summary),
+            required_for_closed_loop=True,
+            feeds_learning=False,
+            next_action_if_missing="live_apply_gateway_sends_controlled_apply_result_receipt",
+        ),
+    ]
+    required = [stage for stage in stages if stage["required_for_closed_loop"]]
+    missing_required = [
+        str(stage["stage_id"])
+        for stage in required
+        if stage["recorded"] is not True
+    ]
+    recorded_count = sum(1 for stage in stages if stage["recorded"] is True)
+    return {
+        "schema_version": "ai_agent_autonomous_trace_ledger_v1",
+        "purpose": (
+            "把 AI 自動化每個節點的 public-safe receipt 收斂成同一份 ledger；"
+            "這些紀錄是後續 RAG、KM、PlayBook trust 與報告學習的依據。"
+        ),
+        "latest_flow_closed": latest_flow_closure.get("closed") is True,
+        "latest_loop_closed": loop_ledger.get("closed") is True,
+        "stage_count": len(stages),
+        "recorded_stage_count": recorded_count,
+        "required_stage_count": len(required),
+        "missing_required_stage_ids": missing_required,
+        "learning_source_stage_ids": [
+            str(stage["stage_id"])
+            for stage in stages
+            if stage["feeds_learning"] is True
+        ],
+        "public_safety": {
+            "reads_raw_sessions": False,
+            "stores_secret_values": False,
+            "stores_unredacted_telegram_payload": False,
+            "stores_internal_reasoning": False,
+        },
+        "stages": stages,
+    }
+
+
+def _build_log_integration_taxonomy(
+    *,
+    operation_summary: Mapping[str, Any],
+    auto_repair_summary: Mapping[str, Any],
+    verifier_summary: Mapping[str, Any],
+    km_summary: Mapping[str, Any],
+    telegram_summary: Mapping[str, Any],
+    mcp_gateway_summary: Mapping[str, Any],
+    legacy_mcp_summary: Mapping[str, Any],
+    service_log_summary: Mapping[str, Any],
+    executor_log_summary: Mapping[str, Any],
+    timeline_summary: Mapping[str, Any],
+    playbook_trust_summary: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Expose how logs are normalized, labeled, grouped, and fed to agents."""
+
+    operation_total = sum(_trace_total(operation_summary, item) for item in _EXECUTOR_OPERATION_TYPES)
+    operation_recent = sum(_trace_recent(operation_summary, item) for item in _EXECUTOR_OPERATION_TYPES)
+    source_families = [
+        {
+            "source_family_id": "mcp_gateway_tool_calls",
+            "source_tables": ["awooop_mcp_gateway_audit"],
+            "normalized_event_schema": "ToolCallEvidence",
+            "label_dimensions": ["project", "run", "trace", "agent", "tool", "policy_gate"],
+            "total": _trace_total(mcp_gateway_summary),
+            "recent": _trace_recent(mcp_gateway_summary),
+            "feeds_learning": True,
+            "public_safe": True,
+            "raw_payload_policy": "hash_only_no_raw_input_output",
+            "next_action_if_empty": "route_first_class_tools_through_awooop_mcp_gateway",
+        },
+        {
+            "source_family_id": "legacy_mcp_tool_calls",
+            "source_tables": ["mcp_audit_log"],
+            "normalized_event_schema": "LegacyToolCallEvidence",
+            "label_dimensions": ["incident", "session_ref", "flywheel_node", "agent", "tool"],
+            "total": _trace_total(legacy_mcp_summary),
+            "recent": _trace_recent(legacy_mcp_summary),
+            "feeds_learning": True,
+            "public_safe": True,
+            "raw_payload_policy": "bridge_to_gateway_hash_or_redacted_summary",
+            "next_action_if_empty": "keep_legacy_bridge_until_all_callers_use_gateway",
+        },
+        {
+            "source_family_id": "service_package_logs",
+            "source_tables": [
+                "incident_evidence.recent_logs",
+                "incident_evidence.evidence_summary",
+                "incident_evidence.anomaly_context",
+            ],
+            "normalized_event_schema": "ServiceLogEvidence",
+            "label_dimensions": ["project", "product", "website", "service", "package", "incident"],
+            "total": _trace_total(service_log_summary),
+            "recent": _trace_recent(service_log_summary),
+            "feeds_learning": True,
+            "public_safe": True,
+            "raw_payload_policy": "sanitized_summary_only",
+            "next_action_if_empty": "collect_sanitized_service_package_logs_before_decision",
+        },
+        {
+            "source_family_id": "executor_operation_logs",
+            "source_tables": ["automation_operation_log"],
+            "normalized_event_schema": "ExecutorOperationEvidence",
+            "label_dimensions": [
+                "project",
+                "service",
+                "package",
+                "tool",
+                "incident",
+                "operation",
+                "playbook",
+                "risk",
+            ],
+            "total": max(operation_total, _trace_total(executor_log_summary)),
+            "recent": max(operation_recent, _trace_recent(executor_log_summary)),
+            "feeds_learning": True,
+            "public_safe": True,
+            "raw_payload_policy": "stdout_stderr_tail_or_structured_result_only",
+            "next_action_if_empty": "persist_executor_operation_log_for_candidate_check_apply",
+        },
+        {
+            "source_family_id": "auto_repair_receipts",
+            "source_tables": ["auto_repair_executions"],
+            "normalized_event_schema": "RepairExecutionReceipt",
+            "label_dimensions": ["incident", "service", "playbook", "risk", "result"],
+            "total": _trace_total(auto_repair_summary),
+            "recent": _trace_recent(auto_repair_summary),
+            "feeds_learning": True,
+            "public_safe": True,
+            "raw_payload_policy": "execution_step_refs_not_raw_secrets",
+            "next_action_if_empty": "write_auto_repair_execution_receipt_after_apply",
+        },
+        {
+            "source_family_id": "post_apply_verifier",
+            "source_tables": ["incident_evidence.post_execution_state"],
+            "normalized_event_schema": "VerifierEvidence",
+            "label_dimensions": ["incident", "operation", "playbook", "service", "result"],
+            "total": _trace_total(verifier_summary),
+            "recent": _trace_recent(verifier_summary),
+            "feeds_learning": True,
+            "public_safe": True,
+            "raw_payload_policy": "post_state_summary_no_secret_values",
+            "next_action_if_empty": "run_post_apply_verifier_for_each_apply",
+        },
+        {
+            "source_family_id": "rag_km_entries",
+            "source_tables": ["knowledge_entries"],
+            "normalized_event_schema": "KnowledgeWritebackEvidence",
+            "label_dimensions": ["project", "incident", "playbook", "path_type", "status"],
+            "total": _trace_total(km_summary),
+            "recent": _trace_recent(km_summary),
+            "feeds_learning": True,
+            "public_safe": True,
+            "raw_payload_policy": "curated_summary_and_refs_only",
+            "next_action_if_empty": "write_km_entry_after_verifier",
+        },
+        {
+            "source_family_id": "playbook_trust_signals",
+            "source_tables": ["playbooks"],
+            "normalized_event_schema": "PlayBookTrustSignal",
+            "label_dimensions": ["project", "playbook", "status", "trust_band", "review_required"],
+            "total": _trace_total(playbook_trust_summary),
+            "recent": _trace_recent(playbook_trust_summary),
+            "feeds_learning": True,
+            "public_safe": True,
+            "raw_payload_policy": "aggregate_trust_counters_only",
+            "next_action_if_empty": "write_trust_delta_after_verified_execution",
+        },
+        {
+            "source_family_id": "operator_timeline_projection",
+            "source_tables": ["timeline_events"],
+            "normalized_event_schema": "OperatorTimelineEvent",
+            "label_dimensions": ["incident", "event_type", "status", "actor", "actor_role"],
+            "total": _trace_total(timeline_summary),
+            "recent": _trace_recent(timeline_summary),
+            "feeds_learning": True,
+            "public_safe": True,
+            "raw_payload_policy": "short_public_safe_status_projection",
+            "next_action_if_empty": "project_ai_runtime_stage_to_timeline_events",
+        },
+        {
+            "source_family_id": "telegram_delivery_receipts",
+            "source_tables": ["awooop_outbound_message"],
+            "normalized_event_schema": "NotificationReceipt",
+            "label_dimensions": ["project", "channel", "incident", "action", "send_status"],
+            "total": _trace_total(telegram_summary),
+            "recent": _trace_recent(telegram_summary),
+            "feeds_learning": False,
+            "public_safe": True,
+            "raw_payload_policy": "provider_message_ref_no_unredacted_payload",
+            "next_action_if_empty": "send_controlled_apply_result_via_gateway",
+        },
+    ]
+    label_dimensions = sorted(
+        {
+            str(dimension)
+            for source in source_families
+            for dimension in source["label_dimensions"]
+        }
+    )
+    active_source_count = sum(1 for source in source_families if _int_value(source["total"]) > 0)
+    return {
+        "schema_version": "ai_agent_log_integration_taxonomy_v1",
+        "purpose": (
+            "將專案、產品、網站、服務、套件、工具與通知來源的 log "
+            "統一轉成可貼標、可分群、可回放、可餵 RAG/KM/PlayBook 的 evidence。"
+        ),
+        "normalized_event_flow": [
+            "collect_source_log_or_receipt",
+            "redact_and_hash_sensitive_fields",
+            "assign_labels",
+            "correlate_incident_operation_playbook",
+            "write_trace_ledger",
+            "retrieve_similar_context_via_rag",
+            "select_or_repair_playbook",
+            "run_check_mode_then_controlled_apply",
+            "verify_and_write_learning_back",
+        ],
+        "label_dimensions": label_dimensions,
+        "required_label_dimensions": [
+            "project",
+            "source_family",
+            "incident",
+            "operation",
+            "service",
+            "tool",
+            "playbook",
+        ],
+        "source_families": source_families,
+        "rollups": {
+            "source_family_count": len(source_families),
+            "active_source_family_count": active_source_count,
+            "inactive_source_family_count": len(source_families) - active_source_count,
+            "label_dimension_count": len(label_dimensions),
+            "classified_event_total": sum(_int_value(source["total"]) for source in source_families),
+            "recent_classified_event_total": sum(_int_value(source["recent"]) for source in source_families),
+            "learning_source_family_count": sum(
+                1 for source in source_families if source["feeds_learning"] is True
+            ),
+        },
+        "public_safety": {
+            "raw_secret_collection_allowed": False,
+            "raw_session_collection_allowed": False,
+            "unredacted_payload_storage_allowed": False,
+            "internal_reasoning_storage_allowed": False,
+        },
+    }
+
+
 def _first_operation(
     rows: Iterable[Mapping[str, Any]],
     operation_type: str,
@@ -868,6 +1313,12 @@ def build_runtime_receipt_readback_from_rows(
     km_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
     telegram_count_rows: Iterable[Mapping[str, Any] | Any] = (),
     telegram_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
+    mcp_gateway_count_rows: Iterable[Mapping[str, Any] | Any] = (),
+    legacy_mcp_count_rows: Iterable[Mapping[str, Any] | Any] = (),
+    service_log_count_rows: Iterable[Mapping[str, Any] | Any] = (),
+    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] = (),
     error_type: str | None = None,
 ) -> dict[str, Any]:
     """Build the live executor receipt readback from already-fetched rows."""
@@ -888,6 +1339,12 @@ def build_runtime_receipt_readback_from_rows(
     )
     km_summary = _status_counts(km_count_rows, status_key="status")
     telegram_summary = _status_counts(telegram_count_rows, status_key="send_status")
+    mcp_gateway_summary = _status_counts(mcp_gateway_count_rows, status_key="status")
+    legacy_mcp_summary = _status_counts(legacy_mcp_count_rows, status_key="status")
+    service_log_summary = _status_counts(service_log_count_rows, status_key="status")
+    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")
     latest_closure = _latest_flow_closure(
         operation_latest_rows=operation_latest,
         verifier_latest_rows=verifier_latest,
@@ -911,6 +1368,34 @@ def build_runtime_receipt_readback_from_rows(
         latest_failure_classification=latest_failure,
         controlled_retry_package=retry_package,
     )
+    trace_ledger = _build_trace_ledger(
+        operation_summary=operation_summary,
+        auto_repair_summary=auto_repair_summary,
+        verifier_summary=verifier_summary,
+        km_summary=km_summary,
+        telegram_summary=telegram_summary,
+        mcp_gateway_summary=mcp_gateway_summary,
+        legacy_mcp_summary=legacy_mcp_summary,
+        service_log_summary=service_log_summary,
+        executor_log_summary=executor_log_summary,
+        timeline_summary=timeline_summary,
+        playbook_trust_summary=playbook_trust_summary,
+        latest_flow_closure=latest_closure,
+        loop_ledger=loop_ledger,
+    )
+    log_integration_taxonomy = _build_log_integration_taxonomy(
+        operation_summary=operation_summary,
+        auto_repair_summary=auto_repair_summary,
+        verifier_summary=verifier_summary,
+        km_summary=km_summary,
+        telegram_summary=telegram_summary,
+        mcp_gateway_summary=mcp_gateway_summary,
+        legacy_mcp_summary=legacy_mcp_summary,
+        service_log_summary=service_log_summary,
+        executor_log_summary=executor_log_summary,
+        timeline_summary=timeline_summary,
+        playbook_trust_summary=playbook_trust_summary,
+    )
     apply_summary = operation_summary.get("ansible_apply_executed") or {}
     readback = {
         "schema_version": _LIVE_READBACK_SCHEMA_VERSION,
@@ -1014,10 +1499,22 @@ def build_runtime_receipt_readback_from_rows(
                 ),
             ),
         },
+        "mcp_context": {
+            "gateway": mcp_gateway_summary,
+            "legacy": legacy_mcp_summary,
+            "total": _trace_total(mcp_gateway_summary) + _trace_total(legacy_mcp_summary),
+            "recent": _trace_recent(mcp_gateway_summary) + _trace_recent(legacy_mcp_summary),
+        },
+        "service_log_evidence": service_log_summary,
+        "executor_log_projection": executor_log_summary,
+        "timeline_projection": timeline_summary,
+        "playbook_trust": playbook_trust_summary,
         "latest_flow_closure": latest_closure,
         "latest_failure_classification": latest_failure,
         "controlled_retry_package": retry_package,
         "autonomous_execution_loop_ledger": loop_ledger,
+        "trace_ledger": trace_ledger,
+        "log_integration_taxonomy": log_integration_taxonomy,
     }
     if error_type:
         readback["error"] = {
@@ -1076,6 +1573,50 @@ def _attach_runtime_receipt_readback(
             == "ready_for_no_write_check_mode_replay"
             else 0
         ),
+        "live_mcp_context_count": _int_value(readback.get("mcp_context", {}).get("total")),
+        "live_service_log_evidence_count": _int_value(
+            readback.get("service_log_evidence", {}).get("total")
+        ),
+        "live_executor_log_projection_count": _int_value(
+            readback.get("executor_log_projection", {}).get("total")
+        ),
+        "live_timeline_projection_count": _int_value(
+            readback.get("timeline_projection", {}).get("total")
+        ),
+        "live_playbook_trust_signal_count": _int_value(
+            readback.get("playbook_trust", {}).get("total")
+        ),
+        "live_trace_recorded_stage_count": _int_value(
+            readback.get("trace_ledger", {}).get("recorded_stage_count")
+        ),
+        "live_trace_required_missing_count": len(
+            (readback.get("trace_ledger") or {}).get("missing_required_stage_ids") or []
+        ),
+        "live_log_source_family_count": _int_value(
+            ((readback.get("log_integration_taxonomy") or {}).get("rollups") or {}).get(
+                "source_family_count"
+            )
+        ),
+        "live_log_active_source_family_count": _int_value(
+            ((readback.get("log_integration_taxonomy") or {}).get("rollups") or {}).get(
+                "active_source_family_count"
+            )
+        ),
+        "live_log_label_dimension_count": _int_value(
+            ((readback.get("log_integration_taxonomy") or {}).get("rollups") or {}).get(
+                "label_dimension_count"
+            )
+        ),
+        "live_log_classified_event_total": _int_value(
+            ((readback.get("log_integration_taxonomy") or {}).get("rollups") or {}).get(
+                "classified_event_total"
+            )
+        ),
+        "live_log_recent_classified_event_total": _int_value(
+            ((readback.get("log_integration_taxonomy") or {}).get("rollups") or {}).get(
+                "recent_classified_event_total"
+            )
+        ),
     })
     return payload
 
@@ -1321,6 +1862,19 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
     try:
         async with get_db_context(project_id) as db:
             await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
+
+            async def _safe_aux_rows(query_name: str, sql: str) -> list[Mapping[str, Any]]:
+                try:
+                    return (await db.execute(text(sql), params)).mappings().all()
+                except Exception as exc:  # pragma: no cover - depends on live schema drift
+                    logger.warning(
+                        "ai_agent_autonomous_runtime_trace_aux_read_failed",
+                        project_id=project_id,
+                        query_name=query_name,
+                        error_type=type(exc).__name__,
+                    )
+                    return []
+
             operation_counts = (
                 await db.execute(text(_RUNTIME_OPERATION_COUNTS_SQL), params)
             ).mappings().all()
@@ -1351,6 +1905,30 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
             telegram_latest = (
                 await db.execute(text(_RUNTIME_TELEGRAM_LATEST_SQL), params)
             ).mappings().all()
+            mcp_gateway_counts = await _safe_aux_rows(
+                "mcp_gateway_counts",
+                _RUNTIME_MCP_GATEWAY_COUNTS_SQL,
+            )
+            legacy_mcp_counts = await _safe_aux_rows(
+                "legacy_mcp_counts",
+                _RUNTIME_LEGACY_MCP_COUNTS_SQL,
+            )
+            service_log_counts = await _safe_aux_rows(
+                "service_log_counts",
+                _RUNTIME_SERVICE_LOG_COUNTS_SQL,
+            )
+            executor_log_counts = await _safe_aux_rows(
+                "executor_log_counts",
+                _RUNTIME_EXECUTOR_LOG_COUNTS_SQL,
+            )
+            timeline_counts = await _safe_aux_rows(
+                "timeline_counts",
+                _RUNTIME_TIMELINE_COUNTS_SQL,
+            )
+            playbook_trust_counts = await _safe_aux_rows(
+                "playbook_trust_counts",
+                _RUNTIME_PLAYBOOK_TRUST_COUNTS_SQL,
+            )
     except Exception as exc:
         logger.warning(
             "ai_agent_autonomous_runtime_receipt_readback_failed",
@@ -1378,6 +1956,12 @@ async def load_ai_agent_autonomous_runtime_receipt_readback(
         km_latest_rows=km_latest,
         telegram_count_rows=telegram_counts,
         telegram_latest_rows=telegram_latest,
+        mcp_gateway_count_rows=mcp_gateway_counts,
+        legacy_mcp_count_rows=legacy_mcp_counts,
+        service_log_count_rows=service_log_counts,
+        executor_log_count_rows=executor_log_counts,
+        timeline_count_rows=timeline_counts,
+        playbook_trust_count_rows=playbook_trust_counts,
     )
 
 
@@ -1590,6 +2174,136 @@ _RUNTIME_TELEGRAM_LATEST_SQL = """
 """
 
 
+_RUNTIME_MCP_GATEWAY_COUNTS_SQL = """
+    SELECT
+        coalesce(result_status, 'unknown') AS status,
+        count(*) AS total,
+        count(*) FILTER (
+            WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
+        ) AS recent
+    FROM awooop_mcp_gateway_audit
+    WHERE project_id = :project_id
+    GROUP BY coalesce(result_status, 'unknown')
+    ORDER BY status
+"""
+
+
+_RUNTIME_LEGACY_MCP_COUNTS_SQL = """
+    SELECT
+        CASE
+            WHEN success IS TRUE THEN 'success'
+            WHEN success IS FALSE THEN 'failed'
+            ELSE 'unknown'
+        END AS status,
+        count(*) AS total,
+        count(*) FILTER (
+            WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
+        ) AS recent
+    FROM mcp_audit_log
+    GROUP BY
+        CASE
+            WHEN success IS TRUE THEN 'success'
+            WHEN success IS FALSE THEN 'failed'
+            ELSE 'unknown'
+        END
+    ORDER BY status
+"""
+
+
+_RUNTIME_SERVICE_LOG_COUNTS_SQL = """
+    SELECT
+        'sanitized_recent_logs' AS status,
+        count(*) AS total,
+        count(*) FILTER (
+            WHERE collected_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
+        ) AS recent
+    FROM incident_evidence
+    WHERE recent_logs IS NOT NULL
+       OR evidence_summary IS NOT NULL
+       OR mcp_health IS NOT NULL
+       OR anomaly_context IS NOT NULL
+"""
+
+
+_RUNTIME_EXECUTOR_LOG_COUNTS_SQL = """
+    SELECT
+        coalesce(status, 'unknown') AS status,
+        count(*) AS total,
+        count(*) FILTER (
+            WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
+        ) AS recent
+    FROM automation_operation_log
+    WHERE operation_type IN (
+        'ansible_candidate_matched',
+        'ansible_check_mode_executed',
+        'ansible_apply_executed',
+        'ansible_rollback_executed',
+        'ansible_execution_skipped'
+    )
+      AND (
+        output IS NOT NULL
+        OR error IS NOT NULL
+        OR stderr_feed_back IS NOT NULL
+        OR dry_run_result IS NOT NULL
+      )
+    GROUP BY coalesce(status, 'unknown')
+    ORDER BY status
+"""
+
+
+_RUNTIME_TIMELINE_COUNTS_SQL = """
+    SELECT
+        coalesce(status, 'unknown') AS status,
+        count(*) AS total,
+        count(*) FILTER (
+            WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
+        ) AS recent
+    FROM timeline_events
+    WHERE event_type IN (
+        'mcp_call',
+        'verifier',
+        'ai_agent_deploy_control_plane_decision',
+        'controlled_apply',
+        'auto_repair',
+        'km_writeback'
+    )
+       OR actor IN (
+        'ansible_check_mode_worker',
+        'ansible_controlled_apply_worker',
+        'post_apply_verifier',
+        'truth_chain_reconciliation'
+    )
+       OR actor_role IN ('mcp', 'replay', 'verifier', 'executor')
+"""
+
+
+_RUNTIME_PLAYBOOK_TRUST_COUNTS_SQL = """
+    SELECT
+        CASE
+            WHEN review_required IS TRUE THEN 'review_required'
+            WHEN trust_score >= 0.8 THEN 'high_trust'
+            WHEN trust_score < 0.3 THEN 'low_trust'
+            WHEN success_count > 0 OR failure_count > 0 THEN 'learning_active'
+            ELSE 'seeded_not_used'
+        END AS status,
+        count(*) AS total,
+        count(*) FILTER (
+            WHERE updated_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
+        ) AS recent
+    FROM playbooks
+    WHERE project_id = :project_id
+    GROUP BY
+        CASE
+            WHEN review_required IS TRUE THEN 'review_required'
+            WHEN trust_score >= 0.8 THEN 'high_trust'
+            WHEN trust_score < 0.3 THEN 'low_trust'
+            WHEN success_count > 0 OR failure_count > 0 THEN 'learning_active'
+            ELSE 'seeded_not_used'
+        END
+    ORDER BY status
+"""
+
+
 def _validate_payload(payload: dict[str, Any]) -> None:
     if payload.get("schema_version") != _SCHEMA_VERSION:
         raise ValueError(f"schema_version must be {_SCHEMA_VERSION}")