From a63e645513f05b5cfb5d290df520b479e19543c3 Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 01:49:47 +0800 Subject: [PATCH] fix(km): recover primary knowledge readback via direct fallback --- apps/api/src/services/knowledge_service.py | 546 +++++++++++++++++- ...est_knowledge_service_readback_degraded.py | 117 ++++ apps/web/messages/en.json | 6 +- apps/web/messages/zh-TW.json | 6 +- .../src/app/[locale]/knowledge-base/page.tsx | 45 +- 5 files changed, 704 insertions(+), 16 deletions(-) diff --git a/apps/api/src/services/knowledge_service.py b/apps/api/src/services/knowledge_service.py index 63152d15d..02df79ace 100644 --- a/apps/api/src/services/knowledge_service.py +++ b/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, diff --git a/apps/api/tests/test_knowledge_service_readback_degraded.py b/apps/api/tests/test_knowledge_service_readback_degraded.py index 15b9a6e99..94a120774 100644 --- a/apps/api/tests/test_knowledge_service_readback_degraded.py +++ b/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) diff --git a/apps/web/messages/en.json b/apps/web/messages/en.json index b7634b84b..d48e5d1e0 100644 --- a/apps/web/messages/en.json +++ b/apps/web/messages/en.json @@ -2222,11 +2222,15 @@ "sourceBackedTitle": "AI source-backed KM 已接管", "sourceBackedSummary": "資料未歸零", "sourceBackedDescription": "Primary KM DB readback 正在受控修復；目前顯示 source-backed 條目、分類與資產 taxonomy，不是知識庫歸零。原因：{reason}；階段：{stage}；下一步：{next}。", + "directRecoveredTitle": "Primary KM 已由 direct readback 救回", + "directRecoveredSummary": "真實 KM 條目已恢復顯示；session pool 仍在受控修復。", + "directRecoveredDescription": "主知識條目 API 透過短連線 direct readback 取回 primary KM、分類與 AI taxonomy；讀取不寫入，下一步修復 session pool。", "degradedTitle": "知識條目讀取降級", "degradedSummary": "Primary readback 降級；列表保留可用資料。", "degradedDescription": "主知識條目 API 已回 degraded：{stage}；下一步：{next}。目前列表不視為真實歸零，AI 受控修復隊列與 RAG / PlayBook readback 仍會保留。", "mode": { - "sourceBacked": "接管" + "sourceBacked": "接管", + "directRecovered": "Direct 救回" }, "chip": { "mode": "模式", diff --git a/apps/web/messages/zh-TW.json b/apps/web/messages/zh-TW.json index ccf5b91e7..a5e836668 100644 --- a/apps/web/messages/zh-TW.json +++ b/apps/web/messages/zh-TW.json @@ -2222,11 +2222,15 @@ "sourceBackedTitle": "AI source-backed KM 已接管", "sourceBackedSummary": "資料未歸零", "sourceBackedDescription": "Primary KM DB readback 正在受控修復；目前顯示 source-backed 條目、分類與資產 taxonomy，不是知識庫歸零。原因：{reason}；階段：{stage}；下一步：{next}。", + "directRecoveredTitle": "Primary KM 已由 direct readback 救回", + "directRecoveredSummary": "真實 KM 條目已恢復顯示；session pool 仍在受控修復。", + "directRecoveredDescription": "主知識條目 API 透過短連線 direct readback 取回 primary KM、分類與 AI taxonomy；讀取不寫入，下一步修復 session pool。", "degradedTitle": "知識條目讀取降級", "degradedSummary": "Primary readback 降級；列表保留可用資料。", "degradedDescription": "主知識條目 API 已回 degraded：{stage}；下一步：{next}。目前列表不視為真實歸零，AI 受控修復隊列與 RAG / PlayBook readback 仍會保留。", "mode": { - "sourceBacked": "接管" + "sourceBacked": "接管", + "directRecovered": "Direct 救回" }, "chip": { "mode": "模式", diff --git a/apps/web/src/app/[locale]/knowledge-base/page.tsx b/apps/web/src/app/[locale]/knowledge-base/page.tsx index 3cf92d620..090d04dbd 100644 --- a/apps/web/src/app/[locale]/knowledge-base/page.tsx +++ b/apps/web/src/app/[locale]/knowledge-base/page.tsx @@ -64,6 +64,7 @@ interface ListResponse { categories: CategoryCount[] asset_taxonomy?: AssetTaxonomyCount[] readback_status?: 'ready' | 'degraded' | string + primary_readback_ready?: boolean degraded_reason_code?: string | null operator_stage?: string | null next_step?: string | null @@ -441,8 +442,12 @@ const humanizeSignalTag = (tag: string) => { const compactReadbackToken = (value: string | null | undefined) => { if (!value) return '--' if (value === 'primary_km_db_timeout_or_pool_exhausted') return 'DB pool' + if (value === 'ready_direct_connection_after_session_timeout') return 'Direct KM' + if (value === 'knowledge_readback_direct_connection_recovered') return '已救回' + if (value === 'repair_session_pool_readback_without_hiding_primary_km') return '修復 pool' if (value === 'knowledge_readback_source_backed_ai_controlled_repair') return '接管' if (value === 'repair_primary_km_db_readback_then_promote_source_backed_receipts_to_persistent_km') return '修復 KM' + if (value === 'source_backed_db_empty_direct_connection') return 'Direct 空值' if (value === 'primary_km_taxonomy_readback_degraded') return 'Taxonomy 降級' if (value === 'knowledge_primary_entries_ready_taxonomy_degraded') return 'Primary 可讀' if (value === 'repair_primary_km_taxonomy_readback_without_hiding_entries') return '修復 taxonomy' @@ -509,7 +514,7 @@ export default function KnowledgeBasePage({ const [assetTaxonomy, setAssetTaxonomy] = useState(ASSET_TAXONOMY_FALLBACK_COUNTS) const [loading, setLoading] = useState(true) const [entryFetchError, setEntryFetchError] = useState(null) - const [entryReadback, setEntryReadback] = useState | null>(null) + const [entryReadback, setEntryReadback] = useState | null>(null) const [governanceTelemetry, setGovernanceTelemetry] = useState({ staleCandidates: null, ownerReviews: null, @@ -554,6 +559,7 @@ export default function KnowledgeBasePage({ setAssetTaxonomy(data.asset_taxonomy?.length ? data.asset_taxonomy : ASSET_TAXONOMY_FALLBACK_COUNTS) setEntryReadback({ readback_status: data.readback_status ?? 'ready', + primary_readback_ready: data.primary_readback_ready ?? false, degraded_reason_code: data.degraded_reason_code ?? null, operator_stage: data.operator_stage ?? null, next_step: data.next_step ?? null, @@ -701,10 +707,16 @@ export default function KnowledgeBasePage({ const totalCount = categories.reduce((sum, c) => sum + c.count, 0) const isKnowledgeListUnavailable = Boolean(entryFetchError) const readbackStatus = entryReadback?.readback_status ?? null + const isKnowledgeDirectRecovered = readbackStatus === 'ready_direct_connection_after_session_timeout' const isKnowledgeSourceBacked = Boolean(readbackStatus?.startsWith('source_backed')) - const isKnowledgeListDegraded = Boolean(readbackStatus && readbackStatus !== 'ready') + const isKnowledgeListDegraded = Boolean(readbackStatus && readbackStatus !== 'ready' && !isKnowledgeDirectRecovered) const hasKnowledgePayload = entries.length > 0 || total > 0 || categories.some(category => category.count > 0) const knowledgeReadbackReady = !loading && !isKnowledgeListUnavailable && (!isKnowledgeListDegraded || hasKnowledgePayload) + const showKnowledgeReadbackNotice = Boolean(entryFetchError || isKnowledgeListDegraded || isKnowledgeDirectRecovered) + const readbackNoticeTone = isKnowledgeDirectRecovered + ? 'border-status-healthy/20 bg-status-healthy/5' + : 'border-status-warning/20 bg-status-warning/5' + const readbackNoticeTitleTone = isKnowledgeDirectRecovered ? 'text-status-healthy' : 'text-status-warning' const localeCode = params.locale === 'en' ? 'en-US' : 'zh-TW' const formatCount = useCallback( (value: number) => value.toLocaleString(localeCode), @@ -714,7 +726,11 @@ export default function KnowledgeBasePage({ () => [ { label: t('dataChain.chip.mode'), - value: isKnowledgeSourceBacked ? t('dataChain.mode.sourceBacked') : compactReadbackToken(readbackStatus), + value: isKnowledgeSourceBacked + ? t('dataChain.mode.sourceBacked') + : isKnowledgeDirectRecovered + ? t('dataChain.mode.directRecovered') + : compactReadbackToken(readbackStatus), icon: CheckCircle2, tone: 'border-status-healthy/20 bg-status-healthy/10 text-status-healthy', }, @@ -757,6 +773,7 @@ export default function KnowledgeBasePage({ entryReadback?.operator_stage, formatCount, isKnowledgeSourceBacked, + isKnowledgeDirectRecovered, readbackStatus, t, total, @@ -1485,27 +1502,31 @@ export default function KnowledgeBasePage({ - {(entryFetchError || isKnowledgeListDegraded) && ( + {showKnowledgeReadbackNotice && (
-
+
-
+

{entryFetchError ? t('dataChain.errorDescription', { reason: entryFetchError }) - : isKnowledgeSourceBacked - ? t('dataChain.sourceBackedSummary') - : t('dataChain.degradedSummary')} + : isKnowledgeDirectRecovered + ? t('dataChain.directRecoveredSummary') + : isKnowledgeSourceBacked + ? t('dataChain.sourceBackedSummary') + : t('dataChain.degradedSummary')}

{!entryFetchError && (