fix(km): expose degraded readback reason

apps/api/src/models/knowledge.py

@@ -22,7 +22,6 @@ from pydantic import BaseModel, Field
 
 from src.utils.timezone import now_taipei
 
-
 # =============================================================================
 # Enums
 # =============================================================================
@@ -135,6 +134,8 @@ class KnowledgeListResponse(BaseModel):
     categories: list[CategoryCount] = Field(default_factory=list)
     asset_taxonomy: list[KnowledgeAssetTaxonomyCount] = Field(default_factory=list)
     readback_status: str = "ready"
+    primary_readback_ready: bool = True
+    degraded_reason_code: str | None = None
     operator_stage: str | None = None
     next_step: str | None = None
     writes_on_read: bool = False

apps/api/src/services/knowledge_service.py

@@ -210,6 +210,19 @@ _SOURCE_BACKED_KNOWLEDGE_SPECS: tuple[dict[str, object], ...] = (
     },
 )
 
+
+def _classify_knowledge_readback_degraded_reason(reason: str) -> str:
+    normalized = reason.lower()
+    if "pool" in normalized or "timeout" in normalized or "timed out" in normalized:
+        return "primary_km_db_timeout_or_pool_exhausted"
+    if "missing tenant context" in normalized or "project_id" in normalized or "unauthorized" in normalized:
+        return "primary_km_project_context_missing"
+    if "undefinedcolumn" in normalized or "does not exist" in normalized or "missing column" in normalized:
+        return "primary_km_schema_mismatch"
+    if "validation" in normalized or "pydantic" in normalized:
+        return "primary_km_legacy_row_validation"
+    return "primary_km_readback_exception"
+
 # =============================================================================
 # Singleton
 # =============================================================================
@@ -245,6 +258,8 @@ def build_knowledge_list_readback_degraded_response(
         categories=_source_category_counts(entries),
         asset_taxonomy=_source_asset_taxonomy_counts(entries),
         readback_status=readback_status,
+        primary_readback_ready=False,
+        degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason),
         operator_stage="knowledge_readback_source_backed_ai_controlled_repair",
         next_step=(
             "repair_primary_km_db_readback_then_promote_source_backed_receipts_to_persistent_km"
@@ -496,6 +511,7 @@ class KnowledgeService:
                     total=total,
                     categories=categories,
                     asset_taxonomy=asset_taxonomy,
+                    primary_readback_ready=True,
                 )
         except Exception as exc:  # noqa: BLE001 - production readback must fail soft
             logger.warning(
@@ -555,9 +571,19 @@ class KnowledgeService:
 
     async def search(self, query: str, limit: int = 20) -> list[KnowledgeEntry]:
         """關鍵字搜尋"""
+        try:
             async with get_db_context() as db:
                 repo: IKnowledgeRepository = KnowledgeDBRepository(db)
                 return await repo.search(query, limit)
+        except Exception as exc:  # noqa: BLE001 - KM search must not 500 the UI
+            logger.warning(
+                "knowledge_search_readback_degraded",
+                error=str(exc),
+                q=query,
+                limit=limit,
+                degraded_reason_code=_classify_knowledge_readback_degraded_reason(str(exc)),
+            )
+            return _filter_source_backed_entries(q=query)[:limit]
 
     async def semantic_search(
         self,
@@ -630,7 +656,9 @@ class KnowledgeService:
             list[KnowledgeEntry] — ANTI_PATTERN 條目，空表示無已知失敗案例
         """
         from datetime import timedelta
+
         from sqlalchemy import text as sa_text
+
         from src.utils.timezone import now_taipei
 
         cutoff = now_taipei() - timedelta(days=days)

apps/api/tests/test_knowledge_service_readback_degraded.py

@@ -60,6 +60,8 @@ async def test_knowledge_list_entries_fails_soft_when_readback_breaks(monkeypatc
     ]
     assert all(row.count >= 1 for row in response.asset_taxonomy)
     assert response.readback_status == "source_backed_degraded"
+    assert response.primary_readback_ready is False
+    assert response.degraded_reason_code == "primary_km_db_timeout_or_pool_exhausted"
     assert response.operator_stage == "knowledge_readback_source_backed_ai_controlled_repair"
     assert response.next_step == "repair_primary_km_db_readback_then_promote_source_backed_receipts_to_persistent_km"
     assert response.writes_on_read is False
@@ -83,6 +85,26 @@ async def test_knowledge_list_entries_source_backed_filter_and_search(monkeypatc
     assert response.asset_taxonomy[[row.key for row in response.asset_taxonomy].index("alert")].count == 1
 
 
+@pytest.mark.asyncio
+async def test_knowledge_search_fails_soft_to_source_backed_entries(monkeypatch) -> None:
+    monkeypatch.setattr(
+        knowledge_service_module,
+        "get_db_context",
+        lambda: _BrokenDbContext(),
+    )
+    service = KnowledgeService.__new__(KnowledgeService)
+
+    entries = await service.search("Telegram", limit=10)
+
+    entry_ids = {entry.id for entry in entries}
+    assert all(entry.id.startswith("source-backed-") for entry in entries)
+    assert {
+        "source-backed-service-telegram-alert-receipts",
+        "source-backed-alert-telegram-monitoring-coverage",
+        "source-backed-schedule-report-monitoring",
+    }.issubset(entry_ids)
+
+
 @pytest.mark.asyncio
 async def test_knowledge_categories_fails_soft_when_readback_breaks(monkeypatch) -> None:
     monkeypatch.setattr(