fix(km): recover primary knowledge readback via direct fallback

apps/api/src/services/knowledge_service.py

@@ -13,9 +13,14 @@ Knowledge Base Phase 1: CRUD + 狀態流轉 + 搜尋
 """
 
 import asyncio
+import json
+from collections.abc import Mapping
+from typing import Any, cast
 
 import structlog
 
+from src.core.config import settings
+from src.core.context import get_current_project_id
 from src.db.base import get_db_context
 from src.models.knowledge import (
     CategoryCount,
@@ -32,6 +37,7 @@ from src.repositories.interfaces import IKnowledgeRepository
 from src.repositories.knowledge_repository import KnowledgeDBRepository
 from src.services.embedding_service import OllamaEmbeddingService
 from src.services.model_registry import get_model as _get_model
+from src.utils.timezone import now_taipei
 
 logger = structlog.get_logger(__name__)
 
@@ -57,6 +63,10 @@ _PRIMARY_KM_LIST_RETRY_TIMEOUT_SECONDS = 1.0
 _PRIMARY_KM_RETRY_DELAY_SECONDS = 0.1
 _PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS = 1.2
 _PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS = 1.5
+_PRIMARY_KM_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5
+_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS = 3.0
+_PRIMARY_KM_DIRECT_CLOSE_TIMEOUT_SECONDS = 0.5
+_PRIMARY_KM_DIRECT_STATEMENT_TIMEOUT_MS = 3500
 _ASSET_TAXONOMY_CATEGORY_HINTS: dict[str, tuple[str, ...]] = {
     "service": (
         "infrastructure",
@@ -71,6 +81,81 @@ _ASSET_TAXONOMY_CATEGORY_HINTS: dict[str, tuple[str, ...]] = {
     "alert": ("alert_handling",),
     "playbook": ("auto_repair",),
 }
+_ASSET_TAXONOMY_TERMS: dict[str, tuple[str, ...]] = {
+    "project": (
+        "project",
+        "repo",
+        "repository",
+        "gitea",
+        "branch",
+        "workflow",
+        "source_control",
+    ),
+    "product": (
+        "product",
+        "awoooi",
+        "awooop",
+        "iwooos",
+        "stockplatform",
+        "momo",
+        "awooogo",
+        "agent-bounty",
+        "agent_bounty",
+        "tsenyang",
+        "vibework",
+        "2026fifa",
+    ),
+    "website": ("website", "site", "nginx", "ssl", "domain", "route", "frontend"),
+    "service": ("service", "daemon", "api", "worker", "runtime", "container", "pod"),
+    "package": (
+        "package",
+        "dependency",
+        "npm",
+        "pnpm",
+        "node",
+        "python",
+        "pip",
+        "prisma",
+        "next",
+        "library",
+    ),
+    "tool": (
+        "tool",
+        "ansible",
+        "mcp",
+        "playbook",
+        "telegram",
+        "wazuh",
+        "kali",
+        "sentry",
+        "signoz",
+        "runner",
+    ),
+    "log": (
+        "log",
+        "logs",
+        "event",
+        "timeline",
+        "trace",
+        "audit",
+        "callback",
+        "conversation_event",
+        "telemetry",
+    ),
+    "alert": (
+        "alert",
+        "telegram",
+        "sentry",
+        "signoz",
+        "notification",
+        "warning",
+        "critical",
+    ),
+    "playbook": ("playbook", "runbook", "sop"),
+    "rag": ("rag", "vector", "embedding", "semantic", "retrieval"),
+    "mcp": ("mcp", "connector", "gateway", "tool integration", "tool-integration"),
+    "schedule": ("schedule", "cron", "job", "worker", "patrol", "recurrence", "cadence"),
+}
 
 _SOURCE_BACKED_KNOWLEDGE_SPECS: tuple[dict[str, object], ...] = (
     {
@@ -254,6 +339,155 @@ def _knowledge_readback_exception_reason(exc: BaseException) -> str:
         return "TimeoutError"
     return str(exc) or type(exc).__name__
 
+
+def _current_knowledge_project_id() -> str | None:
+    project_id = get_current_project_id()
+    return str(project_id).strip() if project_id else None
+
+
+def _normalize_direct_tags(value: object) -> list[str]:
+    if value is None:
+        return []
+    if isinstance(value, list | tuple | set):
+        return [str(tag).strip() for tag in value if str(tag).strip()]
+    if isinstance(value, str):
+        stripped = value.strip()
+        if not stripped:
+            return []
+        try:
+            parsed = json.loads(stripped)
+        except json.JSONDecodeError:
+            parsed = None
+        if isinstance(parsed, list):
+            return [str(tag).strip() for tag in parsed if str(tag).strip()]
+        return [stripped]
+    return [str(value).strip()] if str(value).strip() else []
+
+
+def _direct_enum_member(value: object, enum_cls: type[Any], fallback: Any) -> Any:
+    raw = str(getattr(value, "value", value) or "").lower()
+    for member in enum_cls:
+        if raw in {member.value.lower(), member.name.lower()}:
+            return member
+    return fallback
+
+
+def _direct_knowledge_entry_from_row(row: Mapping[str, Any]) -> KnowledgeEntry:
+    return KnowledgeEntry(
+        id=str(row.get("id")),
+        title=str(row.get("title") or "Untitled KM entry"),
+        content=str(row.get("content") or ""),
+        entry_type=_direct_enum_member(
+            row.get("entry_type"),
+            EntryType,
+            EntryType.BEST_PRACTICE,
+        ),
+        category=str(row.get("category") or "general").strip() or "general",
+        tags=_normalize_direct_tags(row.get("tags")),
+        source=_direct_enum_member(
+            row.get("source"),
+            EntrySource,
+            EntrySource.AI_EXTRACTED,
+        ),
+        status=_direct_enum_member(
+            row.get("status"),
+            EntryStatus,
+            EntryStatus.REVIEW,
+        ),
+        related_incident_id=row.get("related_incident_id"),
+        related_playbook_id=row.get("related_playbook_id"),
+        related_approval_id=row.get("related_approval_id"),
+        path_type=row.get("path_type"),
+        symptoms_hash=row.get("symptoms_hash"),
+        view_count=int(row.get("view_count") or 0),
+        created_by=row.get("created_by"),
+        created_at=row.get("created_at") or now_taipei(),
+        updated_at=row.get("updated_at") or row.get("created_at") or now_taipei(),
+    )
+
+
+def _direct_tag_needle(tag: str) -> str:
+    return f'"{tag.replace(chr(34), chr(92) + chr(34))}"'
+
+
+def _direct_base_where(project_id: str) -> tuple[list[str], list[Any]]:
+    return [
+        "project_id = $1",
+        "lower(status::text) != 'archived'",
+    ], [project_id]
+
+
+def _append_direct_filter(
+    clauses: list[str],
+    params: list[Any],
+    *,
+    category: str | None = None,
+    entry_type: EntryType | None = None,
+    status: EntryStatus | None = None,
+    tags: list[str] | None = None,
+    q: str | None = None,
+) -> None:
+    if category:
+        params.append(str(category).strip() or "general")
+        clauses.append(
+            "coalesce(nullif(trim(category), ''), 'general') = "
+            f"${len(params)}"
+        )
+    if entry_type:
+        params.append(entry_type.value)
+        clauses.append(f"lower(entry_type::text) = ${len(params)}")
+    if status:
+        params.append(status.value)
+        clauses.append(f"lower(status::text) = ${len(params)}")
+    for tag in tags or []:
+        params.append(_direct_tag_needle(tag))
+        clauses.append(f"strpos(tags::text, ${len(params)}) > 0")
+    if q:
+        params.append(f"%{q}%")
+        index = len(params)
+        clauses.append(
+            "("
+            f"title ILIKE ${index} OR "
+            f"content ILIKE ${index} OR "
+            f"tags::text ILIKE ${index} OR "
+            f"coalesce(nullif(trim(category), ''), 'general') ILIKE ${index}"
+            ")"
+        )
+
+
+def _direct_where_sql(clauses: list[str]) -> str:
+    return " WHERE " + " AND ".join(clauses)
+
+
+def _sql_literal(value: str) -> str:
+    return "'" + value.replace("'", "''") + "'"
+
+
+def _asset_taxonomy_direct_condition(key: str) -> str:
+    category_expr = "coalesce(nullif(trim(category), ''), 'general')"
+    category_text = f"lower({category_expr}::text)"
+    conditions: list[str] = []
+    category_hints = _ASSET_TAXONOMY_CATEGORY_HINTS.get(key, ())
+    if category_hints:
+        conditions.append(
+            f"{category_text} IN ("
+            + ", ".join(_sql_literal(hint.lower()) for hint in category_hints)
+            + ")"
+        )
+    for term in _ASSET_TAXONOMY_TERMS.get(key, ()):
+        pattern = _sql_literal(f"%{term.lower()}%")
+        conditions.extend(
+            [
+                f"lower(coalesce(title, '')) LIKE {pattern}",
+                f"lower(coalesce(content, '')) LIKE {pattern}",
+                f"lower(coalesce(tags::text, '')) LIKE {pattern}",
+                f"{category_text} LIKE {pattern}",
+            ]
+        )
+    if key == "playbook":
+        conditions.append("related_playbook_id IS NOT NULL")
+    return " OR ".join(conditions) if conditions else f"{category_text} = {_sql_literal(key)}"
+
 # =============================================================================
 # Singleton
 # =============================================================================
@@ -303,14 +537,14 @@ def build_knowledge_list_readback_degraded_response(
 def _source_backed_entries() -> list[KnowledgeEntry]:
     entries: list[KnowledgeEntry] = []
     for spec in _SOURCE_BACKED_KNOWLEDGE_SPECS:
-        tags = [str(tag) for tag in spec.get("tags", [])]
+        tags = [str(tag) for tag in cast(list[object], spec.get("tags", []))]
         related_playbook_id = spec.get("related_playbook_id")
         entries.append(
             KnowledgeEntry(
                 id=str(spec["id"]),
                 title=str(spec["title"]),
                 content=str(spec["content"]),
-                entry_type=spec["entry_type"],
+                entry_type=cast(EntryType, spec["entry_type"]),
                 category=str(spec["category"]),
                 tags=tags,
                 source=EntrySource.AI_EXTRACTED,
@@ -639,6 +873,295 @@ class KnowledgeService:
             timeout=timeout,
         )
 
+    async def _list_entries_from_direct(
+        self,
+        *,
+        category: str | None,
+        entry_type: EntryType | None,
+        status: EntryStatus | None,
+        tags: list[str] | None,
+        q: str | None,
+        limit: int,
+        offset: int,
+    ) -> KnowledgeListResponse | None:
+        """Pool 壓力下以短連線只讀救回 primary KM，不把 727 筆誤降成 13 筆。"""
+        project_id = _current_knowledge_project_id()
+        if not project_id:
+            return None
+
+        db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
+        try:
+            import asyncpg  # type: ignore[import-untyped]
+
+            conn = await asyncio.wait_for(
+                asyncpg.connect(db_url),
+                timeout=_PRIMARY_KM_DIRECT_CONNECT_TIMEOUT_SECONDS,
+            )
+            try:
+                await asyncio.wait_for(
+                    conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id),
+                    timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+                )
+                await asyncio.wait_for(
+                    conn.execute(
+                        "SET statement_timeout = "
+                        f"'{int(_PRIMARY_KM_DIRECT_STATEMENT_TIMEOUT_MS)}ms'"
+                    ),
+                    timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+                )
+                clauses, params = _direct_base_where(project_id)
+                _append_direct_filter(
+                    clauses,
+                    params,
+                    category=category,
+                    entry_type=entry_type,
+                    status=status,
+                    tags=tags,
+                    q=q,
+                )
+                where_sql = _direct_where_sql(clauses)
+                total = int(
+                    await asyncio.wait_for(
+                        conn.fetchval(
+                            f"SELECT count(*)::int FROM knowledge_entries{where_sql}",
+                            *params,
+                        ),
+                        timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+                    )
+                    or 0
+                )
+                if total == 0:
+                    return build_knowledge_list_readback_degraded_response(
+                        "knowledge_db_empty_direct_connection",
+                        category=category,
+                        entry_type=entry_type,
+                        status=status,
+                        tags=tags,
+                        q=q,
+                        limit=limit,
+                        offset=offset,
+                        readback_status="source_backed_db_empty_direct_connection",
+                    )
+
+                list_params = [*params, limit, offset]
+                limit_index = len(params) + 1
+                offset_index = len(params) + 2
+                rows = await asyncio.wait_for(
+                    conn.fetch(
+                        f"""
+                        SELECT
+                            id::text AS id,
+                            title,
+                            content,
+                            entry_type::text AS entry_type,
+                            coalesce(nullif(trim(category), ''), 'general') AS category,
+                            tags,
+                            source::text AS source,
+                            status::text AS status,
+                            related_incident_id,
+                            related_playbook_id,
+                            related_approval_id,
+                            path_type,
+                            symptoms_hash,
+                            coalesce(view_count, 0)::int AS view_count,
+                            created_by,
+                            created_at,
+                            updated_at
+                        FROM knowledge_entries
+                        {where_sql}
+                        ORDER BY updated_at DESC NULLS LAST, created_at DESC NULLS LAST
+                        LIMIT ${limit_index} OFFSET ${offset_index}
+                        """,
+                        *list_params,
+                    ),
+                    timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+                )
+                items = [_direct_knowledge_entry_from_row(dict(row)) for row in rows]
+                categories = await self._read_categories_direct_with_conn(
+                    conn,
+                    project_id=project_id,
+                )
+                asset_taxonomy = await self._read_asset_taxonomy_direct_with_conn(
+                    conn,
+                    project_id=project_id,
+                )
+                return KnowledgeListResponse(
+                    items=items,
+                    total=total,
+                    categories=categories or _source_category_counts(items),
+                    asset_taxonomy=asset_taxonomy or _source_asset_taxonomy_counts(items),
+                    readback_status="ready_direct_connection_after_session_timeout",
+                    primary_readback_ready=True,
+                    operator_stage="knowledge_readback_direct_connection_recovered",
+                    next_step="repair_session_pool_readback_without_hiding_primary_km",
+                    writes_on_read=False,
+                    manual_review_required=False,
+                )
+            finally:
+                try:
+                    await asyncio.wait_for(
+                        conn.close(),
+                        timeout=_PRIMARY_KM_DIRECT_CLOSE_TIMEOUT_SECONDS,
+                    )
+                except Exception as close_exc:  # pragma: no cover - live cleanup
+                    logger.warning(
+                        "knowledge_direct_readback_close_failed",
+                        project_id=project_id,
+                        error_type=type(close_exc).__name__,
+                    )
+        except Exception as exc:  # pragma: no cover - live DB pressure
+            logger.warning(
+                "knowledge_direct_readback_failed",
+                project_id=project_id,
+                error_type=type(exc).__name__,
+            )
+            return None
+
+    async def _read_categories_direct(self) -> list[CategoryCount] | None:
+        project_id = _current_knowledge_project_id()
+        if not project_id:
+            return None
+        db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
+        try:
+            import asyncpg  # type: ignore[import-untyped]
+
+            conn = await asyncio.wait_for(
+                asyncpg.connect(db_url),
+                timeout=_PRIMARY_KM_DIRECT_CONNECT_TIMEOUT_SECONDS,
+            )
+            try:
+                await asyncio.wait_for(
+                    conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id),
+                    timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+                )
+                await asyncio.wait_for(
+                    conn.execute(
+                        "SET statement_timeout = "
+                        f"'{int(_PRIMARY_KM_DIRECT_STATEMENT_TIMEOUT_MS)}ms'"
+                    ),
+                    timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+                )
+                return await self._read_categories_direct_with_conn(
+                    conn,
+                    project_id=project_id,
+                )
+            finally:
+                await asyncio.wait_for(
+                    conn.close(),
+                    timeout=_PRIMARY_KM_DIRECT_CLOSE_TIMEOUT_SECONDS,
+                )
+        except Exception as exc:  # pragma: no cover - live DB pressure
+            logger.warning(
+                "knowledge_categories_direct_readback_failed",
+                project_id=project_id,
+                error_type=type(exc).__name__,
+            )
+            return None
+
+    async def _read_asset_taxonomy_direct(self) -> list[KnowledgeAssetTaxonomyCount] | None:
+        project_id = _current_knowledge_project_id()
+        if not project_id:
+            return None
+        db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://")
+        try:
+            import asyncpg  # type: ignore[import-untyped]
+
+            conn = await asyncio.wait_for(
+                asyncpg.connect(db_url),
+                timeout=_PRIMARY_KM_DIRECT_CONNECT_TIMEOUT_SECONDS,
+            )
+            try:
+                await asyncio.wait_for(
+                    conn.execute("SELECT set_config('app.project_id', $1, TRUE)", project_id),
+                    timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+                )
+                await asyncio.wait_for(
+                    conn.execute(
+                        "SET statement_timeout = "
+                        f"'{int(_PRIMARY_KM_DIRECT_STATEMENT_TIMEOUT_MS)}ms'"
+                    ),
+                    timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+                )
+                return await self._read_asset_taxonomy_direct_with_conn(
+                    conn,
+                    project_id=project_id,
+                )
+            finally:
+                await asyncio.wait_for(
+                    conn.close(),
+                    timeout=_PRIMARY_KM_DIRECT_CLOSE_TIMEOUT_SECONDS,
+                )
+        except Exception as exc:  # pragma: no cover - live DB pressure
+            logger.warning(
+                "knowledge_asset_taxonomy_direct_readback_failed",
+                project_id=project_id,
+                error_type=type(exc).__name__,
+            )
+            return None
+
+    async def _read_categories_direct_with_conn(
+        self,
+        conn: Any,
+        *,
+        project_id: str,
+    ) -> list[CategoryCount]:
+        rows = await asyncio.wait_for(
+            conn.fetch(
+                """
+                SELECT
+                    coalesce(nullif(trim(category), ''), 'general') AS category,
+                    count(*)::int AS cnt
+                FROM knowledge_entries
+                WHERE project_id = $1
+                  AND lower(status::text) != 'archived'
+                GROUP BY 1
+                ORDER BY count(*) DESC, category ASC
+                """,
+                project_id,
+            ),
+            timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+        )
+        return [
+            CategoryCount(
+                category=str(row["category"] or "general"),
+                count=int(row["cnt"] or 0),
+            )
+            for row in rows
+        ]
+
+    async def _read_asset_taxonomy_direct_with_conn(
+        self,
+        conn: Any,
+        *,
+        project_id: str,
+    ) -> list[KnowledgeAssetTaxonomyCount]:
+        select_columns = ",\n".join(
+            (
+                "count(*) FILTER (WHERE "
+                f"({_asset_taxonomy_direct_condition(key)}))::int AS {key}"
+            )
+            for key in _ASSET_TAXONOMY_FALLBACKS
+        )
+        row = await asyncio.wait_for(
+            conn.fetchrow(
+                f"""
+                SELECT
+                    {select_columns}
+                FROM knowledge_entries
+                WHERE project_id = $1
+                  AND lower(status::text) != 'archived'
+                """,
+                project_id,
+            ),
+            timeout=_PRIMARY_KM_DIRECT_QUERY_TIMEOUT_SECONDS,
+        )
+        if row is None:
+            return []
+        return [
+            KnowledgeAssetTaxonomyCount(key=key, count=int(row[key] or 0))
+            for key in _ASSET_TAXONOMY_FALLBACKS
+        ]
+
     async def list_entries(
         self,
         category: str | None = None,
@@ -689,6 +1212,17 @@ class KnowledgeService:
                         limit=limit,
                         offset=offset,
                     )
+                direct_response = await self._list_entries_from_direct(
+                    category=category,
+                    entry_type=entry_type,
+                    status=status,
+                    tags=tags,
+                    q=q,
+                    limit=limit,
+                    offset=offset,
+                )
+                if direct_response is not None:
+                    return direct_response
             logger.warning(
                 "knowledge_list_readback_degraded",
                 error=degraded_reason,
@@ -723,6 +1257,10 @@ class KnowledgeService:
             return _source_asset_taxonomy_counts(_source_backed_entries())
         except Exception as exc:  # noqa: BLE001 - taxonomy must not 500 the KM UI
             reason = _knowledge_readback_exception_reason(exc)
+            if _classify_knowledge_readback_degraded_reason(reason) == "primary_km_db_timeout_or_pool_exhausted":
+                direct_taxonomy = await self._read_asset_taxonomy_direct()
+                if direct_taxonomy and any(row.count > 0 for row in direct_taxonomy):
+                    return direct_taxonomy
             logger.warning(
                 "knowledge_asset_taxonomy_readback_degraded",
                 error=reason,
@@ -742,6 +1280,10 @@ class KnowledgeService:
             return _source_category_counts(_source_backed_entries())
         except Exception as exc:  # noqa: BLE001 - categories must not 500 the KM UI
             reason = _knowledge_readback_exception_reason(exc)
+            if _classify_knowledge_readback_degraded_reason(reason) == "primary_km_db_timeout_or_pool_exhausted":
+                direct_categories = await self._read_categories_direct()
+                if direct_categories and any(row.count > 0 for row in direct_categories):
+                    return direct_categories
             logger.warning(
                 "knowledge_categories_readback_degraded",
                 error=reason,

apps/api/tests/test_knowledge_service_readback_degraded.py

@@ -3,9 +3,11 @@ import asyncio
 import pytest
 
 from src.models.knowledge import (
+    CategoryCount,
     EntrySource,
     EntryStatus,
     EntryType,
+    KnowledgeAssetTaxonomyCount,
     KnowledgeEntry,
     KnowledgeListResponse,
 )
@@ -148,6 +150,71 @@ async def test_knowledge_list_entries_bounds_primary_timeout_to_source_backed(mo
     assert response.total == 13
 
 
+@pytest.mark.asyncio
+async def test_knowledge_list_entries_uses_direct_readback_after_session_pool_timeout(monkeypatch) -> None:
+    service = KnowledgeService.__new__(KnowledgeService)
+    primary_calls = 0
+    direct_calls = 0
+    primary_entry = KnowledgeEntry(
+        id="km-primary-live-1",
+        title="Primary KM live row",
+        content="Persistent KM row recovered through direct readback",
+        entry_type=EntryType.RUNBOOK,
+        category="alert_handling",
+        tags=["telegram", "ai_agent", "log"],
+        source=EntrySource.AI_EXTRACTED,
+        status=EntryStatus.APPROVED,
+    )
+
+    async def pool_exhausted_primary(**_kwargs):
+        nonlocal primary_calls
+        primary_calls += 1
+        raise RuntimeError("db pool exhausted")
+
+    async def direct_readback(**kwargs):
+        nonlocal direct_calls
+        direct_calls += 1
+        assert kwargs["limit"] == 50
+        return KnowledgeListResponse(
+            items=[primary_entry],
+            total=727,
+            categories=[
+                CategoryCount(category="alert_handling", count=113),
+                CategoryCount(category="AI自動化/Ansible受控修復", count=87),
+            ],
+            asset_taxonomy=[
+                KnowledgeAssetTaxonomyCount(key="log", count=727),
+                KnowledgeAssetTaxonomyCount(key="alert", count=443),
+                KnowledgeAssetTaxonomyCount(key="mcp", count=36),
+            ],
+            readback_status="ready_direct_connection_after_session_timeout",
+            primary_readback_ready=True,
+            operator_stage="knowledge_readback_direct_connection_recovered",
+            next_step="repair_session_pool_readback_without_hiding_primary_km",
+            writes_on_read=False,
+            manual_review_required=False,
+        )
+
+    monkeypatch.setattr(service, "_list_entries_from_primary", pool_exhausted_primary)
+    monkeypatch.setattr(service, "_list_entries_from_direct", direct_readback)
+    monkeypatch.setattr(knowledge_service_module, "_PRIMARY_KM_RETRY_DELAY_SECONDS", 0)
+
+    response = await service.list_entries(limit=50)
+
+    assert primary_calls == 2
+    assert direct_calls == 1
+    assert response.total == 727
+    assert response.items == [primary_entry]
+    assert response.readback_status == "ready_direct_connection_after_session_timeout"
+    assert response.primary_readback_ready is True
+    assert response.operator_stage == "knowledge_readback_direct_connection_recovered"
+    assert response.next_step == "repair_session_pool_readback_without_hiding_primary_km"
+    assert response.categories[0].category == "alert_handling"
+    assert response.asset_taxonomy[0].key == "log"
+    assert response.writes_on_read is False
+    assert response.manual_review_required is False
+
+
 @pytest.mark.asyncio
 async def test_knowledge_list_entries_keeps_primary_items_when_taxonomy_degrades(monkeypatch) -> None:
     primary_entry = KnowledgeEntry(
@@ -247,6 +314,30 @@ async def test_knowledge_categories_fails_soft_when_readback_breaks(monkeypatch)
     assert all(row.count == 1 for row in categories)
 
 
+@pytest.mark.asyncio
+async def test_knowledge_categories_use_direct_readback_when_pool_is_busy(monkeypatch) -> None:
+    service = KnowledgeService.__new__(KnowledgeService)
+
+    async def pool_exhausted_categories():
+        raise RuntimeError("db pool exhausted")
+
+    async def direct_categories():
+        return [
+            CategoryCount(category="alert_handling", count=113),
+            CategoryCount(category="AI自動化/Ansible受控修復", count=87),
+        ]
+
+    monkeypatch.setattr(service, "_read_primary_categories", pool_exhausted_categories)
+    monkeypatch.setattr(service, "_read_categories_direct", direct_categories)
+
+    categories = await service.get_categories()
+
+    assert [(row.category, row.count) for row in categories] == [
+        ("alert_handling", 113),
+        ("AI自動化/Ansible受控修復", 87),
+    ]
+
+
 @pytest.mark.asyncio
 async def test_knowledge_asset_taxonomy_fails_soft_when_readback_breaks(monkeypatch) -> None:
     monkeypatch.setattr(
@@ -275,6 +366,32 @@ async def test_knowledge_asset_taxonomy_fails_soft_when_readback_breaks(monkeypa
     assert all(row.count >= 1 for row in taxonomy)
 
 
+@pytest.mark.asyncio
+async def test_knowledge_asset_taxonomy_uses_direct_readback_when_pool_is_busy(monkeypatch) -> None:
+    service = KnowledgeService.__new__(KnowledgeService)
+
+    async def pool_exhausted_taxonomy():
+        raise RuntimeError("db pool exhausted")
+
+    async def direct_taxonomy():
+        return [
+            KnowledgeAssetTaxonomyCount(key="log", count=727),
+            KnowledgeAssetTaxonomyCount(key="alert", count=443),
+            KnowledgeAssetTaxonomyCount(key="mcp", count=36),
+        ]
+
+    monkeypatch.setattr(service, "_read_primary_asset_taxonomy", pool_exhausted_taxonomy)
+    monkeypatch.setattr(service, "_read_asset_taxonomy_direct", direct_taxonomy)
+
+    taxonomy = await service.get_asset_taxonomy()
+
+    assert [(row.key, row.count) for row in taxonomy] == [
+        ("log", 727),
+        ("alert", 443),
+        ("mcp", 36),
+    ]
+
+
 @pytest.mark.asyncio
 async def test_knowledge_side_readbacks_bound_timeouts_to_source_backed(monkeypatch) -> None:
     service = KnowledgeService.__new__(KnowledgeService)