fix: complete admin knowledge audit improvements

ai_knowledge_base_v2/apps/backend/alembic/versions/0009_log_retention_storage_indexes.py

@@ -0,0 +1,59 @@
+"""log retention, storage snapshots and audit indexes
+
+Revision ID: 0009_logs_storage
+Revises: 0008_unique_kb_node
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0009_logs_storage"
+down_revision = "0008_unique_kb_node"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    tables = set(inspector.get_table_names())
+    if "sys_log_retention_policy" not in tables:
+        op.create_table(
+        "sys_log_retention_policy",
+        sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
+        sa.Column("retention_days", sa.Integer(), nullable=True),
+        sa.Column("enabled", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("last_run_at", sa.DateTime(), nullable=True),
+        sa.Column("last_result", sa.Text(), nullable=True),
+        sa.Column("updated_by", sa.BigInteger(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
+    )
+    if "sys_storage_snapshot" not in tables:
+        op.create_table(
+        "sys_storage_snapshot",
+        sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
+        sa.Column("scope", sa.String(50), nullable=False),
+        sa.Column("total_bytes", sa.BigInteger(), nullable=True),
+        sa.Column("detail_json", sa.Text(), nullable=False),
+        sa.Column("status", sa.String(20), nullable=False),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
+    )
+    _create_index_if_missing("sys_knowledge_retrieval_log", "ix_retrieval_status_created", ["status", "created_at"])
+    _create_index_if_missing("sys_knowledge_retrieval_log", "ix_retrieval_created_id", ["created_at", "id"])
+    _create_index_if_missing("sys_ai_request_log", "ix_ai_log_status_created", ["status", "created_at"])
+    _create_index_if_missing("sys_operation_log", "ix_operation_module_created", ["module", "created_at"])
+    _create_index_if_missing("sys_chat_session", "ix_chat_session_deleted_updated", ["is_deleted", "updated_at"])
+
+
+def _create_index_if_missing(table: str, name: str, columns: list[str]) -> None:
+    if name not in {item["name"] for item in sa.inspect(op.get_bind()).get_indexes(table)}:
+        op.create_index(name, table, columns)
+
+
+def downgrade() -> None:
+    op.drop_index("ix_chat_session_deleted_updated", table_name="sys_chat_session")
+    op.drop_index("ix_operation_module_created", table_name="sys_operation_log")
+    op.drop_index("ix_ai_log_status_created", table_name="sys_ai_request_log")
+    op.drop_index("ix_retrieval_created_id", table_name="sys_knowledge_retrieval_log")
+    op.drop_index("ix_retrieval_status_created", table_name="sys_knowledge_retrieval_log")
+    op.drop_table("sys_storage_snapshot")
+    op.drop_table("sys_log_retention_policy")

ai_knowledge_base_v2/apps/backend/app/api/admin_agent_records.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
 import json
+from datetime import datetime
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
-from sqlalchemy import select
+from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 
 from app.core.database import get_db
@@ -18,6 +19,8 @@ from app.models.knowledge import (
 )
 from app.schemas.knowledge import AttentionUpdateRequest
 from app.services.admin_service import OperationLogService
+from app.models.logs import LogRetentionPolicy
+from app.services.maintenance_service import MaintenanceService
 
 router = APIRouter()
 
@@ -29,10 +32,12 @@ def retrieval_logs(
     knowledgeCalled: bool | None = Query(default=None),
     statusValue: str = Query(default="", alias="status"),
     attentionCreated: bool | None = Query(default=None),
+    page: int = Query(default=1, ge=1),
+    pageSize: int = Query(default=50, ge=1, le=100),
     db: Session = Depends(get_db),
     current_admin: Admin = Depends(get_current_admin),
 ) -> dict:
-    query = select(KnowledgeRetrievalLog).order_by(KnowledgeRetrievalLog.id.desc()).limit(300)
+    query = select(KnowledgeRetrievalLog).order_by(KnowledgeRetrievalLog.id.desc())
     if sessionId is not None:
         query = query.where(KnowledgeRetrievalLog.session_id == sessionId)
     if knowledgeId is not None:
@@ -43,9 +48,55 @@ def retrieval_logs(
         query = query.where(KnowledgeRetrievalLog.status == statusValue)
     if attentionCreated is not None:
         query = query.where(KnowledgeRetrievalLog.attention_created == (1 if attentionCreated else 0))
+    query = query.offset((page - 1) * pageSize).limit(pageSize)
     return api_success([_retrieval_dict(item, detail=False) for item in db.scalars(query).all()])
 
 
+@router.post("/retrieval-log/cleanup/estimate")
+def estimate_cleanup(payload: dict, db: Session = Depends(get_db), current_admin: Admin = Depends(get_current_admin)) -> dict:
+    before = _cleanup_before(payload)
+    count = db.scalar(select(func.count()).select_from(KnowledgeRetrievalLog).where(KnowledgeRetrievalLog.created_at < before)) or 0
+    return api_success({"before": before, "estimatedCount": count})
+
+
+@router.post("/retrieval-log/cleanup")
+def cleanup_logs(payload: dict, db: Session = Depends(get_db), current_admin: Admin = Depends(get_current_admin)) -> dict:
+    before = _cleanup_before(payload)
+    total = MaintenanceService.delete_retrieval_logs(db, before)
+    OperationLogService.write(db, admin_id=current_admin.id, module="retrieval_log", action=f"cleanup:{total}", target_id=None)
+    db.commit()
+    return api_success({"deleted": total, "before": before})
+
+
+@router.get("/retrieval-log/retention")
+def get_retention(db: Session = Depends(get_db), current_admin: Admin = Depends(get_current_admin)) -> dict:
+    policy = db.scalar(select(LogRetentionPolicy).order_by(LogRetentionPolicy.id).limit(1))
+    return api_success({"enabled": bool(policy and policy.enabled), "retentionDays": policy.retention_days if policy else None, "lastRunAt": policy.last_run_at if policy else None, "lastResult": policy.last_result if policy else None})
+
+
+@router.put("/retrieval-log/retention")
+def set_retention(payload: dict, db: Session = Depends(get_db), current_admin: Admin = Depends(get_current_admin)) -> dict:
+    days = payload.get("retentionDays")
+    if days is not None and (not isinstance(days, int) or days < 1):
+        raise HTTPException(status_code=400, detail="保留天数必须为正整数，永久保留请传 null")
+    policy = db.scalar(select(LogRetentionPolicy).order_by(LogRetentionPolicy.id).limit(1)) or LogRetentionPolicy()
+    policy.retention_days, policy.enabled, policy.updated_by = days, 0 if days is None else 1, current_admin.id
+    db.add(policy)
+    OperationLogService.write(db, admin_id=current_admin.id, module="retrieval_log", action="retention_update", target_id=None)
+    db.commit()
+    return api_success({"enabled": bool(policy.enabled), "retentionDays": policy.retention_days})
+
+
+def _cleanup_before(payload: dict) -> datetime:
+    raw = str(payload.get("before") or "").strip()
+    if not raw:
+        raise HTTPException(status_code=400, detail="必须指定清理日期")
+    try:
+        return datetime.fromisoformat(raw.replace("Z", "+00:00")).replace(tzinfo=None)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail="清理日期格式错误") from exc
+
+
 @router.get("/retrieval-log/{log_id}")
 def retrieval_detail(
     log_id: int,

ai_knowledge_base_v2/apps/backend/app/api/admin_dashboard.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timedelta
 
 from fastapi import APIRouter, Depends, Query
+import json
+from pathlib import Path
+from sqlalchemy import select, text
 from sqlalchemy.orm import Session
 
 from app.core.database import get_db
@@ -11,6 +14,8 @@ from app.core.responses import api_success
 from app.models.admin import Admin
 from app.schemas.admin import DashboardStats
 from app.services.admin_service import AdminDashboardService
+from app.models.logs import StorageSnapshot
+from app.services.redis_client import get_sync_redis_client
 
 router = APIRouter()
 
@@ -30,3 +35,53 @@ def dashboard(
         end_dt = None
     stats = AdminDashboardService.stats(db, start_dt, end_dt)
     return api_success(DashboardStats.model_validate(stats).model_dump())
+
+
+@router.get("/dashboard/storage")
+def storage_stats(
+    refresh: bool = Query(default=False),
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    latest = db.scalar(select(StorageSnapshot).where(StorageSnapshot.scope == "project").order_by(StorageSnapshot.id.desc()).limit(1))
+    if latest is not None and not refresh:
+        return api_success(_snapshot_dict(latest, db))
+    detail: dict = {"database": {"status": "unavailable", "bytes": None, "tables": []}, "redis": {"status": "unavailable", "bytes": None}, "files": {"status": "unavailable", "bytes": None}}
+    errors: list[str] = []
+    try:
+        if db.bind and db.bind.dialect.name == "mysql":
+            rows = db.execute(text("SELECT table_name, data_length + index_length AS bytes FROM information_schema.tables WHERE table_schema = DATABASE() ORDER BY bytes DESC")).all()
+            tables = [{"name": row[0], "bytes": int(row[1] or 0)} for row in rows]
+            detail["database"] = {"status": "ok", "bytes": sum(x["bytes"] for x in tables), "tables": tables}
+        else:
+            detail["database"] = {"status": "unavailable", "bytes": None, "tables": [], "message": "当前数据库不支持精确表占用统计"}
+    except Exception as exc:
+        errors.append(f"database: {exc}")
+    try:
+        redis = get_sync_redis_client()
+        if redis is not None:
+            info = redis.info("memory")
+            detail["redis"] = {"status": "ok", "bytes": int(info.get("used_memory", 0))}
+    except Exception as exc:
+        errors.append(f"redis: {exc}")
+    try:
+        root = Path(".uploads")
+        if root.exists():
+            detail["files"] = {"status": "ok", "bytes": sum(p.stat().st_size for p in root.rglob("*") if p.is_file())}
+        else:
+            detail["files"] = {"status": "unavailable", "bytes": None, "message": "项目未配置本地附件目录"}
+    except Exception as exc:
+        errors.append(f"files: {exc}")
+    known = [x.get("bytes") for x in detail.values() if x.get("status") == "ok"]
+    snapshot = StorageSnapshot(scope="project", total_bytes=sum(known) if known else None, detail_json=json.dumps(detail, ensure_ascii=False), status="partial" if errors or any(x.get("status") != "ok" for x in detail.values()) else "ok", error_message="; ".join(errors) or None)
+    db.add(snapshot); db.commit(); db.refresh(snapshot)
+    return api_success(_snapshot_dict(snapshot, db))
+
+
+def _snapshot_dict(item: StorageSnapshot, db: Session) -> dict:
+    with_growth = {"totalBytes": item.total_bytes, "detail": json.loads(item.detail_json), "status": item.status, "errorMessage": item.error_message, "createdAt": item.created_at}
+    # Growth is intentionally unknown until enough snapshots exist; never report a misleading zero.
+    for days, key in ((7, "growth7DaysBytes"), (30, "growth30DaysBytes")):
+        previous = db.scalar(select(StorageSnapshot).where(StorageSnapshot.scope == item.scope, StorageSnapshot.created_at <= item.created_at - timedelta(days=days)).order_by(StorageSnapshot.created_at.desc()).limit(1))
+        with_growth[key] = item.total_bytes - previous.total_bytes if previous and item.total_bytes is not None and previous.total_bytes is not None else None
+    return with_growth

ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py

@@ -36,6 +36,31 @@ def list_knowledge(db: Session = Depends(get_db), current_admin: Admin = Depends
     return api_success([_knowledge_dict(item) for item in items])
 
 
+@router.post("/knowledge/resolve-node")
+def resolve_knowledge_node(
+    payload: dict,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    node_id = str(payload.get("nodeId") or "").strip()
+    if not node_id or len(node_id) > 100 or not node_id.replace("_", "").replace("-", "").isalnum():
+        raise HTTPException(status_code=400, detail="Node ID 格式错误")
+    existing = db.scalar(select(Knowledge).where(Knowledge.feishu_node_id == node_id))
+    if existing is not None:
+        raise HTTPException(status_code=409, detail=f"该 Node ID 已添加（知识库：{existing.name}）")
+    try:
+        info = FeishuKnowledgeService.get_node_info(node_id, db)
+    except Exception as exc:
+        raise HTTPException(status_code=502, detail=f"飞书节点信息获取失败：{exc}") from exc
+    space_id, title = str(info.get("space_id") or "").strip(), str(info.get("title") or "").strip()
+    if not space_id:
+        raise HTTPException(status_code=422, detail="飞书接口未返回 Space ID，请检查节点是否移动或当前应用权限")
+    if not title:
+        raise HTTPException(status_code=422, detail="飞书接口未返回文章标题，请检查文章是否已删除")
+    return api_success({"nodeId": node_id, "spaceId": space_id, "sourceTitle": title, "name": title, "remark": str(info.get("description") or ""), "objType": info.get("obj_type")})
+
+
 @router.post("/knowledge/import")
 def import_knowledge_from_space(
     payload: dict,
@@ -91,12 +116,7 @@ def create_knowledge(
     current_admin: Admin = Depends(get_current_admin),
 ) -> dict:
     _require_super_admin(current_admin)
-    existing = db.scalar(
-        select(Knowledge).where(
-            Knowledge.feishu_space_id == payload.feishuSpaceId,
-            Knowledge.feishu_node_id == payload.feishuNodeId,
-        )
-    )
+    existing = db.scalar(select(Knowledge).where(Knowledge.feishu_node_id == payload.feishuNodeId))
     if existing is not None:
         raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="该飞书文章已创建知识库")
     knowledge = Knowledge(

ai_knowledge_base_v2/apps/backend/app/main.py

@@ -11,6 +11,8 @@ from app.core.database import create_tables
 from app.core.exception_handlers import register_exception_handlers
 from app.core.observability import RequestObservabilityMiddleware, configure_logging
 from app.services.secret_service import SecretService
+from app.services.maintenance_service import MaintenanceService
+import asyncio
 
 
 @asynccontextmanager
@@ -19,7 +21,15 @@ async def lifespan(app: FastAPI):
     SecretService.validate_production_key()
     if settings.auto_create_tables:
         create_tables()
-    yield
+    maintenance_task = asyncio.create_task(MaintenanceService.run_forever())
+    try:
+        yield
+    finally:
+        maintenance_task.cancel()
+        try:
+            await maintenance_task
+        except asyncio.CancelledError:
+            pass
 
 
 settings = get_settings()

ai_knowledge_base_v2/apps/backend/app/models/__init__.py

@@ -19,7 +19,7 @@ from app.models.knowledge import (
     KnowledgeVersion,
     UserKnowledgePermission,
 )
-from app.models.logs import AiRequestLog, OperationLog
+from app.models.logs import AiRequestLog, LogRetentionPolicy, OperationLog, StorageSnapshot
 from app.models.user import User
 
 __all__ = [
@@ -44,6 +44,8 @@ __all__ = [
     "HumanAttentionRecord",
     "ModelConfig",
     "OperationLog",
+    "LogRetentionPolicy",
+    "StorageSnapshot",
     "Prompt",
     "Role",
     "SystemConfig",

ai_knowledge_base_v2/apps/backend/app/models/logs.py

@@ -44,3 +44,25 @@ class OperationLog(Base):
     ip: Mapped[str | None] = mapped_column(String(50), nullable=True)
     result: Mapped[str] = mapped_column(String(20), nullable=False)
     created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False)
+
+
+class LogRetentionPolicy(Base):
+    __tablename__ = "sys_log_retention_policy"
+    id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True)
+    retention_days: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    enabled: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+    last_run_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    last_result: Mapped[str | None] = mapped_column(Text, nullable=True)
+    updated_by: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False)
+
+
+class StorageSnapshot(Base):
+    __tablename__ = "sys_storage_snapshot"
+    id: Mapped[int] = mapped_column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True)
+    scope: Mapped[str] = mapped_column(String(50), nullable=False)
+    total_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    detail_json: Mapped[str] = mapped_column(Text, nullable=False)
+    status: Mapped[str] = mapped_column(String(20), nullable=False)
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False)

ai_knowledge_base_v2/apps/backend/app/services/knowledge_agent_service.py

@@ -23,6 +23,7 @@ from app.models.knowledge import (
     KnowledgeVersion,
 )
 from app.services.knowledge_pipeline_service import expand_synonyms, extract_terms
+from app.services.knowledge_catalog_cache_service import KnowledgeCatalogCacheService
 from app.services.knowledge_service import KnowledgeScope
 from app.services.model_service import _call_configured_model, _system_config_bool
 from app.services.rag_service import PromptService, RagResult, RetrievedChunk
@@ -76,16 +77,16 @@ class KnowledgeAgentService:
         db.add(log)
         db.flush()
         scopes = [KnowledgeScope(id=item["knowledgeId"], name=item["name"], feishu_space_id="", feishu_node_id="") for item in catalog]
-        trace: list[dict] = [{"tool": "get_knowledge_catalog", "count": len(catalog)}]
+        trace: list[dict] = [cls._trace("get_knowledge_catalog", 1, {}, {"items": catalog, "count": len(catalog)}, started)]
         need_knowledge, selected_ids, reason = cls._decide(question, catalog)
-        trace.append({"tool": "agent_decision", "needKnowledge": need_knowledge, "selectedKnowledgeIds": selected_ids, "reason": reason})
+        trace.append(cls._trace("agent_decision", 2, {"question": question}, {"needKnowledge": need_knowledge, "selectedKnowledgeIds": selected_ids, "reason": reason}, started))
         chunks: list[RetrievedChunk] = []
         try:
             if need_knowledge and selected_ids:
                 terms = cls._query_terms(question)
                 candidates = cls.search_knowledge(db, terms, selected_ids, catalog)
-                trace.append({"tool": "search_knowledge", "queryTerms": terms, "candidateCount": len(candidates)})
-                await cls._rerank(db, question, candidates, trace)
+                trace.append(cls._trace("search_knowledge", len(trace) + 1, {"queryTerms": terms, "knowledgeIds": selected_ids}, {"candidateCount": len(candidates), "candidates": [cls._candidate_trace(x) for x in candidates]}, started))
+                await cls._rerank(db, question, candidates, trace, started)
                 selected = cls._select_sections(candidates)
                 chunks = [
                     RetrievedChunk(
@@ -100,7 +101,7 @@ class KnowledgeAgentService:
                     )
                     for item in selected
                 ]
-                trace.append({"tool": "read_knowledge_sections", "sectionIds": [item.section.id for item in selected]})
+                trace.append(cls._trace("read_knowledge_sections", len(trace) + 1, {"sectionIds": [item.section.id for item in selected]}, {"count": len(selected), "sections": [{"knowledgeId": x.knowledge.id, "knowledgeName": x.knowledge.name, "sectionId": x.section.id, "title": x.section.title, "content": cls._protect_content(x.section.content), "adopted": True} for x in selected]}, started))
                 cls._persist_candidates(db, log.id, candidates)
                 log.knowledge_called = 1
                 log.selected_knowledge_ids = ",".join(str(item) for item in selected_ids)
@@ -134,6 +135,13 @@ class KnowledgeAgentService:
         version_overrides: dict[int, int] | None = None,
         preview_knowledge_ids: list[int] | None = None,
     ) -> list[dict]:
+        cacheable = version_overrides is None and preview_knowledge_ids is None
+        if cacheable:
+            return KnowledgeCatalogCacheService.get_or_build("open", lambda: KnowledgeAgentService._catalog_from_db(db, None, None))
+        return KnowledgeAgentService._catalog_from_db(db, version_overrides, preview_knowledge_ids)
+
+    @staticmethod
+    def _catalog_from_db(db: Session, version_overrides: dict[int, int] | None, preview_knowledge_ids: list[int] | None) -> list[dict]:
         query = select(Knowledge).where(Knowledge.lifecycle_status == "active")
         if preview_knowledge_ids is None:
             query = query.where(
@@ -171,6 +179,15 @@ class KnowledgeAgentService:
             )
         return catalog
 
+    @staticmethod
+    def _trace(tool: str, order: int, request: dict, response: dict, started: float, *, status: str = "success", error: str | None = None, retries: int = 0) -> dict:
+        now = int((perf_counter() - started) * 1000)
+        return {"tool": tool, "order": order, "request": request, "response": response, **response, "status": status, "startedAtMs": now, "endedAtMs": now, "durationMs": 0, "responseCount": response.get("count", response.get("candidateCount")), "error": error, "retries": retries}
+
+    @staticmethod
+    def _candidate_trace(item: Candidate) -> dict:
+        return {"knowledgeId": item.knowledge.id, "knowledgeName": item.knowledge.name, "versionId": item.version.id, "chunkId": item.chunk.id, "sectionId": item.section.id, "title": item.chunk.title, "lexicalScore": item.lexical_score, "rerankScore": item.rerank_score, "selected": item.selected, "discardReason": item.discard_reason}
+
     @classmethod
     def search_knowledge(cls, db: Session, terms: list[str], selected_ids: list[int], catalog: list[dict]) -> list[Candidate]:
         versions_by_kb = {item["knowledgeId"]: item["versionId"] for item in catalog}
@@ -200,12 +217,12 @@ class KnowledgeAgentService:
         return candidates[:MAX_LEXICAL_CANDIDATES]
 
     @classmethod
-    async def _rerank(cls, db: Session, question: str, candidates: list[Candidate], trace: list[dict]) -> None:
+    async def _rerank(cls, db: Session, question: str, candidates: list[Candidate], trace: list[dict], started: float) -> None:
         if not candidates:
             return
         model = db.scalar(select(ModelConfig).where(ModelConfig.enabled == 1).order_by(ModelConfig.id.desc()).limit(1))
         if model is None or _system_config_bool(db, "mock_model_enabled", get_settings().mock_model_enabled):
-            trace.append({"tool": "model_rerank", "status": "lexical_fallback", "reason": "模型未启用或处于 mock"})
+            trace.append(cls._trace("Rerank", len(trace) + 1, {"candidateCount": len(candidates)}, {"count": len(candidates), "mode": "lexical_fallback", "candidates": [cls._candidate_trace(x) for x in candidates]}, started))
             return
         prompt = (
             "你是知识检索重排器。根据用户问题给候选打0到100相关分，只返回JSON："
@@ -221,9 +238,9 @@ class KnowledgeAgentService:
             for index, item in enumerate(candidates, 1):
                 item.rerank_score = score_map.get(index)
             candidates.sort(key=lambda item: item.rerank_score if item.rerank_score is not None else item.lexical_score, reverse=True)
-            trace.append({"tool": "model_rerank", "status": "success", "candidateCount": len(candidates)})
+            trace.append(cls._trace("Rerank", len(trace) + 1, {"candidateCount": len(candidates)}, {"count": len(candidates), "candidates": [cls._candidate_trace(x) for x in candidates]}, started))
         except Exception as exc:
-            trace.append({"tool": "model_rerank", "status": "lexical_fallback", "reason": str(exc)[:300]})
+            trace.append(cls._trace("Rerank", len(trace) + 1, {"candidateCount": len(candidates)}, {"count": len(candidates), "mode": "lexical_fallback"}, started, status="fallback", error=str(exc)[:300]))
 
     @staticmethod
     def _select_sections(candidates: list[Candidate]) -> list[Candidate]:

ai_knowledge_base_v2/apps/backend/app/services/knowledge_catalog_cache_service.py

@@ -1,12 +1,46 @@
 from __future__ import annotations
 
+import json
+import time
+from typing import Callable
+
 from app.services.redis_client import get_sync_redis_client
 
 
 class KnowledgeCatalogCacheService:
     """Central invalidation point for current and future KnowledgeList caches."""
 
-    VERSION_KEY = "knowledge:catalog:version"
+    VERSION_KEY = "kb:catalog:generation"
+    KEY_PREFIX = "kb:catalog:v1"
+    METRICS_KEY = "kb:catalog:metrics"
+    TTL_SECONDS = 300
+
+    @classmethod
+    def get_or_build(cls, variant: str, builder: Callable[[], list[dict]]) -> list[dict]:
+        """Read-through cache. Redis is optional and never the source of truth."""
+        client = get_sync_redis_client()
+        if client is None:
+            return builder()
+        started = time.perf_counter()
+        try:
+            generation = client.get(cls.VERSION_KEY) or "1"
+            key = f"{cls.KEY_PREFIX}:{generation}:{variant}"
+            cached = client.get(key)
+            if cached:
+                client.hincrby(cls.METRICS_KEY, "hits", 1)
+                client.hincrbyfloat(cls.METRICS_KEY, "hit_ms", (time.perf_counter() - started) * 1000)
+                return json.loads(cached)
+            client.hincrby(cls.METRICS_KEY, "misses", 1)
+            value = builder()
+            # Short lock prevents concurrent cold requests from repeatedly rebuilding.
+            lock = client.set(f"{key}:lock", "1", nx=True, ex=15)
+            if lock:
+                client.setex(key, cls.TTL_SECONDS, json.dumps(value, ensure_ascii=False, default=str))
+                client.delete(f"{key}:lock")
+            client.hincrbyfloat(cls.METRICS_KEY, "miss_ms", (time.perf_counter() - started) * 1000)
+            return value
+        except Exception:
+            return builder()
 
     @classmethod
     def invalidate(cls) -> None:

ai_knowledge_base_v2/apps/backend/app/services/maintenance_service.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from datetime import datetime, timedelta
+
+from sqlalchemy import delete, select
+
+from app.core.database import SessionLocal
+from app.models.knowledge import KnowledgeRetrievalCandidate, KnowledgeRetrievalLog
+from app.models.logs import LogRetentionPolicy
+from app.services.redis_client import get_sync_redis_client
+
+
+class MaintenanceService:
+    """Single-runner background maintenance. MySQL remains authoritative."""
+
+    @classmethod
+    async def run_forever(cls) -> None:
+        while True:
+            await asyncio.to_thread(cls.run_due_jobs)
+            await asyncio.sleep(3600)
+
+    @classmethod
+    def run_due_jobs(cls) -> None:
+        redis = get_sync_redis_client()
+        lock_acquired = False
+        try:
+            if redis is not None:
+                lock_acquired = bool(redis.set("maintenance:daily:lock", "1", nx=True, ex=3300))
+                if not lock_acquired:
+                    return
+            with SessionLocal() as db:
+                policy = db.scalar(select(LogRetentionPolicy).order_by(LogRetentionPolicy.id).limit(1))
+                if not policy or not policy.enabled or not policy.retention_days:
+                    return
+                if policy.last_run_at and policy.last_run_at.date() >= datetime.now().date():
+                    return
+                before = datetime.now() - timedelta(days=policy.retention_days)
+                deleted = cls.delete_retrieval_logs(db, before)
+                policy.last_run_at = datetime.now()
+                policy.last_result = json.dumps({"status": "success", "deleted": deleted, "before": before.isoformat()}, ensure_ascii=False)
+                db.add(policy)
+                db.commit()
+        except Exception as exc:
+            with SessionLocal() as db:
+                policy = db.scalar(select(LogRetentionPolicy).order_by(LogRetentionPolicy.id).limit(1))
+                if policy:
+                    policy.last_run_at = datetime.now()
+                    policy.last_result = json.dumps({"status": "failed", "error": str(exc)[:1000]}, ensure_ascii=False)
+                    db.add(policy)
+                    db.commit()
+        finally:
+            if redis is not None and lock_acquired:
+                try:
+                    redis.delete("maintenance:daily:lock")
+                except Exception:
+                    pass
+
+    @staticmethod
+    def delete_retrieval_logs(db, before: datetime, batch_size: int = 500) -> int:
+        total = 0
+        while True:
+            ids = list(db.scalars(select(KnowledgeRetrievalLog.id).where(KnowledgeRetrievalLog.created_at < before).order_by(KnowledgeRetrievalLog.id).limit(batch_size)).all())
+            if not ids:
+                return total
+            db.execute(delete(KnowledgeRetrievalCandidate).where(KnowledgeRetrievalCandidate.retrieval_log_id.in_(ids)))
+            db.execute(delete(KnowledgeRetrievalLog).where(KnowledgeRetrievalLog.id.in_(ids)))
+            total += len(ids)
+            db.commit()