feat(awooop): surface ai route repair evidence

apps/api/src/api/v1/platform/operator_runs.py

@@ -151,6 +151,7 @@ class AiRouteStatusResponse(BaseModel):
     active_lane: dict[str, Any] | None = None
     skipped_lanes: list[dict[str, Any]] = Field(default_factory=list)
     operator_action: dict[str, Any] | None = None
+    repair_evidence: dict[str, Any] | None = None
     checked_at: datetime
 
 

apps/api/src/services/platform_operator_service.py

@@ -99,6 +99,8 @@ _CICD_STATUS_FILTERS = {"running", "success", "failed", "pending"}
 _CICD_STAGE_RE = re.compile(r"^[a-z0-9_:-]{1,64}$", re.IGNORECASE)
 _AI_ROUTE_STATUS_SCHEMA_VERSION = "awooop_ai_route_status_v1"
 _AI_ROUTE_WORKLOADS = set(get_args(OllamaWorkloadType))
+_AI_ROUTE_REPAIR_EVIDENCE_PROVIDER = "ai_route_repair"
+_AI_ROUTE_REPAIR_EVIDENCE_STAGE = "repair_diagnosis"
 _SOURCE_CORRELATION_SCHEMA_VERSION = "source_provider_correlation_v1"
 _SOURCE_CORRELATION_PROVIDERS = ("sentry", "signoz")
 _SOURCE_CORRELATION_EVENT_LIMIT = 200
@@ -642,7 +644,7 @@ async def get_ai_route_status(
         selected_provider=route.primary.provider_name,
         health=health,
     ))
-    return response
+    return await _ai_route_response_with_repair_evidence(response)
 
 
 def _validate_ai_route_workload(workload_type: str | None) -> OllamaWorkloadType:
@@ -696,7 +698,7 @@ async def _ai_route_lightweight_status_from_policy(
             route_reason=route_reason,
             error=str(exc),
         )
-        return _ai_route_unavailable_status(
+        response = _ai_route_unavailable_status(
             workload=workload,
             policy_order=policy_order,
             checked_at=checked_at,
@@ -704,6 +706,7 @@ async def _ai_route_lightweight_status_from_policy(
             route_error=route_error,
             route_source="ollama_failover_manager",
         )
+        return await _ai_route_response_with_repair_evidence(response)
 
     health_by_provider = {
         endpoint.provider_name: _ai_route_health_item(report)
@@ -741,7 +744,7 @@ async def _ai_route_lightweight_status_from_policy(
             selected_provider="gemini",
             health=health_by_provider,
         ))
-        return response
+        return await _ai_route_response_with_repair_evidence(response)
 
     selected = endpoints[selected_index]
     model = get_settings().OLLAMA_HEALTH_CHECK_MODEL
@@ -783,7 +786,7 @@ async def _ai_route_lightweight_status_from_policy(
         selected_provider=selected.provider_name,
         health=health_by_provider,
     ))
-    return response
+    return await _ai_route_response_with_repair_evidence(response)
 
 
 async def _ai_route_probe_connectivity(
@@ -873,6 +876,208 @@ def _ai_route_unavailable_status(
     return response
 
 
+async def _ai_route_response_with_repair_evidence(
+    response: dict[str, Any],
+) -> dict[str, Any]:
+    """Attach latest read-only repair dossier evidence when a lane is degraded."""
+    response["repair_evidence"] = None
+    if response.get("lane_mode") not in {
+        "degraded_failover",
+        "cloud_fallback",
+        "unavailable",
+    }:
+        return response
+
+    target_provider = _ai_route_repair_evidence_target(response)
+    response["repair_evidence"] = await _latest_ai_route_repair_evidence(
+        target_provider=target_provider,
+    )
+    return response
+
+
+def _ai_route_repair_evidence_target(response: Mapping[str, Any]) -> str | None:
+    skipped_lanes = response.get("skipped_lanes")
+    if isinstance(skipped_lanes, list):
+        for lane in skipped_lanes:
+            if not isinstance(lane, dict):
+                continue
+            provider_name = str(lane.get("provider_name") or "").strip()
+            if provider_name and lane.get("action_required") is True:
+                return provider_name
+        for lane in skipped_lanes:
+            if isinstance(lane, dict):
+                provider_name = str(lane.get("provider_name") or "").strip()
+                if provider_name:
+                    return provider_name
+    policy_order = response.get("policy_order")
+    if isinstance(policy_order, list):
+        for item in policy_order:
+            if not isinstance(item, dict):
+                continue
+            if item.get("runtime") == "ollama":
+                provider_name = str(item.get("provider_name") or "").strip()
+                if provider_name:
+                    return provider_name
+    return None
+
+
+async def _latest_ai_route_repair_evidence(
+    *,
+    project_id: str = "awoooi",
+    target_provider: str | None = None,
+) -> dict[str, Any] | None:
+    """Fetch the newest AI route repair diagnosis stored in AwoooP event DB."""
+    params: dict[str, Any] = {
+        "project_id": project_id,
+        "provider": _AI_ROUTE_REPAIR_EVIDENCE_PROVIDER,
+        "stage": _AI_ROUTE_REPAIR_EVIDENCE_STAGE,
+    }
+    target_clause = ""
+    if target_provider:
+        target_clause = """
+          AND COALESCE(
+                NULLIF(source_envelope #>> '{log_correlation,target_resource}', ''),
+                NULLIF(source_envelope #>> '{extra,payload,target_resource}', '')
+              ) = :target_provider
+        """
+        params["target_provider"] = target_provider
+
+    try:
+        item = await _fetch_latest_ai_route_repair_evidence(
+            params=params,
+            target_clause=target_clause,
+        )
+        if item is None and target_provider:
+            params.pop("target_provider", None)
+            item = await _fetch_latest_ai_route_repair_evidence(
+                params=params,
+                target_clause="",
+            )
+        return item
+    except Exception as exc:
+        logger.warning(
+            "ai_route_repair_evidence_fetch_failed",
+            project_id=project_id,
+            target_provider=target_provider,
+            error=str(exc),
+        )
+        return None
+
+
+async def _fetch_latest_ai_route_repair_evidence(
+    *,
+    params: dict[str, Any],
+    target_clause: str,
+) -> dict[str, Any] | None:
+    sql = text(f"""
+        SELECT
+            event_id,
+            run_id,
+            provider_event_id,
+            source_envelope,
+            provider_ts,
+            received_at
+        FROM awooop_conversation_event
+        WHERE project_id = :project_id
+          AND LOWER(COALESCE(
+                NULLIF(source_envelope->>'provider', ''),
+                NULLIF(split_part(provider_event_id, ':', 1), ''),
+                channel_type
+              )) = :provider
+          AND LOWER(COALESCE(NULLIF(source_envelope->>'stage', ''), 'received')) = :stage
+          {target_clause}
+        ORDER BY received_at DESC, event_id DESC
+        LIMIT 1
+    """)
+    async with get_db_context("awoooi") as db:
+        result = await db.execute(sql, params)
+        row = result.mappings().first()
+    return _ai_route_repair_evidence_item(row) if row else None
+
+
+def _ai_route_repair_evidence_item(
+    row: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Project route-repair source envelopes into a compact operator-safe view."""
+    envelope = _as_dict(row.get("source_envelope"))
+    extra = _as_dict(envelope.get("extra"))
+    payload = _as_dict(extra.get("payload"))
+    log_correlation = _as_dict(envelope.get("log_correlation"))
+    live_probe = _as_dict(payload.get("live_probe"))
+    observed_state = _as_dict(payload.get("observed_state"))
+    side_effects = _ai_route_repair_side_effects(payload.get("side_effects"))
+
+    return {
+        "schema_version": (
+            payload.get("schema_version")
+            or envelope.get("schema_version")
+            or "ai_route_repair_evidence_projection_v1"
+        ),
+        "provider": (
+            envelope.get("provider")
+            or str(row.get("provider_event_id") or "").split(":", 1)[0]
+        ),
+        "stage": envelope.get("stage") or _AI_ROUTE_REPAIR_EVIDENCE_STAGE,
+        "provider_event_id": row.get("provider_event_id"),
+        "conversation_event_id": _string_or_none(row.get("event_id")),
+        "run_id": _string_or_none(row.get("run_id")),
+        "alertname": log_correlation.get("alertname"),
+        "severity": log_correlation.get("severity"),
+        "fingerprint": log_correlation.get("fingerprint"),
+        "target_resource": (
+            log_correlation.get("target_resource")
+            or payload.get("target_resource")
+            or observed_state.get("target_resource")
+        ),
+        "observed_state": observed_state,
+        "live_probe": live_probe,
+        "access_blockers": _as_string_list(payload.get("access_blockers")),
+        "side_effects": side_effects,
+        "source_ref_count": _source_ref_count(envelope),
+        "provider_ts": row.get("provider_ts"),
+        "received_at": row.get("received_at"),
+    }
+
+
+def _ai_route_repair_side_effects(value: Any) -> dict[str, bool | None]:
+    raw = _as_dict(value)
+    return {
+        "incident_created": _bool_or_none(raw.get("incident_created")),
+        "telegram_sent": _bool_or_none(raw.get("telegram_sent")),
+        "approval_created": _bool_or_none(raw.get("approval_created")),
+        "runtime_route_changed": _bool_or_none(raw.get("runtime_route_changed")),
+    }
+
+
+def _as_string_list(value: Any) -> list[str]:
+    if isinstance(value, list):
+        return [str(item) for item in value if str(item or "").strip()]
+    if value not in (None, ""):
+        return [str(value)]
+    return []
+
+
+def _string_or_none(value: Any) -> str | None:
+    if value in (None, ""):
+        return None
+    return str(value)
+
+
+def _bool_or_none(value: Any) -> bool | None:
+    return value if isinstance(value, bool) else None
+
+
+def _source_ref_count(envelope: Any) -> int:
+    source_refs = _as_dict(_as_dict(envelope).get("source_refs"))
+    total = 0
+    for value in source_refs.values():
+        if isinstance(value, list):
+            total += len([item for item in value if str(item or "").strip()])
+        elif value not in (None, ""):
+            total += 1
+    return total
+
+
 def _ai_route_lane_state(
     *,
     policy_order: list[dict[str, Any]],