fix(api): keep operator read-model gaps from breaking ai loop UI

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

@@ -10,6 +10,7 @@ from datetime import UTC, datetime
 from typing import Annotated, Any, Literal
 from uuid import UUID
 
+import structlog
 from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel, Field
 
@@ -36,6 +37,45 @@ from src.services.channel_hub import record_external_alert_event
 from src.services.platform_operator_service import list_recent_channel_events
 
 router = APIRouter()
+logger = structlog.get_logger(__name__)
+
+
+def _empty_recurrence_summary() -> dict[str, Any]:
+    return {
+        "source_event_total": 0,
+        "recurrence_group_total": 0,
+        "recurrent_group_total": 0,
+        "duplicate_event_total": 0,
+        "linked_run_total": 0,
+        "unlinked_event_total": 0,
+        "auto_repair_linked_total": 0,
+        "verified_repair_group_total": 0,
+        "open_work_item_group_total": 0,
+        "manual_gate_group_total": 0,
+        "controlled_apply_gate_group_total": 0,
+        "automation_gap_group_total": 0,
+        "failed_repair_group_total": 0,
+        "source_correlation_review_group_total": 0,
+        "source_correlation_decision_recorded_group_total": 0,
+        "source_correlation_applied_group_total": 0,
+        "latest_received_at": None,
+    }
+
+
+def _recurrence_source_unavailable_response(
+    *,
+    project_id: str | None,
+    limit: int,
+    error: str,
+) -> dict[str, Any]:
+    return {
+        "project_id": project_id or "awoooi",
+        "limit": limit,
+        "summary": _empty_recurrence_summary(),
+        "items": [],
+        "source_status": "source_unavailable",
+        "source_error": error,
+    }
 
 
 class _BodyProjectContext:
@@ -440,11 +480,25 @@ async def get_event_dossier_recurrence(
     ),
     limit: int = Query(100, ge=1, le=300, description="最多納入統計筆數"),
 ) -> dict[str, Any]:
-    return await fetch_channel_event_dossier_recurrence(
-        project_id=project_id,
-        provider=provider,
-        limit=limit,
-    )
+    try:
+        return await fetch_channel_event_dossier_recurrence(
+            project_id=project_id,
+            provider=provider,
+            limit=limit,
+        )
+    except Exception as exc:
+        logger.warning(
+            "channel_event_dossier_recurrence_source_unavailable",
+            project_id=project_id or "awoooi",
+            provider=provider,
+            limit=limit,
+            error=exc.__class__.__name__,
+        )
+        return _recurrence_source_unavailable_response(
+            project_id=project_id,
+            limit=limit,
+            error=exc.__class__.__name__,
+        )
 
 
 @router.get(

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

@@ -15,6 +15,7 @@ from decimal import Decimal
 from typing import Any, Literal
 from uuid import UUID
 
+import structlog
 from fastapi import APIRouter, Depends, Query
 from pydantic import BaseModel, Field
 
@@ -51,6 +52,7 @@ from src.services.platform_operator_service import (
 )
 
 router = APIRouter()
+logger = structlog.get_logger(__name__)
 
 _DEFAULT_PER_PAGE = 50
 _MAX_PER_PAGE = 200
@@ -239,6 +241,22 @@ class ListCallbackRepliesResponse(BaseModel):
     cache: OperatorSummaryCacheInfo | None = None
 
 
+def _callback_replies_source_unavailable_response(
+    *,
+    page: int,
+    per_page: int,
+) -> dict[str, Any]:
+    return {
+        "items": [],
+        "total": 0,
+        "page": page,
+        "per_page": per_page,
+        "summary": None,
+        "cache": None,
+        "source_status": "source_unavailable",
+    }
+
+
 class CicdEventItem(BaseModel):
     id: str
     project_id: str
@@ -378,15 +396,31 @@ async def list_callback_replies(
     per_page: int = Query(20, ge=1, le=_MAX_PER_PAGE, description="每頁筆數"),
     refresh: bool = Query(False, description="略過短 TTL 快取並重新聚合"),
 ) -> dict[str, Any]:
-    return await list_callback_replies_svc(
-        project_id=project_id,
-        callback_reply_status=callback_reply_status,
-        action=action,
-        incident_id=incident_id,
-        page=page,
-        per_page=per_page,
-        refresh=refresh,
-    )
+    try:
+        return await list_callback_replies_svc(
+            project_id=project_id,
+            callback_reply_status=callback_reply_status,
+            action=action,
+            incident_id=incident_id,
+            page=page,
+            per_page=per_page,
+            refresh=refresh,
+        )
+    except Exception as exc:
+        logger.warning(
+            "operator_callback_replies_source_unavailable",
+            project_id=project_id or "awoooi",
+            callback_reply_status=callback_reply_status,
+            action=action,
+            incident_id=incident_id,
+            page=page,
+            per_page=per_page,
+            error=exc.__class__.__name__,
+        )
+        return _callback_replies_source_unavailable_response(
+            page=page,
+            per_page=per_page,
+        )
 
 
 @router.get(

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

@@ -5,6 +5,7 @@ from __future__ import annotations
 from time import perf_counter
 from typing import Any
 
+import structlog
 from fastapi import APIRouter, Depends, Query
 
 from src.core.awooop_operator_auth import (
@@ -18,6 +19,49 @@ from src.services.awooop_truth_chain_service import (
 )
 
 router = APIRouter()
+logger = structlog.get_logger(__name__)
+
+
+def _quality_summary_unavailable_response(
+    *,
+    project_id: str,
+    hours: int,
+    limit: int,
+    error: str,
+) -> dict[str, Any]:
+    return {
+        "schema_version": "automation_quality_summary_v1",
+        "project_id": project_id or "awoooi",
+        "window_hours": hours,
+        "limit": limit,
+        "incident_total": 0,
+        "evaluated_total": 0,
+        "verified_auto_repair_total": 0,
+        "average_score": 0.0,
+        "score_buckets": {},
+        "by_verdict": {},
+        "gate_failures": [
+            {
+                "gate": "quality_summary_read_model",
+                "total": 1,
+                "reason": "source_unavailable",
+            }
+        ],
+        "automation_flow_gates": {},
+        "execution_backend_summary": {},
+        "ansible_runtime": {},
+        "examples": [],
+        "production_claim": {
+            "can_claim_full_auto_repair": False,
+            "reason": "quality_summary_source_unavailable",
+        },
+        "source_status": "source_unavailable",
+        "source_error": error,
+        "visibility_note": (
+            "Quality summary source unavailable. AI loop receipts remain visible from "
+            "the independent runtime readback panels."
+        ),
+    }
 
 
 @router.get(
@@ -53,7 +97,19 @@ async def get_automation_quality_summary(
             duration_seconds=perf_counter() - started_at,
             error=exc.__class__.__name__,
         )
-        raise
+        logger.warning(
+            "awooop_automation_quality_summary_source_unavailable",
+            project_id=project_id,
+            hours=hours,
+            limit=limit,
+            error=exc.__class__.__name__,
+        )
+        return _quality_summary_unavailable_response(
+            project_id=project_id,
+            hours=hours,
+            limit=limit,
+            error=exc.__class__.__name__,
+        )
     summary["examples"] = []
     summary["visibility_note"] = (
         "Aggregate only. Use /truth-chain/{source_id} with operator auth for source-level details."

apps/api/tests/test_platform_read_model_fallbacks.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import pytest
+
+from src.api.v1.platform import events, operator_runs, truth_chain
+
+
+@pytest.mark.asyncio
+async def test_quality_summary_returns_source_unavailable_when_read_model_fails(monkeypatch):
+    async def broken_fetch(**kwargs):
+        raise RuntimeError("quality read model unavailable")
+
+    observations: list[dict] = []
+
+    monkeypatch.setattr(truth_chain, "fetch_automation_quality_summary", broken_fetch)
+    monkeypatch.setattr(
+        truth_chain,
+        "record_quality_summary_observation",
+        lambda **kwargs: observations.append(kwargs),
+    )
+
+    response = await truth_chain.get_automation_quality_summary(
+        project_id="awoooi",
+        hours=24,
+        limit=30,
+    )
+
+    assert response["source_status"] == "source_unavailable"
+    assert response["evaluated_total"] == 0
+    assert response["verified_auto_repair_total"] == 0
+    assert response["production_claim"]["can_claim_full_auto_repair"] is False
+    assert observations[0]["success"] is False
+
+
+@pytest.mark.asyncio
+async def test_recurrence_endpoint_returns_empty_summary_when_read_model_fails(monkeypatch):
+    async def broken_fetch(**kwargs):
+        raise RuntimeError("recurrence read model unavailable")
+
+    monkeypatch.setattr(events, "fetch_channel_event_dossier_recurrence", broken_fetch)
+
+    response = await events.get_event_dossier_recurrence(
+        project_id="awoooi",
+        provider=None,
+        limit=100,
+    )
+
+    assert response["source_status"] == "source_unavailable"
+    assert response["project_id"] == "awoooi"
+    assert response["items"] == []
+    assert response["summary"]["source_event_total"] == 0
+    assert response["summary"]["open_work_item_group_total"] == 0
+    validated = events.ChannelEventRecurrenceResponse.model_validate(response)
+    assert validated.summary.source_event_total == 0
+
+
+@pytest.mark.asyncio
+async def test_callback_replies_endpoint_returns_empty_page_when_read_model_fails(monkeypatch):
+    async def broken_fetch(**kwargs):
+        raise RuntimeError("callback reply read model unavailable")
+
+    monkeypatch.setattr(operator_runs, "list_callback_replies_svc", broken_fetch)
+
+    response = await operator_runs.list_callback_replies(
+        project_id="awoooi",
+        callback_reply_status=None,
+        action=None,
+        incident_id=None,
+        page=1,
+        per_page=100,
+        refresh=False,
+    )
+
+    assert response["source_status"] == "source_unavailable"
+    assert response["items"] == []
+    assert response["total"] == 0
+    assert response["page"] == 1
+    assert response["per_page"] == 100
+    assert response["summary"] is None
+    validated = operator_runs.ListCallbackRepliesResponse.model_validate(response)
+    assert validated.total == 0