projectAIpopular/gateway/api.py

"""FastAPI 网关：对外提供 v2（端云协同）与 v1（legacy 路由）两套接口。

v2 端点（《实现方案_v2》5.3）：
  POST /chat                        端云协同协作管线
  POST /chat/legacy                 原 v1 L0 行为（离线降级，测试封闭）
  GET  /runs/{request_id}/workspace 交流文本
  GET  /runs/{request_id}/artifacts/{name}  工件下载
  GET  /review/queue、POST /review/{id}     人工检验
  GET  /metrics                     含 v2 token/快路径统计

启动：uvicorn gateway.api:app --host 127.0.0.1 --port 8000
（默认仅回环；如需局域网访问改 --host 0.0.0.0 并设置 GATEWAY_TRUSTED_HOSTS）
"""
from __future__ import annotations

import os as _os
from pathlib import Path as _Path

# 加载项目根目录的 .env 文件（包含 DEEPSEEK_API_KEY 等密钥）
_dotenv_path = _Path(__file__).resolve().parent.parent / ".env"
if _dotenv_path.exists():
    from dotenv import load_dotenv
    load_dotenv(_dotenv_path)

import asyncio
import re
from pathlib import Path
from typing import List, Optional

from pydantic import BaseModel, Field

from router_system.config import load_config
from router_system.router import Router, build_router

from gateway.jobs import get_job_store, JobStore
from gateway.model_pool import (
    compute_cost,
    entry_to_architect_cfg,
    entry_to_worker_cfg,
    get_pool,
)

# 路径参数 ID 白名单（runs/agent/sessions 的 ID 都由此系统生成；
# 拒绝任意其他字符可一并杀灭 Windows 反斜杠穿越 ../..%5C 等变体）
_ID_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")

# ---- v2 依赖（惰性导入，缺依赖时降级提示） ----
try:
    from router_system.architect import build_architect
    from router_system.knowledge import KnowledgeBase
    from router_system.pipeline import CollaborativePipeline, build_pipeline
    from router_system.review import ReviewQueue
    from router_system.v2stats import V2Stats
    from router_system.worker import build_worker
    _V2_OK = True
except Exception:  # pragma: no cover
    _V2_OK = False

# ---- v1 全局单例 ----
_router: Optional[Router] = None

# ---- v2 全局单例 ----
_pipeline: Optional["CollaborativePipeline"] = None
_v2stats = V2Stats() if _V2_OK else None
_review = None
_settings = None


def get_router() -> Router:
    global _router
    if _router is None:
        _router = build_router()
    return _router


def get_review() -> "ReviewQueue":
    global _review
    if _review is None and _V2_OK:
        cfg = load_config().get("review", {})
        _review = ReviewQueue(db_path=cfg.get("queue_db", "data/review.sqlite3"))
    return _review


def settings_store():
    """用户可调整设置（懒加载单例）。"""
    global _settings
    if _settings is None:
        from gateway.settings import load_settings
        _settings = load_settings()
    return _settings


def rebuild_pipeline() -> None:
    """清除管线单例，下次调用重建（配置改动后生效）。"""
    global _pipeline
    _pipeline = None


def build_v2_pipeline(worker_cfg_override: Optional[dict] = None):
    """从配置 + 用户设置构建 v2 协作管线（architect + worker + pipeline）。

    worker_cfg_override 可注入（测试/演示用 mock）。无 API key 时 /chat 会走
    本地降级路径（不崩溃）。
    """
    global _pipeline
    if _pipeline is None:
        cfg = load_config()
        s = settings_store().to_dict() if _V2_OK else {}
        kb = KnowledgeBase()

        # architect（合并用户设置；模型池指派优先——多价位模型，D1）
        acfg = dict(cfg.get("architect", {}))
        acfg.update(s.get("architect", {}))
        pool = get_pool()
        pe = pool.resolve("architect")
        if pe is not None:
            acfg.update(entry_to_architect_cfg(pe))
        architect = build_architect(acfg)

        # worker（合并用户设置；backend 可 mock/openai/llama_server；
        # 模型池指派优先，测试注入 override 最后生效）
        wcfg = dict(cfg.get("worker", {}))
        wcfg.update(s.get("worker", {}))
        pw = pool.resolve("worker")
        if pw is not None:
            wcfg.update(entry_to_worker_cfg(pw))
        if worker_cfg_override:
            wcfg.update(worker_cfg_override)
        worker = build_worker(wcfg, kb=kb)

        # pipeline（合并用户设置）
        cfg2 = dict(cfg)
        pcfg = dict(cfg.get("pipeline", {}))
        pcfg.update(s.get("pipeline", {}))
        cfg2["pipeline"] = pcfg
        _pipeline = build_pipeline(cfg2, architect, worker)
    return _pipeline


def set_pipeline(pipe) -> None:
    """测试注入替身管线。"""
    global _pipeline
    _pipeline = pipe


def get_pipeline():
    global _pipeline
    if _pipeline is None:
        return build_v2_pipeline()
    return _pipeline


# ---- 请求/响应模型 ----
class QueryRequest(BaseModel):
    query: str = Field(..., min_length=1, max_length=8000, description="用户查询")
    domain_group: Optional[str] = Field(
        None, description="大领域组（两级路由第一级）：tech | professional | lifestyle | general；不指定则自动检测"
    )
    mode: str = Field(
        "fast",
        description="执行模式：fast（快路径，小模型直答优先）| full（完整协作，DeepSeek 架构师全程参与）"
    )


class HealthResponse(BaseModel):
    status: str
    domains: List[str]
    classifier: str
    judge: str
    fallback: str


# ---- FastAPI 应用 ----
try:
    from fastapi import FastAPI, HTTPException
    from fastapi.staticfiles import StaticFiles
    from fastapi.responses import HTMLResponse

    _INDEX_PATH = Path(__file__).resolve().parent / "static" / "index.html"
    _STATIC_DIR = Path(__file__).resolve().parent / "static"

    app = FastAPI(
        title="端云协同 LLM 协作系统",
        description="大模型(Architect) + 本地小模型(Worker) 通过交流文本协作；v1 保留为 legacy 路由",
        version="2.0.0",
    )

    # Web 信任围栏（dsh browser-auth 同款思路）：只信任本机/显式放行的 Host，
    # 防 DNS rebinding 把浏览器请求打到本网关。GATEWAY_TRUSTED_HOSTS 可覆盖（"*" = 放行全部）。
    _trusted = _os.environ.get(
        "GATEWAY_TRUSTED_HOSTS",
        "localhost,127.0.0.1,0.0.0.0,[::1],testserver,testclient",
    )
    try:
        from starlette.middleware.trustedhost import TrustedHostMiddleware
        app.add_middleware(TrustedHostMiddleware,
                           allowed_hosts=[h.strip() for h in _trusted.split(",")])
    except ImportError:  # pragma: no cover
        pass

    # Vue SPA 静态资源（html=True：对不存在的路径 fallback 到 index.html，支持 SPA 路由）
    app.mount("/static", StaticFiles(directory=str(_STATIC_DIR), html=True), name="static")

    @app.get("/", response_class=HTMLResponse, tags=["ui"])
    async def index():
        """Vue SPA 的 index.html（FastAPI API 路由优先，此处仅作 fallback）。"""
        if _INDEX_PATH.exists():
            return HTMLResponse(_INDEX_PATH.read_text(encoding="utf-8"))
        return HTMLResponse("<h1>端云协同 LLM 系统</h1><p>请先构建前端：cd webapp && npm run build</p>")

    def _check_id(value: str, what: str = "ID") -> str:
        """校验路径参数 ID（防目录穿越/注入：仅允许系统生成的字符集）。

        非法格式一律按 404 处理——此类 ID 在本系统里不可能存在，
        不泄露校验规则本身。
        """
        if not _ID_RE.fullmatch(value or ""):
            raise HTTPException(status_code=404, detail=f"未找到 {what}: {value!r}")
        return value

    @app.get("/health", response_model=HealthResponse, tags=["system"])
    async def health():
        return get_router().health()

    # Vue SPA（由 StaticFiles mount 在 / 路径提供）
    @app.post("/chat", tags=["chat"])
    async def chat(req: QueryRequest):
        """立即返回 request_id，协作管线在后台 asyncio.Task 中运行。

        完成后结果写入 runs/{id}/workspace.json（由 pipeline 内部完成），
        状态通过 GET /runs/{id}/status 查询，SSE 通过 /runs/{id}/stream 订阅。
        """
        import uuid

        request_id = uuid.uuid4().hex[:12]
        run_dir = Path("runs") / request_id
        run_dir.mkdir(parents=True, exist_ok=True)

        # 注册任务（容量满时拒绝）
        ok, msg = get_job_store().register(request_id)
        if not ok:
            raise HTTPException(status_code=503, detail=msg)

        # 提交后台协程
        async def _run():
            job_store = get_job_store()
            try:
                pipeline = get_pipeline()
                # mode=full 时临时跳过快路径，直接走架构师协作（DeepSeek V4）
                if req.mode == "full":
                    old_fast = pipeline.fast_path
                    pipeline.fast_path = False
                    try:
                        result = await pipeline.run(req.query, request_id=request_id)
                    finally:
                        pipeline.fast_path = old_fast
                else:
                    result = await pipeline.run(req.query, request_id=request_id)
                if _v2stats is not None:
                    # 按模型池单价分账（条目未命中时 cost_est 保持管线原值）
                    entry = get_pool().find_by_model(result.model_used or "")
                    if entry is not None:
                        result.cost_est = compute_cost(
                            entry, result.api_input_tokens, result.api_output_tokens)
                    _v2stats.record(result)
                _maybe_enqueue(result)
                job_store._task_done(request_id, result)
            except Exception as exc:
                import traceback
                traceback.print_exc()
                job_store._task_failed(request_id, "pipeline_error", detail=str(exc))
                # 写一份失败状态到 workspace.json（pipeline 异常时）
                try:
                    import json
                    fail_ws = {
                        "request_id": request_id,
                        "query": req.query,
                        "status": "failed",
                        "error": str(exc),
                        "brief": None, "plan": [], "progress": [],
                        "issues": [], "decisions": [], "archive": [],
                        "meta": {"state": "failed", "round": 0, "rounds_cap": 6,
                                 "api_input_tokens": 0, "api_output_tokens": 0},
                    }
                    (run_dir / "workspace.json").write_text(
                        json.dumps(fail_ws, ensure_ascii=False, indent=2), encoding="utf-8")
                except Exception:
                    pass

        get_job_store().submit(request_id, _run())
        return {"request_id": request_id, "status": "pending"}

    # ---------------- v2：任务状态 ----------------
    @app.get("/runs/{request_id}/status", tags=["v2"])
    async def get_run_status(request_id: str):
        """查询任务当前状态（pending / running / done / failed）。"""
        _check_id(request_id, "request_id")
        info = get_job_store().get(request_id)
        if info is None:
            raise HTTPException(status_code=404, detail=f"任务 {request_id} 不存在")
        ws_state = None
        ws_path = Path("runs") / request_id / "workspace.json"
        if ws_path.exists():
            import json
            try:
                ws_data = json.loads(ws_path.read_text(encoding="utf-8"))
                ws_state = ws_data.get("meta", {}).get("state") or ws_data.get("status")
            except Exception:
                ws_state = "done"
        return {
            "request_id": request_id,
            "status": info.state,       # pending | running | done | failed
            "ws_state": ws_state,
            "started_at": info.started_at,
            "finished_at": info.finished_at,
            "error": info.error,
            # PipelineResult 等效字段（扁平存储在 TaskInfo 中）
            "response": info.response,
            "pipeline_status": info.status,
            "fast_path": info.fast_path,
            "rounds_used": info.rounds_used,
            "api_input_tokens": info.api_input_tokens,
            "api_output_tokens": info.api_output_tokens,
            "cost_est": info.cost_est,
            "model_used": info.model_used,
            "latency_ms": round(info.latency_ms, 2) if info.latency_ms else None,
            "route": info.route,
            "workspace_path": info.workspace_path,
        }

    # ---------------- v2：SSE 实时流 ----------------
    @app.get("/runs/{request_id}/stream", tags=["v2"])
    async def stream_run(request_id: str):
        """SSE 端点：实时推送 workspace.json 状态变化（供前端协作可视化）。"""
        _check_id(request_id, "request_id")
        from fastapi.responses import StreamingResponse

        async def event_generator():
            import json  # noqa: F401
            cancel_event = get_job_store().new_cancel(request_id)
            try:
                async for ev in get_job_store().watch_workspace(request_id, cancel_event):
                    payload = json.dumps(ev, ensure_ascii=False)
                    yield f"data: {payload}\n\n"
            finally:
                cancel_event.set()

        return StreamingResponse(
            event_generator(),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",
            },
        )

    # ---------------- v1：/chat/legacy ----------------
    @app.post("/chat/legacy", tags=["chat"])
    async def chat_legacy(req: QueryRequest):
        """原 v1 L0 专家系统行为（离线可跑，测试封闭）。"""
        try:
            result = await get_router().route(req.query, domain_group=req.domain_group)
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))
        return {
            "response": result.response,
            "domain": result.domain,
            "difficulty": result.difficulty,
            "confidence": round(result.confidence, 4),
            "upgraded": result.upgraded,
            "quality_score": round(result.quality_score, 4),
            "model_used": result.model_used,
            "route": result.route,
            "latency_ms": round(result.latency_ms, 2),
            "cache_hit": result.cache_hit,
            "cache_level": result.cache_level,
            "cost_est": round(result.cost_est, 6),
            "error": result.error,
            "subdomain": result.subdomain,
            "subdomain2": result.subdomain2,
            "domain_group": result.domain_group,
            "request_id": result.request_id,
        }

    @app.get("/traces/{request_id}", tags=["system"])
    async def get_trace(request_id: str):
        _check_id(request_id, "request_id")
        trace = get_router().trace_store.get(request_id)
        if trace is None:
            raise HTTPException(status_code=404, detail=f"未找到请求 {request_id} 的推理链")
        return trace

    # ---------------- v2：workspace / artifacts ----------------
    @app.get("/runs/{request_id}/workspace", tags=["v2"])
    async def get_workspace(request_id: str):
        _check_id(request_id, "request_id")
        p = Path("runs") / request_id / "workspace.json"
        if not p.exists():
            raise HTTPException(status_code=404, detail=f"未找到运行 {request_id}")
        import json
        return json.loads(p.read_text(encoding="utf-8"))

    @app.get("/runs/{request_id}/artifacts/{name}", tags=["v2"])
    async def get_artifact(request_id: str, name: str):
        _check_id(request_id, "request_id")
        d = (Path("runs") / request_id / "artifacts").resolve()
        # 工件名关押：解析后必须仍在 artifacts 目录内（防 ..\ 与绝对路径逃逸）
        p = (d / name).resolve()
        if p != d and d not in p.parents:
            raise HTTPException(status_code=400, detail=f"非法工件名: {name!r}")
        if not p.exists():
            raise HTTPException(status_code=404, detail=f"未找到工件 {name}")
        return {"name": name, "content": p.read_text(encoding="utf-8")}

    # ---------------- v2：人工检验 ----------------
    @app.get("/review/queue", tags=["v2"])
    async def review_queue(status: Optional[str] = None, limit: int = 50):
        return get_review().list(status=status, limit=limit)

    @app.post("/review/{review_id}", tags=["v2"])
    async def review_submit(review_id: int, verdict: str, correction: Optional[str] = None):
        try:
            ok = get_review().submit(review_id, verdict, correction=correction)
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))
        if not ok:
            raise HTTPException(status_code=404, detail=f"审核记录不存在或已审核: {review_id}")
        return {"ok": True, "review_id": review_id, "verdict": verdict}

    # ---------------- 模型池（多价位异构模型） ----------------
    @app.get("/pool", tags=["pool"])
    async def pool_list():
        """读取模型池（角色指派 + 条目列表，api_key 打码）。"""
        return get_pool().list()

    @app.post("/pool", tags=["pool"])
    async def pool_upsert(entry: dict):
        """新增或更新池条目（按 id）。api_key 留空表示保留原值。"""
        try:
            get_pool().upsert(entry)
        except ValueError as e:
            raise HTTPException(status_code=400, detail=f"条目非法: {e}")
        rebuild_pipeline()
        return get_pool().list()

    @app.delete("/pool/{entry_id}", tags=["pool"])
    async def pool_delete(entry_id: str):
        ok = get_pool().delete(entry_id)
        if ok:
            rebuild_pipeline()
        return {"ok": ok, **get_pool().list()}

    @app.put("/pool/roles", tags=["pool"])
    async def pool_roles(roles: dict):
        """指派角色：{"architect": "<id|>", "worker": "<id|>", "agent": "<id|>"}。
        空串 = 沿用经典单模型设置。"""
        try:
            get_pool().set_roles(roles)
        except ValueError as e:
            raise HTTPException(status_code=400, detail=f"指派非法: {e}")
        rebuild_pipeline()
        return get_pool().list()

    @app.post("/pool/{entry_id}/test", tags=["pool"])
    async def pool_test(entry_id: str):
        """连通性测试：用条目自身的端点/凭据探测 /models。"""
        entry = get_pool().get(entry_id)
        if entry is None:
            raise HTTPException(status_code=404, detail=f"条目不存在: {entry_id}")
        if entry["backend"] == "mock":
            return {"ok": True, "detail": "mock 后端无需测试"}
        r = await _probe_backend(entry["backend"], entry["base_url"], entry.get("api_key", ""))
        return r

    @app.get("/pool/{entry_id}/models", tags=["pool"])
    async def pool_models(entry_id: str):
        """探测该条目端点下的可用模型列表。"""
        entry = get_pool().get(entry_id)
        if entry is None:
            raise HTTPException(status_code=404, detail=f"条目不存在: {entry_id}")
        if entry["backend"] == "mock":
            return {"models": [{"id": "mock", "name": "mock（内置模拟）"}]}
        return await _list_backend_models(
            entry["backend"], entry["base_url"], entry.get("api_key", ""))

    # ---------------- 智能体（工具调用，zcode 式文件操作） ----------------
    def _resolve_agent_chat(pool_id: str = ""):
        """解析智能体模型：显式 pool_id > 池 agent 角色 > 经典 Architect 设置。

        返回 (OpenAICompatChat, model_name, pool_id)；无法解析时抛 HTTPException。
        """
        from gateway.agent import OpenAICompatChat

        pool = get_pool()
        entry = pool.get(pool_id) if pool_id else pool.resolve("agent")
        if entry is not None:
            if entry["backend"] == "llama_server" and not entry.get("api_key"):
                chat = OpenAICompatChat(
                    base_url=entry["base_url"] or "http://127.0.0.1:8901/v1",
                    api_key=None, model=entry["model"],
                    temperature=float(entry.get("temperature", 0.3)),
                    max_tokens=int(entry.get("max_tokens", 4096)))
            elif entry["backend"] == "mock":
                raise HTTPException(status_code=400,
                                    detail="mock 模型不支持智能体工具调用，请选择真实模型")
            else:
                chat = OpenAICompatChat(
                    base_url=entry["base_url"], api_key=entry.get("api_key") or None,
                    model=entry["model"],
                    temperature=float(entry.get("temperature", 0.3)),
                    max_tokens=int(entry.get("max_tokens", 4096)))
            return chat, entry["model"], entry["id"]
        # 经典回退：Architect 设置（api_key 走 .env）
        import os
        s = settings_store().to_dict()
        acfg = dict(load_config().get("architect", {}))
        acfg.update(s.get("architect", {}))
        key = acfg.get("api_key") or os.environ.get(acfg.get("api_key_env", "DEEPSEEK_API_KEY"))
        if not key:
            raise HTTPException(
                status_code=400,
                detail="未配置大模型 API Key：请在模型池指派 agent 角色（填 key），或在设置中配置 Architect API Key")
        chat = build_agent_chat(acfg)
        return chat, acfg.get("model", "unknown"), ""

    def build_agent_chat(acfg: dict):
        """经典设置 -> OpenAICompatChat（独立函数便于测试注入替身）。"""
        from gateway.agent import OpenAICompatChat
        import os
        key = acfg.get("api_key") or os.environ.get(acfg.get("api_key_env", "DEEPSEEK_API_KEY"))
        return OpenAICompatChat(
            base_url=acfg.get("base_url", "https://api.deepseek.com"),
            api_key=key,
            model=acfg.get("model", "deepseek-v4-flash"),
            temperature=float(acfg.get("temperature", 0.2)),
            max_tokens=int(acfg.get("max_tokens", 4096)),
        )

    @app.post("/agent", tags=["agent"])
    async def agent_run(req: dict):
        """提交智能体任务：{"task", "pool_id"?, "workspace"?, "executor_pool_id"?, "session_id"?}。

        workspace 为用户选择的工作目录；缺省继承会话目录，再缺省用设置默认值。
        session_id 提供时任务在会话内执行（多轮上下文 + 轮次记录）。
        立即返回 request_id；过程事件经 GET /agent/{id}/stream (SSE) 推送。
        """
        from gateway.agent import get_agent_service, get_session_store, new_request_id
        from router_system.tools import WorkspaceTools

        task = str((req or {}).get("task") or "").strip()
        if not task:
            raise HTTPException(status_code=400, detail="task 不能为空")
        if len(task) > 8000:
            raise HTTPException(status_code=400, detail="task 过长（>8000）")
        pool_id = str((req or {}).get("pool_id") or "")

        s = settings_store().to_dict()
        agent_cfg = s.get("agent", {})

        # 会话（可选）：须存在且空闲；工作区缺省继承会话目录
        session = None
        session_id = str((req or {}).get("session_id") or "").strip()
        if session_id:
            session = get_session_store().get(session_id)
            if session is None:
                raise HTTPException(status_code=404, detail=f"会话不存在: {session_id}")
            if session.data.get("busy"):
                raise HTTPException(status_code=409, detail="该会话有任务正在运行，请稍候")
            # 会话级配置继承（创建时指定，后续轮次沿用）
            if not pool_id:
                pool_id = str(session.data.get("pool_id") or "")

        ws_raw = str((req or {}).get("workspace") or "").strip()
        if not ws_raw and session is not None:
            ws_raw = str(session.data.get("workspace") or "")
        if ws_raw:
            ws_path = Path(ws_raw)
            if not ws_path.exists():
                raise HTTPException(status_code=400, detail=f"工作目录不存在: {ws_raw}")
            if not ws_path.is_dir():
                raise HTTPException(status_code=400, detail=f"不是目录: {ws_raw}")
            workspace_dir = str(ws_path.resolve())
        else:
            workspace_dir = agent_cfg.get("workspace_dir", "agent_workspace")

        chat, model, used_pool_id = _resolve_agent_chat(pool_id)

        # 两级模式（D7）：显式指定执行者（本地小模型）时，规划=chat、执行=executor_chat
        executor_pool_id = str((req or {}).get("executor_pool_id") or "").strip()
        if not executor_pool_id and session is not None:
            executor_pool_id = str(session.data.get("executor_pool_id") or "")
        executor_chat = None
        executor_model = ""
        if executor_pool_id:
            entry = get_pool().get(executor_pool_id)
            if entry is None:
                raise HTTPException(status_code=400,
                                    detail=f"执行者条目不存在: {executor_pool_id}")
            if entry["backend"] == "mock":
                raise HTTPException(status_code=400,
                                    detail="mock 模型不能担任执行者，请选择 llama_server 或 openai 条目")
            from gateway.agent import OpenAICompatChat
            executor_chat = OpenAICompatChat(
                base_url=entry["base_url"] or "http://127.0.0.1:8901/v1",
                api_key=entry.get("api_key") or None,
                model=entry["model"],
                temperature=float(entry.get("temperature", 0.3)),
                max_tokens=int(entry.get("max_tokens", 4096)))
            executor_model = f"{entry['name']}（{entry['model']}）"

        if executor_chat is not None and executor_pool_id == used_pool_id:
            raise HTTPException(status_code=400,
                                detail="规划者与执行者是同一个模型，两级模式无意义；请更换执行者条目")

        service = get_agent_service()
        request_id = new_request_id()
        mode = "dual" if executor_chat is not None else "single"
        info = service.register(request_id, task, model, used_pool_id,
                                workspace=workspace_dir,
                                executor_model=executor_model, mode=mode)
        if info is None:
            raise HTTPException(status_code=503, detail="智能体同时运行任务已达上限")

        if session is not None:
            session.data["busy"] = True
            if not session.data.get("workspace"):
                session.data["workspace"] = workspace_dir
            get_session_store().save(session)

        async def _run():
            try:
                await service.run(
                    info, chat,
                    workspace_dir=workspace_dir,
                    max_rounds=int(agent_cfg.get("max_rounds", 8)),
                    token_cap=int(agent_cfg.get("token_cap", 20000)),
                    allow_shell=bool(agent_cfg.get("allow_shell", False)),
                    shell_timeout_s=int(agent_cfg.get("shell_timeout_s", 20)),
                    allow_net=bool(agent_cfg.get("allow_net", True)),
                    executor_chat=executor_chat,
                    max_handoffs=int(agent_cfg.get("max_handoffs", 2)),
                    session=session,
                    approval_policy=str(agent_cfg.get("approval_policy", "dangerous")),
                    approval_timeout_s=int(agent_cfg.get("approval_timeout_s", 120)),
                )
            except Exception as exc:
                import traceback
                traceback.print_exc()
                info.state = "failed"
                info.error = str(exc)
                info.finished_at = __import__("time").time()
                service._write_status(info)
            finally:
                if session is not None:
                    session.data["busy"] = False
                    get_session_store().save(session)

        info.asyncio_task = asyncio.create_task(_run())
        return {"request_id": request_id, "status": "running", "model": model,
                "workspace": workspace_dir, "mode": mode,
                "executor_model": executor_model, "session_id": session_id or None}

    @app.post("/agent/{request_id}/cancel", tags=["agent"])
    async def agent_cancel(request_id: str):
        """停止运行中的智能体任务。"""
        _check_id(request_id, "request_id")
        from gateway.agent import get_agent_service
        service = get_agent_service()
        info = service.get(request_id)
        if info is None:
            raise HTTPException(status_code=404, detail=f"智能体任务不存在: {request_id}")
        if info.state != "running":
            return {"ok": False, "detail": f"任务已结束（{info.state}）"}
        if info.asyncio_task is not None:
            info.asyncio_task.cancel()
        info.state = "failed"
        info.error = "cancelled_by_user"
        info.finished_at = __import__("time").time()
        service._write_status(info)
        return {"ok": True}

    @app.post("/agent/{request_id}/approve", tags=["agent"])
    async def agent_approve(request_id: str, req: dict):
        """裁决待审批操作：{"approval_id", "allowed"}（dsh 式 allow-once / deny）。"""
        _check_id(request_id, "request_id")
        from gateway.agent import get_agent_service
        info = get_agent_service().get(request_id)
        if info is None:
            raise HTTPException(status_code=404, detail=f"智能体任务不存在: {request_id}")
        approval_id = str((req or {}).get("approval_id") or "")
        allowed = bool((req or {}).get("allowed", False))
        mgr = getattr(info, "_approval_manager", None)
        if mgr is None:
            raise HTTPException(status_code=409, detail="该任务无审批流程")
        ok = mgr.decide(approval_id, allowed)
        if not ok:
            raise HTTPException(status_code=404, detail=f"审批单不存在或已裁决: {approval_id}")
        return {"ok": True, "approval_id": approval_id, "allowed": allowed}

    # ---------------- 会话（dsh 式多轮对话） ----------------
    @app.post("/agent/sessions", tags=["agent"])
    async def agent_session_create(req: dict = None):
        """创建会话：{"title"?, "workspace"?, "pool_id"?, "executor_pool_id"?}"""
        from gateway.agent import get_session_store
        r = req or {}
        sess = get_session_store().create(
            title=str(r.get("title") or "").strip(),
            workspace=str(r.get("workspace") or "").strip(),
            pool_id=str(r.get("pool_id") or ""),
            executor_pool_id=str(r.get("executor_pool_id") or ""))
        return sess.view()

    @app.get("/agent/sessions", tags=["agent"])
    async def agent_sessions():
        """会话列表（按更新时间倒序）。"""
        from gateway.agent import get_session_store
        return get_session_store().list()

    @app.get("/agent/sessions/{sid}", tags=["agent"])
    async def agent_session_detail(sid: str):
        """会话详情（含轮次）。"""
        _check_id(sid, "会话 ID")
        from gateway.agent import get_session_store
        sess = get_session_store().get(sid)
        if sess is None:
            raise HTTPException(status_code=404, detail=f"会话不存在: {sid}")
        return sess.view()

    @app.patch("/agent/sessions/{sid}", tags=["agent"])
    async def agent_session_rename(sid: str, req: dict = None):
        """重命名会话：{"title"}（dsh session.rename 对齐）。"""
        _check_id(sid, "会话 ID")
        from gateway.agent import get_session_store
        title = str((req or {}).get("title") or "").strip()
        if not title:
            raise HTTPException(status_code=400, detail="title 不能为空")
        sess = get_session_store().rename(sid, title)
        if sess is None:
            raise HTTPException(status_code=404, detail=f"会话不存在: {sid}")
        return sess.view()

    @app.delete("/agent/sessions/{sid}", tags=["agent"])
    async def agent_session_delete(sid: str):
        _check_id(sid, "会话 ID")
        from gateway.agent import get_session_store
        return {"ok": get_session_store().delete(sid)}

    @app.get("/agent/fs", tags=["agent"])
    async def agent_fs_browse(path: str = ""):
        """目录选择器：浏览本地文件系统（只列子目录，不读文件内容）。"""
        from router_system.tools import browse_directories
        return browse_directories(path)

    @app.get("/agent/workspaces", tags=["agent"])
    async def agent_workspaces():
        """当前工作区 + 最近打开列表。"""
        s = settings_store().to_dict()
        agent_cfg = s.get("agent", {})
        current = agent_cfg.get("workspace_dir", "agent_workspace")
        return {"current": current, "recent": list(agent_cfg.get("recent_workspaces", []))}

    @app.post("/agent/workspaces", tags=["agent"])
    async def agent_open_workspace(req: dict):
        """打开（或创建）一个工作目录：设为当前并记入最近列表。"""
        path = str((req or {}).get("path") or "").strip()
        create = bool((req or {}).get("create", False))
        if not path:
            raise HTTPException(status_code=400, detail="path 不能为空")
        p = Path(path)
        if not p.exists():
            if not create:
                raise HTTPException(status_code=400,
                                    detail=f"目录不存在: {path}（可勾选“新建目录”）")
            try:
                p.mkdir(parents=True, exist_ok=True)
            except OSError as e:
                raise HTTPException(status_code=400, detail=f"创建失败: {e}")
        elif not p.is_dir():
            raise HTTPException(status_code=400, detail=f"不是目录: {path}")
        resolved = str(p.resolve())
        store = settings_store()
        store.update({"agent": {"workspace_dir": resolved}})
        merged = store.to_dict().get("agent", {})
        recent = [w for w in merged.get("recent_workspaces", []) if w != resolved]
        recent.insert(0, resolved)
        store.update({"agent": {"recent_workspaces": recent[:8]}})
        return {"ok": True, "current": resolved,
                "recent": store.to_dict().get("agent", {}).get("recent_workspaces", [])}

    @app.get("/agent/{request_id}/status", tags=["agent"])
    async def agent_status(request_id: str):
        _check_id(request_id, "request_id")
        from gateway.agent import get_agent_service
        info = get_agent_service().get(request_id)
        if info is None:
            # 尝试磁盘恢复（服务重启后仍可查历史）
            p = get_agent_service().status_path(request_id)
            if p.exists():
                import json
                return json.loads(p.read_text(encoding="utf-8"))
            raise HTTPException(status_code=404, detail=f"智能体任务不存在: {request_id}")
        return info.to_dict()

    @app.get("/agent/{request_id}/events", tags=["agent"])
    async def agent_events(request_id: str):
        """完整事件列表（JSON，刷新后恢复用）。"""
        _check_id(request_id, "request_id")
        from gateway.agent import get_agent_service
        return get_agent_service().read_events(request_id)

    @app.get("/agent/{request_id}/stream", tags=["agent"])
    async def agent_stream(request_id: str):
        """SSE：实时推送智能体过程事件（round/tool_call/tool_result/usage/final）。"""
        _check_id(request_id, "request_id")
        from fastapi.responses import StreamingResponse
        from gateway.agent import get_agent_service

        async def event_generator():
            import json
            cancel = asyncio.Event()
            try:
                async for ev in get_agent_service().watch_events(request_id, cancel):
                    yield f"data: {json.dumps(ev, ensure_ascii=False)}\n\n"
            finally:
                cancel.set()

        return StreamingResponse(
            event_generator(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive",
                     "X-Accel-Buffering": "no"},
        )

    @app.get("/agent/workspace", tags=["agent"])
    async def agent_workspace(path: str = "", root: str = ""):
        """列出智能体工作区内容。root 可指定其他已选工作目录（默认用设置值）。越界/非法返回 400。"""
        from router_system.tools import ToolError, WorkspaceTools
        base = _agent_workspace_root(root)
        tools = WorkspaceTools(base)
        try:
            return tools.list_dir(path)
        except ToolError as e:
            raise HTTPException(status_code=400, detail=str(e))

    @app.get("/agent/file", tags=["agent"])
    async def agent_file(path: str, root: str = ""):
        """读取智能体工作区内文件（前端预览，越界即 400）。root 同 /agent/workspace。"""
        from router_system.tools import ToolError, WorkspaceTools
        base = _agent_workspace_root(root)
        tools = WorkspaceTools(base)
        try:
            result = tools.read_file(path)
        except ToolError as e:
            raise HTTPException(status_code=400, detail=str(e))
        if not result.get("ok"):
            raise HTTPException(status_code=404, detail=result.get("error", "读取失败"))
        return result

    def _agent_workspace_root(root: str = "") -> str:
        """解析工作区根：显式 root（须为已存在目录）> 设置值。"""
        root = (root or "").strip()
        if root:
            p = Path(root)
            if not p.is_dir():
                raise HTTPException(status_code=400, detail=f"工作目录不存在: {root}")
            return str(p.resolve())
        s = settings_store().to_dict()
        return s.get("agent", {}).get("workspace_dir", "agent_workspace")

    # ---------------- 模型设置（用户可调整） ----------------
    @app.get("/config", tags=["settings"])
    async def get_config():
        """读取当前可调整设置（小模型 / 大模型 / 管线）。

        architect.api_key 打码返回（api_key_set + 前 6 位，对齐 D2 池条目语义）；
        修改时留空/不传 = 保留服务端已存值。
        """
        out = settings_store().to_dict()
        arch = out.get("architect") or {}
        key = arch.get("api_key") or ""
        arch["api_key_set"] = bool(key)
        arch["api_key"] = (key[:6] + "…") if key else ""
        return out

    @app.put("/config", tags=["settings"])
    async def put_config(patch: dict):
        """部分更新设置并重建管线。示例：
        {"worker": {"backend": "openai", "base_url": "http://127.0.0.1:11434/v1", "temperature": 0.4}}

        architect.api_key 传空串 = 保留原值（与打码返回配套）。
        """
        arch_patch = (patch or {}).get("architect")
        if isinstance(arch_patch, dict) and not str(arch_patch.get("api_key") or "").strip():
            arch_patch.pop("api_key", None)
        try:
            merged = settings_store().update(patch)
        except Exception as e:
            raise HTTPException(status_code=400, detail=f"设置非法: {e}")
        rebuild_pipeline()
        key = merged.get("architect", {}).get("api_key") or ""
        merged["architect"]["api_key_set"] = bool(key)
        merged["architect"]["api_key"] = (key[:6] + "…") if key else ""
        return merged

    @app.post("/config/reset", tags=["settings"])
    async def reset_config():
        """恢复默认设置并重建管线。"""
        merged = settings_store().reset()
        rebuild_pipeline()
        return merged

    # ---------------- 模型发现 & 验证（/config 与 /pool 共用） ----------------
    async def _list_backend_models(backend: str, base_url: str = "", api_key: str = "") -> dict:
        """探测 OpenAI 兼容端点的模型列表。返回 {"models":[...]} 或 {"error": "..."}。"""
        import httpx

        if backend == "llama_server":
            url = (base_url or f"http://127.0.0.1:{settings_store().get('worker','port',8901)}/v1") + "/models"
        elif backend == "openai":
            if not base_url:
                return {"error": "openai 后端需要填写 API 地址"}
            url = base_url.rstrip("/") + "/models"
        else:
            return {"error": f"不支持的后端类型: {backend}"}

        headers = {}
        if api_key:
            headers["Authorization"] = f"Bearer {api_key}"

        try:
            async with httpx.AsyncClient(timeout=10.0) as client:
                resp = await client.get(url, headers=headers)
                resp.raise_for_status()
                data = resp.json()
                models = []
                # OpenAI /models 格式：{"object": "list", "data": [{"id": "..."}]}
                # Ollama /models 格式：{"models": [{"name": "..."}]}
                raw = data.get("data") or data.get("models") or []
                for m in raw:
                    mid = m.get("id") or m.get("name") or ""
                    if mid:
                        models.append({"id": mid, "name": mid})
                return {"models": models}
        except httpx.TimeoutException:
            return {"error": "连接超时，模型服务可能未启动"}
        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP {e.response.status_code}：{e.response.text[:200]}"}
        except Exception as e:
            return {"error": str(e)}

    async def _probe_backend(backend: str, base_url: str = "", api_key: str = "") -> dict:
        """后端连通性探测。返回 {"ok": bool, ...}。"""
        import httpx

        if backend == "llama_server":
            url = (base_url or f"http://127.0.0.1:{settings_store().get('worker','port',8901)}/v1") + "/models"
        elif backend == "openai":
            if not base_url:
                return {"ok": False, "detail": "openai 后端需要填写 API 地址"}
            url = base_url.rstrip("/") + "/models"
        else:
            return {"ok": False, "detail": f"不支持的后端: {backend}"}

        headers = {}
        if api_key:
            headers["Authorization"] = f"Bearer {api_key}"

        try:
            async with httpx.AsyncClient(timeout=10.0) as client:
                resp = await client.get(url, headers=headers)
                resp.raise_for_status()
                return {"ok": True, "status_code": resp.status_code}
        except Exception as e:
            return {"ok": False, "detail": str(e)}

    @app.get("/config/models", tags=["settings"])
    async def list_models(backend: str, base_url: str = "", api_key: str = ""):
        """探测指定后端支持的模型列表。

        backend: llama_server | openai
        base_url: 端点地址（llama_server 默认 http://127.0.0.1:8901/v1）
        api_key: 可选（云端 API 需要）

        返回 {"models": [{"id": "...", "name": "..."}]}
        """
        return await _list_backend_models(backend, base_url, api_key)

    @app.get("/config/ping", tags=["settings"])
    async def ping_backend(backend: str, base_url: str = "", api_key: str = ""):
        """验证后端连接是否可用（健康检查）。"""
        return await _probe_backend(backend, base_url, api_key)

    # ─────────────────────────────────────────────────────────────────────
    # llama-server 内置管理
    # ─────────────────────────────────────────────────────────────────────
    from gateway.llama_manager import get_llama_manager

    @app.get("/llama/status", tags=["llama"])
    async def llama_status():
        """查询 llama-server 运行状态。"""
        lm = get_llama_manager()
        s = lm.status()
        return {
            "running": s.running,
            "pid": s.pid,
            "model": s.model,
            "port": s.port,
            "base_url": s.base_url,
            "started_at": s.started_at,
            "error": s.error,
        }

    @app.get("/llama/models", tags=["llama"])
    async def llama_local_models():
        """列出本地已有模型文件（models/*.gguf）。"""
        lm = get_llama_manager()
        return {"models": lm.list_local_models()}

    @app.post("/llama/start", tags=["llama"])
    async def llama_start(
        model: str,
        port: int = 8901,
        ngl: int = 99,
        ctx: int = 4096,
    ):
        """启动本地 llama-server。

        model: 模型文件路径（相对于项目根，或绝对路径）
        port/ngl/ctx: 服务参数
        """
        lm = get_llama_manager()
        s = await lm.start(model=model, port=port, ngl=ngl, ctx=ctx)
        return {
            "running": s.running,
            "pid": s.pid,
            "model": s.model,
            "port": s.port,
            "base_url": s.base_url,
            "error": s.error,
        }

    @app.post("/llama/stop", tags=["llama"])
    async def llama_stop():
        """停止本地 llama-server。"""
        lm = get_llama_manager()
        await lm.stop()
        return {"running": False}

    @app.post("/llama/download", tags=["llama"])
    async def llama_download(
        url: str,
        dest: Optional[str] = None,
    ):
        """从 HuggingFace 或直链下载 .gguf 模型到 models/ 目录。

        支持 HuggingFace 路径别名，如 "Qwen/Qwen3-4B-GGUF/Qwen3-4B-Q4_K_M.gguf"
        """
        lm = get_llama_manager()
        # 检查是否已在下载
        existing = lm.get_download_progress(url)
        if existing and not existing.done:
            return {
                "url": url,
                "dest": existing.dest,
                "downloaded_bytes": existing.downloaded_bytes,
                "total_bytes": existing.total_bytes,
                "progress_pct": existing.progress_pct,
                "speed": existing.speed,
                "eta": existing.eta,
                "done": False,
                "error": None,
            }
        prog = await lm.download_model(url=url, dest=dest)
        return {
            "url": prog.url,
            "dest": prog.dest,
            "downloaded_bytes": prog.downloaded_bytes,
            "total_bytes": prog.total_bytes,
            "progress_pct": prog.progress_pct,
            "speed": prog.speed,
            "eta": prog.eta,
            "done": prog.done,
            "error": prog.error,
        }

    @app.get("/llama/download/stream", tags=["llama"])
    async def llama_download_progress(url: str):
        """SSE：推送模型下载进度。"""
        from fastapi.responses import StreamingResponse

        async def event_gen():
            lm = get_llama_manager()
            last_pct = -1.0
            while True:
                prog = lm.get_download_progress(url)
                if prog and (prog.done or abs(prog.progress_pct - last_pct) > 0.1):
                    payload = json.dumps({
                        "url": prog.url,
                        "dest": prog.dest,
                        "downloaded_bytes": prog.downloaded_bytes,
                        "total_bytes": prog.total_bytes,
                        "progress_pct": round(prog.progress_pct, 2),
                        "speed": prog.speed,
                        "eta": prog.eta,
                        "done": prog.done,
                        "error": prog.error,
                    }, ensure_ascii=False)
                    yield f"data: {payload}\n\n"
                    last_pct = prog.progress_pct
                if prog and prog.done:
                    break
                await asyncio.sleep(0.5)

        return StreamingResponse(
            event_gen(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive",
                     "X-Accel-Buffering": "no"},
        )

    # ─────────────────────────────────────────────────────────────────
    # Vue SPA 页面路由（非 API，交给前端 Vue Router 处理）
    # 这些路径返回 index.html，浏览器加载后由 Vue Router 渲染对应页面
    # ─────────────────────────────────────────────────────────────────
    @app.get("/", include_in_schema=False)
    @app.get("/chat", include_in_schema=False)
    @app.get("/collaboration", include_in_schema=False)
    @app.get("/review", include_in_schema=False)
    @app.get("/settings", include_in_schema=False)
    @app.get("/agent", include_in_schema=False)
    async def spa_page():
        """所有前端页面路径返回 Vue SPA index.html。"""
        if _INDEX_PATH.exists():
            return HTMLResponse(_INDEX_PATH.read_text(encoding="utf-8"))
        return HTMLResponse("<h1>端云协同 LLM 系统</h1><p>请构建前端：cd webapp && npm run build</p>")

    # ---------------- metrics API（JSON REST 端点）--------------------
    # 前端通过 axios GET /api/metrics 调用此端点获取指标数据
    @app.get("/metrics", include_in_schema=False)
    async def metrics_page():
        return await metrics()

    @app.get("/api/metrics", tags=["system"])
    async def metrics():
        out = {
            "router": get_router().stats.summary(),
            "cache": get_router().cache.stats(),
        }
        if _V2_OK and _v2stats is not None:
            out["v2"] = _v2stats.summary()
            out["review"] = {
                "pending": get_review().count(status="pending"),
                "total": get_review().count(),
            }
        return out

except ImportError:
    app = None
    print("[gateway] 未安装 fastapi，请执行: pip install -r requirements.txt")


def _maybe_enqueue(result) -> None:
    """按 review 配置抽样/强制入队（异步场景用同步快速调用）。"""
    if not _V2_OK or get_review() is None:
        return
    if result.status not in ("done", "escalated", "fast_path"):
        return
    cfg = load_config().get("review", {})
    sample_rate = float(cfg.get("sample_rate", 0.10))
    force_tags = cfg.get("force_tags", ["safety"])
    # 从交流文本读取 tags（若可）
    tags: List[str] = []
    if result.workspace_path:
        try:
            import json
            ws = json.loads(Path(result.workspace_path).read_text(encoding="utf-8"))
            tags = (ws.get("brief") or {}).get("tags") or []
        except Exception:
            tags = []
    if ReviewQueue.should_enqueue(tags, sample_rate=sample_rate, force_tags=force_tags):
        get_review().enqueue(
            request_id=result.request_id, query=result.query, response=result.response,
            tags=tags, reason="auto", workspace_path=result.workspace_path,
        )


if __name__ == "__main__":
    import uvicorn
    # 默认只绑定回环地址：网关能读写工作区文件/执行命令，不宜默认暴露到局域网
    # （需要局域网访问时显式 --host 0.0.0.0，并设置 GATEWAY_TRUSTED_HOSTS 放行对应主机名）
    uvicorn.run("gateway.api:app", host="127.0.0.1", port=8000, reload=False)