fix(km): keep primary entries during taxonomy degradation

apps/api/src/main.py

@@ -20,9 +20,9 @@ Date: 2026-03-20
 
 import asyncio
 import os
-from uuid import uuid4
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
+from uuid import uuid4
 
 import sentry_sdk
 import structlog
@@ -303,8 +303,8 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     try:
         from sqlalchemy import select
 
-        from src.db.base import get_db_context
         from src.core.context import clear_project_context, set_project_context
+        from src.db.base import get_db_context
         from src.db.models import IncidentRecord
         from src.models.incident import IncidentStatus
         from src.services.incident_service import get_incident_service
@@ -950,7 +950,11 @@ async def request_logging_middleware(request: Request, call_next):
     """
     import time
 
-    from src.core.context import clear_project_context, get_current_project_context, set_project_context
+    from src.core.context import (
+        clear_project_context,
+        get_current_project_context,
+        set_project_context,
+    )
 
     request_id = request.headers.get("X-Request-ID") or str(uuid4())
     project_id, source = _resolve_request_project_context(request)
@@ -1008,12 +1012,44 @@ async def db_context_guard() -> dict:
     from src.core.context import get_current_project_context
     from src.db.base import get_db_context
 
-    async with get_db_context():
+    async def _probe_db_context() -> None:
+        async with get_db_context():
+            return None
+
+    try:
+        await asyncio.wait_for(_probe_db_context(), timeout=1.5)
         return {
             "status": "ok",
             "project_context": get_current_project_context(),
             "source": "runtime_guard",
         }
+    except HTTPException:
+        raise
+    except Exception as exc:  # noqa: BLE001 - guard readback should not become a 500
+        message = f"{type(exc).__name__}: {exc}".lower()
+        timeout_or_pool = any(
+            marker in message
+            for marker in (
+                "pool",
+                "timeout",
+                "timed out",
+                "too many connections",
+                "connection limit",
+            )
+        )
+        return {
+            "status": "degraded",
+            "project_context": get_current_project_context(),
+            "source": "runtime_guard",
+            "db_context_ready": False,
+            "degraded_reason_code": (
+                "primary_db_timeout_or_pool_exhausted"
+                if timeout_or_pool
+                else "primary_db_context_readback_exception"
+            ),
+            "writes_on_read": False,
+            "manual_review_required": False,
+        }
 
 
 # =============================================================================

apps/api/src/repositories/knowledge_repository.py

@@ -474,18 +474,20 @@ class KnowledgeDBRepository:
         將 KM 依專案、產品、網站、服務、套件、工具、Log、Alert、PlayBook、
         RAG、MCP 與排程分群。
         """
-        rows: list[tuple[str, int]] = []
-        for key in _ASSET_TAXONOMY_KEYS:
-            result = await self.db.execute(
-                select(func.count())
-                .select_from(KnowledgeEntryRecord)
-                .where(
-                    _active_status_filter(),
-                    _asset_taxonomy_condition(key),
+        result = await self.db.execute(
+            select(
+                *(
+                    func.count()
+                    .filter(_asset_taxonomy_condition(key))
+                    .label(key)
+                    for key in _ASSET_TAXONOMY_KEYS
                 )
             )
-            rows.append((key, int(result.scalar() or 0)))
-        return rows
+            .select_from(KnowledgeEntryRecord)
+            .where(_active_status_filter())
+        )
+        row = result.mappings().one()
+        return [(key, int(row.get(key) or 0)) for key in _ASSET_TAXONOMY_KEYS]
 
     async def search(self, query: str, limit: int = 20) -> list[KnowledgeEntry]:
         """關鍵字搜尋 (title + content + tags)"""

apps/api/src/services/knowledge_service.py

@@ -52,6 +52,21 @@ _DEGRADED_CATEGORY_FALLBACKS = (
 )
 
 _ASSET_TAXONOMY_FALLBACKS = _DEGRADED_CATEGORY_FALLBACKS[:-1]
+_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS = 1.5
+_ASSET_TAXONOMY_CATEGORY_HINTS: dict[str, tuple[str, ...]] = {
+    "service": (
+        "infrastructure",
+        "application",
+        "ai_system",
+        "database",
+        "host_resource",
+        "kubernetes",
+        "alert_handling",
+    ),
+    "tool": ("devops_tool", "AI自動化/Ansible受控修復"),
+    "alert": ("alert_handling",),
+    "playbook": ("auto_repair",),
+}
 
 _SOURCE_BACKED_KNOWLEDGE_SPECS: tuple[dict[str, object], ...] = (
     {
@@ -366,6 +381,10 @@ def _source_asset_taxonomy_counts(
 def _entry_matches_asset_key(entry: KnowledgeEntry, key: str) -> bool:
     if entry.category == key:
         return True
+    if entry.category.lower() in {
+        hint.lower() for hint in _ASSET_TAXONOMY_CATEGORY_HINTS.get(key, ())
+    }:
+        return True
     if key in {tag.lower() for tag in entry.tags}:
         return True
     if key == "playbook" and entry.related_playbook_id:
@@ -484,34 +503,91 @@ class KnowledgeService:
                 limit=limit,
                 offset=offset,
             )
+        if total == 0:
+            return build_knowledge_list_readback_degraded_response(
+                "knowledge_db_empty",
+                category=category,
+                entry_type=entry_type,
+                status=status,
+                tags=tags,
+                q=q,
+                limit=limit,
+                offset=offset,
+                readback_status="source_backed_db_empty",
+            )
+
+        readback_status = "ready"
+        degraded_reason_code: str | None = None
+        operator_stage: str | None = None
+        next_step: str | None = None
+
+        try:
+            categories = await asyncio.wait_for(
+                self._read_primary_categories(),
+                timeout=_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS,
+            )
+        except Exception as exc:  # noqa: BLE001 - primary entries must stay visible
+            logger.warning(
+                "knowledge_list_categories_readback_degraded",
+                error=str(exc),
+                total=total,
+                limit=limit,
+                offset=offset,
+            )
+            categories = _source_category_counts(items)
+            readback_status = "ready_partial_taxonomy_degraded"
+            degraded_reason_code = "primary_km_taxonomy_readback_degraded"
+            operator_stage = "knowledge_primary_entries_ready_taxonomy_degraded"
+            next_step = "repair_primary_km_taxonomy_readback_without_hiding_entries"
+
+        try:
+            asset_taxonomy = await asyncio.wait_for(
+                self._read_primary_asset_taxonomy(),
+                timeout=_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS,
+            )
+        except Exception as exc:  # noqa: BLE001 - primary entries must stay visible
+            logger.warning(
+                "knowledge_list_asset_taxonomy_readback_degraded",
+                error=str(exc),
+                total=total,
+                limit=limit,
+                offset=offset,
+            )
+            asset_taxonomy = _source_asset_taxonomy_counts(items)
+            readback_status = "ready_partial_taxonomy_degraded"
+            degraded_reason_code = "primary_km_taxonomy_readback_degraded"
+            operator_stage = "knowledge_primary_entries_ready_taxonomy_degraded"
+            next_step = "repair_primary_km_taxonomy_readback_without_hiding_entries"
+
+        return KnowledgeListResponse(
+            items=items,
+            total=total,
+            categories=categories,
+            asset_taxonomy=asset_taxonomy,
+            readback_status=readback_status,
+            primary_readback_ready=True,
+            degraded_reason_code=degraded_reason_code,
+            operator_stage=operator_stage,
+            next_step=next_step,
+        )
+
+    async def _read_primary_categories(self) -> list[CategoryCount]:
+        async with get_db_context() as db:
+            repo: IKnowledgeRepository = KnowledgeDBRepository(db)
             categories_raw = await repo.get_categories()
             categories = [
                 CategoryCount(category=cat, count=cnt) for cat, cnt in categories_raw
             ]
+            return categories
+
+    async def _read_primary_asset_taxonomy(self) -> list[KnowledgeAssetTaxonomyCount]:
+        async with get_db_context() as db:
+            repo: IKnowledgeRepository = KnowledgeDBRepository(db)
             asset_taxonomy_raw = await repo.get_asset_taxonomy_counts()
-            asset_taxonomy = [
+            return [
                 KnowledgeAssetTaxonomyCount(key=key, count=count)
                 for key, count in asset_taxonomy_raw
             ]
-            if total == 0:
-                return build_knowledge_list_readback_degraded_response(
-                    "knowledge_db_empty",
-                    category=category,
-                    entry_type=entry_type,
-                    status=status,
-                    tags=tags,
-                    q=q,
-                    limit=limit,
-                    offset=offset,
-                    readback_status="source_backed_db_empty",
-                )
-            return KnowledgeListResponse(
-                items=items,
-                total=total,
-                categories=categories,
-                asset_taxonomy=asset_taxonomy,
-                primary_readback_ready=True,
-            )
 
     async def list_entries(
         self,

apps/api/tests/test_db_context_guard.py

@@ -2,15 +2,14 @@
 from __future__ import annotations
 
 from contextlib import asynccontextmanager
-from fastapi import HTTPException
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
 from unittest.mock import patch
 
+import pytest
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
+
 from src.db.base import get_db_context
-from src.main import db_context_guard, app, http_exception_handler
+from src.main import app, db_context_guard, http_exception_handler
 
 
 def test_db_context_guard_without_project_id_is_unauthorized():
@@ -45,6 +44,16 @@ class _UnauthorizedDbContext:
         return False
 
 
+class _BrokenDbContext:
+    """Simulate production DB pool pressure without hiding project context."""
+
+    async def __aenter__(self):
+        raise RuntimeError("QueuePool limit timeout")
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):  # noqa: ARG001
+        return False
+
+
 def _build_guard_app() -> FastAPI:
     app = FastAPI()
 
@@ -82,6 +91,22 @@ def test_db_context_guard_with_project_id_returns_snapshot():
     assert body["project_context"]["source"] == "test.guard"
 
 
+def test_db_context_guard_with_project_id_db_pressure_returns_degraded_snapshot():
+    """DB pool 壓力應回傳 evidence，不應把安全 guard 變成 500。"""
+    app = _build_guard_app()
+    with patch("src.db.base.get_db_context", return_value=_BrokenDbContext()):
+        client = TestClient(app)
+        response = client.get("/api/v1/security/db-context-guard", headers={"X-Project-ID": "awoooi"})
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["status"] == "degraded"
+    assert body["project_context"]["project_id"] == "awoooi"
+    assert body["db_context_ready"] is False
+    assert body["degraded_reason_code"] == "primary_db_timeout_or_pool_exhausted"
+    assert body["manual_review_required"] is False
+
+
 def test_http_exception_handler_is_registered():
     assert app.exception_handlers[HTTPException] is http_exception_handler
 

apps/api/tests/test_knowledge_repository_read_model.py

@@ -128,6 +128,9 @@ class _MappingResult:
     def all(self):
         return self.rows
 
+    def one(self):
+        return self.rows[0]
+
 
 class _RowsResult:
     def __init__(self, rows: list[SimpleNamespace]):
@@ -224,11 +227,14 @@ async def test_get_asset_taxonomy_counts_uses_stable_ai_automation_lenses() -> N
     class _TaxonomyDb:
         def __init__(self):
             self.statements = []
-            self.values = list(range(10, 10 + len(_ASSET_TAXONOMY_KEYS)))
+            self.values = {
+                key: count
+                for count, key in enumerate(_ASSET_TAXONOMY_KEYS, start=10)
+            }
 
         async def execute(self, statement):
             self.statements.append(statement)
-            return _ScalarResult(self.values.pop(0))
+            return _MappingResult([self.values])
 
     db = _TaxonomyDb()
     repo = KnowledgeDBRepository(db=db)  # type: ignore[arg-type]
@@ -237,6 +243,7 @@ async def test_get_asset_taxonomy_counts_uses_stable_ai_automation_lenses() -> N
 
     assert [key for key, _ in counts] == list(_ASSET_TAXONOMY_KEYS)
     assert [count for _, count in counts] == list(range(10, 10 + len(_ASSET_TAXONOMY_KEYS)))
+    assert len(db.statements) == 1
     compiled = "\n".join(
         str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
         for stmt in db.statements

apps/api/tests/test_knowledge_service_readback_degraded.py

@@ -1,6 +1,12 @@
 import pytest
 
-from src.models.knowledge import KnowledgeListResponse
+from src.models.knowledge import (
+    EntrySource,
+    EntryStatus,
+    EntryType,
+    KnowledgeEntry,
+    KnowledgeListResponse,
+)
 from src.services import knowledge_service as knowledge_service_module
 from src.services.knowledge_service import KnowledgeService
 
@@ -13,6 +19,14 @@ class _BrokenDbContext:
         return False
 
 
+class _OkDbContext:
+    async def __aenter__(self):
+        return object()
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False
+
+
 @pytest.mark.asyncio
 async def test_knowledge_list_entries_fails_soft_when_readback_breaks(monkeypatch) -> None:
     monkeypatch.setattr(
@@ -108,6 +122,56 @@ async def test_knowledge_list_entries_retries_transient_pool_timeout(monkeypatch
     assert response.operator_stage == "knowledge_readback_primary_retry_recovered"
 
 
+@pytest.mark.asyncio
+async def test_knowledge_list_entries_keeps_primary_items_when_taxonomy_degrades(monkeypatch) -> None:
+    primary_entry = KnowledgeEntry(
+        id="km-primary-1",
+        title="Primary KM row",
+        content="Persistent AI automation memory row",
+        entry_type=EntryType.RUNBOOK,
+        category="alert_handling",
+        tags=["telegram", "ai_agent"],
+        source=EntrySource.AI_EXTRACTED,
+        status=EntryStatus.APPROVED,
+    )
+
+    class _TaxonomyBrokenRepo:
+        def __init__(self, _db):
+            pass
+
+        async def list_entries(self, **_kwargs):
+            return [primary_entry], 638
+
+        async def get_categories(self):
+            raise RuntimeError("db pool exhausted")
+
+        async def get_asset_taxonomy_counts(self):
+            raise RuntimeError("db pool exhausted")
+
+    monkeypatch.setattr(
+        knowledge_service_module,
+        "get_db_context",
+        lambda: _OkDbContext(),
+    )
+    monkeypatch.setattr(
+        knowledge_service_module,
+        "KnowledgeDBRepository",
+        _TaxonomyBrokenRepo,
+    )
+    service = KnowledgeService.__new__(KnowledgeService)
+
+    response = await service.list_entries(limit=50)
+
+    assert response.total == 638
+    assert response.items == [primary_entry]
+    assert response.readback_status == "ready_partial_taxonomy_degraded"
+    assert response.primary_readback_ready is True
+    assert response.degraded_reason_code == "primary_km_taxonomy_readback_degraded"
+    assert response.operator_stage == "knowledge_primary_entries_ready_taxonomy_degraded"
+    assert response.categories[[row.category for row in response.categories].index("alert_handling")].count == 1
+    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(