From def64351408e7068101964b808cf3a368862a5a7 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 18:45:20 +0800 Subject: [PATCH] fix(km): keep primary entries during taxonomy degradation --- apps/api/src/main.py | 44 ++++++- .../src/repositories/knowledge_repository.py | 22 ++-- apps/api/src/services/knowledge_service.py | 116 +++++++++++++++--- apps/api/tests/test_db_context_guard.py | 37 +++++- .../test_knowledge_repository_read_model.py | 11 +- ...est_knowledge_service_readback_degraded.py | 66 +++++++++- 6 files changed, 253 insertions(+), 43 deletions(-) diff --git a/apps/api/src/main.py b/apps/api/src/main.py index 047c5c1ec..899e53279 100644 --- a/apps/api/src/main.py +++ b/apps/api/src/main.py @@ -20,9 +20,9 @@ Date: 2026-03-20 import asyncio import os -from uuid import uuid4 from collections.abc import AsyncGenerator from contextlib import asynccontextmanager +from uuid import uuid4 import sentry_sdk import structlog @@ -303,8 +303,8 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: try: from sqlalchemy import select - from src.db.base import get_db_context from src.core.context import clear_project_context, set_project_context + from src.db.base import get_db_context from src.db.models import IncidentRecord from src.models.incident import IncidentStatus from src.services.incident_service import get_incident_service @@ -950,7 +950,11 @@ async def request_logging_middleware(request: Request, call_next): """ import time - from src.core.context import clear_project_context, get_current_project_context, set_project_context + from src.core.context import ( + clear_project_context, + get_current_project_context, + set_project_context, + ) request_id = request.headers.get("X-Request-ID") or str(uuid4()) project_id, source = _resolve_request_project_context(request) @@ -1008,12 +1012,44 @@ async def db_context_guard() -> dict: from src.core.context import get_current_project_context from src.db.base import get_db_context - async with get_db_context(): + async def _probe_db_context() -> None: + async with get_db_context(): + return None + + try: + await asyncio.wait_for(_probe_db_context(), timeout=1.5) return { "status": "ok", "project_context": get_current_project_context(), "source": "runtime_guard", } + except HTTPException: + raise + except Exception as exc: # noqa: BLE001 - guard readback should not become a 500 + message = f"{type(exc).__name__}: {exc}".lower() + timeout_or_pool = any( + marker in message + for marker in ( + "pool", + "timeout", + "timed out", + "too many connections", + "connection limit", + ) + ) + return { + "status": "degraded", + "project_context": get_current_project_context(), + "source": "runtime_guard", + "db_context_ready": False, + "degraded_reason_code": ( + "primary_db_timeout_or_pool_exhausted" + if timeout_or_pool + else "primary_db_context_readback_exception" + ), + "writes_on_read": False, + "manual_review_required": False, + } # ============================================================================= diff --git a/apps/api/src/repositories/knowledge_repository.py b/apps/api/src/repositories/knowledge_repository.py index b9c0ce847..f76a497cf 100644 --- a/apps/api/src/repositories/knowledge_repository.py +++ b/apps/api/src/repositories/knowledge_repository.py @@ -474,18 +474,20 @@ class KnowledgeDBRepository: 將 KM 依專案、產品、網站、服務、套件、工具、Log、Alert、PlayBook、 RAG、MCP 與排程分群。 """ - rows: list[tuple[str, int]] = [] - for key in _ASSET_TAXONOMY_KEYS: - result = await self.db.execute( - select(func.count()) - .select_from(KnowledgeEntryRecord) - .where( - _active_status_filter(), - _asset_taxonomy_condition(key), + result = await self.db.execute( + select( + *( + func.count() + .filter(_asset_taxonomy_condition(key)) + .label(key) + for key in _ASSET_TAXONOMY_KEYS ) ) - rows.append((key, int(result.scalar() or 0))) - return rows + .select_from(KnowledgeEntryRecord) + .where(_active_status_filter()) + ) + row = result.mappings().one() + return [(key, int(row.get(key) or 0)) for key in _ASSET_TAXONOMY_KEYS] async def search(self, query: str, limit: int = 20) -> list[KnowledgeEntry]: """關鍵字搜尋 (title + content + tags)""" diff --git a/apps/api/src/services/knowledge_service.py b/apps/api/src/services/knowledge_service.py index 3e936a15e..453071531 100644 --- a/apps/api/src/services/knowledge_service.py +++ b/apps/api/src/services/knowledge_service.py @@ -52,6 +52,21 @@ _DEGRADED_CATEGORY_FALLBACKS = ( ) _ASSET_TAXONOMY_FALLBACKS = _DEGRADED_CATEGORY_FALLBACKS[:-1] +_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS = 1.5 +_ASSET_TAXONOMY_CATEGORY_HINTS: dict[str, tuple[str, ...]] = { + "service": ( + "infrastructure", + "application", + "ai_system", + "database", + "host_resource", + "kubernetes", + "alert_handling", + ), + "tool": ("devops_tool", "AI自動化/Ansible受控修復"), + "alert": ("alert_handling",), + "playbook": ("auto_repair",), +} _SOURCE_BACKED_KNOWLEDGE_SPECS: tuple[dict[str, object], ...] = ( { @@ -366,6 +381,10 @@ def _source_asset_taxonomy_counts( def _entry_matches_asset_key(entry: KnowledgeEntry, key: str) -> bool: if entry.category == key: return True + if entry.category.lower() in { + hint.lower() for hint in _ASSET_TAXONOMY_CATEGORY_HINTS.get(key, ()) + }: + return True if key in {tag.lower() for tag in entry.tags}: return True if key == "playbook" and entry.related_playbook_id: @@ -484,34 +503,91 @@ class KnowledgeService: limit=limit, offset=offset, ) + if total == 0: + return build_knowledge_list_readback_degraded_response( + "knowledge_db_empty", + category=category, + entry_type=entry_type, + status=status, + tags=tags, + q=q, + limit=limit, + offset=offset, + readback_status="source_backed_db_empty", + ) + + readback_status = "ready" + degraded_reason_code: str | None = None + operator_stage: str | None = None + next_step: str | None = None + + try: + categories = await asyncio.wait_for( + self._read_primary_categories(), + timeout=_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS, + ) + except Exception as exc: # noqa: BLE001 - primary entries must stay visible + logger.warning( + "knowledge_list_categories_readback_degraded", + error=str(exc), + total=total, + limit=limit, + offset=offset, + ) + categories = _source_category_counts(items) + readback_status = "ready_partial_taxonomy_degraded" + degraded_reason_code = "primary_km_taxonomy_readback_degraded" + operator_stage = "knowledge_primary_entries_ready_taxonomy_degraded" + next_step = "repair_primary_km_taxonomy_readback_without_hiding_entries" + + try: + asset_taxonomy = await asyncio.wait_for( + self._read_primary_asset_taxonomy(), + timeout=_PRIMARY_KM_TAXONOMY_TIMEOUT_SECONDS, + ) + except Exception as exc: # noqa: BLE001 - primary entries must stay visible + logger.warning( + "knowledge_list_asset_taxonomy_readback_degraded", + error=str(exc), + total=total, + limit=limit, + offset=offset, + ) + asset_taxonomy = _source_asset_taxonomy_counts(items) + readback_status = "ready_partial_taxonomy_degraded" + degraded_reason_code = "primary_km_taxonomy_readback_degraded" + operator_stage = "knowledge_primary_entries_ready_taxonomy_degraded" + next_step = "repair_primary_km_taxonomy_readback_without_hiding_entries" + + return KnowledgeListResponse( + items=items, + total=total, + categories=categories, + asset_taxonomy=asset_taxonomy, + readback_status=readback_status, + primary_readback_ready=True, + degraded_reason_code=degraded_reason_code, + operator_stage=operator_stage, + next_step=next_step, + ) + + async def _read_primary_categories(self) -> list[CategoryCount]: + 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 ] + return categories + + async def _read_primary_asset_taxonomy(self) -> list[KnowledgeAssetTaxonomyCount]: + async with get_db_context() as db: + repo: IKnowledgeRepository = KnowledgeDBRepository(db) asset_taxonomy_raw = await repo.get_asset_taxonomy_counts() - asset_taxonomy = [ + return [ KnowledgeAssetTaxonomyCount(key=key, count=count) for key, count in asset_taxonomy_raw ] - if total == 0: - return build_knowledge_list_readback_degraded_response( - "knowledge_db_empty", - category=category, - entry_type=entry_type, - status=status, - tags=tags, - q=q, - limit=limit, - offset=offset, - readback_status="source_backed_db_empty", - ) - return KnowledgeListResponse( - items=items, - total=total, - categories=categories, - asset_taxonomy=asset_taxonomy, - primary_readback_ready=True, - ) async def list_entries( self, diff --git a/apps/api/tests/test_db_context_guard.py b/apps/api/tests/test_db_context_guard.py index b2cb1810d..4b400b70e 100644 --- a/apps/api/tests/test_db_context_guard.py +++ b/apps/api/tests/test_db_context_guard.py @@ -2,15 +2,14 @@ from __future__ import annotations from contextlib import asynccontextmanager -from fastapi import HTTPException - -import pytest -from fastapi import FastAPI -from fastapi.testclient import TestClient from unittest.mock import patch +import pytest +from fastapi import FastAPI, HTTPException +from fastapi.testclient import TestClient + from src.db.base import get_db_context -from src.main import db_context_guard, app, http_exception_handler +from src.main import app, db_context_guard, http_exception_handler def test_db_context_guard_without_project_id_is_unauthorized(): @@ -45,6 +44,16 @@ class _UnauthorizedDbContext: return False +class _BrokenDbContext: + """Simulate production DB pool pressure without hiding project context.""" + + async def __aenter__(self): + raise RuntimeError("QueuePool limit timeout") + + async def __aexit__(self, exc_type, exc_val, exc_tb): # noqa: ARG001 + return False + + def _build_guard_app() -> FastAPI: app = FastAPI() @@ -82,6 +91,22 @@ def test_db_context_guard_with_project_id_returns_snapshot(): assert body["project_context"]["source"] == "test.guard" +def test_db_context_guard_with_project_id_db_pressure_returns_degraded_snapshot(): + """DB pool 壓力應回傳 evidence，不應把安全 guard 變成 500。""" + app = _build_guard_app() + with patch("src.db.base.get_db_context", return_value=_BrokenDbContext()): + client = TestClient(app) + response = client.get("/api/v1/security/db-context-guard", headers={"X-Project-ID": "awoooi"}) + + assert response.status_code == 200 + body = response.json() + assert body["status"] == "degraded" + assert body["project_context"]["project_id"] == "awoooi" + assert body["db_context_ready"] is False + assert body["degraded_reason_code"] == "primary_db_timeout_or_pool_exhausted" + assert body["manual_review_required"] is False + + def test_http_exception_handler_is_registered(): assert app.exception_handlers[HTTPException] is http_exception_handler diff --git a/apps/api/tests/test_knowledge_repository_read_model.py b/apps/api/tests/test_knowledge_repository_read_model.py index a15e5c7fc..a2e68a5f6 100644 --- a/apps/api/tests/test_knowledge_repository_read_model.py +++ b/apps/api/tests/test_knowledge_repository_read_model.py @@ -128,6 +128,9 @@ class _MappingResult: def all(self): return self.rows + def one(self): + return self.rows[0] + class _RowsResult: def __init__(self, rows: list[SimpleNamespace]): @@ -224,11 +227,14 @@ async def test_get_asset_taxonomy_counts_uses_stable_ai_automation_lenses() -> N class _TaxonomyDb: def __init__(self): self.statements = [] - self.values = list(range(10, 10 + len(_ASSET_TAXONOMY_KEYS))) + self.values = { + key: count + for count, key in enumerate(_ASSET_TAXONOMY_KEYS, start=10) + } async def execute(self, statement): self.statements.append(statement) - return _ScalarResult(self.values.pop(0)) + return _MappingResult([self.values]) db = _TaxonomyDb() repo = KnowledgeDBRepository(db=db) # type: ignore[arg-type] @@ -237,6 +243,7 @@ async def test_get_asset_taxonomy_counts_uses_stable_ai_automation_lenses() -> N assert [key for key, _ in counts] == list(_ASSET_TAXONOMY_KEYS) assert [count for _, count in counts] == list(range(10, 10 + len(_ASSET_TAXONOMY_KEYS))) + assert len(db.statements) == 1 compiled = "\n".join( str(stmt.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True})) for stmt in db.statements diff --git a/apps/api/tests/test_knowledge_service_readback_degraded.py b/apps/api/tests/test_knowledge_service_readback_degraded.py index dfb122036..661a4ff75 100644 --- a/apps/api/tests/test_knowledge_service_readback_degraded.py +++ b/apps/api/tests/test_knowledge_service_readback_degraded.py @@ -1,6 +1,12 @@ import pytest -from src.models.knowledge import KnowledgeListResponse +from src.models.knowledge import ( + EntrySource, + EntryStatus, + EntryType, + KnowledgeEntry, + KnowledgeListResponse, +) from src.services import knowledge_service as knowledge_service_module from src.services.knowledge_service import KnowledgeService @@ -13,6 +19,14 @@ class _BrokenDbContext: return False +class _OkDbContext: + async def __aenter__(self): + return object() + + async def __aexit__(self, exc_type, exc, tb): + return False + + @pytest.mark.asyncio async def test_knowledge_list_entries_fails_soft_when_readback_breaks(monkeypatch) -> None: monkeypatch.setattr( @@ -108,6 +122,56 @@ 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_keeps_primary_items_when_taxonomy_degrades(monkeypatch) -> None: + primary_entry = KnowledgeEntry( + id="km-primary-1", + title="Primary KM row", + content="Persistent AI automation memory row", + entry_type=EntryType.RUNBOOK, + category="alert_handling", + tags=["telegram", "ai_agent"], + source=EntrySource.AI_EXTRACTED, + status=EntryStatus.APPROVED, + ) + + class _TaxonomyBrokenRepo: + def __init__(self, _db): + pass + + async def list_entries(self, **_kwargs): + return [primary_entry], 638 + + async def get_categories(self): + raise RuntimeError("db pool exhausted") + + async def get_asset_taxonomy_counts(self): + raise RuntimeError("db pool exhausted") + + monkeypatch.setattr( + knowledge_service_module, + "get_db_context", + lambda: _OkDbContext(), + ) + monkeypatch.setattr( + knowledge_service_module, + "KnowledgeDBRepository", + _TaxonomyBrokenRepo, + ) + service = KnowledgeService.__new__(KnowledgeService) + + response = await service.list_entries(limit=50) + + assert response.total == 638 + assert response.items == [primary_entry] + assert response.readback_status == "ready_partial_taxonomy_degraded" + assert response.primary_readback_ready is True + assert response.degraded_reason_code == "primary_km_taxonomy_readback_degraded" + assert response.operator_stage == "knowledge_primary_entries_ready_taxonomy_degraded" + assert response.categories[[row.category for row in response.categories].index("alert_handling")].count == 1 + assert response.asset_taxonomy[[row.key for row in response.asset_taxonomy].index("alert")].count == 1 + + @pytest.mark.asyncio async def test_knowledge_search_fails_soft_to_source_backed_entries(monkeypatch) -> None: monkeypatch.setattr(