Agent/knowhub/server.py

"""
KnowHub Server

Agent 工具使用经验的共享平台。
FastAPI + Milvus Lite（知识）+ SQLite（资源），单文件部署。
"""

import os
import re
import json
import asyncio
import base64
import time
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Optional, List, Dict
from pathlib import Path
import httpx
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

from fastapi import FastAPI, HTTPException, Query, Header, Body, BackgroundTasks, Request
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

# 导入 LLM 调用（需要 agent 模块在 Python path 中）
import sys
sys.path.insert(0, str(Path(__file__).parent.parent))

# 加载环境变量
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent.parent / ".env")

from agent.llm import create_openrouter_llm_call, create_qwen_llm_call
from knowhub.kb_manage_prompts import (
    DEDUP_RELATION_PROMPT,
    TOOL_ANALYSIS_PROMPT,
    RERANK_PROMPT_TEMPLATE,
    KNOWLEDGE_EVOLVE_PROMPT_TEMPLATE,
    KNOWLEDGE_SLIM_PROMPT_TEMPLATE,
    MESSAGE_EXTRACT_PROMPT_TEMPLATE,
)

_dedup_llm = create_openrouter_llm_call(model="google/gemini-2.5-flash-lite")
_tool_analysis_llm = create_qwen_llm_call(model="qwen3.5-plus")

# 导入向量存储和 embedding
from knowhub.knowhub_db.pg_store import PostgreSQLStore
from knowhub.knowhub_db.pg_resource_store import PostgreSQLResourceStore
from knowhub.knowhub_db.pg_tool_store import PostgreSQLToolStore
from knowhub.knowhub_db.pg_capability_store import PostgreSQLCapabilityStore
from knowhub.knowhub_db.pg_requirement_store import PostgreSQLRequirementStore
from knowhub.knowhub_db.pg_strategy_store import PostgreSQLStrategyStore
from knowhub.knowhub_db.version_context import set_version as _set_active_version, DEFAULT_VERSION as _DEFAULT_VERSION
from knowhub.embeddings import get_embedding, get_embeddings_batch

BRAND_NAME    = os.getenv("BRAND_NAME", "KnowHub")
BRAND_API_ENV = os.getenv("BRAND_API_ENV", "KNOWHUB_API")
BRAND_DB      = os.getenv("BRAND_DB", "knowhub.db")

# 组织密钥配置（格式：org1:key1_base64,org2:key2_base64）
ORG_KEYS_RAW = os.getenv("ORG_KEYS", "")
ORG_KEYS = {}
if ORG_KEYS_RAW:
    for pair in ORG_KEYS_RAW.split(","):
        if ":" in pair:
            org, key_b64 = pair.split(":", 1)
            ORG_KEYS[org.strip()] = key_b64.strip()

DB_PATH = Path(__file__).parent / BRAND_DB

# 全局 PostgreSQL 存储实例
pg_store: Optional[PostgreSQLStore] = None
pg_resource_store: Optional[PostgreSQLResourceStore] = None
pg_tool_store: Optional[PostgreSQLToolStore] = None
pg_capability_store: Optional[PostgreSQLCapabilityStore] = None
pg_requirement_store: Optional[PostgreSQLRequirementStore] = None
pg_strategy_store: Optional[PostgreSQLStrategyStore] = None

# --- 加密/解密 ---

def get_org_key(resource_id: str) -> Optional[bytes]:
    """从content_id提取组织前缀，返回对应密钥"""
    if "/" in resource_id:
        org = resource_id.split("/")[0]
        if org in ORG_KEYS:
            return base64.b64decode(ORG_KEYS[org])
    return None


def encrypt_content(resource_id: str, plaintext: str) -> str:
    """加密内容，返回格式：encrypted:AES256-GCM:{base64_data}"""
    if not plaintext:
        return ""

    key = get_org_key(resource_id)
    if not key:
        # 没有配置密钥，明文存储（不推荐）
        return plaintext

    aesgcm = AESGCM(key)
    nonce = os.urandom(12)  # 96-bit nonce
    ciphertext = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)

    # 组合 nonce + ciphertext
    encrypted_data = nonce + ciphertext
    encoded = base64.b64encode(encrypted_data).decode("ascii")

    return f"encrypted:AES256-GCM:{encoded}"


def decrypt_content(resource_id: str, encrypted_text: str, provided_key: Optional[str] = None) -> str:
    """解密内容，如果没有提供密钥或密钥错误，返回[ENCRYPTED]"""
    if not encrypted_text:
        return ""

    if not encrypted_text.startswith("encrypted:AES256-GCM:"):
        # 未加密的内容，直接返回
        return encrypted_text

    # 提取加密数据
    encoded = encrypted_text.split(":", 2)[2]
    encrypted_data = base64.b64decode(encoded)

    nonce = encrypted_data[:12]
    ciphertext = encrypted_data[12:]

    # 获取密钥
    key = None
    if provided_key:
        # 使用提供的密钥
        try:
            key = base64.b64decode(provided_key)
        except Exception:
            return "[ENCRYPTED]"
    else:
        # 从配置中获取
        key = get_org_key(resource_id)

    if not key:
        return "[ENCRYPTED]"

    try:
        aesgcm = AESGCM(key)
        plaintext = aesgcm.decrypt(nonce, ciphertext, None)
        return plaintext.decode("utf-8")
    except Exception:
        return "[ENCRYPTED]"


def to_serializable(data):
    """通用序列化工具：把任意 Python 对象转换为 JSON 可序列化的原生类型"""
    # 基本类型直接返回
    if data is None or isinstance(data, (str, int, float, bool)):
        return data

    # 字典类型递归处理
    if isinstance(data, dict):
        return {k: to_serializable(v) for k, v in data.items()}

    # 列表/元组类型递归处理
    if isinstance(data, (list, tuple)):
        return [to_serializable(item) for item in data]

    # 尝试转换为字典（对于有 to_dict 方法的对象）
    if hasattr(data, 'to_dict') and callable(getattr(data, 'to_dict')):
        try:
            return to_serializable(data.to_dict())
        except:
            pass

    # 尝试转换为列表（对于可迭代对象，如 RepeatedScalarContainer）
    if hasattr(data, '__iter__') and not isinstance(data, (str, bytes, dict)):
        try:
            # 强制转换为列表并递归处理
            result = []
            for item in data:
                result.append(to_serializable(item))
            return result
        except:
            pass

    # 尝试获取对象的属性字典
    if hasattr(data, '__dict__'):
        try:
            return to_serializable(vars(data))
        except:
            pass

    # 最后的 fallback：对于无法处理的类型，返回 None 而不是字符串表示
    # 这样可以避免产生无法序列化的字符串
    return None


# --- Models ---

class ResourceIn(BaseModel):
    id: str
    title: str = ""
    body: str
    secure_body: str = ""
    content_type: str = "text"  # text|code|credential|cookie
    metadata: dict = {}
    sort_order: int = 0
    submitted_by: str = ""


class ResourcePatchIn(BaseModel):
    """PATCH /api/resource/{id} 请求体"""
    title: Optional[str] = None
    body: Optional[str] = None
    secure_body: Optional[str] = None
    content_type: Optional[str] = None
    metadata: Optional[dict] = None


class PostBatchRequest(BaseModel):
    post_ids: List[str] = Field(default_factory=list)


# Knowledge Models
class KnowledgeIn(BaseModel):
    task: str
    content: str
    types: list[str] = ["strategy"]
    tags: dict = {}
    scopes: list[str] = ["org:cybertogether"]
    owner: str = ""
    message_id: str = ""
    resource_ids: list[str] = []
    source: dict = {}  # {name, category, urls, agent_id, submitted_by, timestamp}
    eval: dict = {}    # {score, helpful, harmful, confidence}
    capability_ids: list[str] = []
    tool_ids: list[str] = []


class KnowledgeOut(BaseModel):
    id: str
    message_id: str
    types: list[str]
    task: str
    tags: dict
    scopes: list[str]
    owner: str
    content: str
    source: dict
    eval: dict
    created_at: str
    updated_at: str


class KnowledgeUpdateIn(BaseModel):
    add_helpful_case: Optional[dict] = None
    add_harmful_case: Optional[dict] = None
    update_score: Optional[int] = Field(default=None, ge=1, le=5)
    evolve_feedback: Optional[str] = None


class KnowledgePatchIn(BaseModel):
    """PATCH /api/knowledge/{id} 请求体（直接字段编辑）"""
    task: Optional[str] = None
    content: Optional[str] = None
    types: Optional[list[str]] = None
    tags: Optional[dict] = None
    scopes: Optional[list[str]] = None
    owner: Optional[str] = None
    capability_ids: Optional[list[str]] = None
    tool_ids: Optional[list[str]] = None


class MessageExtractIn(BaseModel):
    """POST /api/extract 请求体（消息历史提取）"""
    messages: list[dict]  # [{role: str, content: str}, ...]
    agent_id: str = "unknown"
    submitted_by: str  # 必填，作为 owner
    session_key: str = ""


class KnowledgeBatchUpdateIn(BaseModel):
    feedback_list: list[dict]



class KnowledgeVerifyIn(BaseModel):
    action: str  # "approve" | "reject"
    verified_by: str = "user"


class KnowledgeBatchVerifyIn(BaseModel):
    knowledge_ids: List[str]
    action: str  # "approve"
    verified_by: str


class KnowledgeSearchResponse(BaseModel):
    results: list[dict]
    count: int


# --- Tool Models ---

class ToolIn(BaseModel):
    id: str
    name: str = ""
    version: Optional[str] = None
    introduction: str = ""
    tutorial: str = ""
    input: dict | str = ""
    output: dict | str = ""
    status: str = "未接入"
    capability_ids: list[str] = []
    knowledge_ids: list[str] = []
    provider_ids: list[str] = []


class ToolPatchIn(BaseModel):
    name: Optional[str] = None
    version: Optional[str] = None
    introduction: Optional[str] = None
    tutorial: Optional[str] = None
    input: Optional[dict | str] = None
    output: Optional[dict | str] = None
    status: Optional[str] = None
    capability_ids: Optional[list[str]] = None
    knowledge_ids: Optional[list[str]] = None
    provider_ids: Optional[list[str]] = None


# --- Capability Models ---

class CapabilityIn(BaseModel):
    id: str
    name: str = ""
    criterion: str = ""
    description: str = ""
    requirement_ids: list[str] = []
    implements: dict = {}
    tool_ids: list[str] = []
    knowledge_ids: list[str] = []


class CapabilityPatchIn(BaseModel):
    name: Optional[str] = None
    criterion: Optional[str] = None
    description: Optional[str] = None
    requirement_ids: Optional[list[str]] = None
    implements: Optional[dict] = None
    tool_ids: Optional[list[str]] = None
    knowledge_ids: Optional[list[str]] = None


# --- Requirement Models ---

class RequirementIn(BaseModel):
    id: str
    description: str = ""
    capability_ids: list[str] = []
    knowledge_ids: list[str] = []
    source_nodes: list[dict] = []
    status: str = "未满足"
    match_result: str = ""


class RequirementPatchIn(BaseModel):
    description: Optional[str] = None
    capability_ids: Optional[list[str]] = None
    knowledge_ids: Optional[list[str]] = None
    source_nodes: Optional[list[dict]] = None
    status: Optional[str] = None
    match_result: Optional[str] = None


# --- Strategy Models ---

class StrategyIn(BaseModel):
    id: str
    name: str = ""
    description: str = ""
    body: str = ""
    status: str = "draft"
    version: str = "v0"
    capability_ids: list[str] = []
    knowledge_ids: list[str] = []
    resource_ids: list[str] = []
    requirement_ids: list[str] = []


class StrategyPatchIn(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    body: Optional[str] = None
    status: Optional[str] = None
    version: Optional[str] = None
    capability_ids: Optional[list[str]] = None
    knowledge_ids: Optional[list[str]] = None
    resource_ids: Optional[list[str]] = None
    requirement_ids: Optional[list[str]] = None


class ResourceNode(BaseModel):
    id: str
    title: str


class ResourceOut(BaseModel):
    id: str
    title: str
    body: str
    secure_body: str = ""
    content_type: str = "text"
    metadata: dict = {}
    images: list[str] = []
    toc: Optional[ResourceNode] = None
    children: list[ResourceNode]
    prev: Optional[ResourceNode] = None
    next: Optional[ResourceNode] = None


# --- Dedup: Globals & Prompt ---

knowledge_processor: Optional["KnowledgeProcessor"] = None


# --- Dedup: RelationCache ---

class RelationCache:
    """关系缓存，存储在内存中"""

    def __init__(self):
        self._cache: Dict[str, List[str]] = {}

    def load(self) -> dict:
        return self._cache

    def save(self, cache: dict):
        self._cache = cache

    def add_relation(self, relation_type: str, knowledge_id: str):
        if relation_type not in self._cache:
            self._cache[relation_type] = []
        if knowledge_id not in self._cache[relation_type]:
            self._cache[relation_type].append(knowledge_id)


# --- Dedup: KnowledgeProcessor ---

class KnowledgeProcessor:
    def __init__(self):
        self._lock = asyncio.Lock()
        self._relation_cache = RelationCache()

    async def process_pending(self):
        """持续处理 pending 和 dedup_passed 知识直到队列为空，有锁防并发"""
        if self._lock.locked():
            return
        async with self._lock:
            # 第一阶段：处理 pending（去重）
            while True:
                try:
                    pending = pg_store.query('status == "pending"', limit=50)
                except Exception as e:
                    print(f"[KnowledgeProcessor] 查询 pending 失败: {e}")
                    break
                if not pending:
                    break
                for knowledge in pending:
                    await self._process_one(knowledge)
            # 第二阶段：处理 dedup_passed（工具关联）
            while True:
                try:
                    dedup_passed = pg_store.query('status == "dedup_passed"', limit=50)
                except Exception as e:
                    print(f"[KnowledgeProcessor] 查询 dedup_passed 失败: {e}")
                    break
                if not dedup_passed:
                    break
                for knowledge in dedup_passed:
                    await self._analyze_tool_relation(knowledge)

    async def _process_one(self, knowledge: dict):
        kid = knowledge["id"]
        now = int(time.time())
        # 乐观锁：pending → processing（时间戳存秒级）
        try:
            pg_store.update(kid, {"status": "processing", "updated_at": now})
        except Exception as e:
            print(f"[KnowledgeProcessor] 锁定 {kid} 失败: {e}")
            return
        try:
            # 向量召回 top-10（只召回 approved/checked）
            embedding = knowledge.get("task_embedding") or knowledge.get("embedding")
            if not embedding:
                embedding = await get_embedding(knowledge["task"])
            candidates = pg_store.search(
                query_embedding=embedding,
                filters='(status == "approved" or status == "checked")',
                limit=10
            )
            candidates = [c for c in candidates if c["id"] != kid]
            # 只保留相似度 >= 0.75 的候选，低于阈值的 task 语义差异太大，直接视为 none
            candidates = [c for c in candidates if c.get("score", 0) >= 0.75]

            if not candidates:
                pg_store.update(kid, {"status": "dedup_passed", "updated_at": now})
                return

            llm_result = await self._llm_judge_relations(knowledge, candidates)
            await self._apply_decision(knowledge, llm_result)

        except Exception as e:
            print(f"[KnowledgeProcessor] 处理 {kid} 失败: {e}，回退到 pending")
            try:
                pg_store.update(kid, {"status": "pending", "updated_at": int(time.time())})
            except Exception:
                pass

    async def _llm_judge_relations(self, new_knowledge: dict, candidates: list) -> dict:
        existing_list = "\n".join([
            f"[{i+1}] ID: {c['id']} | Task: {c['task']} | Content: {c['content'][:300]}"
            for i, c in enumerate(candidates)
        ])
        prompt = DEDUP_RELATION_PROMPT.format(
            new_task=new_knowledge["task"],
            new_content=new_knowledge["content"],
            existing_list=existing_list
        )
        for attempt in range(3):
            try:
                response = await _dedup_llm(
                    messages=[{"role": "user", "content": prompt}],
                )
                content = response.get("content", "").strip()
                # 清理 markdown 代码块
                if "```" in content:
                    parts = content.split("```")
                    for part in parts:
                        part = part.strip()
                        if part.startswith("json"):
                            part = part[4:].strip()
                        try:
                            result = json.loads(part)
                            if "final_decision" in result:
                                content = part
                                break
                        except Exception:
                            continue
                result = json.loads(content)
                assert result.get("final_decision") in ("approved", "rejected")
                return result
            except Exception as e:
                print(f"[LLM Judge] 第{attempt+1}次失败: {e}")
                if attempt < 2:
                    await asyncio.sleep(1)
        return {"final_decision": "approved", "relations": []}

    async def _apply_decision(self, new_knowledge: dict, llm_result: dict):
        kid = new_knowledge["id"]
        final_decision = llm_result.get("final_decision", "approved")
        relations = llm_result.get("relations", [])
        now = int(time.time())

        # 强制规则：如果存在 duplicate 或 subset 关系，必须 rejected
        if any(rel.get("type") in ("duplicate", "subset") for rel in relations):
            final_decision = "rejected"

        if final_decision == "rejected":
            # 记录 rejected 知识的关系到 knowledge_relation 表
            for rel in relations:
                old_id = rel.get("old_id")
                rel_type = rel.get("type", "none")
                if old_id and rel_type != "none":
                    pg_store.add_relation(kid, old_id, rel_type)
                if rel_type in ("duplicate", "subset") and old_id:
                    try:
                        old = pg_store.get_by_id(old_id)
                        if not old:
                            continue
                        eval_data = old.get("eval") or {}
                        eval_data["helpful"] = eval_data.get("helpful", 0) + 1
                        helpful_history = eval_data.get("helpful_history") or []
                        helpful_history.append({
                            "source": "dedup",
                            "related_id": kid,
                            "relation_type": rel_type,
                            "timestamp": now
                        })
                        eval_data["helpful_history"] = helpful_history
                        pg_store.update(old_id, {"eval": eval_data, "updated_at": now})
                    except Exception as e:
                        print(f"[Apply Decision] 更新旧知识 {old_id} helpful 失败: {e}")
            pg_store.update(kid, {"status": "rejected", "updated_at": now})
        else:
            for rel in relations:
                rel_type = rel.get("type", "none")
                reverse_type = rel.get("reverse_type", "none")
                old_id = rel.get("old_id")
                if not old_id or rel_type == "none":
                    continue
                pg_store.add_relation(kid, old_id, rel_type)
                self._relation_cache.add_relation(rel_type, kid)
                self._relation_cache.add_relation(rel_type, old_id)
                if reverse_type and reverse_type != "none":
                    try:
                        pg_store.add_relation(old_id, kid, reverse_type)
                        self._relation_cache.add_relation(reverse_type, old_id)
                        self._relation_cache.add_relation(reverse_type, kid)
                    except Exception as e:
                        print(f"[Apply Decision] 写入反向关系 {old_id} 失败: {e}")
            pg_store.update(kid, {
                "status": "dedup_passed",
                "updated_at": now
            })

    async def _llm_analyze_tools(self, knowledge: dict) -> dict:
        """使用 LLM 分析知识中涉及的工具（复用迁移脚本逻辑）"""
        task = (knowledge.get("task") or "")[:600]
        content = (knowledge.get("content") or "")[:1200]
        prompt = TOOL_ANALYSIS_PROMPT.format(task=task, content=content)
        try:
            response = await _tool_analysis_llm(
                messages=[{"role": "user", "content": prompt}],
                max_tokens=2048,
                temperature=0.1,
            )
            raw = (response.get("content") or "").strip()
            if raw.startswith("```"):
                lines = raw.split("\n")
                inner = []
                in_block = False
                for line in lines:
                    if line.startswith("```"):
                        in_block = not in_block
                        continue
                    if in_block:
                        inner.append(line)
                raw = "\n".join(inner).strip()
            return json.loads(raw)
        except Exception as e:
            print(f"[Tool Analysis LLM] 调用失败: {e}")
            raise

    async def _create_or_get_tool_resource(self, tool_info: dict) -> Optional[str]:
        """创建或获取工具资源（存入 PostgreSQL tool 表）"""
        category = tool_info.get("category", "other")
        slug = tool_info.get("slug", "")
        if not slug:
            return None
        tool_id = f"tools/{category}/{slug}"
        now_ts = int(time.time())
        cursor = pg_store._get_cursor()
        try:
            cursor.execute("SELECT id FROM tool WHERE id = %s", (tool_id,))
            if cursor.fetchone():
                return tool_id
            cursor.execute("""
                INSERT INTO tool (id, name, version, introduction, tutorial, input, output,
                                  updated_time, status)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
            """, (
                tool_id,
                tool_info.get("name", slug),
                tool_info.get("version") or None,
                tool_info.get("description", ""),
                tool_info.get("usage", ""),
                json.dumps(tool_info.get("input", "")),
                json.dumps(tool_info.get("output", "")),
                now_ts,
                tool_info.get("status", "未接入"),
            ))
            pg_store.conn.commit()
            print(f"[Tool Resource] 创建新工具: {tool_id}")
            return tool_id
        finally:
            cursor.close()

    async def _update_tool_knowledge_index(self, tool_id: str, knowledge_id: str):
        """向工具添加知识关联（写入 tool_knowledge 关联表）"""
        pg_tool_store.add_knowledge(tool_id, knowledge_id)

    async def _analyze_tool_relation(self, knowledge: dict):
        """分析知识与工具的关联关系"""
        kid = knowledge["id"]
        now = int(time.time())
        # 乐观锁：dedup_passed → analyzing
        try:
            pg_store.update(kid, {"status": "analyzing", "updated_at": now})
        except Exception as e:
            print(f"[Tool Analysis] 锁定 {kid} 失败: {e}")
            return
        try:
            tool_analysis = await self._llm_analyze_tools(knowledge)
            has_tools = bool(tool_analysis and tool_analysis.get("has_tools"))

            existing_tags = knowledge.get("tags") or {}
            has_tool_tag = existing_tags.get("tool") is True

            # 情况1：LLM 判定无工具，但有 tool tag → 重新分析一次
            if not has_tools and has_tool_tag:
                print(f"[Tool Analysis] {kid} LLM 判定无工具但有 tool tag，重新分析")
                tool_analysis = await self._llm_analyze_tools(knowledge)
                has_tools = bool(tool_analysis and tool_analysis.get("has_tools"))
                # 重新分析后仍然不一致 → 知识模糊，rejected
                if not has_tools:
                    pg_store.update(kid, {"status": "rejected", "updated_at": now})
                    print(f"[Tool Analysis] {kid} 两次判定不一致，知识模糊，rejected")
                    return

            # 情况2：无工具且无 tool tag → 直接 approved
            if not has_tools:
                pg_store.update(kid, {"status": "approved", "updated_at": now})
                return

            # 情况3/4：有工具 → 创建工具并关联
            tool_ids = []
            for tool_info in (tool_analysis.get("tools") or []):
                tool_id = await self._create_or_get_tool_resource(tool_info)
                if tool_id:
                    tool_ids.append(tool_id)

            updates: dict = {
                "status": "approved",
                "updated_at": now
            }
            # 有工具但无 tool tag → 添加 tag
            if not has_tool_tag:
                updated_tags = dict(existing_tags)
                updated_tags["tool"] = True
                updates["tags"] = updated_tags
                print(f"[Tool Analysis] {kid} 添加 tool tag")

            pg_store.update(kid, updates)

            # 写入 tool_knowledge 关联
            for tool_id in tool_ids:
                await self._update_tool_knowledge_index(tool_id, kid)

            print(f"[Tool Analysis] {kid} 关联了 {len(tool_ids)} 个工具")

        except Exception as e:
            print(f"[Tool Analysis] {kid} 分析失败: {e}，回退到 dedup_passed")
            try:
                pg_store.update(kid, {"status": "dedup_passed", "updated_at": int(time.time())})
            except Exception:
                pass


async def _periodic_processor():
    """每60秒检测超时条目并回滚：processing(>5min)→pending，analyzing(>10min)→dedup_passed"""
    while True:
        await asyncio.sleep(60)
        try:
            now = int(time.time())
            # 回滚超时的 processing（5分钟 → pending）
            timeout_5min = now - 300
            processing = pg_store.query('status == "processing"', limit=200)
            for item in processing:
                updated_at = item.get("updated_at", 0) or 0
                updated_at_sec = updated_at // 1000 if updated_at > 1_000_000_000_000 else updated_at
                if updated_at_sec < timeout_5min:
                    print(f"[Periodic] 回滚超时 processing → pending: {item['id']}")
                    pg_store.update(item["id"], {"status": "pending", "updated_at": int(time.time())})
            # 回滚超时的 analyzing（10分钟 → dedup_passed）
            timeout_10min = now - 600
            analyzing = pg_store.query('status == "analyzing"', limit=200)
            for item in analyzing:
                updated_at = item.get("updated_at", 0) or 0
                updated_at_sec = updated_at // 1000 if updated_at > 1_000_000_000_000 else updated_at
                if updated_at_sec < timeout_10min:
                    print(f"[Periodic] 回滚超时 analyzing → dedup_passed: {item['id']}")
                    pg_store.update(item["id"], {"status": "dedup_passed", "updated_at": int(time.time())})
        except Exception as e:
            print(f"[Periodic] 定时任务错误: {e}")


# --- App ---

@asynccontextmanager
async def lifespan(app: FastAPI):
    global pg_store, pg_resource_store, pg_tool_store, pg_capability_store, pg_requirement_store, pg_strategy_store, knowledge_processor

    # 初始化 PostgreSQL（knowledge + resources + tools + capabilities + requirements + strategy）
    pg_store = PostgreSQLStore()
    pg_resource_store = PostgreSQLResourceStore()
    pg_tool_store = PostgreSQLToolStore()
    pg_capability_store = PostgreSQLCapabilityStore()
    pg_requirement_store = PostgreSQLRequirementStore()
    pg_strategy_store = PostgreSQLStrategyStore()

    # 初始化去重处理器 + 启动定时兜底任务
    knowledge_processor = KnowledgeProcessor()
    periodic_task = asyncio.create_task(_periodic_processor())

    yield

    # 清理
    periodic_task.cancel()
    try:
        await periodic_task
    except asyncio.CancelledError:
        pass
    pg_store.close()
    pg_resource_store.close()
    pg_tool_store.close()
    pg_capability_store.close()
    pg_requirement_store.close()
    pg_strategy_store.close()


app = FastAPI(title=BRAND_NAME, lifespan=lifespan)

# 挂载静态文件
STATIC_DIR = Path(__file__).parent / "frontend" / "dist"
if STATIC_DIR.exists():
    app.mount("/assets", StaticFiles(directory=str(STATIC_DIR / "assets")), name="assets")

# 提供 itemsets_all.json
@app.get("/itemsets_all.json")
async def serve_itemsets():
    itemsets_file = STATIC_DIR / "itemsets_all.json"
    if itemsets_file.exists():
        return FileResponse(itemsets_file)
    return {"error": "itemsets_all.json not found"}


# --- 数据版本上下文中间件 ---
# 从请求头 X-KnowHub-Version 或 query ?version=xxx 读版本，写入 contextvar；
# 全链路的 store 查询都从 contextvar 读 active version 做过滤。
_ALLOWED_VERSIONS = {'dev_abstract', 'dev_dedup', 'tao_dev', 'v0'}

@app.middleware("http")
async def set_active_version(request: Request, call_next):
    v = request.headers.get('X-KnowHub-Version') or request.query_params.get('version') or _DEFAULT_VERSION
    if v not in _ALLOWED_VERSIONS:
        v = _DEFAULT_VERSION
    _set_active_version(v)
    response = await call_next(request)
    response.headers['X-KnowHub-Version'] = v
    return response


# --- 缓存自动失效中间件 ---
# 任何对核心实体的写操作（POST/PATCH/DELETE）自动清除对应缓存
_DASHBOARD_INVALIDATE_PREFIXES = ("/api/requirement", "/api/capability", "/api/tool", "/api/strategy", "/api/knowledge")

@app.middleware("http")
async def auto_invalidate_caches(request: Request, call_next):
    response = await call_next(request)
    if request.method in ("POST", "PATCH", "PUT", "DELETE") and response.status_code < 400:
        path = request.url.path
        if any(path.startswith(p) for p in _DASHBOARD_INVALIDATE_PREFIXES):
            _invalidate_dashboard_cache()
        if path.startswith("/api/resource") and not path.endswith("/batch"):
            _invalidate_resource_cache()
    return response


# --- Knowledge API ---

# --- Resource Batch API ---

# --- Resource 缓存（与 Dashboard 同 TTL，写入时失效） ---
_resource_cache: Dict[str, dict] = {}
_resource_cache_ts: float = 0


def _invalidate_resource_cache():
    global _resource_cache, _resource_cache_ts
    _resource_cache.clear()
    _resource_cache_ts = 0


def _get_cached_resource(rid: str) -> Optional[dict]:
    """从缓存取 resource，miss 时查 DB 并写入缓存"""
    global _resource_cache_ts
    now = time.time()
    if now - _resource_cache_ts > _DASHBOARD_CACHE_TTL:
        _resource_cache.clear()
        _resource_cache_ts = now
    if rid in _resource_cache:
        return _resource_cache[rid]
    row = pg_resource_store.get_by_id(rid)
    if not row:
        return None
    entry = {
        "id": row["id"],
        "title": row["title"],
        "body": row["body"],
        "content_type": row.get("content_type", "text"),
        "metadata": row.get("metadata", {}),
        "images": row.get("images", []),
    }
    _resource_cache[rid] = entry
    return entry


@app.post("/api/resource/batch")
def batch_get_resources(body: dict = Body(...)):
    """批量获取 resource 基本信息（不含 toc/children/siblings 导航），用于 Dashboard 等场景。
    带后端内存缓存（24h TTL，resource 写入时失效）。"""
    ids = body.get("ids", [])
    if not ids:
        return {"resources": {}}
    resources: Dict[str, dict] = {}
    for rid in ids:
        try:
            entry = _get_cached_resource(rid)
            if entry:
                resources[rid] = entry
        except Exception as e:
            print(f"[batch_get_resources] Failed to fetch {rid}: {e}")
    return {"resources": resources}


@app.post("/api/resource", status_code=201)
def submit_resource(resource: ResourceIn):
    """提交资源（存入 PostgreSQL resources 表）"""
    try:
        # 加密敏感内容
        encrypted_secure_body = encrypt_content(resource.id, resource.secure_body)

        pg_resource_store.insert_or_update({
            'id': resource.id,
            'title': resource.title,
            'body': resource.body,
            'secure_body': encrypted_secure_body,
            'content_type': resource.content_type,
            'metadata': resource.metadata,
            'sort_order': resource.sort_order,
            'submitted_by': resource.submitted_by
        })
        return {"status": "ok", "id": resource.id}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/resource/{resource_id:path}", response_model=ResourceOut)
def get_resource(resource_id: str, x_org_key: Optional[str] = Header(None)):
    """获取资源详情（从 PostgreSQL）"""
    try:
        row = pg_resource_store.get_by_id(resource_id)
        if not row:
            raise HTTPException(status_code=404, detail=f"Resource not found: {resource_id}")

        # 解密敏感内容
        secure_body = decrypt_content(resource_id, row.get("secure_body", ""), x_org_key)

        # 计算导航上下文
        root_id = resource_id.split("/")[0] if "/" in resource_id else resource_id

        # TOC (根节点)
        toc = None
        if "/" in resource_id:
            toc_row = pg_resource_store.get_by_id(root_id)
            if toc_row:
                toc = ResourceNode(id=toc_row["id"], title=toc_row["title"])

        # Children (子节点)
        children_rows = pg_resource_store.list_resources(prefix=f"{resource_id}/", limit=1000)
        children = [ResourceNode(id=r["id"], title=r["title"]) for r in children_rows
                    if r["id"].count("/") == resource_id.count("/") + 1]

        # Prev/Next (同级节点)
        prev_node, next_node = pg_resource_store.get_siblings(resource_id)
        prev = ResourceNode(id=prev_node["id"], title=prev_node["title"]) if prev_node else None
        next = ResourceNode(id=next_node["id"], title=next_node["title"]) if next_node else None

        return ResourceOut(
            id=row["id"],
            title=row["title"],
            body=row["body"],
            secure_body=secure_body,
            content_type=row["content_type"],
            metadata=row.get("metadata", {}),
            images=row.get("images", []),
            toc=toc,
            children=children,
            prev=prev,
            next=next,
        )
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.patch("/api/resource/{resource_id:path}")
def patch_resource(resource_id: str, patch: ResourcePatchIn):
    """更新resource字段（PostgreSQL）"""
    try:
        # 检查是否存在
        if not pg_resource_store.get_by_id(resource_id):
            raise HTTPException(status_code=404, detail=f"Resource not found: {resource_id}")

        # 构建更新字典
        updates = {}

        if patch.title is not None:
            updates['title'] = patch.title
        if patch.body is not None:
            updates['body'] = patch.body
        if patch.secure_body is not None:
            updates['secure_body'] = encrypt_content(resource_id, patch.secure_body)
        if patch.content_type is not None:
            updates['content_type'] = patch.content_type
        if patch.metadata is not None:
            updates['metadata'] = patch.metadata

        if not updates:
            return {"status": "ok", "message": "No fields to update"}

        pg_resource_store.update(resource_id, updates)
        return {"status": "ok", "id": resource_id}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/resource")
def list_resources(
    content_type: Optional[str] = Query(None),
    limit: int = Query(100, ge=1, le=1000)
):
    """列出所有resource（PostgreSQL）"""
    try:
        results = pg_resource_store.list_resources(
            content_type=content_type,
            limit=limit
        )
        return {"results": results, "count": len(results)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.delete("/api/resource/{resource_id:path}")
def delete_resource(resource_id: str):
    """删除单个resource（PostgreSQL）"""
    try:
        if not pg_resource_store.get_by_id(resource_id):
            raise HTTPException(status_code=404, detail=f"Resource not found: {resource_id}")

        pg_resource_store.delete(resource_id)
        return {"status": "ok", "id": resource_id}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# --- Knowledge API ---


# ===== Knowledge API =====

async def _llm_rerank(query: str, candidates: list[dict], top_k: int) -> list[str]:
    """
    使用 LLM 对候选知识进行精排

    Args:
        query: 查询文本
        candidates: 候选知识列表
        top_k: 返回数量

    Returns:
        排序后的知识 ID 列表
    """
    if not candidates:
        return []

    # 构造 prompt
    candidates_text = "\n".join([
        f"[{i+1}] ID: {c['id']}\nTask: {c['task']}\nContent: {c['content'][:200]}..."
        for i, c in enumerate(candidates)
    ])

    prompt = RERANK_PROMPT_TEMPLATE.format(
        top_k=top_k,
        query=query,
        candidates_text=candidates_text
    )

    try:
        response = await _dedup_llm(
            messages=[{"role": "user", "content": prompt}],
        )

        content = response.get("content", "").strip()
        # 解析 ID 列表
        selected_ids = [
            idx.strip()
            for idx in re.split(r'[,\s]+', content)
            if idx.strip().startswith(("knowledge-", "research-"))
        ]

        return selected_ids[:top_k]

    except Exception as e:
        print(f"[LLM Rerank] 失败: {e}")
        return []


# --- Knowledge Ask / Upload API (Librarian Agent HTTP 接口) ---


class AgentRequest(BaseModel):
    """POST /api/agent 请求体（统一远端 Agent 入口）"""
    agent_type: str  # 必填，必须是 remote_ 前缀
    task: str  # 必填，单任务描述
    messages: Optional[list[dict]] = None  # 预置 OpenAI 格式消息
    continue_from: Optional[str] = None  # 已有 sub_trace_id，传入则续跑
    skills: Optional[list[str]] = None  # 调用方指定 skill，服务器按 agent_type 的白名单过滤


class AgentResponse(BaseModel):
    """POST /api/agent 响应体（标准 Agent 结果）"""
    sub_trace_id: Optional[str] = None
    status: str
    summary: str = ""
    stats: dict = {}
    error: Optional[str] = None


# agent_type → handler 的映射。handler 签名: (query, continue_from, skills) -> dict
# 新增远端 Agent 在这里登记。
_REMOTE_AGENT_DISPATCH = {}


def _get_remote_agent_dispatch():
    """延迟导入 agent 实现，避免循环导入和启动时加载重型依赖"""
    global _REMOTE_AGENT_DISPATCH
    if not _REMOTE_AGENT_DISPATCH:
        from agents.librarian import run_librarian
        from agents.research import research as _research
        _REMOTE_AGENT_DISPATCH = {
            "remote_librarian": run_librarian,
            "remote_research": _research,
        }
    return _REMOTE_AGENT_DISPATCH


@app.post("/api/agent", response_model=AgentResponse)
async def agent_api(req: AgentRequest):
    """
    统一远端 Agent 入口。按 agent_type 分发到对应实现。

    全部同步：Agent 运行完成后返回标准 {status, sub_trace_id, summary, stats}。
    续跑由 caller 传入 continue_from 显式指定（服务器不维护 caller→trace 映射）。
    Skills 由 caller 指定，每个 handler 用自己的白名单过滤。
    """
    if not req.agent_type.startswith("remote_"):
        raise HTTPException(
            status_code=400,
            detail=f"agent_type 必须以 'remote_' 开头，收到: {req.agent_type}",
        )

    dispatch = _get_remote_agent_dispatch()
    handler = dispatch.get(req.agent_type)
    if handler is None:
        raise HTTPException(
            status_code=404,
            detail=f"未注册的 agent_type: {req.agent_type}，可用: {list(dispatch.keys())}",
        )

    try:
        result = await handler(
            query=req.task,
            continue_from=req.continue_from,
            skills=req.skills,
        )
        return AgentResponse(**result)
    except Exception as e:
        import traceback
        traceback.print_exc()
        print(f"[/api/agent] {req.agent_type} 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/knowledge/upload/pending")
async def list_pending_uploads_api():
    """列出所有未处理或失败的 upload 任务（用于排查和重跑）"""
    from agents.librarian import list_pending_uploads
    pending = list_pending_uploads()
    return {"pending": pending, "count": len(pending)}


@app.post("/api/knowledge/upload/retry")
async def retry_pending_uploads_api():
    """重跑所有失败的 upload 任务（同步执行，按顺序处理）"""
    from agents.librarian import list_pending_uploads, run_librarian

    pending = list_pending_uploads()
    failed = [p for p in pending if p["status"] == "failed"]

    results = []
    for item in failed:
        buffer_file = item["file"]
        data = json.loads(Path(buffer_file).read_text(encoding="utf-8"))
        payload = data.get("data", {})
        result = await run_librarian(
            query=json.dumps(payload),
            continue_from=None,
            skills=["upload_strategy"],
        )
        results.append({"file": buffer_file, "status": result.get("status"), "error": result.get("error")})

    return {"retried": len(failed), "results": results}


@app.get("/api/knowledge/search")
async def search_knowledge_api(
    q: str = Query(..., description="查询文本"),
    top_k: int = Query(default=5, ge=1, le=20),
    min_score: int = Query(default=3, ge=1, le=5),
    types: Optional[str] = None,
    owner: Optional[str] = None,
    requirement_id: Optional[str] = None,
    capability_id: Optional[str] = None,
    tool_id: Optional[str] = None
):
    """检索知识（向量召回 + LLM 精排）"""
    try:
        # 1. 生成查询向量
        query_embedding = await get_embedding(q)

        # 2. 构建过滤表达式
        filters = []
        if types:
            type_list = [t.strip() for t in types.split(',') if t.strip()]
            if len(type_list) == 1:
                filters.append(f'array_contains(types, "{type_list[0]}")')
            elif len(type_list) > 1:
                type_filters = [f'array_contains(types, "{t}")' for t in type_list]
                filters.append(f'({" or ".join(type_filters)})')
        if owner:
            owner_list = [o.strip() for o in owner.split(',') if o.strip()]
            if len(owner_list) == 1:
                filters.append(f'owner == "{owner_list[0]}"')
            else:
                # 多个owner用OR连接
                owner_filters = [f'owner == "{o}"' for o in owner_list]
                filters.append(f'({" or ".join(owner_filters)})')

        # 添加 min_score 过滤
        filters.append(f'eval["score"] >= {min_score}')

        # 只搜索 approved 和 checked 的知识
        filters.append('(status == "approved" or status == "checked")')

        filter_expr = ' and '.join(filters) if filters else None
        
        relation_filters = {}
        if requirement_id: relation_filters['requirement_id'] = requirement_id
        if capability_id: relation_filters['capability_id'] = capability_id
        if tool_id: relation_filters['tool_id'] = tool_id

        # 3. 向量召回（3*k 个候选）
        recall_limit = top_k * 3
        candidates = pg_store.search(
            query_embedding=query_embedding,
            filters=filter_expr,
            limit=recall_limit,
            relation_filters=relation_filters
        )

        if not candidates:
            return {"results": [], "count": 0, "reranked": False}

        # 转换为可序列化的格式
        serialized_candidates = [to_serializable(c) for c in candidates]

        # 为了保证搜索的极致速度，直接返回向量召回的 top-k（跳过缓慢的 LLM 精排）
        return {"results": serialized_candidates[:top_k], "count": len(serialized_candidates[:top_k]), "reranked": False}
        
    except Exception as e:
        print(f"[Knowledge Search] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/knowledge", status_code=201)
async def save_knowledge(knowledge: KnowledgeIn, background_tasks: BackgroundTasks):
    """保存新知识"""
    try:
        # 生成 ID
        timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
        random_suffix = uuid.uuid4().hex[:4]
        knowledge_id = f"knowledge-{timestamp}-{random_suffix}"

        now = int(time.time())

        # 设置默认值
        owner = knowledge.owner or f"agent:{knowledge.source.get('agent_id', 'unknown')}"

        # 准备 source
        source = {
            "name": knowledge.source.get("name", ""),
            "category": knowledge.source.get("category", ""),
            "urls": knowledge.source.get("urls", []),
            "agent_id": knowledge.source.get("agent_id", "unknown"),
            "submitted_by": knowledge.source.get("submitted_by", ""),
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "message_id": knowledge.message_id
        }

        # 准备 eval
        eval_data = {
            "score": knowledge.eval.get("score", 3),
            "helpful": knowledge.eval.get("helpful", 1),
            "harmful": knowledge.eval.get("harmful", 0),
            "confidence": knowledge.eval.get("confidence", 0.5),
            "helpful_history": [],
            "harmful_history": []
        }

        # 生成向量（task_embedding + content_embedding 双向量）
        task_embedding = await get_embedding(knowledge.task)
        content_embedding = await get_embedding(knowledge.content)

        # 提取 tag keys（用于高效筛选）
        tag_keys = list(knowledge.tags.keys()) if isinstance(knowledge.tags, dict) else []

        # 准备插入数据
        insert_data = {
            "id": knowledge_id,
            "task_embedding": task_embedding,
            "content_embedding": content_embedding,
            "message_id": knowledge.message_id,
            "task": knowledge.task,
            "content": knowledge.content,
            "types": knowledge.types,
            "tags": knowledge.tags,
            "tag_keys": tag_keys,
            "scopes": knowledge.scopes,
            "owner": owner,
            "source": source,
            "eval": eval_data,
            "created_at": now,
            "updated_at": now,
            "status": "pending",
            "capability_ids": knowledge.capability_ids,
            "tool_ids": knowledge.tool_ids,
            "resource_ids": knowledge.resource_ids,
        }

        print(f"[Save Knowledge] 插入数据: {json.dumps({k: v for k, v in insert_data.items() if k != 'embedding'}, ensure_ascii=False)}")

        # 插入 Milvus
        pg_store.insert(insert_data)

        # 触发后台去重处理
        background_tasks.add_task(knowledge_processor.process_pending)

        return {"status": "pending", "knowledge_id": knowledge_id, "message": "知识已入队，正在处理去重..."}

    except Exception as e:
        print(f"[Save Knowledge] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/knowledge")
def list_knowledge(
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=1000),
    types: Optional[str] = None,
    scopes: Optional[str] = None,
    owner: Optional[str] = None,
    tags: Optional[str] = None,
    status: Optional[str] = None,
    requirement_id: Optional[str] = None,
    capability_id: Optional[str] = None,
    tool_id: Optional[str] = None
):
    """列出知识（支持后端筛选和分页）"""
    try:
        # 构建过滤表达式
        filters = []

        # types 支持多个，改为用 OR 连接（并集：包含任意选中 type 即可）
        if types:
            type_list = [t.strip() for t in types.split(',') if t.strip()]
            if len(type_list) == 1:
                filters.append(f'array_contains(types, "{type_list[0]}")')
            elif len(type_list) > 1:
                type_filters = [f'array_contains(types, "{t}")' for t in type_list]
                filters.append(f'({" or ".join(type_filters)})')

        if scopes:
            filters.append(f'array_contains(scopes, "{scopes}")')

        if owner:
            owner_list = [o.strip() for o in owner.split(',') if o.strip()]
            if len(owner_list) == 1:
                filters.append(f'owner == "{owner_list[0]}"')
            else:
                # 多个owner用OR连接
                owner_filters = [f'owner == "{o}"' for o in owner_list]
                filters.append(f'({" or ".join(owner_filters)})')

        # tags 支持多个，用 AND 连接（使用 tag_keys 数组进行高效筛选）
        if tags:
            tag_list = [t.strip() for t in tags.split(',') if t.strip()]
            for t in tag_list:
                filters.append(f'array_contains(tag_keys, "{t}")')

        # 只返回指定 status 的知识（默认 approved 和 checked）
        status_list = [s.strip() for s in (status or "approved,checked").split(',') if s.strip()]
        status_conditions = ' or '.join([f'status == "{s}"' for s in status_list])
        filters.append(f'({status_conditions})')

        # 如果没有过滤条件，查询所有
        filter_expr = ' and '.join(filters) if filters else 'id != ""'

        relation_filters = {}
        if requirement_id: relation_filters['requirement_id'] = requirement_id
        if capability_id: relation_filters['capability_id'] = capability_id
        if tool_id: relation_filters['tool_id'] = tool_id

        # 查询 Milvus/PG（先获取所有符合条件的数据）
        # limit 是总数限制，我们需要获取足够多的数据来支持分页
        max_limit = 10000  # 设置一个合理的上限
        results = pg_store.query(filter_expr, limit=max_limit, relation_filters=relation_filters)

        # 转换为可序列化的格式
        serialized_results = [to_serializable(r) for r in results]

        # 按 created_at 降序排序（最新的在前）
        serialized_results.sort(key=lambda x: x.get('created_at', 0), reverse=True)

        # 计算分页
        total = len(serialized_results)
        total_pages = (total + page_size - 1) // page_size  # 向上取整
        start_idx = (page - 1) * page_size
        end_idx = start_idx + page_size
        page_results = serialized_results[start_idx:end_idx]

        return {
            "results": page_results,
            "pagination": {
                "page": page,
                "page_size": page_size,
                "total": total,
                "total_pages": total_pages
            }
        }

    except Exception as e:
        print(f"[List Knowledge] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/knowledge/meta/tags")
def get_all_tags():
    """获取所有已有的 tags"""
    try:
        # 查询所有知识
        results = pg_store.query('id != ""', limit=10000)

        all_tags = set()
        for item in results:
            # 转换为标准字典
            serialized_item = to_serializable(item)
            tags_dict = serialized_item.get("tags", {})
            if isinstance(tags_dict, dict):
                for key in tags_dict.keys():
                    all_tags.add(key)

        return {"tags": sorted(list(all_tags))}

    except Exception as e:
        print(f"[Get Tags] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/knowledge/pending")
def get_pending_knowledge(limit: int = Query(default=50, ge=1, le=200)):
    """查询待处理队列（pending + processing + dedup_passed + analyzing）"""
    try:
        pending = pg_store.query(
            'status == "pending" or status == "processing" or status == "dedup_passed" or status == "analyzing"',
            limit=limit
        )
        serialized = [to_serializable(r) for r in pending]
        return {"results": serialized, "count": len(serialized)}
    except Exception as e:
        print(f"[Pending] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/knowledge/process")
async def trigger_process(force: bool = Query(default=False)):
    """手动触发去重处理。force=true 时先回滚所有 processing → pending，analyzing → dedup_passed"""
    try:
        if force:
            processing = pg_store.query('status == "processing"', limit=200)
            for item in processing:
                pg_store.update(item["id"], {"status": "pending", "updated_at": int(time.time())})
            print(f"[Manual Process] 回滚 {len(processing)} 条 processing → pending")
            analyzing = pg_store.query('status == "analyzing"', limit=200)
            for item in analyzing:
                pg_store.update(item["id"], {"status": "dedup_passed", "updated_at": int(time.time())})
            print(f"[Manual Process] 回滚 {len(analyzing)} 条 analyzing → dedup_passed")
        asyncio.create_task(knowledge_processor.process_pending())
        return {"status": "ok", "message": "处理任务已触发"}
    except Exception as e:
        print(f"[Manual Process] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/knowledge/migrate")
async def migrate_knowledge_schema():
    """手动触发 schema 迁移（PostgreSQL不需要此功能）"""
    return {"status": "ok", "message": "PostgreSQL不需要schema迁移"}


@app.get("/api/knowledge/status/{knowledge_id}")
def get_knowledge_status(knowledge_id: str):
    """查询单条知识的处理状态和关系"""
    try:
        result = pg_store.get_by_id(knowledge_id)
        if not result:
            raise HTTPException(status_code=404, detail=f"Knowledge not found: {knowledge_id}")
        serialized = to_serializable(result)
        return {
            "id": knowledge_id,
            "status": serialized.get("status", "approved"),
            "relations": serialized.get("relations", []),
            "created_at": serialized.get("created_at"),
            "updated_at": serialized.get("updated_at"),
        }
    except HTTPException:
        raise
    except Exception as e:
        print(f"[Knowledge Status] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/knowledge/{knowledge_id}")
def get_knowledge(knowledge_id: str):
    """获取单条知识"""
    try:
        result = pg_store.get_by_id(knowledge_id)
        if not result:
            raise HTTPException(status_code=404, detail=f"Knowledge not found: {knowledge_id}")

        return to_serializable(result)

    except HTTPException:
        raise
    except Exception as e:
        print(f"[Get Knowledge] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


async def _evolve_knowledge_with_llm(old_content: str, feedback: str) -> str:
    """使用 LLM 进行知识进化重写"""
    prompt = KNOWLEDGE_EVOLVE_PROMPT_TEMPLATE.format(
        old_content=old_content,
        feedback=feedback
    )
    try:
        response = await _dedup_llm(
            messages=[{"role": "user", "content": prompt}],
        )
        evolved = response.get("content", "").strip()
        if len(evolved) < 5:
            raise ValueError("LLM output too short")
        return evolved
    except Exception as e:
        print(f"知识进化失败，采用追加模式回退: {e}")
        return f"{old_content}\n\n---\n[Update {datetime.now().strftime('%Y-%m-%d')}]: {feedback}"


@app.put("/api/knowledge/{knowledge_id}")
async def update_knowledge(knowledge_id: str, update: KnowledgeUpdateIn):
    """更新知识评估，支持知识进化"""
    try:
        # 获取现有知识
        existing = pg_store.get_by_id(knowledge_id)
        if not existing:
            raise HTTPException(status_code=404, detail=f"Knowledge not found: {knowledge_id}")

        eval_data = existing.get("eval", {})

        # 更新评分
        if update.update_score is not None:
            eval_data["score"] = update.update_score

        # 添加有效案例
        if update.add_helpful_case:
            eval_data["helpful"] = eval_data.get("helpful", 0) + 1
            if "helpful_history" not in eval_data:
                eval_data["helpful_history"] = []
            eval_data["helpful_history"].append(update.add_helpful_case)

        # 添加有害案例
        if update.add_harmful_case:
            eval_data["harmful"] = eval_data.get("harmful", 0) + 1
            if "harmful_history" not in eval_data:
                eval_data["harmful_history"] = []
            eval_data["harmful_history"].append(update.add_harmful_case)

        # 知识进化
        content = existing["content"]
        need_reembed = False

        if update.evolve_feedback:
            content = await _evolve_knowledge_with_llm(content, update.evolve_feedback)
            eval_data["helpful"] = eval_data.get("helpful", 0) + 1
            need_reembed = True

        # 准备更新数据
        updates = {
            "content": content,
            "eval": eval_data,
        }

        # 如果内容变化，重新生成向量
        if need_reembed:
            embedding = await get_embedding(existing['task'])
            updates["task_embedding"] = embedding

        # 更新 Milvus
        pg_store.update(knowledge_id, updates)

        return {"status": "ok", "knowledge_id": knowledge_id}

    except HTTPException:
        raise
    except Exception as e:
        print(f"[Update Knowledge] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.patch("/api/knowledge/{knowledge_id}")
async def patch_knowledge(knowledge_id: str, patch: KnowledgePatchIn):
    """直接编辑知识字段"""
    try:
        # 获取现有知识
        existing = pg_store.get_by_id(knowledge_id)
        if not existing:
            raise HTTPException(status_code=404, detail=f"Knowledge not found: {knowledge_id}")

        updates = {}
        need_reembed = False
        need_content_reembed = False

        if patch.task is not None:
            updates["task"] = patch.task
            need_reembed = True

        if patch.content is not None:
            updates["content"] = patch.content
            need_content_reembed = True

        if patch.types is not None:
            updates["types"] = patch.types

        if patch.tags is not None:
            updates["tags"] = patch.tags
            # 同时更新 tag_keys
            updates["tag_keys"] = list(patch.tags.keys()) if isinstance(patch.tags, dict) else []

        if patch.scopes is not None:
            updates["scopes"] = patch.scopes

        if patch.owner is not None:
            updates["owner"] = patch.owner

        if patch.capability_ids is not None:
            updates["capability_ids"] = patch.capability_ids

        if patch.tool_ids is not None:
            updates["tool_ids"] = patch.tool_ids

        if not updates:
            return {"status": "ok", "knowledge_id": knowledge_id}

        # 如果 task 变化，重新生成 task_embedding
        if need_reembed:
            task = updates.get("task", existing["task"])
            embedding = await get_embedding(task)
            updates["task_embedding"] = embedding

        # 如果 content 变化，重新生成 content_embedding
        if need_content_reembed:
            content = updates.get("content", existing["content"])
            content_embedding = await get_embedding(content)
            updates["content_embedding"] = content_embedding

        # 更新 Milvus
        pg_store.update(knowledge_id, updates)

        return {"status": "ok", "knowledge_id": knowledge_id}

    except HTTPException:
        raise
    except Exception as e:
        print(f"[Patch Knowledge] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.delete("/api/knowledge/{knowledge_id}")
def delete_knowledge(knowledge_id: str):
    """删除单条知识"""
    try:
        # 检查知识是否存在
        existing = pg_store.get_by_id(knowledge_id)
        if not existing:
            raise HTTPException(status_code=404, detail=f"Knowledge not found: {knowledge_id}")

        # 从 PostgreSQL 删除
        pg_store.delete(knowledge_id)
        print(f"[Delete Knowledge] 已删除知识: {knowledge_id}")

        return {"status": "ok", "knowledge_id": knowledge_id}

    except HTTPException:
        raise
    except Exception as e:
        print(f"[Delete Knowledge] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/knowledge/batch_delete")
def batch_delete_knowledge(knowledge_ids: List[str] = Body(...)):
    """批量删除知识"""
    try:
        if not knowledge_ids:
            raise HTTPException(status_code=400, detail="knowledge_ids cannot be empty")

        deleted_count = 0
        for kid in knowledge_ids:
            pg_store.delete(kid)
            deleted_count += 1
        print(f"[Batch Delete] 已删除 {deleted_count} 条知识")
        return {"status": "ok", "deleted_count": deleted_count}

    except HTTPException:
        raise
    except Exception as e:
        print(f"[Batch Delete] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/knowledge/batch_verify")
async def batch_verify_knowledge(batch: KnowledgeBatchVerifyIn):
    """批量验证通过（approved → checked）"""
    if not batch.knowledge_ids:
        return {"status": "ok", "updated": 0}

    try:
        now_iso = datetime.now(timezone.utc).isoformat()
        updated_count = 0

        for kid in batch.knowledge_ids:
            existing = pg_store.get_by_id(kid)
            if not existing:
                continue

            eval_data = existing.get("eval") or {}
            eval_data["verification"] = {
                "status": "checked",
                "verified_by": batch.verified_by,
                "verified_at": now_iso,
                "note": None,
                "issue_type": None,
                "issue_action": None,
            }
            pg_store.update(kid, {"eval": eval_data, "status": "checked", "updated_at": int(time.time())})
            updated_count += 1

        return {"status": "ok", "updated": updated_count}

    except Exception as e:
        print(f"[Batch Verify] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/knowledge/{knowledge_id}/verify")
async def verify_knowledge(knowledge_id: str, verify: KnowledgeVerifyIn):
    """知识验证：approve 切换 approved↔checked，reject 设为 rejected"""
    try:
        existing = pg_store.get_by_id(knowledge_id)
        if not existing:
            raise HTTPException(status_code=404, detail=f"Knowledge not found: {knowledge_id}")

        current_status = existing.get("status", "approved")

        if verify.action == "approve":
            # checked → approved（取消验证），其他 → checked
            new_status = "approved" if current_status == "checked" else "checked"
            pg_store.update(knowledge_id, {
                "status": new_status,
                "updated_at": int(time.time())
            })
            return {"status": "ok", "new_status": new_status,
                    "message": "已取消验证" if new_status == "approved" else "验证通过"}

        elif verify.action == "reject":
            pg_store.update(knowledge_id, {
                "status": "rejected",
                "updated_at": int(time.time())
            })
            return {"status": "ok", "new_status": "rejected", "message": "已拒绝"}

        else:
            raise HTTPException(status_code=400, detail=f"Unknown action: {verify.action}")

    except HTTPException:
        raise
    except Exception as e:
        print(f"[Verify Knowledge] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/knowledge/batch_update")
async def batch_update_knowledge(batch: KnowledgeBatchUpdateIn):
    """批量反馈知识有效性"""
    if not batch.feedback_list:
        return {"status": "ok", "updated": 0}

    try:
        # 先处理无需进化的，收集需要进化的
        evolution_tasks = []   # [(knowledge_id, old_content, feedback, eval_data)]
        simple_updates = []    # [(knowledge_id, is_effective, eval_data)]

        for item in batch.feedback_list:
            knowledge_id = item.get("knowledge_id")
            is_effective = item.get("is_effective")
            feedback = item.get("feedback", "")

            if not knowledge_id:
                continue

            existing = pg_store.get_by_id(knowledge_id)
            if not existing:
                continue

            eval_data = existing.get("eval", {})

            if is_effective and feedback:
                evolution_tasks.append((knowledge_id, existing["content"], feedback, eval_data, existing["task"]))
            else:
                simple_updates.append((knowledge_id, is_effective, eval_data))

        # 执行简单更新
        for knowledge_id, is_effective, eval_data in simple_updates:
            if is_effective:
                eval_data["helpful"] = eval_data.get("helpful", 0) + 1
            else:
                eval_data["harmful"] = eval_data.get("harmful", 0) + 1

            pg_store.update(knowledge_id, {"eval": eval_data})

        # 并发执行知识进化
        if evolution_tasks:
            print(f"🧬 并发处理 {len(evolution_tasks)} 条知识进化...")
            evolved_results = await asyncio.gather(
                *[_evolve_knowledge_with_llm(old, fb) for _, old, fb, _, _ in evolution_tasks]
            )

            for (knowledge_id, _, _, eval_data, task), evolved_content in zip(evolution_tasks, evolved_results):
                eval_data["helpful"] = eval_data.get("helpful", 0) + 1

                # 重新生成向量（只基于 task）
                embedding = await get_embedding(task)

                pg_store.update(knowledge_id, {
                    "content": evolved_content,
                    "eval": eval_data,
                    "task_embedding": embedding
                })

        return {"status": "ok", "updated": len(simple_updates) + len(evolution_tasks)}

    except Exception as e:
        print(f"[Batch Update] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/knowledge/slim")
async def slim_knowledge(model: str = "google/gemini-2.5-flash-lite"):
    """知识库瘦身：合并语义相似知识"""
    try:
        # 获取所有知识
        all_knowledge = pg_store.query('id != ""', limit=10000)
        # 转换为可序列化的格式
        all_knowledge = [to_serializable(item) for item in all_knowledge]

        if len(all_knowledge) < 2:
            return {"status": "ok", "message": f"知识库仅有 {len(all_knowledge)} 条，无需瘦身"}

        # 构造发给大模型的内容
        entries_text = ""
        for item in all_knowledge:
            eval_data = item.get("eval", {})
            types = item.get("types", [])
            entries_text += f"[ID: {item['id']}] [Types: {','.join(types)}] "
            entries_text += f"[Helpful: {eval_data.get('helpful', 0)}, Harmful: {eval_data.get('harmful', 0)}] [Score: {eval_data.get('score', 3)}]\n"
            entries_text += f"Task: {item['task']}\n"
            entries_text += f"Content: {item['content'][:200]}...\n\n"

        prompt = KNOWLEDGE_SLIM_PROMPT_TEMPLATE.format(entries_text=entries_text)

        print(f"\n[知识瘦身] 正在调用 {model} 分析 {len(all_knowledge)} 条知识...")
        slim_llm = create_openrouter_llm_call(model=model)
        response = await slim_llm(
            messages=[{"role": "user", "content": prompt}],
        )
        content = response.get("content", "").strip()
        if not content:
            raise HTTPException(status_code=500, detail="LLM 返回为空")

        # 解析大模型输出
        report_line = ""
        new_entries = []
        blocks = [b.strip() for b in content.split("===") if b.strip()]

        for block in blocks:
            if block.startswith("REPORT:"):
                report_line = block
                continue

            lines = block.split("\n")
            kid, types, helpful, harmful, score, task, content_lines = None, [], 0, 0, 3, "", []
            current_field = None

            for line in lines:
                if line.startswith("ID:"):
                    kid = line[3:].strip()
                    current_field = None
                elif line.startswith("TYPES:"):
                    types_str = line[6:].strip()
                    types = [t.strip() for t in types_str.split(",") if t.strip()]
                    current_field = None
                elif line.startswith("HELPFUL:"):
                    try:
                        helpful = int(line[8:].strip())
                    except Exception:
                        helpful = 0
                    current_field = None
                elif line.startswith("HARMFUL:"):
                    try:
                        harmful = int(line[8:].strip())
                    except Exception:
                        harmful = 0
                    current_field = None
                elif line.startswith("SCORE:"):
                    try:
                        score = int(line[6:].strip())
                    except Exception:
                        score = 3
                    current_field = None
                elif line.startswith("TASK:"):
                    task = line[5:].strip()
                    current_field = "task"
                elif line.startswith("CONTENT:"):
                    content_lines.append(line[8:].strip())
                    current_field = "content"
                elif current_field == "task":
                    task += "\n" + line
                elif current_field == "content":
                    content_lines.append(line)

            if kid and content_lines:
                new_entries.append({
                    "id": kid,
                    "types": types if types else ["strategy"],
                    "helpful": helpful,
                    "harmful": harmful,
                    "score": score,
                    "task": task.strip(),
                    "content": "\n".join(content_lines).strip()
                })

        if not new_entries:
            raise HTTPException(status_code=500, detail="解析大模型输出失败")

        # 生成向量并重建知识库
        print(f"[知识瘦身] 正在为 {len(new_entries)} 条知识生成向量...")

        # 批量生成向量（只基于 task）
        texts = [e['task'] for e in new_entries]
        embeddings = await get_embeddings_batch(texts)

        # 清空并重建（PostgreSQL使用TRUNCATE）
        cursor = pg_store._get_cursor()
        try:
            # 先清关联表再清主表
            for jt in ('requirement_knowledge', 'capability_knowledge', 'tool_knowledge',
                        'knowledge_resource', 'knowledge_relation'):
                cursor.execute(f"TRUNCATE TABLE {jt}")
            cursor.execute("TRUNCATE TABLE knowledge")
            pg_store.conn.commit()
        finally:
            cursor.close()

        knowledge_list = []
        for e, embedding in zip(new_entries, embeddings):
            eval_data = {
                "score": e["score"],
                "helpful": e["helpful"],
                "harmful": e["harmful"],
                "confidence": 0.9,
                "helpful_history": [],
                "harmful_history": []
            }
            source = {
                "name": "slim",
                "category": "exp",
                "urls": [],
                "agent_id": "slim",
                "submitted_by": "system",
                "timestamp": datetime.now(timezone.utc).isoformat()
            }
            knowledge_list.append({
                "id": e["id"],
                "task_embedding": embedding,
                "message_id": "",
                "task": e["task"],
                "content": e["content"],
                "types": e["types"],
                "tags": {},
                "tag_keys": [],
                "scopes": ["org:cybertogether"],
                "owner": "agent:slim",
                "source": source,
                "eval": eval_data,
                "created_at": now,
                "updated_at": now,
                "status": "approved",
            })

        pg_store.insert_batch(knowledge_list)

        result_msg = f"瘦身完成：{len(all_knowledge)} → {len(new_entries)} 条知识"
        if report_line:
            result_msg += f"\n{report_line}"
        print(f"[知识瘦身] {result_msg}")

        return {"status": "ok", "before": len(all_knowledge), "after": len(new_entries), "report": report_line}

    except HTTPException:
        raise
    except Exception as e:
        print(f"[Slim Knowledge] 错误: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/extract")
async def extract_knowledge_from_messages(extract_req: MessageExtractIn, background_tasks: BackgroundTasks):
    """从消息历史中提取知识（LLM 分析）"""
    if not extract_req.submitted_by:
        raise HTTPException(status_code=400, detail="submitted_by is required")

    messages = extract_req.messages
    if not messages or len(messages) == 0:
        return {"status": "ok", "extracted_count": 0, "knowledge_ids": []}

    # 构造消息历史文本
    messages_text = ""
    for msg in messages:
        role = msg.get("role", "unknown")
        content = msg.get("content", "")
        messages_text += f"[{role}]: {content}\n\n"

    # LLM 提取知识
    prompt = MESSAGE_EXTRACT_PROMPT_TEMPLATE.format(messages_text=messages_text)

    try:
        print(f"\n[Extract] 正在从 {len(messages)} 条消息中提取知识...")

        response = await _dedup_llm(
            messages=[{"role": "user", "content": prompt}],
        )

        content = response.get("content", "").strip()

        # 尝试解析 JSON
        # 移除可能的 markdown 代码块标记
        if content.startswith("```json"):
            content = content[7:]
        if content.startswith("```"):
            content = content[3:]
        if content.endswith("```"):
            content = content[:-3]
        content = content.strip()

        extracted_knowledge = json.loads(content)

        if not isinstance(extracted_knowledge, list):
            raise ValueError("LLM output is not a list")

        if not extracted_knowledge:
            return {"status": "ok", "extracted_count": 0, "knowledge_ids": []}

        # 批量生成向量（只基于 task）
        texts = [item.get('task', '') for item in extracted_knowledge]
        embeddings = await get_embeddings_batch(texts)

        # 保存提取的知识
        knowledge_ids = []
        now = int(time.time())
        knowledge_list = []

        for item, embedding in zip(extracted_knowledge, embeddings):
            task = item.get("task", "")
            knowledge_content = item.get("content", "")
            types = item.get("types", ["strategy"])
            score = item.get("score", 3)

            if not task or not knowledge_content:
                continue

            # 生成 ID
            timestamp = datetime.now().strftime('%Y%m%d-%H%M%S')
            random_suffix = uuid.uuid4().hex[:4]
            knowledge_id = f"knowledge-{timestamp}-{random_suffix}"

            # 准备数据
            source = {
                "name": "message_extraction",
                "category": "exp",
                "urls": [],
                "agent_id": extract_req.agent_id,
                "submitted_by": extract_req.submitted_by,
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "session_key": extract_req.session_key
            }

            eval_data = {
                "score": score,
                "helpful": 1,
                "harmful": 0,
                "confidence": 0.7,
                "helpful_history": [],
                "harmful_history": []
            }

            knowledge_list.append({
                "id": knowledge_id,
                "task_embedding": embedding,
                "message_id": "",
                "task": task,
                "content": knowledge_content,
                "types": types,
                "tags": {},
                "tag_keys": [],
                "scopes": ["org:cybertogether"],
                "owner": extract_req.submitted_by,
                "source": source,
                "eval": eval_data,
                "created_at": now,
                "updated_at": now,
                "status": "pending",
            })
            knowledge_ids.append(knowledge_id)

        # 批量插入
        if knowledge_list:
            pg_store.insert_batch(knowledge_list)
            background_tasks.add_task(knowledge_processor.process_pending)

        print(f"[Extract] 成功提取并保存 {len(knowledge_ids)} 条知识")

        return {
            "status": "ok",
            "extracted_count": len(knowledge_ids),
            "knowledge_ids": knowledge_ids
        }

    except json.JSONDecodeError as e:
        print(f"[Extract] JSON 解析失败: {e}")
        print(f"[Extract] LLM 输出: {content[:500]}")
        return {"status": "error", "error": "Failed to parse LLM output", "extracted_count": 0}
    except Exception as e:
        print(f"[Extract] 提取失败: {e}")
        return {"status": "error", "error": str(e), "extracted_count": 0}


# ===== Tool API =====

@app.post("/api/tool", status_code=201)
async def create_tool(tool: ToolIn):
    """创建或更新工具"""
    try:
        now = int(time.time())
        embedding = await get_embedding(f"{tool.name} {tool.introduction}")
        pg_tool_store.insert_or_update({
            'id': tool.id,
            'name': tool.name,
            'version': tool.version,
            'introduction': tool.introduction,
            'tutorial': tool.tutorial,
            'input': tool.input,
            'output': tool.output,
            'updated_time': now,
            'status': tool.status,
            'capability_ids': tool.capability_ids,
            'knowledge_ids': tool.knowledge_ids,
            'provider_ids': tool.provider_ids,
            'embedding': embedding,
        })
        return {"status": "ok", "id": tool.id}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/tool")
def list_tools(
    status: Optional[str] = Query(None),
    limit: int = Query(100, ge=1, le=1000),
    offset: int = Query(0, ge=0),
):
    """列出工具"""
    try:
        results = pg_tool_store.list_all(limit=limit, offset=offset, status=status)
        total = pg_tool_store.count(status=status)
        return {"results": results, "total": total}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/tool/search")
async def search_tools(
    q: str = Query(..., description="查询文本"),
    top_k: int = Query(5, ge=1, le=100),
    status: Optional[str] = None,
):
    """向量检索工具"""
    try:
        query_embedding = await get_embedding(q)
        results = pg_tool_store.search(query_embedding, limit=top_k, status=status)
        return {"results": results, "count": len(results)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/tool/{tool_id:path}")
def get_tool(tool_id: str):
    """获取单个工具详情"""
    try:
        result = pg_tool_store.get_by_id(tool_id)
        if not result:
            raise HTTPException(status_code=404, detail=f"Tool not found: {tool_id}")
        return result
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.patch("/api/tool/{tool_id:path}")
async def patch_tool(tool_id: str, patch: ToolPatchIn):
    """更新工具字段"""
    try:
        if not pg_tool_store.get_by_id(tool_id):
            raise HTTPException(status_code=404, detail=f"Tool not found: {tool_id}")

        updates = {}
        need_reembed = False

        for field in ('name', 'version', 'introduction', 'tutorial', 'input', 'output',
                       'status', 'capability_ids', 'knowledge_ids', 'provider_ids'):
            value = getattr(patch, field)
            if value is not None:
                updates[field] = value
                if field in ('name', 'introduction'):
                    need_reembed = True

        if not updates:
            return {"status": "ok", "id": tool_id}

        updates['updated_time'] = int(time.time())

        if need_reembed:
            existing = pg_tool_store.get_by_id(tool_id)
            name = updates.get('name', existing['name'])
            intro = updates.get('introduction', existing['introduction'])
            updates['embedding'] = await get_embedding(f"{name} {intro}")

        pg_tool_store.update(tool_id, updates)
        return {"status": "ok", "id": tool_id}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.delete("/api/tool/{tool_id:path}")
def delete_tool(tool_id: str):
    """删除工具"""
    try:
        if not pg_tool_store.get_by_id(tool_id):
            raise HTTPException(status_code=404, detail=f"Tool not found: {tool_id}")
        pg_tool_store.delete(tool_id)
        return {"status": "ok", "id": tool_id}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# ===== Capability API =====

@app.post("/api/capability", status_code=201)
async def create_capability(cap: CapabilityIn):
    """创建或更新原子能力"""
    try:
        embedding = await get_embedding(f"{cap.name} {cap.description}")
        pg_capability_store.insert_or_update({
            'id': cap.id,
            'name': cap.name,
            'criterion': cap.criterion,
            'description': cap.description,
            'requirement_ids': cap.requirement_ids,
            'implements': cap.implements,
            'tool_ids': cap.tool_ids,
            'knowledge_ids': cap.knowledge_ids,
            'embedding': embedding,
        })
        return {"status": "ok", "id": cap.id}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/capability")
def list_capabilities(
    limit: int = Query(100, ge=1, le=1000),
    offset: int = Query(0, ge=0),
):
    """列出原子能力"""
    try:
        results = pg_capability_store.list_all(limit=limit, offset=offset)
        total = pg_capability_store.count()
        return {"results": results, "total": total}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/capability/search")
async def search_capabilities(
    q: str = Query(..., description="查询文本"),
    top_k: int = Query(5, ge=1, le=100),
):
    """向量检索原子能力"""
    try:
        query_embedding = await get_embedding(q)
        results = pg_capability_store.search(query_embedding, limit=top_k)
        return {"results": results, "count": len(results)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/capability/{cap_id}")
def get_capability(cap_id: str):
    """获取单个原子能力"""
    try:
        result = pg_capability_store.get_by_id(cap_id)
        if not result:
            raise HTTPException(status_code=404, detail=f"Capability not found: {cap_id}")
        return result
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.patch("/api/capability/{cap_id}")
async def patch_capability(cap_id: str, patch: CapabilityPatchIn):
    """更新原子能力字段"""
    try:
        existing = pg_capability_store.get_by_id(cap_id)
        if not existing:
            raise HTTPException(status_code=404, detail=f"Capability not found: {cap_id}")

        updates = {}
        need_reembed = False

        for field in ('name', 'criterion', 'description', 'requirement_ids',
                       'implements', 'tool_ids', 'knowledge_ids'):
            value = getattr(patch, field)
            if value is not None:
                updates[field] = value
                if field in ('name', 'description'):
                    need_reembed = True

        if not updates:
            return {"status": "ok", "id": cap_id}

        if need_reembed:
            name = updates.get('name', existing['name'])
            desc = updates.get('description', existing['description'])
            updates['embedding'] = await get_embedding(f"{name} {desc}")

        pg_capability_store.update(cap_id, updates)
        return {"status": "ok", "id": cap_id}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.delete("/api/capability/{cap_id}")
def delete_capability(cap_id: str):
    """删除原子能力"""
    try:
        if not pg_capability_store.get_by_id(cap_id):
            raise HTTPException(status_code=404, detail=f"Capability not found: {cap_id}")
        pg_capability_store.delete(cap_id)
        return {"status": "ok", "id": cap_id}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# ===== Requirement API =====

@app.post("/api/requirement", status_code=201)
async def create_requirement(req: RequirementIn):
    """创建或更新需求"""
    try:
        embedding = await get_embedding(req.description)
        pg_requirement_store.insert_or_update({
            'id': req.id,
            'description': req.description,
            'capability_ids': req.capability_ids,
            'knowledge_ids': req.knowledge_ids,
            'source_nodes': req.source_nodes,
            'status': req.status,
            'match_result': req.match_result,
            'embedding': embedding,
        })
        return {"status": "ok", "id": req.id}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/requirement")
def list_requirements(
    status: Optional[str] = Query(None),
    limit: int = Query(100, ge=1, le=1000),
    offset: int = Query(0, ge=0),
):
    """列出需求"""
    try:
        results = pg_requirement_store.list_all(limit=limit, offset=offset, status=status)
        total = pg_requirement_store.count(status=status)
        return {"results": results, "total": total}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/requirement/search")
async def search_requirements(
    q: str = Query(..., description="查询文本"),
    top_k: int = Query(5, ge=1, le=100),
):
    """向量检索需求"""
    try:
        query_embedding = await get_embedding(q)
        results = pg_requirement_store.search(query_embedding, limit=top_k)
        return {"results": results, "count": len(results)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/api/requirement/{req_id}")
def get_requirement(req_id: str):
    """获取单个需求"""
    try:
        result = pg_requirement_store.get_by_id(req_id)
        if not result:
            raise HTTPException(status_code=404, detail=f"Requirement not found: {req_id}")
        return result
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.patch("/api/requirement/{req_id}")
async def patch_requirement(req_id: str, patch: RequirementPatchIn):
    """更新需求字段"""
    try:
        existing = pg_requirement_store.get_by_id(req_id)
        if not existing:
            raise HTTPException(status_code=404, detail=f"Requirement not found: {req_id}")

        updates = {}
        need_reembed = False

        for field in ('description', 'capability_ids', 'knowledge_ids', 'source_nodes', 'status', 'match_result'):
            value = getattr(patch, field)
            if value is not None:
                updates[field] = value
                if field == 'description':
                    need_reembed = True

        if not updates:
            return {"status": "ok", "id": req_id}

        if need_reembed:
            updates['embedding'] = await get_embedding(updates['description'])

        pg_requirement_store.update(req_id, updates)
        return {"status": "ok", "id": req_id}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.delete("/api/requirement/{req_id}")
def delete_requirement(req_id: str):
    """删除需求"""
    try:
        if not pg_requirement_store.get_by_id(req_id):
            raise HTTPException(status_code=404, detail=f"Requirement not found: {req_id}")
        pg_requirement_store.delete(req_id)
        return {"status": "ok", "id": req_id}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/pattern/posts/batch")
async def proxy_pattern_posts_batch(payload: PostBatchRequest):
    """代理帖子批量查询，避免前端直接请求外部域名失败后静默回退为纯 ID。"""
    post_ids = [pid for pid in payload.post_ids if pid]
    if not post_ids:
        return {"success": True, "posts": []}

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.post(
                "https://pattern.aiddit.com/api/pattern/posts/batch",
                json={"post_ids": post_ids},
            )
        resp.raise_for_status()
        return resp.json()
    except httpx.HTTPStatusError as e:
        raise HTTPException(status_code=e.response.status_code, detail="Pattern posts API returned an error")
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Failed to fetch pattern posts: {e}")

@app.get("/api/pattern/itemsets")
async def proxy_pattern_itemsets(execution_id: int):
    """代理获取 itemsets，避免前端直接请求外部域名"""
    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.get(
                f"https://pattern.aiddit.com/api/pattern/itemsets?execution_id={execution_id}&page_size=1000",
            )
        resp.raise_for_status()
        return resp.json()
    except httpx.HTTPStatusError as e:
        raise HTTPException(status_code=e.response.status_code, detail="Pattern itemsets API returned an error")
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Failed to fetch pattern itemsets: {e}")

@app.get("/")
def frontend():
    """KnowHub 管理前端"""
    index_file = STATIC_DIR / "index.html"
    if not index_file.exists():
        return HTMLResponse("<h1>KnowHub Frontend Not Found</h1><p>Please ensure knowhub/frontend/dist/index.html exists. Run 'yarn build' in frontend directory.</p>", status_code=404)
    return FileResponse(str(index_file))


# ===== Strategy API =====

@app.post("/api/strategy", status_code=201)
async def submit_strategy(strategy: StrategyIn):
    """创建或更新策略（自动填时间戳 + name/description 向量）"""
    try:
        now = int(time.time())
        data = strategy.model_dump()
        data['created_at'] = data.get('created_at') or now
        data['updated_at'] = now
        data['embedding'] = await get_embedding(f"{strategy.name} {strategy.description}")
        pg_strategy_store.insert_or_update(data)
        return {"success": True, "id": strategy.id}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/strategy")
def get_strategies(limit: int = 100, offset: int = 0, status: Optional[str] = None):
    try:
        results = pg_strategy_store.list_all(limit=limit, offset=offset, status=status)
        total = pg_strategy_store.count(status=status)
        return {"strategies": results, "total": total}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/strategy/search")
async def search_strategies(q: str = Query(...), top_k: int = 20, status: Optional[str] = None):
    try:
        query_embedding = await get_embedding(q)
        results = pg_strategy_store.search(query_embedding, limit=top_k, status=status)
        return {"results": results, "count": len(results)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/strategy/{strategy_id:path}")
def get_strategy(strategy_id: str):
    try:
        result = pg_strategy_store.get_by_id(strategy_id)
        if not result:
            raise HTTPException(status_code=404, detail="Strategy not found")
        return result
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.patch("/api/strategy/{strategy_id:path}")
async def patch_strategy(strategy_id: str, updates: StrategyPatchIn):
    """更新策略字段。若改了 name/description，会重算向量。"""
    try:
        existing = pg_strategy_store.get_by_id(strategy_id)
        if not existing:
            raise HTTPException(status_code=404, detail="Strategy not found")

        update_dict = updates.model_dump(exclude_unset=True)
        if not update_dict:
            return {"success": True}

        if 'name' in update_dict or 'description' in update_dict:
            name = update_dict.get('name', existing.get('name', ''))
            desc = update_dict.get('description', existing.get('description', ''))
            update_dict['embedding'] = await get_embedding(f"{name} {desc}")

        update_dict['updated_at'] = int(time.time())
        pg_strategy_store.update(strategy_id, update_dict)
        return {"success": True}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.delete("/api/strategy/{strategy_id:path}")
def delete_strategy(strategy_id: str):
    try:
        if not pg_strategy_store.get_by_id(strategy_id):
            raise HTTPException(status_code=404, detail="Strategy not found")
        pg_strategy_store.delete(strategy_id)
        return {"success": True}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# ===== Relation API =====

@app.get("/api/relation/{table_name}")
async def get_relations(table_name: str, request: Request):
    """通用关系表查询接口"""
    allowed_tables = {
        "capability_knowledge",
        "capability_tool",
        "knowledge_relation",
        "knowledge_resource",
        "requirement_capability",
        "requirement_knowledge",
        "tool_knowledge",
        "tool_provider"
    }
    table_name = table_name.lower()
    if table_name not in allowed_tables:
        raise HTTPException(status_code=400, detail="Invalid table name")
        
    try:
        params = dict(request.query_params)
        
        where_clauses = []
        values = []
        for k, v in params.items():
            if k in ["limit", "offset"]: continue
            where_clauses.append(f"{k} = %s")
            values.append(v)
        
        query = f"SELECT * FROM {table_name}"
        if where_clauses:
            query += " WHERE " + " AND ".join(where_clauses)
            
        limit = int(params.get("limit", 100))
        query += " LIMIT %s"
        values.append(limit)
        
        cursor = pg_store._get_cursor()
        try:
            cursor.execute(query, tuple(values))
            rows = cursor.fetchall()
            if not rows:
                 return {"results": [], "count": 0}
            colnames = [desc[0] for desc in cursor.description]
            results = [dict(zip(colnames, row)) for row in rows]
            return {"results": results, "count": len(results)}
        finally:
            cursor.close()
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# --- Dashboard Snapshot (缓存聚合接口) ---

_dashboard_snapshot_cache: Dict[str, dict] = {}   # 按 version 分桶
_dashboard_snapshot_ts: Dict[str, float] = {}
_DASHBOARD_CACHE_TTL = 24 * 3600  # 24 小时


def _invalidate_dashboard_cache():
    """数据写入后调用，清除 dashboard 快照缓存（所有版本）"""
    _dashboard_snapshot_cache.clear()
    _dashboard_snapshot_ts.clear()


def _build_dashboard_snapshot() -> dict:
    """在后端一次性构建 Dashboard 所需的全部数据"""
    # 兼容三种部署形态：production build / dev (vite public) / 仓库 static
    _PROJECT_ROOT = Path(__file__).parent
    tree_candidates = [
        STATIC_DIR / "category_tree.json",
        _PROJECT_ROOT / "frontend" / "public" / "category_tree.json",
        _PROJECT_ROOT / "static" / "category_tree.json",
    ]
    tree_data = None
    for tree_file in tree_candidates:
        if tree_file.exists():
            tree_data = json.loads(tree_file.read_text(encoding="utf-8"))
            break
    if tree_data is None:
        raise HTTPException(
            status_code=500,
            detail=f"category_tree.json not found in any of: {[str(p) for p in tree_candidates]}",
        )

    reqs = pg_requirement_store.list_all(limit=1000, offset=0)
    caps = pg_capability_store.list_all(limit=1000, offset=0)
    tools = pg_tool_store.list_all(limit=1000, offset=0)
    procs = pg_strategy_store.list_all(limit=1000, offset=0)
    know_raw = pg_store.query('(status == "approved" or status == "checked")', limit=1000)
    know = [to_serializable(r) for r in know_raw]

    return {
        "tree": tree_data,
        "reqs": reqs,
        "caps": caps,
        "tools": tools,
        "procs": procs,
        "know": know,
        "built_at": time.time(),
    }


@app.get("/api/dashboard/snapshot")
def get_dashboard_snapshot():
    """返回 Dashboard 所需的全部数据快照，带服务端内存缓存（24h TTL，写入时失效）。按 version 分桶。"""
    from knowhub.knowhub_db.version_context import get_version
    v = get_version()
    now = time.time()
    cached = _dashboard_snapshot_cache.get(v)
    ts = _dashboard_snapshot_ts.get(v, 0)
    if cached and (now - ts < _DASHBOARD_CACHE_TTL):
        return cached
    try:
        snap = _build_dashboard_snapshot()
        snap['version'] = v
        _dashboard_snapshot_cache[v] = snap
        _dashboard_snapshot_ts[v] = now
        return snap
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/api/dashboard/invalidate")
def invalidate_dashboard_cache():
    """手动清除 dashboard 缓存"""
    _invalidate_dashboard_cache()
    return {"status": "ok"}


@app.get("/category_tree.json")
def serve_category_tree():
    """类别树JSON数据"""
    tree_file = STATIC_DIR / "category_tree.json"
    if not tree_file.exists():
        return {"error": "Not Found"}
    return FileResponse(str(tree_file))


@app.get("/{frontend_path:path}")
def frontend_spa_fallback(frontend_path: str):
    """SPA 路由兜底：将非 API 的前端子路径回退到 index.html，由 React Router 处理。"""
    if frontend_path.startswith("api/") or frontend_path.startswith("assets/"):
        raise HTTPException(status_code=404, detail="Not Found")

    # 带扩展名的路径尝试作为静态文件直接返回，不存在则抛 404
    if "." in Path(frontend_path).name:
        full_path = STATIC_DIR / frontend_path
        if full_path.exists() and full_path.is_file():
            return FileResponse(str(full_path))
        raise HTTPException(status_code=404, detail="Not Found")

    index_file = STATIC_DIR / "index.html"
    if not index_file.exists():
        return HTMLResponse("<h1>KnowHub Frontend Not Found</h1><p>Please ensure knowhub/frontend/dist/index.html exists. Run 'yarn build' in frontend directory.</p>", status_code=404)
    return FileResponse(str(index_file))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=9999)