feat(awooop): surface status chain in operator console

apps/api/src/services/platform_operator_service.py

@@ -10,8 +10,8 @@ from __future__ import annotations
 
 import re
 import uuid
-from collections.abc import Mapping
 from collections import defaultdict
+from collections.abc import Mapping
 from datetime import UTC, datetime
 from typing import Any
 from uuid import UUID
@@ -36,6 +36,7 @@ from src.services.awooop_approval_token import issue_approval_token, record_appr
 from src.services.awooop_truth_chain_service import (
     _summarize_gateway_mcp,
     _summarize_mcp,
+    fetch_truth_chain,
 )
 from src.services.run_state_machine import transition
 
@@ -357,8 +358,37 @@ async def list_callback_replies(
         rows_result = await db.execute(list_sql, params)
         rows = list(rows_result.mappings().all())
 
+    items = [_callback_reply_event_item(row) for row in rows]
+    status_chain_cache: dict[tuple[str, str], dict[str, Any]] = {}
+    for item in items:
+        incident = item.get("incident_id")
+        if not incident:
+            item["awooop_status_chain"] = _build_awooop_status_chain(
+                incident_ids=[],
+                source_id=None,
+            )
+            continue
+        incident_id = str(incident)
+        item_project_id = str(item.get("project_id") or project_id or "awoooi")
+        cache_key = (item_project_id, incident_id)
+        cached = status_chain_cache.get(cache_key)
+        if cached is not None:
+            item["awooop_status_chain"] = cached
+            continue
+        remediation_history = await _fetch_run_remediation_history(
+            [incident_id],
+            limit=5,
+        )
+        chain = await _fetch_awooop_status_chain(
+            incident_ids=[incident_id],
+            project_id=item_project_id,
+            remediation_history=remediation_history,
+        )
+        status_chain_cache[cache_key] = chain
+        item["awooop_status_chain"] = chain
+
     return {
-        "items": [_callback_reply_event_item(row) for row in rows],
+        "items": items,
         "total": total,
         "page": page,
         "per_page": per_page,
@@ -910,6 +940,214 @@ def _run_remediation_list_summary(
     }
 
 
+def _safe_int(value: Any) -> int:
+    try:
+        return int(value or 0)
+    except (TypeError, ValueError):
+        return 0
+
+
+def _latest_remediation_history_item(
+    history: dict[str, Any] | None,
+) -> dict[str, Any]:
+    if not isinstance(history, dict):
+        return {}
+    items = history.get("items") if isinstance(history.get("items"), list) else []
+    latest = items[0] if items and isinstance(items[0], dict) else {}
+    return latest
+
+
+def _remediation_evidence_state(history: dict[str, Any] | None) -> str:
+    """Classify ADR-100 evidence with the same operator semantics as Telegram."""
+    if not isinstance(history, dict):
+        return "missing"
+
+    total = _safe_int(history.get("total"))
+    if total <= 0:
+        if history.get("status") == "fetch_failed":
+            return "fetch_failed"
+        return "missing"
+
+    latest = _latest_remediation_history_item(history)
+    if latest.get("writes_incident_state") or latest.get("writes_auto_repair_result"):
+        return "write_observed"
+    if latest.get("allowed") is False or latest.get("success") is False:
+        return "blocked"
+    if (
+        str(latest.get("safety_level") or "").lower() == "read_only"
+        or str(latest.get("required_scope") or "").lower() == "read"
+    ):
+        return "read_only"
+    return "observed"
+
+
+def _select_status_chain_source_id(
+    incident_ids: list[str],
+    remediation_history: dict[str, Any] | None,
+) -> str | None:
+    latest_incident_id = str(
+        _latest_remediation_history_item(remediation_history).get("incident_id") or ""
+    ).strip()
+    if latest_incident_id and latest_incident_id in incident_ids:
+        return latest_incident_id
+    return incident_ids[0] if incident_ids else latest_incident_id or None
+
+
+def _build_awooop_status_chain(
+    *,
+    incident_ids: list[str],
+    truth_chain: dict[str, Any] | None = None,
+    remediation_history: dict[str, Any] | None = None,
+    source_id: str | None = None,
+    fetch_error: str | None = None,
+) -> dict[str, Any]:
+    """Build the shared read-only status chain used by Telegram and Operator UI."""
+    truth_status = (
+        truth_chain.get("truth_status")
+        if isinstance(truth_chain, dict) and isinstance(truth_chain.get("truth_status"), dict)
+        else {}
+    )
+    quality = (
+        truth_chain.get("automation_quality")
+        if isinstance(truth_chain, dict) and isinstance(truth_chain.get("automation_quality"), dict)
+        else {}
+    )
+    facts = quality.get("facts") if isinstance(quality.get("facts"), dict) else {}
+    latest = _latest_remediation_history_item(remediation_history)
+    remediation_state = _remediation_evidence_state(remediation_history)
+    remediation_total = (
+        _safe_int(remediation_history.get("total"))
+        if isinstance(remediation_history, dict)
+        else 0
+    )
+    latest_route = _route_label_from_remediation(latest) if latest else "--"
+
+    current_stage = str(truth_status.get("current_stage") or "unknown")
+    stage_status = str(truth_status.get("stage_status") or "unknown")
+    verdict = str(quality.get("verdict") or "unknown")
+    verification = (
+        facts.get("verification_result")
+        or latest.get("verification_result_preview")
+        or "missing"
+    )
+    auto_repair_records = _safe_int(facts.get("auto_repair_execution_records"))
+    operation_records = _safe_int(facts.get("automation_operation_records"))
+    gateway_total = _safe_int(facts.get("mcp_gateway_total"))
+    km_entries = _safe_int(facts.get("knowledge_entries"))
+    needs_human = bool(truth_status.get("needs_human"))
+
+    if verdict == "auto_repaired_verified":
+        repair_state = "auto_repaired_verified"
+        next_step = "monitor_for_regression"
+    elif auto_repair_records > 0 or operation_records > 0:
+        repair_state = (
+            "executed_pending_verification"
+            if str(verification) == "missing"
+            else "executed"
+        )
+        next_step = "verify_execution_result"
+    elif remediation_state == "read_only":
+        repair_state = "read_only_dry_run"
+        next_step = "approve_or_escalate_from_awooop"
+    elif remediation_state == "write_observed":
+        repair_state = "write_observed_manual_review"
+        next_step = "review_write_evidence"
+    elif remediation_state == "blocked":
+        repair_state = "blocked_manual_required"
+        next_step = "manual_investigation"
+    elif needs_human:
+        repair_state = "manual_required"
+        next_step = "manual_investigation"
+    else:
+        repair_state = "no_execution_evidence"
+        next_step = "collect_evidence_or_wait"
+
+    if remediation_state in {"blocked", "fetch_failed"}:
+        needs_human = True
+    if (
+        remediation_state == "write_observed"
+        and repair_state != "auto_repaired_verified"
+    ):
+        needs_human = True
+
+    blockers = [
+        str(item)
+        for item in [
+            *(truth_status.get("blockers") if isinstance(truth_status.get("blockers"), list) else []),
+            *(quality.get("blockers") if isinstance(quality.get("blockers"), list) else []),
+        ]
+        if item
+    ]
+    if fetch_error:
+        blockers.append("truth_chain_fetch_failed")
+
+    return {
+        "schema_version": "awooop_status_chain_v1",
+        "source": "truth_chain+adr100_history",
+        "source_id": source_id,
+        "incident_ids": incident_ids,
+        "current_stage": current_stage,
+        "stage_status": stage_status,
+        "verdict": verdict,
+        "repair_state": repair_state,
+        "verification": str(verification),
+        "needs_human": needs_human,
+        "next_step": next_step,
+        "blockers": blockers[:8],
+        "fetch_error": fetch_error,
+        "evidence": {
+            "auto_repair_records": auto_repair_records,
+            "operation_records": operation_records,
+            "mcp_gateway_total": gateway_total,
+            "knowledge_entries": km_entries,
+            "remediation_total": remediation_total,
+            "remediation_state": remediation_state,
+            "latest_route": latest_route,
+            "latest_mode": latest.get("mode"),
+            "latest_at": latest.get("created_at"),
+            "latest_preview": latest.get("verification_result_preview"),
+        },
+        "writes": {
+            "incident": latest.get("writes_incident_state"),
+            "auto_repair": latest.get("writes_auto_repair_result"),
+        },
+    }
+
+
+async def _fetch_awooop_status_chain(
+    *,
+    incident_ids: list[str],
+    project_id: str,
+    remediation_history: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Fetch read-only truth-chain state and merge it with ADR-100 evidence."""
+    source_id = _select_status_chain_source_id(incident_ids, remediation_history)
+    truth_chain: dict[str, Any] | None = None
+    fetch_error: str | None = None
+    if source_id:
+        try:
+            truth_chain = await fetch_truth_chain(
+                source_id=source_id,
+                project_id=project_id or "awoooi",
+            )
+        except Exception as exc:
+            fetch_error = str(exc)
+            logger.warning(
+                "operator_awooop_status_chain_fetch_failed",
+                source_id=source_id,
+                project_id=project_id,
+                error=fetch_error,
+            )
+
+    return _build_awooop_status_chain(
+        incident_ids=incident_ids,
+        truth_chain=truth_chain,
+        remediation_history=remediation_history,
+        source_id=source_id,
+        fetch_error=fetch_error,
+    )
+
+
 def _validate_remediation_status_filter(value: str | None) -> None:
     if value is None:
         return
@@ -1384,6 +1622,11 @@ async def get_run_detail(
     )
     legacy_mcp_history = await _fetch_run_legacy_mcp_history(incident_ids)
     remediation_history = await _fetch_run_remediation_history(incident_ids)
+    awooop_status_chain = await _fetch_awooop_status_chain(
+        incident_ids=incident_ids,
+        project_id=run.project_id,
+        remediation_history=remediation_history,
+    )
 
     timeline: list[dict[str, Any]] = [
         _timeline_item(
@@ -1550,6 +1793,7 @@ async def get_run_detail(
         "mcp_gateway": mcp_gateway_summary,
         "mcp_legacy": legacy_mcp_history,
         "remediation_history": remediation_history,
+        "awooop_status_chain": awooop_status_chain,
         "timeline": timeline,
         "counts": {
             "steps": len(step_items),