From a9d85d0d62915a9657f815e7701622a9e7753a58 Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 22 Jul 2026 17:35:18 +0800 Subject: [PATCH] fix(knowledge): bind canonical project readback --- apps/api/src/api/v1/knowledge.py | 17 +- apps/api/src/services/knowledge_service.py | 125 ++++++++++-- ...est_knowledge_service_readback_degraded.py | 184 +++++++++++++++++- 3 files changed, 301 insertions(+), 25 deletions(-) diff --git a/apps/api/src/api/v1/knowledge.py b/apps/api/src/api/v1/knowledge.py index eb40ff098..40c27982c 100644 --- a/apps/api/src/api/v1/knowledge.py +++ b/apps/api/src/api/v1/knowledge.py @@ -32,6 +32,8 @@ logger = structlog.get_logger(__name__) router = APIRouter(prefix="/knowledge", tags=["Knowledge Base"]) +CANONICAL_KNOWLEDGE_PROJECT_ID = "awoooi" + # ============================================================================= # Endpoints @@ -49,6 +51,7 @@ async def list_entries( """列出知識條目""" service = get_knowledge_service() return await service.list_entries( + project_id=CANONICAL_KNOWLEDGE_PROJECT_ID, category=category, entry_type=entry_type, status=status, @@ -65,7 +68,11 @@ async def search_entries( ) -> list[KnowledgeEntry]: """關鍵字搜尋 (title + content + tags)""" service = get_knowledge_service() - return await service.search(q, limit) + return await service.search( + q, + limit, + project_id=CANONICAL_KNOWLEDGE_PROJECT_ID, + ) @router.get("/semantic-search") @@ -102,7 +109,9 @@ async def embed_all_entries() -> dict: async def get_categories() -> list[dict]: """取得分類樹 (含各類數量)""" service = get_knowledge_service() - cats = await service.get_categories() + cats = await service.get_categories( + project_id=CANONICAL_KNOWLEDGE_PROJECT_ID, + ) return [cat.model_dump() for cat in cats] @@ -110,7 +119,9 @@ async def get_categories() -> list[dict]: async def get_asset_taxonomy() -> list[KnowledgeAssetTaxonomyCount]: """取得 AI 自動化資產維度統計（只讀，不寫入）。""" service = get_knowledge_service() - return await service.get_asset_taxonomy() + return await service.get_asset_taxonomy( + project_id=CANONICAL_KNOWLEDGE_PROJECT_ID, + ) @router.get("/{entry_id}", response_model=KnowledgeEntry) diff --git a/apps/api/src/services/knowledge_service.py b/apps/api/src/services/knowledge_service.py index 96a499664..60b8014f7 100644 --- a/apps/api/src/services/knowledge_service.py +++ b/apps/api/src/services/knowledge_service.py @@ -345,6 +345,21 @@ def _current_knowledge_project_id() -> str | None: return str(project_id).strip() if project_id else None +def _resolve_knowledge_project_id(project_id: str | None) -> str | None: + """Prefer an explicit tenant binding; never replace an explicit blank.""" + if project_id is not None: + normalized = str(project_id).strip() + return normalized or None + return _current_knowledge_project_id() + + +def _knowledge_db_context(project_id: str | None): + """Keep legacy context inheritance while allowing a canonical read binding.""" + if project_id is None: + return get_db_context() + return get_db_context(str(project_id).strip()) + + def _normalize_direct_tags(value: object) -> list[str]: if value is None: return [] @@ -787,6 +802,7 @@ class KnowledgeService: async def _list_entries_from_primary( self, *, + project_id: str | None = None, category: str | None, entry_type: EntryType | None, status: EntryStatus | None, @@ -795,7 +811,7 @@ class KnowledgeService: limit: int, offset: int, ) -> KnowledgeListResponse: - async with get_db_context() as db: + async with _knowledge_db_context(project_id) as db: repo: IKnowledgeRepository = KnowledgeDBRepository(db) items, total = await repo.list_entries( category=category, @@ -826,7 +842,7 @@ class KnowledgeService: try: categories = await asyncio.wait_for( - self._read_primary_categories(), + self._read_primary_categories(project_id=project_id), timeout=_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS, ) except Exception as exc: # noqa: BLE001 - primary entries must stay visible @@ -845,7 +861,7 @@ class KnowledgeService: try: asset_taxonomy = await asyncio.wait_for( - self._read_primary_asset_taxonomy(), + self._read_primary_asset_taxonomy(project_id=project_id), timeout=_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS, ) except Exception as exc: # noqa: BLE001 - primary entries must stay visible @@ -874,8 +890,12 @@ class KnowledgeService: next_step=next_step, ) - async def _read_primary_categories(self) -> list[CategoryCount]: - async with get_db_context() as db: + async def _read_primary_categories( + self, + *, + project_id: str | None = None, + ) -> list[CategoryCount]: + async with _knowledge_db_context(project_id) as db: repo: IKnowledgeRepository = KnowledgeDBRepository(db) categories_raw = await repo.get_categories() categories = [ @@ -883,8 +903,12 @@ class KnowledgeService: ] return categories - async def _read_primary_asset_taxonomy(self) -> list[KnowledgeAssetTaxonomyCount]: - async with get_db_context() as db: + async def _read_primary_asset_taxonomy( + self, + *, + project_id: str | None = None, + ) -> list[KnowledgeAssetTaxonomyCount]: + async with _knowledge_db_context(project_id) as db: repo: IKnowledgeRepository = KnowledgeDBRepository(db) asset_taxonomy_raw = await repo.get_asset_taxonomy_counts() return [ @@ -896,14 +920,17 @@ class KnowledgeService: self, query: str, limit: int, + *, + project_id: str | None = None, ) -> list[KnowledgeEntry]: - async with get_db_context() as db: + async with _knowledge_db_context(project_id) as db: repo: IKnowledgeRepository = KnowledgeDBRepository(db) return await repo.search(query, limit) async def _list_entries_from_primary_bounded( self, *, + project_id: str | None = None, category: str | None, entry_type: EntryType | None, status: EntryStatus | None, @@ -915,6 +942,7 @@ class KnowledgeService: ) -> KnowledgeListResponse: return await asyncio.wait_for( self._list_entries_from_primary( + project_id=project_id, category=category, entry_type=entry_type, status=status, @@ -929,6 +957,7 @@ class KnowledgeService: async def _list_entries_from_direct( self, *, + project_id: str | None = None, category: str | None, entry_type: EntryType | None, status: EntryStatus | None, @@ -938,7 +967,7 @@ class KnowledgeService: offset: int, ) -> KnowledgeListResponse | None: """Pool 壓力下以短連線只讀救回 primary KM，不把 727 筆誤降成 13 筆。""" - project_id = _current_knowledge_project_id() + project_id = _resolve_knowledge_project_id(project_id) if not project_id: return None @@ -1070,8 +1099,12 @@ class KnowledgeService: ) return None - async def _read_categories_direct(self) -> list[CategoryCount] | None: - project_id = _current_knowledge_project_id() + async def _read_categories_direct( + self, + *, + project_id: str | None = None, + ) -> list[CategoryCount] | None: + project_id = _resolve_knowledge_project_id(project_id) if not project_id: return None db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://") @@ -1111,8 +1144,12 @@ class KnowledgeService: ) return None - async def _read_asset_taxonomy_direct(self) -> list[KnowledgeAssetTaxonomyCount] | None: - project_id = _current_knowledge_project_id() + async def _read_asset_taxonomy_direct( + self, + *, + project_id: str | None = None, + ) -> list[KnowledgeAssetTaxonomyCount] | None: + project_id = _resolve_knowledge_project_id(project_id) if not project_id: return None db_url = settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://") @@ -1224,10 +1261,13 @@ class KnowledgeService: q: str | None = None, limit: int = 20, offset: int = 0, + *, + project_id: str | None = None, ) -> KnowledgeListResponse: """列出知識條目 + 分類統計""" try: return await self._list_entries_from_primary_bounded( + project_id=project_id, category=category, entry_type=entry_type, status=status, @@ -1244,6 +1284,7 @@ class KnowledgeService: try: await asyncio.sleep(_PRIMARY_KM_RETRY_DELAY_SECONDS) retry_response = await self._list_entries_from_primary_bounded( + project_id=project_id, category=category, entry_type=entry_type, status=status, @@ -1266,6 +1307,7 @@ class KnowledgeService: offset=offset, ) direct_response = await self._list_entries_from_direct( + project_id=project_id, category=category, entry_type=entry_type, status=status, @@ -1298,11 +1340,20 @@ class KnowledgeService: offset=offset, ) - async def get_asset_taxonomy(self) -> list[KnowledgeAssetTaxonomyCount]: + async def get_asset_taxonomy( + self, + *, + project_id: str | None = None, + ) -> list[KnowledgeAssetTaxonomyCount]: """取得 AI 自動化資產維度統計。""" try: + primary_read = ( + self._read_primary_asset_taxonomy() + if project_id is None + else self._read_primary_asset_taxonomy(project_id=project_id) + ) taxonomy = await asyncio.wait_for( - self._read_primary_asset_taxonomy(), + primary_read, timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS, ) if taxonomy and any(row.count > 0 for row in taxonomy): @@ -1311,7 +1362,11 @@ class KnowledgeService: 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() + direct_taxonomy = await ( + self._read_asset_taxonomy_direct() + if project_id is None + else self._read_asset_taxonomy_direct(project_id=project_id) + ) if direct_taxonomy and any(row.count > 0 for row in direct_taxonomy): return direct_taxonomy logger.warning( @@ -1321,11 +1376,20 @@ class KnowledgeService: ) return _source_asset_taxonomy_counts(_source_backed_entries()) - async def get_categories(self) -> list[CategoryCount]: + async def get_categories( + self, + *, + project_id: str | None = None, + ) -> list[CategoryCount]: """取得分類統計（直接呼叫 repo，不走 list_entries）""" try: + primary_read = ( + self._read_primary_categories() + if project_id is None + else self._read_primary_categories(project_id=project_id) + ) categories = await asyncio.wait_for( - self._read_primary_categories(), + primary_read, timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS, ) if categories: @@ -1334,7 +1398,11 @@ class KnowledgeService: 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() + direct_categories = await ( + self._read_categories_direct() + if project_id is None + else self._read_categories_direct(project_id=project_id) + ) if direct_categories and any(row.count > 0 for row in direct_categories): return direct_categories logger.warning( @@ -1344,11 +1412,26 @@ class KnowledgeService: ) return _source_category_counts(_source_backed_entries()) - async def search(self, query: str, limit: int = 20) -> list[KnowledgeEntry]: + async def search( + self, + query: str, + limit: int = 20, + *, + project_id: str | None = None, + ) -> list[KnowledgeEntry]: """關鍵字搜尋""" try: + primary_read = ( + self._search_primary_entries(query, limit) + if project_id is None + else self._search_primary_entries( + query, + limit, + project_id=project_id, + ) + ) return await asyncio.wait_for( - self._search_primary_entries(query, limit), + primary_read, timeout=_PRIMARY_KM_SIDE_READ_TIMEOUT_SECONDS, ) except Exception as exc: # noqa: BLE001 - KM search must not 500 the UI diff --git a/apps/api/tests/test_knowledge_service_readback_degraded.py b/apps/api/tests/test_knowledge_service_readback_degraded.py index 94a120774..e51b438eb 100644 --- a/apps/api/tests/test_knowledge_service_readback_degraded.py +++ b/apps/api/tests/test_knowledge_service_readback_degraded.py @@ -1,7 +1,11 @@ import asyncio import pytest +from fastapi import FastAPI, HTTPException +from fastapi.testclient import TestClient +from src.api.v1 import knowledge as knowledge_api +from src.core.context import clear_project_context, set_project_context from src.models.knowledge import ( CategoryCount, EntrySource, @@ -175,6 +179,7 @@ async def test_knowledge_list_entries_uses_direct_readback_after_session_pool_ti nonlocal direct_calls direct_calls += 1 assert kwargs["limit"] == 50 + assert kwargs["project_id"] == "awoooi" return KnowledgeListResponse( items=[primary_entry], total=727, @@ -199,7 +204,7 @@ async def test_knowledge_list_entries_uses_direct_readback_after_session_pool_ti 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) + response = await service.list_entries(project_id="awoooi", limit=50) assert primary_calls == 2 assert direct_calls == 1 @@ -441,3 +446,180 @@ async def test_knowledge_side_readbacks_bound_timeouts_to_source_backed(monkeypa "source-backed-service-telegram-alert-receipts", "source-backed-alert-telegram-monitoring-coverage", } + + +@pytest.mark.asyncio +async def test_knowledge_list_binds_explicit_canonical_project_to_all_primary_reads( + monkeypatch, +) -> None: + project_contexts: list[str | None] = [] + primary_entry = KnowledgeEntry( + id="km-primary-bound-1", + title="Canonical AWOOOI KM row", + content="Persistent project-bound knowledge", + entry_type=EntryType.RUNBOOK, + category="project", + tags=["awoooi"], + source=EntrySource.AI_EXTRACTED, + status=EntryStatus.APPROVED, + ) + + class _BoundRepo: + def __init__(self, _db): + pass + + async def list_entries(self, **_kwargs): + return [primary_entry], 1 + + async def get_categories(self): + return [("project", 1)] + + async def get_asset_taxonomy_counts(self): + return [("project", 1)] + + def bound_db_context(project_id=None): + project_contexts.append(project_id) + return _OkDbContext() + + monkeypatch.setattr( + knowledge_service_module, + "get_db_context", + bound_db_context, + ) + monkeypatch.setattr( + knowledge_service_module, + "KnowledgeDBRepository", + _BoundRepo, + ) + service = KnowledgeService.__new__(KnowledgeService) + + response = await service.list_entries( + project_id="awoooi", + limit=50, + ) + + assert response.items == [primary_entry] + assert response.total == 1 + assert response.readback_status == "ready" + assert response.primary_readback_ready is True + assert project_contexts == ["awoooi", "awoooi", "awoooi"] + + +@pytest.mark.asyncio +async def test_knowledge_read_router_uses_only_canonical_awoooi_project( + monkeypatch, +) -> None: + observed_projects: list[tuple[str, str | None]] = [] + + class _BoundService: + async def list_entries(self, **kwargs): + observed_projects.append(("list", kwargs.get("project_id"))) + return KnowledgeListResponse(items=[], total=0) + + async def search(self, _query, _limit, *, project_id=None): + observed_projects.append(("search", project_id)) + return [] + + async def get_categories(self, *, project_id=None): + observed_projects.append(("categories", project_id)) + return [] + + async def get_asset_taxonomy(self, *, project_id=None): + observed_projects.append(("asset_taxonomy", project_id)) + return [] + + monkeypatch.setattr( + knowledge_api, + "get_knowledge_service", + lambda: _BoundService(), + ) + + await knowledge_api.list_entries( + category=None, + entry_type=None, + status=None, + q=None, + limit=20, + offset=0, + ) + await knowledge_api.search_entries(q="Telegram", limit=20) + await knowledge_api.get_categories() + await knowledge_api.get_asset_taxonomy() + + assert observed_projects == [ + ("list", "awoooi"), + ("search", "awoooi"), + ("categories", "awoooi"), + ("asset_taxonomy", "awoooi"), + ] + + +@pytest.mark.asyncio +async def test_explicit_blank_project_fails_closed_without_inheriting_context() -> None: + service = KnowledgeService.__new__(KnowledgeService) + spoofed_tokens = set_project_context( + "spoofed-project", + source="test.knowledge-spoof", + ) + try: + with pytest.raises(HTTPException) as exc_info: + async with knowledge_service_module._knowledge_db_context(" "): + pass + + direct_response = await service._list_entries_from_direct( + project_id=" ", + category=None, + entry_type=None, + status=None, + tags=None, + q=None, + limit=20, + offset=0, + ) + finally: + clear_project_context(spoofed_tokens) + + assert exc_info.value.status_code == 401 + assert direct_response is None + + absent_tokens = set_project_context(None, source="test.knowledge-absent") + try: + absent_response = await service._list_entries_from_direct( + category=None, + entry_type=None, + status=None, + tags=None, + q=None, + limit=20, + offset=0, + ) + finally: + clear_project_context(absent_tokens) + + assert absent_response is None + + +def test_knowledge_http_query_cannot_override_canonical_project(monkeypatch) -> None: + observed_projects: list[str | None] = [] + + class _BoundService: + async def list_entries(self, **kwargs): + observed_projects.append(kwargs.get("project_id")) + return KnowledgeListResponse(items=[], total=0) + + monkeypatch.setattr( + knowledge_api, + "get_knowledge_service", + lambda: _BoundService(), + ) + app = FastAPI() + app.include_router(knowledge_api.router, prefix="/api/v1") + + with TestClient(app) as client: + response = client.get( + "/api/v1/knowledge", + params={"project_id": "evil-project"}, + ) + + assert response.status_code == 200 + assert observed_projects == ["awoooi"]