From 0f3f4af907b284aa57095d496ca32726cc2e27d7 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 19:34:29 +0800 Subject: [PATCH] fix(km): bound knowledge readbacks under pool pressure --- apps/api/src/services/knowledge_service.py | 132 +++++++++++++----- ...est_knowledge_service_readback_degraded.py | 77 ++++++++++ apps/web/messages/en.json | 6 +- apps/web/messages/zh-TW.json | 6 +- .../src/app/[locale]/knowledge-base/page.tsx | 59 ++++---- 5 files changed, 215 insertions(+), 65 deletions(-) diff --git a/apps/api/src/services/knowledge_service.py b/apps/api/src/services/knowledge_service.py index 453071531..63152d15d 100644 --- a/apps/api/src/services/knowledge_service.py +++ b/apps/api/src/services/knowledge_service.py @@ -52,6 +52,10 @@ _DEGRADED_CATEGORY_FALLBACKS = ( ) _ASSET_TAXONOMY_FALLBACKS = _DEGRADED_CATEGORY_FALLBACKS[:-1] +_PRIMARY_KM_LIST_TIMEOUT_SECONDS = 2.0 +_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 _ASSET_TAXONOMY_CATEGORY_HINTS: dict[str, tuple[str, ...]] = { "service": ( @@ -228,7 +232,13 @@ _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: + if ( + not normalized + or "pool" in normalized + or "timeout" in normalized + or "timed out" in normalized + or "timeouterror" 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" @@ -238,6 +248,12 @@ def _classify_knowledge_readback_degraded_reason(reason: str) -> str: return "primary_km_legacy_row_validation" return "primary_km_readback_exception" + +def _knowledge_readback_exception_reason(exc: BaseException) -> str: + if isinstance(exc, TimeoutError): + return "TimeoutError" + return str(exc) or type(exc).__name__ + # ============================================================================= # Singleton # ============================================================================= @@ -589,6 +605,40 @@ class KnowledgeService: for key, count in asset_taxonomy_raw ] + async def _search_primary_entries( + self, + query: str, + limit: int, + ) -> list[KnowledgeEntry]: + async with get_db_context() as db: + repo: IKnowledgeRepository = KnowledgeDBRepository(db) + return await repo.search(query, limit) + + async def _list_entries_from_primary_bounded( + self, + *, + category: str | None, + entry_type: EntryType | None, + status: EntryStatus | None, + tags: list[str] | None, + q: str | None, + limit: int, + offset: int, + timeout: float, + ) -> KnowledgeListResponse: + return await asyncio.wait_for( + self._list_entries_from_primary( + category=category, + entry_type=entry_type, + status=status, + tags=tags, + q=q, + limit=limit, + offset=offset, + ), + timeout=timeout, + ) + async def list_entries( self, category: str | None = None, @@ -601,7 +651,7 @@ class KnowledgeService: ) -> KnowledgeListResponse: """列出知識條目 + 分類統計""" try: - return await self._list_entries_from_primary( + return await self._list_entries_from_primary_bounded( category=category, entry_type=entry_type, status=status, @@ -609,13 +659,15 @@ class KnowledgeService: q=q, limit=limit, offset=offset, + timeout=_PRIMARY_KM_LIST_TIMEOUT_SECONDS, ) except Exception as exc: # noqa: BLE001 - production readback must fail soft - degraded_reason_code = _classify_knowledge_readback_degraded_reason(str(exc)) + degraded_reason = _knowledge_readback_exception_reason(exc) + degraded_reason_code = _classify_knowledge_readback_degraded_reason(degraded_reason) if degraded_reason_code == "primary_km_db_timeout_or_pool_exhausted": try: - await asyncio.sleep(0.25) - retry_response = await self._list_entries_from_primary( + await asyncio.sleep(_PRIMARY_KM_RETRY_DELAY_SECONDS) + retry_response = await self._list_entries_from_primary_bounded( category=category, entry_type=entry_type, status=status, @@ -623,21 +675,23 @@ class KnowledgeService: q=q, limit=limit, offset=offset, + timeout=_PRIMARY_KM_LIST_RETRY_TIMEOUT_SECONDS, ) retry_response.operator_stage = "knowledge_readback_primary_retry_recovered" return retry_response except Exception as retry_exc: # noqa: BLE001 - keep source-backed fallback + retry_reason = _knowledge_readback_exception_reason(retry_exc) logger.warning( "knowledge_list_readback_retry_degraded", - error=str(retry_exc), - degraded_reason_code=_classify_knowledge_readback_degraded_reason(str(retry_exc)), + error=retry_reason, + degraded_reason_code=_classify_knowledge_readback_degraded_reason(retry_reason), q=q, limit=limit, offset=offset, ) logger.warning( "knowledge_list_readback_degraded", - error=str(exc), + error=degraded_reason, degraded_reason_code=degraded_reason_code, category=category, entry_type=entry_type.value if entry_type else None, @@ -647,7 +701,7 @@ class KnowledgeService: offset=offset, ) return build_knowledge_list_readback_degraded_response( - str(exc), + degraded_reason, category=category, entry_type=entry_type, status=status, @@ -660,50 +714,56 @@ class KnowledgeService: async def get_asset_taxonomy(self) -> list[KnowledgeAssetTaxonomyCount]: """取得 AI 自動化資產維度統計。""" try: - async with get_db_context() as db: - repo: IKnowledgeRepository = KnowledgeDBRepository(db) - rows = await repo.get_asset_taxonomy_counts() - taxonomy = [ - KnowledgeAssetTaxonomyCount(key=key, count=count) - for key, count in rows - ] - if taxonomy and any(row.count > 0 for row in taxonomy): - return taxonomy - return _source_asset_taxonomy_counts(_source_backed_entries()) + taxonomy = await asyncio.wait_for( + self._read_primary_asset_taxonomy(), + timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS, + ) + if taxonomy and any(row.count > 0 for row in taxonomy): + return taxonomy + return _source_asset_taxonomy_counts(_source_backed_entries()) except Exception as exc: # noqa: BLE001 - taxonomy must not 500 the KM UI - logger.warning("knowledge_asset_taxonomy_readback_degraded", error=str(exc)) + reason = _knowledge_readback_exception_reason(exc) + logger.warning( + "knowledge_asset_taxonomy_readback_degraded", + error=reason, + degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason), + ) return _source_asset_taxonomy_counts(_source_backed_entries()) async def get_categories(self) -> list[CategoryCount]: """取得分類統計（直接呼叫 repo，不走 list_entries）""" try: - 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 - ] - if categories: - return categories - return _source_category_counts(_source_backed_entries()) + categories = await asyncio.wait_for( + self._read_primary_categories(), + timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS, + ) + if categories: + return categories + return _source_category_counts(_source_backed_entries()) except Exception as exc: # noqa: BLE001 - categories must not 500 the KM UI - logger.warning("knowledge_categories_readback_degraded", error=str(exc)) + reason = _knowledge_readback_exception_reason(exc) + logger.warning( + "knowledge_categories_readback_degraded", + error=reason, + degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason), + ) return _source_category_counts(_source_backed_entries()) 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) + return await asyncio.wait_for( + self._search_primary_entries(query, limit), + timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS, + ) except Exception as exc: # noqa: BLE001 - KM search must not 500 the UI + reason = _knowledge_readback_exception_reason(exc) logger.warning( "knowledge_search_readback_degraded", - error=str(exc), + error=reason, q=query, limit=limit, - degraded_reason_code=_classify_knowledge_readback_degraded_reason(str(exc)), + degraded_reason_code=_classify_knowledge_readback_degraded_reason(reason), ) return _filter_source_backed_entries(q=query)[:limit] diff --git a/apps/api/tests/test_knowledge_service_readback_degraded.py b/apps/api/tests/test_knowledge_service_readback_degraded.py index 661a4ff75..15b9a6e99 100644 --- a/apps/api/tests/test_knowledge_service_readback_degraded.py +++ b/apps/api/tests/test_knowledge_service_readback_degraded.py @@ -1,3 +1,5 @@ +import asyncio + import pytest from src.models.knowledge import ( @@ -122,6 +124,30 @@ 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_bounds_primary_timeout_to_source_backed(monkeypatch) -> None: + service = KnowledgeService.__new__(KnowledgeService) + calls = 0 + + async def never_finishes_primary(**_kwargs): + nonlocal calls + calls += 1 + await asyncio.Event().wait() + + monkeypatch.setattr(service, "_list_entries_from_primary", never_finishes_primary) + monkeypatch.setattr(knowledge_service_module, "_PRIMARY_KM_LIST_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(knowledge_service_module, "_PRIMARY_KM_LIST_RETRY_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(knowledge_service_module, "_PRIMARY_KM_RETRY_DELAY_SECONDS", 0) + + response = await service.list_entries(limit=50) + + assert calls == 2 + 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.total == 13 + + @pytest.mark.asyncio async def test_knowledge_list_entries_keeps_primary_items_when_taxonomy_degrades(monkeypatch) -> None: primary_entry = KnowledgeEntry( @@ -247,3 +273,54 @@ async def test_knowledge_asset_taxonomy_fails_soft_when_readback_breaks(monkeypa "schedule", ] assert all(row.count >= 1 for row in taxonomy) + + +@pytest.mark.asyncio +async def test_knowledge_side_readbacks_bound_timeouts_to_source_backed(monkeypatch) -> None: + service = KnowledgeService.__new__(KnowledgeService) + + async def never_finishes(*_args): + await asyncio.Event().wait() + + monkeypatch.setattr(knowledge_service_module, "_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(service, "_read_primary_categories", never_finishes) + monkeypatch.setattr(service, "_read_primary_asset_taxonomy", never_finishes) + monkeypatch.setattr(service, "_search_primary_entries", never_finishes) + + categories = await service.get_categories() + taxonomy = await service.get_asset_taxonomy() + entries = await service.search("Telegram", limit=10) + + assert [row.category for row in categories] == [ + "project", + "product", + "website", + "service", + "package", + "tool", + "log", + "alert", + "playbook", + "rag", + "mcp", + "schedule", + "general", + ] + assert [row.key for row in taxonomy] == [ + "project", + "product", + "website", + "service", + "package", + "tool", + "log", + "alert", + "playbook", + "rag", + "mcp", + "schedule", + ] + assert {entry.id for entry in entries} >= { + "source-backed-service-telegram-alert-receipts", + "source-backed-alert-telegram-monitoring-coverage", + } diff --git a/apps/web/messages/en.json b/apps/web/messages/en.json index 7c10fc4c3..bc4ea0306 100644 --- a/apps/web/messages/en.json +++ b/apps/web/messages/en.json @@ -2217,8 +2217,10 @@ } }, "dataChain": { - "errorTitle": "知識條目資料鏈路異常", - "errorDescription": "主知識條目 API 未成功回應：{reason}。下方治理軌道仍會顯示 Hermes 受控覆核與陳舊 KM 狀態，避免誤判成知識庫真的歸零。", + "errorTitle": "知識條目入口需重讀", + "errorDescription": "知識條目入口回應：{reason}。頁面會保留 AI source-backed KM、Hermes 受控覆核與陳舊 KM 狀態，避免誤判成知識庫歸零。", + "sourceBackedTitle": "AI source-backed KM 已接管", + "sourceBackedDescription": "Primary KM DB readback 正在受控修復；目前顯示 source-backed 條目、分類與資產 taxonomy，不是知識庫歸零。原因：{reason}；階段：{stage}；下一步：{next}。", "degradedTitle": "知識條目讀取降級", "degradedDescription": "主知識條目 API 已回 degraded：{stage}；下一步：{next}。目前列表不視為真實歸零，AI 受控修復隊列與 RAG / PlayBook readback 仍會保留。", "retry": "重新讀取" diff --git a/apps/web/messages/zh-TW.json b/apps/web/messages/zh-TW.json index a6cb516fb..74f974e91 100644 --- a/apps/web/messages/zh-TW.json +++ b/apps/web/messages/zh-TW.json @@ -2217,8 +2217,10 @@ } }, "dataChain": { - "errorTitle": "知識條目資料鏈路異常", - "errorDescription": "主知識條目 API 未成功回應：{reason}。下方治理軌道仍會顯示 Hermes 受控覆核與陳舊 KM 狀態，避免誤判成知識庫真的歸零。", + "errorTitle": "知識條目入口需重讀", + "errorDescription": "知識條目入口回應：{reason}。頁面會保留 AI source-backed KM、Hermes 受控覆核與陳舊 KM 狀態，避免誤判成知識庫歸零。", + "sourceBackedTitle": "AI source-backed KM 已接管", + "sourceBackedDescription": "Primary KM DB readback 正在受控修復；目前顯示 source-backed 條目、分類與資產 taxonomy，不是知識庫歸零。原因：{reason}；階段：{stage}；下一步：{next}。", "degradedTitle": "知識條目讀取降級", "degradedDescription": "主知識條目 API 已回 degraded：{stage}；下一步：{next}。目前列表不視為真實歸零，AI 受控修復隊列與 RAG / PlayBook readback 仍會保留。", "retry": "重新讀取" diff --git a/apps/web/src/app/[locale]/knowledge-base/page.tsx b/apps/web/src/app/[locale]/knowledge-base/page.tsx index e6dd82b38..2821f1923 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 + degraded_reason_code?: string | null operator_stage?: string | null next_step?: string | null writes_on_read?: boolean @@ -496,7 +497,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, @@ -541,6 +542,7 @@ export default function KnowledgeBasePage({ setAssetTaxonomy(data.asset_taxonomy?.length ? data.asset_taxonomy : ASSET_TAXONOMY_FALLBACK_COUNTS) setEntryReadback({ readback_status: data.readback_status ?? 'ready', + degraded_reason_code: data.degraded_reason_code ?? null, operator_stage: data.operator_stage ?? null, next_step: data.next_step ?? null, }) @@ -574,23 +576,15 @@ export default function KnowledgeBasePage({ setGovernanceLoading(true) try { const apiBase = process.env.NEXT_PUBLIC_API_URL ?? '' - const [staleCandidates, ownerReviews, burnDown, completionQueue, ragStats] = await Promise.all([ - fetch(`${apiBase}/api/v1/ai/governance/km-stale-candidates?project_id=awoooi&limit=5`) - .then(res => res.ok ? res.json() : null) - .catch(() => null), - fetch(`${apiBase}/api/v1/ai/governance/km-stale-owner-reviews?project_id=awoooi&dispatch_status=pending&limit=5`) - .then(res => res.ok ? res.json() : null) - .catch(() => null), - fetch(`${apiBase}/api/v1/ai/governance/km-stale-owner-review-burndown?project_id=awoooi&limit=5`) - .then(res => res.ok ? res.json() : null) - .catch(() => null), - fetch(`${apiBase}/api/v1/ai/governance/km-stale-owner-review-completion-queue?project_id=awoooi&status_bucket=all&limit=5`) - .then(res => res.ok ? res.json() : null) - .catch(() => null), - fetch(`${apiBase}/api/v1/knowledge/rag/stats?project_id=${encodeURIComponent(KNOWLEDGE_PROJECT_ID)}`) - .then(res => res.ok ? res.json() : null) - .catch(() => null), - ]) + const fetchOptionalJson = async (path: string) => fetch(`${apiBase}${path}`) + .then(res => res.ok ? res.json() : null) + .catch(() => null) + + const staleCandidates = await fetchOptionalJson('/api/v1/ai/governance/km-stale-candidates?project_id=awoooi&limit=5') + const ownerReviews = await fetchOptionalJson('/api/v1/ai/governance/km-stale-owner-reviews?project_id=awoooi&dispatch_status=pending&limit=5') + const burnDown = await fetchOptionalJson('/api/v1/ai/governance/km-stale-owner-review-burndown?project_id=awoooi&limit=5') + const completionQueue = await fetchOptionalJson('/api/v1/ai/governance/km-stale-owner-review-completion-queue?project_id=awoooi&status_bucket=all&limit=5') + const ragStats = await fetchOptionalJson(`/api/v1/knowledge/rag/stats?project_id=${encodeURIComponent(KNOWLEDGE_PROJECT_ID)}`) setGovernanceTelemetry({ staleCandidates, @@ -694,8 +688,11 @@ export default function KnowledgeBasePage({ ) const totalCount = categories.reduce((sum, c) => sum + c.count, 0) const isKnowledgeListUnavailable = Boolean(entryFetchError) - const isKnowledgeListDegraded = entryReadback?.readback_status === 'degraded' - const knowledgeReadbackReady = !loading && !isKnowledgeListUnavailable && !isKnowledgeListDegraded + const readbackStatus = entryReadback?.readback_status ?? null + const isKnowledgeSourceBacked = Boolean(readbackStatus?.startsWith('source_backed')) + const isKnowledgeListDegraded = Boolean(readbackStatus && readbackStatus !== 'ready') + const hasKnowledgePayload = entries.length > 0 || total > 0 || categories.some(category => category.count > 0) + const knowledgeReadbackReady = !loading && !isKnowledgeListUnavailable && (!isKnowledgeListDegraded || hasKnowledgePayload) const localeCode = params.locale === 'en' ? 'en-US' : 'zh-TW' const formatCount = useCallback( (value: number) => value.toLocaleString(localeCode), @@ -1430,15 +1427,27 @@ export default function KnowledgeBasePage({

{entryFetchError ? t('dataChain.errorDescription', { reason: entryFetchError }) - : t('dataChain.degradedDescription', { - stage: entryReadback?.operator_stage ?? '--', - next: entryReadback?.next_step ?? '--', - })} + : isKnowledgeSourceBacked + ? t('dataChain.sourceBackedDescription', { + reason: entryReadback?.degraded_reason_code ?? '--', + stage: entryReadback?.operator_stage ?? '--', + next: entryReadback?.next_step ?? '--', + }) + : t('dataChain.degradedDescription', { + stage: entryReadback?.operator_stage ?? '--', + next: entryReadback?.next_step ?? '--', + })}