wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(knowledge): bind canonical project readback

Browse Source
This commit is contained in:
Your Name
2026-07-22 17:35:18 +08:00
parent 783e111801
commit a9d85d0d62
3 changed files with 301 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
apps/api/src/api/v1/knowledge.py
View File

@@ -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)

125
apps/api/src/services/knowledge_service.py
View File

@@ -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

184
apps/api/tests/test_knowledge_service_readback_degraded.py
View File

@@ -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"]