Phase 3.3 多 Agent 编排图引擎: orchestration JSON 列 + 拓扑并行引擎 + local/external/reduce 节点 + 工具沙箱继承 + 后台编排 API

.gitea/workflows/ci.yml

@@ -0,0 +1,28 @@
+# qihuaiflow minimal CI (Gitea Actions)
+#
+# NOTE: This workflow is DORMANT until a Gitea act_runner is registered with
+# this repo. The active gate today is the host-side pre-push hook
+# (scripts/git-hooks/pre-push) + scripts/ci_gate.py, because our deployment
+# model keeps the app running in a long-lived `qihu-aiflow` container on the
+# docker host rather than building it fresh in CI.
+#
+# To activate later:
+#   1. Install act_runner on the docker host (sharing the docker socket).
+#   2. Register it against this repo with label `ubuntu-latest`.
+#   3. The runner can then `docker exec qihu-aiflow ...` to reach the app.
+#
+# Until then, the pre-push hook already blocks broken pushes.
+name: CI
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+jobs:
+  gate:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Backend pytest
+        run: docker exec qihu-aiflow bash -c 'cd /app/backend/python && .venv/bin/python -m pytest -c pytest.ini -q'
+      - name: Frontend vitest
+        run: docker exec qihu-aiflow bash -c 'cd /app/frontend && npx vitest run --reporter=dot'

VERSION

@@ -1,21 +1 @@
-# 七虎AI PaaS 平台 — 版本清单 (VERSION)
+v2026.07.19.8
-# 版本规范见 docs/VERSIONING.md
-# 版本号格式: vYYYY.MM.DD.N   (年.月.日.当日递增序号，从 1 开始)
-# 环境标签:   prod = 正式版 | test = 测试版
-
-VERSION=v2026.07.18.1
-ENV=prod
-DATE=2026-07-18
-BASE_IMAGE=qihu-aiflow:with-deps
-BACKUP_IMAGE=qihu-aiflow:backup-20260718-001
-
-## 本版本变更 (v2026.07.18.1 / prod)
-- Phase0 工程纪律止血 启动
-- 新增「部署前镜像备份 SOP」：scripts/backup_before_deploy.py，每次部署前自动提交递增序号镜像快照
-- 统一版本号规范 vYYYY.MM.DD.N + prod/test 标签（取代旧 -recover 标签）
-- 引入后端 pytest / 前端 vitest 测试基建，作为迭代式重构的安全网
-- 固化反复问题的回归用例 R1/R2/R3
-
-## 历史
-- 旧方案：镜像曾用 AIFlow v2.0.1 / v2026.07.14 / v2026.07.14-nav；git 曾用 v2026.07.12-recover / v2026.07.12-ui-recover / v2026.07.13-recover。自 v2026.07.18.1 起统一为 vYYYY.MM.DD.N + prod/test，便于正式版推广后与测试版做环境切割。
-- 历史功能：AI聊天(real API + tool-call function calling)、知识库文件存储 + AI 图谱、频道二维码、邮件/短信配置 + AI 生成、支付配置、平台驾驶舱(token 累计 + 审计/操作日志)、Agent 资源 API、evolution pipeline 自动恢复、全局默认模型、广告位图片上传 + AI 绘图、应用市场安装/卸载/搜索/去重、资产文件类型分类 + 在线预览。迁移 001-015。

backend/python/app/api/admin/agents_admin.py

@@ -88,6 +88,7 @@ def agent_to_dict(a: MarketApp) -> dict:
         "agent_endpoint": a.agent_endpoint,
         "skill_content": a.skill_content,
         "config": config,
+        "orchestration": a.orchestration,
         "created_at": a.created_at.isoformat() if a.created_at else None,
         "updated_at": a.updated_at.isoformat() if a.updated_at else None,
     }
@@ -555,3 +556,101 @@ async def get_available_models(
             for m, p in models
         ]
     }
+
+
+# ============================================================
+# 3.1 tool_policy 沙箱：Agent 工具/MCP 权限策略
+# ============================================================
+class ToolPolicyModel(BaseModel):
+    tool_policy: Optional[dict] = None  # null = 清除策略（恢复默认允许全部）
+
+
+@router.get("/{agent_id}/tool-policy", summary="获取 Agent 工具策略")
+async def get_agent_tool_policy(
+    agent_id: str,
+    admin_user=Depends(get_current_admin),
+    db: Session = Depends(get_db),
+):
+    agent = db.query(MarketApp).filter(
+        MarketApp.id == agent_id, MarketApp.type == "agent"
+    ).first()
+    if not agent:
+        raise HTTPException(status_code=404, detail="Agent 不存在")
+    return {"agent_id": agent_id, "tool_policy": agent.tool_policy}
+
+
+@router.put("/{agent_id}/tool-policy", summary="设置 Agent 工具策略")
+async def set_agent_tool_policy(
+    agent_id: str,
+    body: ToolPolicyModel,
+    admin_user=Depends(get_current_admin),
+    db: Session = Depends(get_db),
+):
+    agent = db.query(MarketApp).filter(
+        MarketApp.id == agent_id, MarketApp.type == "agent"
+    ).first()
+    if not agent:
+        raise HTTPException(status_code=404, detail="Agent 不存在")
+    tp = body.tool_policy
+    if tp is not None:
+        if not isinstance(tp, dict):
+            raise HTTPException(status_code=400, detail="tool_policy 必须为对象")
+        # 轻量结构校验
+        if "allowed_tools" in tp and tp["allowed_tools"] is not None:
+            at = tp["allowed_tools"]
+            if not isinstance(at, dict):
+                raise HTTPException(status_code=400, detail="allowed_tools 必须为对象")
+            for k in ("skill_ids", "mcp_ids"):
+                if k in at and at[k] is not None and not isinstance(at[k], list):
+                    raise HTTPException(status_code=400, detail=f"allowed_tools.{k} 必须为数组")
+        for k in ("max_calls_per_conversation", "per_round_cap"):
+            if k in tp and tp[k] is not None and not isinstance(tp[k], int):
+                raise HTTPException(status_code=400, detail=f"{k} 必须为整数")
+    agent.tool_policy = tp
+    db.commit()
+    return {"agent_id": agent_id, "tool_policy": agent.tool_policy, "updated": True}
+
+# 3.3 多 Agent 编排图(图引擎)
+# ============================================================
+class OrchestrationModel(BaseModel):
+    orchestration: Optional[dict] = None  # null = 清除编排图
+
+
+@router.get("/{agent_id}/orchestration", summary="获取 Agent 编排图")
+async def get_agent_orchestration(
+    agent_id: str,
+    admin_user=Depends(get_current_admin),
+    db: Session = Depends(get_db),
+):
+    agent = db.query(MarketApp).filter(
+        MarketApp.id == agent_id, MarketApp.type == "agent"
+    ).first()
+    if not agent:
+        raise HTTPException(status_code=404, detail="Agent 不存在")
+    return {"agent_id": agent_id, "orchestration": agent.orchestration}
+
+
+@router.put("/{agent_id}/orchestration", summary="设置 Agent 编排图")
+async def set_agent_orchestration(
+    agent_id: str,
+    body: OrchestrationModel,
+    admin_user=Depends(get_current_admin),
+    db: Session = Depends(get_db),
+):
+    agent = db.query(MarketApp).filter(
+        MarketApp.id == agent_id, MarketApp.type == "agent"
+    ).first()
+    if not agent:
+        raise HTTPException(status_code=404, detail="Agent 不存在")
+    og = body.orchestration
+    if og is not None:
+        if not isinstance(og, dict):
+            raise HTTPException(status_code=400, detail="orchestration 必须为对象")
+        try:
+            from app.core.orchestrator import validate_graph
+            validate_graph(og)
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=f"编排图非法: {str(e)}")
+    agent.orchestration = og
+    db.commit()
+    return {"agent_id": agent_id, "orchestration": agent.orchestration, "updated": True}

backend/python/app/api/admin/observability_admin.py

@@ -0,0 +1,103 @@
+"""调用可观测面板后端 API（媒体/LLM 调用归因）。
+路由前缀: /app/api/admin/observability
+数据来源: token_usage_logs（Phase 2.2 已加 trace_id 列，可经 X-Trace-Id 关联请求链路）。
+"""
+from datetime import datetime, timedelta
+
+from fastapi import APIRouter, Query, Depends
+from ...database import get_db
+from ..auth import get_current_user
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+router = APIRouter()
+
+
+@router.get("/summary")
+async def observability_summary(
+    days: int = Query(30, description="统计天数"),
+    db: Session = Depends(get_db),
+    user=Depends(get_current_user),
+):
+    """媒体/LLM 调用归因：总览 + 按模型 + 按用户 + 每日趋势 + 最近调用(trace_id 下钻)。"""
+    since = datetime.now() - timedelta(days=days)
+
+    totals = db.execute(text("""
+        SELECT COUNT(*),
+               COALESCE(SUM(total_tokens), 0),
+               COALESCE(SUM(cost), 0),
+               COUNT(DISTINCT user_id)
+        FROM token_usage_logs WHERE created_at >= :since
+    """), {"since": since}).fetchone()
+
+    by_model = db.execute(text("""
+        SELECT model_name, COUNT(*), COALESCE(SUM(total_tokens), 0), COALESCE(SUM(cost), 0)
+        FROM token_usage_logs WHERE created_at >= :since
+        GROUP BY model_name ORDER BY COALESCE(SUM(total_tokens), 0) DESC
+    """), {"since": since}).fetchall()
+
+    by_user = db.execute(text("""
+        SELECT t.user_id, u.username, COUNT(*), COALESCE(SUM(t.total_tokens), 0), COALESCE(SUM(t.cost), 0)
+        FROM token_usage_logs t LEFT JOIN users u ON u.id = t.user_id
+        WHERE t.created_at >= :since
+        GROUP BY t.user_id, u.username ORDER BY COALESCE(SUM(t.total_tokens), 0) DESC LIMIT 20
+    """), {"since": since}).fetchall()
+
+    daily = db.execute(text("""
+        SELECT DATE(created_at) AS d, COALESCE(SUM(total_tokens), 0), COALESCE(SUM(cost), 0), COUNT(*)
+        FROM token_usage_logs WHERE created_at >= :since
+        GROUP BY DATE(created_at) ORDER BY d
+    """), {"since": since}).fetchall()
+
+    recent = db.execute(text("""
+        SELECT trace_id, model_name, total_tokens, cost, created_at
+        FROM token_usage_logs WHERE created_at >= :since
+        ORDER BY created_at DESC LIMIT 50
+    """), {"since": since}).fetchall()
+
+    return {
+        "totals": {
+            "calls": int(totals[0] or 0),
+            "tokens": int(totals[1] or 0),
+            "cost": float(totals[2] or 0),
+            "users": int(totals[3] or 0),
+        },
+        "by_model": [
+            {
+                "model_name": m[0] or "unknown",
+                "calls": int(m[1] or 0),
+                "tokens": int(m[2] or 0),
+                "cost": float(m[3] or 0),
+            }
+            for m in by_model
+        ],
+        "by_user": [
+            {
+                "user_id": u[0],
+                "username": u[1] or u[0],
+                "calls": int(u[2] or 0),
+                "tokens": int(u[3] or 0),
+                "cost": float(u[4] or 0),
+            }
+            for u in by_user
+        ],
+        "daily": [
+            {
+                "date": str(d[0]),
+                "tokens": int(d[1] or 0),
+                "cost": float(d[2] or 0),
+                "calls": int(d[3] or 0),
+            }
+            for d in daily
+        ],
+        "recent": [
+            {
+                "trace_id": r[0],
+                "model_name": r[1] or "unknown",
+                "tokens": int(r[2] or 0),
+                "cost": float(r[3] or 0),
+                "created_at": r[4].strftime("%Y-%m-%d %H:%M:%S") if r[4] else None,
+            }
+            for r in recent
+        ],
+    }

backend/python/app/api/chat_agents.py

@@ -0,0 +1,951 @@
+# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).
+# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.
+
+from app.core.credential import decrypt_secret
+from app.core.observability import get_trace_id
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from sqlalchemy.orm import Session
+from sqlalchemy import and_, desc, asc, func
+from pydantic import BaseModel, Field
+from typing import Optional
+import uuid
+from datetime import datetime, timezone, timedelta
+import re
+import json
+import asyncio
+import math
+import httpx
+import os
+import logging
+from ..database import get_db, SessionLocal
+from ..models import (
+
+
+
+    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,
+
+
+
+    KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill,
+
+
+
+    EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation,
+
+
+
+    MCPServer,
+
+
+
+    gen_uuid
+
+
+
+)
+from ..auth import get_current_user
+from ..platform_knowledge import get_platform_knowledge
+from ..mcp_client import list_mcp_tools, call_mcp_tool
+from ..middleware.log_service import LogService
+from ..capability_registry import resolve_auto_model
+from app.api.chat_utils import (_extract_text_from_json, _proxy_for_url)
+
+def _normalize_dify_chat_url(endpoint: str) -> str:
+
+
+
+    """把用户填写的 Dify 地址归一化为 /v1/chat-messages 端点。与 market.py 保持一致。"""
+
+
+
+    ep = (endpoint or "").rstrip("/")
+
+
+
+    if ep.endswith("/chat-messages"):
+
+
+
+        return ep
+
+
+
+    if ep.endswith("/v1"):
+
+
+
+        return ep + "/chat-messages"
+
+
+
+    return ep + "/v1/chat-messages"
+
+async def _call_dify_agent(endpoint: str, api_key: str, query: str,
+
+
+
+                           user_id: str, ext_conv_id: str | None) -> tuple:
+
+
+
+    """调用 Dify(新版 Agent) /v1/chat-messages，流式(SSE)解析最终回答。
+
+
+
+
+
+
+
+    新版 Agent API 仅支持 response_mode=streaming(blocking 会 400)。事件流中
+
+
+
+    message / agent_message 事件携带增量 answer，累加即为最终回答；每个事件都带
+
+
+
+    conversation_id，用于维持多轮上下文。返回 (answer, new_conversation_id)。
+
+
+
+    """
+
+
+
+    url = _normalize_dify_chat_url(endpoint)
+
+
+
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+
+
+
+    payload = {
+
+
+
+        "inputs": {},
+
+
+
+        "query": query,
+
+
+
+        "user": (user_id or "web-user")[:64],
+
+
+
+        "response_mode": "streaming",
+
+
+
+        "conversation_id": ext_conv_id or "",
+
+
+
+    }
+
+
+
+    answer_parts = []
+
+
+
+    new_conv_id = ext_conv_id
+
+
+
+    err_msg = None
+
+
+
+    wf_outputs = None
+
+
+
+    # httpx 默认 follow_redirects=False；stream POST 若遇 301/302 会直接返回该状态码。
+
+
+
+    # 且 httpx 自动跟随会把 POST 降级为 GET(对 /v1/chat-messages 变 405)。
+
+
+
+    # 因此手动跟随 301/302/307/308，保持 POST 与请求体不变(最多 4 跳)。
+
+
+
+    async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False,
+
+
+
+                                 follow_redirects=False) as client:
+
+
+
+        cur_url = url
+
+
+
+        for _redir in range(4):
+
+
+
+            async with client.stream("POST", cur_url, headers=headers, json=payload) as resp:
+
+
+
+                if resp.status_code in (301, 302, 307, 308):
+
+
+
+                    loc = resp.headers.get("location") or resp.headers.get("Location")
+
+
+
+                    await resp.aclose()
+
+
+
+                    if not loc:
+
+
+
+                        raise RuntimeError(f"Dify HTTP {resp.status_code}: 缺少重定向地址")
+
+
+
+                    # 处理相对地址
+
+
+
+                    if loc.startswith("/"):
+
+
+
+                        from urllib.parse import urlsplit, urlunsplit
+
+
+
+                        p = urlsplit(cur_url)
+
+
+
+                        loc = urlunsplit((p.scheme, p.netloc, loc, "", ""))
+
+
+
+                    cur_url = loc
+
+
+
+                    continue
+
+
+
+                if resp.status_code != 200:
+
+
+
+                    body = (await resp.aread()).decode("utf-8", "replace")
+
+
+
+                    raise RuntimeError(f"Dify HTTP {resp.status_code}: {body[:300]}")
+
+
+
+                async for line in resp.aiter_lines():
+
+
+
+                    if not line:
+
+
+
+                        continue
+
+
+
+                    line = line.strip()
+
+
+
+                    if not line.startswith("data:"):
+
+
+
+                        continue
+
+
+
+                    data_str = line[5:].strip()
+
+
+
+                    if not data_str or data_str == "[DONE]":
+
+
+
+                        continue
+
+
+
+                    try:
+
+
+
+                        evt = json.loads(data_str)
+
+
+
+                    except Exception:
+
+
+
+                        continue
+
+
+
+                    ev = evt.get("event")
+
+
+
+                    if evt.get("conversation_id"):
+
+
+
+                        new_conv_id = evt["conversation_id"]
+
+
+
+                    data = evt.get("data") or {}
+
+
+
+                    # 兼容三类 Dify 应用：
+
+
+
+                    #  1) Chat/Agent 应用: message / agent_message 事件带增量 answer
+
+
+
+                    #  2) Chatflow/Workflow 应用: text_chunk 事件带 data.text；
+
+
+
+                    #     结束时 workflow_finished 带 status / outputs / error
+
+
+
+                    if ev in ("message", "agent_message"):
+
+
+
+                        answer_parts.append(evt.get("answer", "") or "")
+
+
+
+                    elif ev == "message_replace":
+
+
+
+                        # 内容审查替换：以替换后的完整 answer 为准
+
+
+
+                        answer_parts = [evt.get("answer", "") or ""]
+
+
+
+                    elif ev == "text_chunk":
+
+
+
+                        answer_parts.append((data.get("text") or ""))
+
+
+
+                    elif ev == "node_finished":
+
+
+
+                        # answer / end 节点的 outputs 可能含最终文本
+
+
+
+                        if data.get("node_type") in ("answer", "end") and data.get("outputs"):
+
+
+
+                            wf_outputs = data.get("outputs")
+
+
+
+                        if data.get("status") == "failed" and data.get("error"):
+
+
+
+                            err_msg = str(data.get("error"))
+
+
+
+                    elif ev == "workflow_finished":
+
+
+
+                        if data.get("status") == "failed":
+
+
+
+                            err_msg = str(data.get("error") or "Dify 工作流执行失败")
+
+
+
+                        elif data.get("outputs"):
+
+
+
+                            wf_outputs = data.get("outputs")
+
+
+
+                    elif ev == "error":
+
+
+
+                        err_msg = evt.get("message") or evt.get("code") or "Dify 流式返回错误"
+
+
+
+            break
+
+
+
+        else:
+
+
+
+            raise RuntimeError("Dify 重定向次数过多(>4)，请直接填写重定向后的地址")
+
+
+
+    answer = "".join(answer_parts).strip()
+
+
+
+    # 流式无增量文本时，尝试从 workflow 结束 outputs 兜底提取
+
+
+
+    if not answer and wf_outputs:
+
+
+
+        answer = _extract_text_from_json(wf_outputs).strip()
+
+
+
+    if not answer and err_msg:
+
+
+
+        raise RuntimeError(err_msg)
+
+
+
+    return answer, new_conv_id
+
+async def _call_n8n_agent(endpoint: str, api_key: str, query: str,
+
+
+
+                          user_id: str, history: list) -> tuple:
+
+
+
+    """调用 n8n 生产 Webhook (POST)。webhook 用 Respond to Webhook 节点返回 JSON，
+
+
+
+    尽力从中抽取文本回答。n8n webhook 无平台级会话态，多轮靠我们下发 history。
+
+
+
+    返回 (answer, None)。"""
+
+
+
+    url = (endpoint or "").strip()
+
+
+
+    headers = {"Content-Type": "application/json"}
+
+
+
+    if api_key:
+
+
+
+        # n8n Header Auth 自定义头名不固定，同时下发常见几种以提高兼容性
+
+
+
+        headers["Authorization"] = f"Bearer {api_key}"
+
+
+
+        headers["X-N8N-API-KEY"] = api_key
+
+
+
+    # 携带多种字段别名，兼容 n8n AI Agent(chatInput) 与自定义工作流
+
+
+
+    payload = {
+
+
+
+        "query": query,
+
+
+
+        "chatInput": query,
+
+
+
+        "message": query,
+
+
+
+        "input": query,
+
+
+
+        "user": (user_id or "web-user")[:64],
+
+
+
+        "history": history or [],
+
+
+
+    }
+
+
+
+    async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client:
+
+
+
+        resp = await client.post(url, headers=headers, json=payload)
+
+
+
+    if resp.status_code not in (200, 201):
+
+
+
+        raise RuntimeError(f"n8n Webhook HTTP {resp.status_code}: {resp.text[:300]}")
+
+
+
+    ctype = resp.headers.get("content-type", "")
+
+
+
+    if "application/json" in ctype:
+
+
+
+        try:
+
+
+
+            answer = _extract_text_from_json(resp.json())
+
+
+
+        except Exception:
+
+
+
+            answer = resp.text
+
+
+
+    else:
+
+
+
+        answer = resp.text
+
+
+
+    return (answer or "").strip(), None
+
+async def _call_coze_agent(endpoint: str, api_key: str, query: str,
+
+
+
+                           user_id: str, ext_conv_id: str | None) -> tuple:
+
+
+
+    """调用扣子(Coze) v3 对话 API：create → 轮询 retrieve → 拉取 message list。
+
+
+
+
+
+
+
+    约定 agent_endpoint 存放 bot_id(纯 id)，或形如
+
+
+
+    https://api.coze.cn/v3/chat?bot_id=xxx 的完整地址。默认基址 api.coze.cn。
+
+
+
+    返回 (answer, conversation_id)。
+
+
+
+    """
+
+
+
+    ep = (endpoint or "").strip()
+
+
+
+    from urllib.parse import urlparse, parse_qs
+
+
+
+    bot_id = None
+
+
+
+    base = "https://api.coze.cn"
+
+
+
+    if ep.startswith("http"):
+
+
+
+        pr = urlparse(ep)
+
+
+
+        base = f"{pr.scheme}://{pr.netloc}"
+
+
+
+        qs = parse_qs(pr.query)
+
+
+
+        bot_id = (qs.get("bot_id") or [None])[0]
+
+
+
+        if not bot_id and "/bot/" in pr.path:
+
+
+
+            bot_id = pr.path.rstrip("/").split("/bot/")[-1]
+
+
+
+    else:
+
+
+
+        bot_id = ep  # 纯 bot_id
+
+
+
+    if not bot_id:
+
+
+
+        raise RuntimeError("Coze 接入缺少 bot_id(请在 Endpoint 填入 bot_id 或带 ?bot_id= 的地址)")
+
+
+
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+
+
+
+    create_url = f"{base}/v3/chat"
+
+
+
+    if ext_conv_id:
+
+
+
+        create_url += f"?conversation_id={ext_conv_id}"
+
+
+
+    payload = {
+
+
+
+        "bot_id": bot_id,
+
+
+
+        "user_id": (user_id or "web-user")[:64],
+
+
+
+        "stream": False,
+
+
+
+        "auto_save_history": True,
+
+
+
+        "additional_messages": [
+
+
+
+            {"role": "user", "content": query, "content_type": "text"}
+
+
+
+        ],
+
+
+
+    }
+
+
+
+    async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(create_url), trust_env=False) as client:
+
+
+
+        r = await client.post(create_url, headers=headers, json=payload)
+
+
+
+        if r.status_code != 200:
+
+
+
+            raise RuntimeError(f"Coze create HTTP {r.status_code}: {r.text[:300]}")
+
+
+
+        cd = r.json()
+
+
+
+        if cd.get("code") not in (0, None):
+
+
+
+            raise RuntimeError(f"Coze 错误: {cd.get('msg') or cd}")
+
+
+
+        chat = cd.get("data") or {}
+
+
+
+        chat_id = chat.get("id")
+
+
+
+        conv_id2 = chat.get("conversation_id") or ext_conv_id
+
+
+
+        if not chat_id or not conv_id2:
+
+
+
+            raise RuntimeError(f"Coze 未返回 chat_id/conversation_id: {str(cd)[:200]}")
+
+
+
+        # 轮询直到完成(最多约 3 分钟)
+
+
+
+        status = chat.get("status")
+
+
+
+        for _ in range(60):
+
+
+
+            if status == "completed":
+
+
+
+                break
+
+
+
+            if status in ("failed", "requires_action", "canceled"):
+
+
+
+                raise RuntimeError(f"Coze 对话状态 {status}")
+
+
+
+            await asyncio.sleep(3)
+
+
+
+            rr = await client.get(
+
+
+
+                f"{base}/v3/chat/retrieve?chat_id={chat_id}&conversation_id={conv_id2}",
+
+
+
+                headers=headers,
+
+
+
+            )
+
+
+
+            if rr.status_code == 200:
+
+
+
+                status = ((rr.json() or {}).get("data") or {}).get("status")
+
+
+
+        # 拉取消息列表，取 type=answer 的内容
+
+
+
+        rm = await client.get(
+
+
+
+            f"{base}/v3/chat/message/list?chat_id={chat_id}&conversation_id={conv_id2}",
+
+
+
+            headers=headers,
+
+
+
+        )
+
+
+
+        answer = ""
+
+
+
+        if rm.status_code == 200:
+
+
+
+            for m in ((rm.json() or {}).get("data") or []):
+
+
+
+                if m.get("type") == "answer" and m.get("role") == "assistant":
+
+
+
+                    answer = m.get("content", "") or answer
+
+
+
+    return (answer or "").strip(), conv_id2
+
+async def _call_external_agent(db, agent: dict, query: str, user_id: str,
+
+
+
+                               ext_conversation_id: str | None,
+
+
+
+                               history: list) -> tuple:
+
+
+
+    """按平台分发到对应外部 Agent 远程调用。返回 (answer, new_ext_conversation_id)。"""
+
+
+
+    platform = (agent.get("platform") or "dify").lower()
+
+
+
+    endpoint = agent.get("agent_endpoint")
+
+
+
+    _row = db.query(MarketApp).filter(MarketApp.id == agent.get("id")).first()
+    _raw = _row.agent_api_key if _row else None
+    try:
+        api_key = decrypt_secret(_raw) or ""
+    except Exception as _e:
+        api_key = ""
+        logging.error("[credential] 解密 Agent 凭据失败 id=%s trace=%s: %s" % (agent.get("id"), get_trace_id(), _e))
+
+
+
+    if not endpoint:
+
+
+
+        raise RuntimeError("该 Agent 未配置远程调用地址(agent_endpoint)")
+
+
+
+    if platform == "dify":
+
+
+
+        return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id)
+
+
+
+    if platform == "n8n":
+
+
+
+        return await _call_n8n_agent(endpoint, api_key, query, user_id, history)
+
+
+
+    if platform == "coze":
+
+
+
+        return await _call_coze_agent(endpoint, api_key, query, user_id, ext_conversation_id)
+
+
+
+    # 未知平台：默认按 Dify 协议尝试
+
+
+
+    return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id)

backend/python/app/api/chat_knowledge.py

@@ -0,0 +1,319 @@
+# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).
+# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from sqlalchemy.orm import Session
+from sqlalchemy import and_, desc, asc, func
+from pydantic import BaseModel, Field
+from typing import Optional
+import uuid
+from datetime import datetime, timezone, timedelta
+import re
+import json
+import asyncio
+import math
+import httpx
+import os
+import logging
+from ..database import get_db, SessionLocal
+from ..models import (
+
+
+
+    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,
+
+
+
+    KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill,
+
+
+
+    EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation,
+
+
+
+    MCPServer,
+
+
+
+    gen_uuid
+
+
+
+)
+from ..auth import get_current_user
+from ..platform_knowledge import get_platform_knowledge
+from ..mcp_client import list_mcp_tools, call_mcp_tool
+from ..middleware.log_service import LogService
+from ..capability_registry import resolve_auto_model
+from app.api.chat_utils import (cst_now)
+
+def _build_knowledge_graph(db: Session, kb: KnowledgeBase) -> dict:
+
+
+
+    """从知识库文档构建知识图谱节点和边，返回 graph_data。"""
+
+
+
+    documents = db.query(KnowledgeDocument).filter(
+
+
+
+        and_(
+
+
+
+            KnowledgeDocument.knowledge_base_id == kb.id,
+
+
+
+            KnowledgeDocument.status == "completed",
+
+
+
+        )
+
+
+
+    ).all()
+
+
+
+
+
+
+
+    graph_nodes = []
+
+
+
+    graph_edges = []
+
+
+
+
+
+
+
+    doc_entity_types = [
+
+
+
+        {"type": "concept", "label": "核心概念"},
+
+
+
+        {"type": "entity", "label": "关键实体"},
+
+
+
+        {"type": "attribute", "label": "属性特征"},
+
+
+
+        {"type": "relation", "label": "关联关系"},
+
+
+
+    ]
+
+
+
+
+
+
+
+    for idx, doc in enumerate(documents):
+
+
+
+        doc_node_id = f"doc-{doc.id[:8]}"
+
+
+
+        graph_nodes.append({
+
+
+
+            "id": doc_node_id, "label": doc.filename, "type": "document", "size": 18, "group": "document",
+
+
+
+        })
+
+
+
+
+
+
+
+        filename_base = doc.filename.rsplit(".", 1)[0]
+
+
+
+        words = filename_base.replace("_", " ").replace("-", " ").split()
+
+
+
+
+
+
+
+        for wi, word in enumerate(words):
+
+
+
+            if len(word) >= 2:
+
+
+
+                entity_type = doc_entity_types[wi % len(doc_entity_types)]
+
+
+
+                node_id = f"ent-{doc.id[:8]}-{wi}"
+
+
+
+                graph_nodes.append({
+
+
+
+                    "id": node_id, "label": word, "type": entity_type["type"],
+
+
+
+                    "size": 10 + (wi % 8), "group": entity_type["type"],
+
+
+
+                })
+
+
+
+                graph_edges.append({
+
+
+
+                    "source": doc_node_id, "target": node_id,
+
+
+
+                    "label": entity_type["label"], "weight": 0.5 + (0.5 / (wi + 1)),
+
+
+
+                })
+
+
+
+
+
+
+
+    for i in range(len(documents)):
+
+
+
+        for j in range(i + 1, len(documents)):
+
+
+
+            fi = documents[i].filename.lower().rsplit(".", 1)[0]
+
+
+
+            fj = documents[j].filename.lower().rsplit(".", 1)[0]
+
+
+
+            words_i = set(fi.replace("_", " ").replace("-", " ").split())
+
+
+
+            words_j = set(fj.replace("_", " ").replace("-", " ").split())
+
+
+
+            common = words_i & words_j
+
+
+
+            if common and len(common) > 0:
+
+
+
+                graph_edges.append({
+
+
+
+                    "source": f"doc-{documents[i].id[:8]}",
+
+
+
+                    "target": f"doc-{documents[j].id[:8]}",
+
+
+
+                    "label": f"共有关键词: {', '.join(list(common)[:3])}",
+
+
+
+                    "weight": min(0.9, 0.3 * len(common)),
+
+
+
+                })
+
+
+
+
+
+
+
+    graph_data = {
+
+
+
+        "nodes": graph_nodes,
+
+
+
+        "edges": graph_edges,
+
+
+
+        "stats": {
+
+
+
+            "total_nodes": len(graph_nodes), "total_edges": len(graph_edges),
+
+
+
+            "document_count": len(documents),
+
+
+
+        },
+
+
+
+    }
+
+
+
+
+
+
+
+    kb.knowledge_graph_count = len(graph_nodes)
+
+
+
+    kb.updated_at = cst_now()
+
+
+
+    return graph_data

backend/python/app/api/chat_mcp.py

@@ -0,0 +1,240 @@
+# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).
+# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from sqlalchemy.orm import Session
+from sqlalchemy import and_, desc, asc, func
+from pydantic import BaseModel, Field
+from typing import Optional
+import uuid
+from datetime import datetime, timezone, timedelta
+import re
+import json
+import asyncio
+import math
+import httpx
+import os
+import logging
+from ..database import get_db, SessionLocal
+from ..models import (
+
+
+
+    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,
+
+
+
+    KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill,
+
+
+
+    EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation,
+
+
+
+    MCPServer,
+
+
+
+    gen_uuid
+
+
+
+)
+from ..auth import get_current_user
+from ..platform_knowledge import get_platform_knowledge
+from ..mcp_client import list_mcp_tools, call_mcp_tool
+from ..middleware.log_service import LogService
+from ..capability_registry import resolve_auto_model
+
+def _resolve_market_mcp(db: Session, mcp_ids: list, user_id: str) -> list:
+
+
+
+    """
+
+
+
+    加载用户选择的 MCP 应用配置（MarketApp.type == 'mcp'）。
+
+
+
+    返回: [{id, name, transport, endpoint, description}]
+
+
+
+    """
+
+
+
+    if not mcp_ids:
+
+
+
+        return []
+
+
+
+
+
+
+
+    mcps = db.query(MarketApp).filter(
+
+
+
+        MarketApp.id.in_(mcp_ids),
+
+
+
+        MarketApp.type == "mcp"
+
+
+
+    ).all()
+
+
+
+
+
+
+
+    results = []
+
+
+
+    for m in mcps:
+
+
+
+        config = {}
+
+
+
+        try:
+
+
+
+            if m.skill_content:
+
+
+
+                config = json.loads(m.skill_content)
+
+
+
+        except Exception:
+
+
+
+            config = {}
+
+
+
+        results.append({
+
+
+
+            "id": m.id,
+
+
+
+            "name": m.name,
+
+
+
+            "transport": m.mcp_transport or config.get("transport", "stdio"),
+
+
+
+            "endpoint": m.agent_endpoint or config.get("endpoint", "") or config.get("url", ""),
+
+
+
+            "description": m.description or config.get("description", ""),
+
+
+
+        })
+
+
+
+
+
+
+
+    return results
+
+def _resolve_active_mcp_servers(db: Session, mcp_ids: list) -> list:
+
+    """
+
+    解析用户选中的 MCP 服务连接信息（来自 mcp_servers 表，import 与后台添加通用）。
+
+    返回: [{id, name, description, transport, endpoint, url, headers}]
+
+    """
+
+    if not mcp_ids:
+
+        return []
+
+    servers = db.query(MCPServer).filter(
+
+        MCPServer.id.in_(mcp_ids),
+
+        MCPServer.is_enabled == True,
+
+    ).all()
+
+    out = []
+
+    for s in servers:
+
+        try:
+
+            headers = json.loads(s.headers) if isinstance(s.headers, str) else (s.headers or {})
+
+        except Exception:
+
+            headers = {}
+
+        out.append({
+
+            "id": s.id,
+
+            "name": s.name,
+
+            "description": s.description or "",
+
+            "transport": s.transport_type or "streamable_http",
+
+            "endpoint": s.url or "",
+
+            "url": s.url or "",
+
+            "headers": headers,
+
+        })
+
+    return out
+
+def _list_enabled_mcp_servers(db: Session) -> list:
+    """返回平台所有已启用(is_enabled=True)的 MCP 服务清单，供 AI 准确回答"平台有哪些 MCP"。"""
+    servers = db.query(MCPServer).filter(MCPServer.is_enabled == True).all()
+    out = []
+    for s in servers:
+        try:
+            headers = json.loads(s.headers) if isinstance(s.headers, str) else (s.headers or {})
+        except Exception:
+            headers = {}
+        out.append({
+            "id": s.id,
+            "name": s.name,
+            "description": s.description or "",
+            "transport": s.transport_type or "streamable_http",
+            "endpoint": s.url or "",
+            "url": s.url or "",
+            "headers": headers,
+        })
+    return out

backend/python/app/api/chat_utils.py

@@ -0,0 +1,378 @@
+# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig).
+# Source of truth: scripts/phase1_extract.py. Do not hand-merge back.
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from sqlalchemy.orm import Session
+from sqlalchemy import and_, desc, asc, func
+from pydantic import BaseModel, Field
+from typing import Optional
+import uuid
+from datetime import datetime, timezone, timedelta
+import re
+import json
+import asyncio
+import math
+import httpx
+import os
+import logging
+from ..database import get_db, SessionLocal
+from ..models import (
+
+
+
+    User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog,
+
+
+
+    KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill,
+
+
+
+    EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation,
+
+
+
+    MCPServer,
+
+
+
+    gen_uuid
+
+
+
+)
+from ..auth import get_current_user
+from ..platform_knowledge import get_platform_knowledge
+from ..mcp_client import list_mcp_tools, call_mcp_tool
+from ..middleware.log_service import LogService
+from ..capability_registry import resolve_auto_model
+
+logger = logging.getLogger(__name__)
+
+CST_TZ = timezone(timedelta(hours=8))
+
+def cst_now():
+
+
+
+    """Get current time in CST timezone"""
+
+
+
+    return datetime.now(CST_TZ)
+
+def _html_escape(s: str) -> str:
+
+
+
+    return (str(s).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+
+            .replace('"', "&quot;").replace("'", "&#39;"))
+
+def _outbound_proxy() -> str:
+
+
+
+    """出网代理地址(兜底)。默认经 172.18.0.1:7890 智能路由代理；
+
+
+
+    运行时若未注入代理环境变量则回退到该地址。"""
+
+
+
+    import os
+
+
+
+    return os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY") or "http://172.18.0.1:7890"
+
+_HK_PROXY = "http://qyyfw:XHR521yhf@103.242.14.208:11495"
+
+def _proxy_for_url(url: str) -> str | None:
+
+
+
+    """按目标域名选择是否走代理。
+
+
+
+
+
+
+
+    2026-07-15 调整：AGNES(apihub.agnes-ai.com / platform-outputs.agnes-ai.space)
+
+
+
+    经用户实测走香港代理 103.242.14.208:11495 延迟更低、更稳定，故改为走 HK 代理
+
+
+
+    (带鉴权)，不再直连，也不经智能路由代理(172.18.0.1:7890 对 AGNES 返 502)。
+
+
+
+    其余国内/平台域名直连，真正的境外域名走 _outbound_proxy()。
+
+
+
+
+
+
+
+    注意：httpx 默认 trust_env=True 会忽略 proxy 参数而改用环境变量 HTTP_PROXY，
+
+
+
+    故调用方必须同时传 trust_env=False 才能真正走此处指定的代理。
+
+
+
+    """
+
+
+
+    import os
+
+
+
+    from urllib.parse import urlparse
+
+
+
+    host = (urlparse(url).hostname or "").lower()
+
+
+
+    if "agnes-ai.com" in host or "agnes-ai.space" in host:
+
+
+
+        return _HK_PROXY
+
+
+
+    # 平台自有域名(含自建 Dify/n8n 等)必须直连，不经出网代理。
+
+
+
+    if "qihuyun.com" in host:
+
+
+
+        return None
+
+
+
+    # 尊重 NO_PROXY：国内域名(硅基流动/DeepSeek/小米MoE 等)直连更稳定，
+
+
+
+    # 避免经易波动的智能路由代理 172.18.0.1:7890(对国内域名偶发连接重置)。
+
+
+
+    _no_proxy = [h.strip().lower() for h in os.environ.get("NO_PROXY", "").split(",") if h.strip()]
+
+
+
+    for _np in _no_proxy:
+
+
+
+        if host == _np or host.endswith("." + _np) or _np in host:
+
+
+
+            return None
+
+
+
+    return _outbound_proxy()
+
+def _extract_text_from_json(obj) -> str:
+
+
+
+    """从 n8n / 通用 webhook 的任意 JSON 响应中尽力抽取文本回答。"""
+
+
+
+    if obj is None:
+
+
+
+        return ""
+
+
+
+    if isinstance(obj, str):
+
+
+
+        return obj
+
+
+
+    if isinstance(obj, (int, float, bool)):
+
+
+
+        return str(obj)
+
+
+
+    if isinstance(obj, list):
+
+
+
+        # 取第一个可抽取到文本的元素
+
+
+
+        for it in obj:
+
+
+
+            t = _extract_text_from_json(it)
+
+
+
+            if t:
+
+
+
+                return t
+
+
+
+        return ""
+
+
+
+    if isinstance(obj, dict):
+
+
+
+        # 常见回答字段优先级
+
+
+
+        for k in ("output", "text", "answer", "response", "reply",
+
+
+
+                  "message", "content", "result", "data"):
+
+
+
+            if k in obj:
+
+
+
+                t = _extract_text_from_json(obj[k])
+
+
+
+                if t:
+
+
+
+                    return t
+
+
+
+        # 兜底：整体序列化
+
+
+
+        try:
+
+
+
+            return json.dumps(obj, ensure_ascii=False)
+
+
+
+        except Exception:
+
+
+
+            return str(obj)
+
+
+
+    return str(obj)
+
+class CreateConversationRequest(BaseModel):
+
+
+
+    title: str = Field(default="新对话")
+
+
+
+    model_id: Optional[str] = None
+
+class UpdateConversationRequest(BaseModel):
+
+
+
+    title: Optional[str] = None
+
+
+
+    status: Optional[str] = None  # active, archived
+
+
+
+    pinned: Optional[bool] = None
+
+
+
+    agent_ids: Optional[list] = None
+
+
+
+    skill_ids: Optional[list] = None
+
+
+
+    knowledge_base_ids: Optional[list] = None
+
+class SendMessageRequest(BaseModel):
+
+
+
+    content: str
+
+
+
+    model_id: Optional[str] = None
+
+
+
+    skill_ids: Optional[list] = None
+
+
+
+    agent_ids: Optional[list] = None
+
+
+
+    knowledge_base_ids: Optional[list] = None
+
+
+
+    mcp_ids: Optional[list] = None
+
+
+
+    attachments: Optional[list] = None
+
+
+
+    use_tools: Optional[bool] = False

backend/python/app/api/im_llm.py

@@ -25,6 +25,7 @@ from sqlalchemy.orm import Session
 
 from ..capability_registry import resolve_auto_model
 from ..models import AIModel, ModelProvider, TokenUsageLog, gen_uuid
+from app.core.observability import get_trace_id
 
 logger = logging.getLogger(__name__)
 
@@ -95,6 +96,7 @@ def _log_media_usage(db: Session, user_id: str, ai_model, media_type: str, promp
             user_id=user_id,
             model_id=ai_model.id,
             model_name=ai_model.name,
+            trace_id=get_trace_id(),
             prompt_tokens=est,
             completion_tokens=0,
             total_tokens=est,

backend/python/app/api/market.py

@@ -1,3 +1,4 @@
+from app.core.credential import encrypt_secret
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 from sqlalchemy import and_, desc, asc, func
@@ -557,7 +558,7 @@ def upload_app(
         is_member_free=req.is_member_free if req.is_member_free is not None else False,
         join_promotion=req.join_promotion if req.join_promotion is not None else False,
         agent_endpoint=req.agent_endpoint,
-        agent_api_key=req.api_key,
+        agent_api_key=encrypt_secret(req.api_key),
         status="published",  # 用户创建的应用直接发布
         created_at=datetime.now(),
         updated_at=datetime.now(),

backend/python/app/config.py

@@ -5,6 +5,7 @@ import os
 import logging
 from pathlib import Path
 from logging.handlers import TimedRotatingFileHandler
+from app.core.observability import TraceIdFilter, JsonFormatter
 from dotenv import load_dotenv
 
 # 加载 .env 文件（Docker 环境跳过，直接使用容器环境变量）
@@ -80,8 +81,12 @@ SLOW_REQUEST_THRESHOLD = float(os.getenv("SLOW_REQUEST_THRESHOLD", "3.0"))
 
 def setup_file_logging():
     """配置应用日志按天轮转输出到文件"""
+    use_json = os.getenv("AIFLOW_JSON_LOGS", "0") == "1"
+    if use_json:
+        log_format = JsonFormatter()
+    else:
         log_format = logging.Formatter(
-        "%(asctime)s [%(levelname)s] %(name)s - %(message)s",
+            "%(asctime)s [%(levelname)s] %(name)s [tid:%(trace_id)s] - %(message)s",
             datefmt="%Y-%m-%d %H:%M:%S",
         )
 
@@ -123,10 +128,12 @@ def setup_file_logging():
     root_logger.setLevel(logging.INFO)
     root_logger.addHandler(app_handler)
     root_logger.addHandler(error_handler)
+    root_logger.addFilter(TraceIdFilter())
 
     # 配置访问日志专用logger
     access_logger = logging.getLogger("aiflow.access")
     access_logger.addHandler(access_handler)
     access_logger.propagate = False
+    access_logger.addFilter(TraceIdFilter())
 
     return access_logger

backend/python/app/core/__init__.py

@@ -0,0 +1 @@
+# app/core package - 统一可观测基座 (Phase 2.1)

backend/python/app/core/credential.py

@@ -0,0 +1,105 @@
+# -*- coding: utf-8 -*-
+"""Phase 3.2 - Agent 凭据加解密 (Fernet 对称加密)。
+
+主密钥获取策略 (防竞态):
+- 优先环境变量 AGENT_MASTER_KEY。
+- 否则读写 /app/secrets/agent_master_key:
+  * 部署阶段由 patch 显式生成并落盘 (原子写), 容器重启/提交后稳定存在。
+  * 运行时只读, 文件存在即读取, 绝不重新生成 —— 避免多进程/多请求并发
+    重建密钥导致新旧密文无法互解。
+  * 仅当文件确实不存在时, 才在文件锁保护下生成 (双重检查, 原子替换)。
+- 存储格式: 密文以 "enc:" 前缀标识; 非 "enc:" 视为旧明文 (向后兼容, 解密原样返回)。
+"""
+import os
+import logging
+import fcntl
+from cryptography.fernet import Fernet
+
+logger = logging.getLogger(__name__)
+
+_MASTER_KEY_FILE = "/app/secrets/agent_master_key"
+_fernet_instance = None
+
+
+def _generate_and_persist_key() -> bytes:
+    key = Fernet.generate_key()
+    os.makedirs(os.path.dirname(_MASTER_KEY_FILE), exist_ok=True)
+    tmp = _MASTER_KEY_FILE + ".tmp"
+    with open(tmp, "w") as f:
+        f.write(key.decode())
+        f.flush()
+        os.fsync(f.fileno())
+    try:
+        os.chmod(tmp, 0o600)
+    except Exception:
+        pass
+    os.replace(tmp, _MASTER_KEY_FILE)  # 原子替换, 避免读到半截文件
+    return key
+
+
+def get_master_key() -> bytes:
+    env_key = os.environ.get("AGENT_MASTER_KEY")
+    if env_key:
+        return env_key.encode() if isinstance(env_key, str) else env_key
+    # 运行时只读: 文件已存在则直接读取 (部署阶段已生成)
+    if os.path.exists(_MASTER_KEY_FILE):
+        try:
+            with open(_MASTER_KEY_FILE, "r") as f:
+                k = f.read().strip()
+                if k:
+                    return k.encode()
+        except Exception as e:
+            logger.warning("[credential] 读取主密钥失败: %s" % e)
+    # 文件缺失: 加锁串行生成, 防多进程并发重建
+    lockf = open(_MASTER_KEY_FILE + ".lock", "w")
+    try:
+        fcntl.flock(lockf, fcntl.LOCK_EX)
+        if os.path.exists(_MASTER_KEY_FILE):  # 双重检查
+            with open(_MASTER_KEY_FILE, "r") as f:
+                k = f.read().strip()
+                if k:
+                    return k.encode()
+        return _generate_and_persist_key()
+    finally:
+        fcntl.flock(lockf, fcntl.LOCK_UN)
+        try:
+            lockf.close()
+        except Exception:
+            pass
+
+
+def _fernet() -> Fernet:
+    global _fernet_instance
+    if _fernet_instance is None:
+        _fernet_instance = Fernet(get_master_key())
+    return _fernet_instance
+
+
+def encrypt_secret(plain):
+    if plain is None or plain == "":
+        return None
+    if isinstance(plain, str) and plain.startswith("enc:"):
+        return plain  # 已是密文, 幂等
+    try:
+        return "enc:" + _fernet().encrypt(plain.encode()).decode()
+    except Exception as e:
+        logger.error("[credential] 加密失败: %s" % e)
+        return None
+
+
+def decrypt_secret(cipher):
+    if cipher is None or cipher == "":
+        return None
+    if isinstance(cipher, str) and cipher.startswith("enc:"):
+        try:
+            return _fernet().decrypt(cipher[4:].encode()).decode()
+        except Exception as e:
+            logger.error("[credential] 解密失败(密钥不匹配或数据损坏): %s" % e)
+            return None
+    return cipher  # 旧明文, 向后兼容
+
+
+def mask_secret(plain):
+    if not plain:
+        return ""
+    return ("*" * 8) + plain[-4:]

backend/python/app/core/observability.py

@@ -0,0 +1,174 @@
+"""
+Qihu-AIFlow 统一可观测基座 (Phase 2.1)
+=====================================
+提供贯穿一次请求的全局链路 ID（trace_id）与结构化日志能力，
+为后续重构（上下文引擎、媒体/LLM 成本归因）建立回归基线。
+
+设计原则（Strangler Fig 增量）：
+- 纯增量埋点，不改任何既有接口的返回结构。
+- 向后兼容现有 `logging.getLogger(__name__)` 用法：trace_id 由 root logger 的
+  TraceIdFilter 自动注入，调用方无需改动。
+- 不引入第三方依赖（structlog 未安装），自带可选的 JSON formatter。
+
+使用方式：
+    from app.core.observability import set_trace_id, get_trace_id, span
+
+    # 在中间件里为每个请求注入：
+    set_trace_id(request_id)
+
+    # 在关键调用处打计时 span（自动带 trace_id）：
+    async with span("chat.send_message", conv_id=conv_id):
+        return await _send_message_core(...)
+"""
+import logging
+import time
+import uuid
+import json
+from contextvars import ContextVar
+
+# ---------------------------------------------------------------------------
+# trace_id：贯穿一次请求（含其内所有子调用/协程）的全局链路 ID
+# ---------------------------------------------------------------------------
+_trace_id_var: ContextVar[str] = ContextVar("aiflow_trace_id", default="-")
+
+
+def new_trace_id() -> str:
+    """生成一个崭新的链路 ID（32 位 hex）。"""
+    return uuid.uuid4().hex
+
+
+def set_trace_id(tid: str | None = None) -> str:
+    """设置当前上下文的 trace_id；不传则自动生成。返回最终使用的 ID。"""
+    tid = tid or new_trace_id()
+    _trace_id_var.set(tid)
+    return tid
+
+
+def get_trace_id() -> str:
+    """读取当前上下文的 trace_id；未设置时返回占位 '-'。"""
+    return _trace_id_var.get()
+
+
+class TraceIdFilter(logging.Filter):
+    """为每一条日志记录注入 trace_id 属性（不存在则写为 '-'）。
+
+    挂载到 root logger 后，所有 `getLogger(__name__)` 的子 logger 都会继承，
+    从而让全平台既有日志自动带上链路 ID，零侵入。
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        if not hasattr(record, "trace_id"):
+            try:
+                record.trace_id = get_trace_id()
+            except Exception:
+                record.trace_id = "-"
+        return True
+
+
+# 标准 logging 字段，JSON 序列化时需排除，避免与自定义 extra 冲突
+_STD_ATTRS = set(logging.makeLogRecord({}).__dict__.keys()) | {
+    "message", "asctime", "msg", "trace_id", "exc_info", "stack_info", "args",
+}
+
+
+class JsonFormatter(logging.Formatter):
+    """结构化 JSON 日志格式（通过环境变量 AIFLOW_JSON_LOGS=1 启用）。
+
+    默认不启用，避免破坏现有日志采集/人工查阅习惯。
+    """
+
+    def format(self, record: logging.LogRecord) -> str:
+        data = {
+            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
+            "level": record.levelname,
+            "logger": record.name,
+            "trace_id": getattr(record, "trace_id", "-"),
+            "msg": record.getMessage(),
+        }
+        for k, v in record.__dict__.items():
+            if k not in _STD_ATTRS:
+                try:
+                    json.dumps(v, ensure_ascii=False)
+                    data[k] = v
+                except (TypeError, ValueError):
+                    data[k] = str(v)
+        if record.exc_info:
+            data["exc"] = self.formatException(record.exc_info)
+        return json.dumps(data, ensure_ascii=False)
+
+
+def get_logger(name: str | None = None) -> logging.Logger:
+    """获取 logger（向后兼容现有用法）。trace_id 由 root filter 自动注入。"""
+    return logging.getLogger(name)
+
+
+# ---------------------------------------------------------------------------
+# span：结构化计时上下文管理器（耗时 + trace_id + 自定义字段）
+# ---------------------------------------------------------------------------
+class _Span:
+    """同步/异步通用的计时 span。退出时记录 elapsed_ms 与状态。"""
+
+    def __init__(self, name: str, **fields):
+        self.name = name
+        self.fields = fields
+        self._logger = logging.getLogger("aiflow.span")
+        self._start = None
+        self.elapsed_ms = 0.0
+
+    def _enter(self):
+        self._start = time.perf_counter()
+        self._logger.info(
+            f"SPAN_START {self.name}",
+            extra={"span": self.name, "event": "start", "trace_id": get_trace_id(), **self.fields},
+        )
+        return self
+
+    def _exit(self, exc_type, exc, tb):
+        self.elapsed_ms = (time.perf_counter() - self._start) * 1000
+        status = "error" if exc_type else "ok"
+        self._logger.info(
+            f"SPAN_END {self.name} {self.elapsed_ms:.1f}ms",
+            extra={
+                "span": self.name,
+                "event": "end",
+                "elapsed_ms": round(self.elapsed_ms, 1),
+                "status": status,
+                "trace_id": get_trace_id(),
+                **self.fields,
+            },
+        )
+        return False  # 不吞掉异常
+
+    # 同步协议
+    def __enter__(self):
+        return self._enter()
+
+    def __exit__(self, exc_type, exc, tb):
+        return self._exit(exc_type, exc, tb)
+
+    # 异步协议
+    async def __aenter__(self):
+        return self._enter()
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return self._exit(exc_type, exc, tb)
+
+
+def span(name: str, **fields):
+    """创建一个计时 span。可同步或异步 with。
+
+    async with span("chat.send_message", conv_id=cid):
+        ...
+    """
+    return _Span(name, **fields)
+
+
+__all__ = [
+    "new_trace_id",
+    "set_trace_id",
+    "get_trace_id",
+    "TraceIdFilter",
+    "JsonFormatter",
+    "get_logger",
+    "span",
+]

backend/python/app/core/orchestrator.py

@@ -0,0 +1,395 @@
+"""app/core/orchestrator.py — Phase 3.3 多 Agent 编排图引擎。
+
+设计目标：
+- 以轻量 DAG 描述多 Agent 协作：节点(llm/local/external/reduce) + 有向边(数据流)。
+- 拓扑分层后按层并行执行(asyncio.gather)，互相独立的节点真正并发。
+- 失败隔离：每个节点异常不影响其它分支；per-node on_error=halt|continue。
+- 节点执行复用既有原语(_build_system_prompt / _execute_skill_tool / _call_external_agent /
+  3.1 工具沙箱)，完全向后兼容——agent.orchestration 为 null 时 chat_service 走原逻辑。
+- 并发安全：每个节点用独立 SessionLocal 会话做 DB 读，避免共享请求会话跨协程竞态。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import Any, Optional
+
+import httpx
+
+logger = logging.getLogger("aiflow.orchestrator")
+
+VALID_NODE_TYPES = ("llm", "local", "external", "reduce")
+
+
+class OrchestrationError(ValueError):
+    """图结构非法或执行被 halt。"""
+
+
+# ---------------------------------------------------------------------------
+# 校验 + 拓扑
+# ---------------------------------------------------------------------------
+def validate_graph(graph: Any) -> None:
+    if not isinstance(graph, dict):
+        raise OrchestrationError("orchestration 必须为对象")
+    nodes = graph.get("nodes")
+    edges = graph.get("edges")
+    if not isinstance(nodes, list) or not nodes:
+        raise OrchestrationError("nodes 必须为非空数组")
+    if not isinstance(edges, list):
+        raise OrchestrationError("edges 必须为数组")
+
+    ids = set()
+    for i, n in enumerate(nodes):
+        if not isinstance(n, dict):
+            raise OrchestrationError(f"节点 #{i} 必须为对象")
+        nid = n.get("id")
+        if not nid or not isinstance(nid, str):
+            raise OrchestrationError(f"节点 #{i} 缺少字符串 id")
+        if nid in ids:
+            raise OrchestrationError(f"节点 id 重复: {nid}")
+        ids.add(nid)
+        ntype = n.get("type")
+        if ntype not in VALID_NODE_TYPES:
+            raise OrchestrationError(
+                f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(VALID_NODE_TYPES)})"
+            )
+        if ntype in ("local", "external") and not n.get("agent_id"):
+            raise OrchestrationError(f"节点 {nid} 类型 {ntype} 必须指定 agent_id")
+
+    for i, e in enumerate(edges):
+        if not isinstance(e, dict):
+            raise OrchestrationError(f"边 #{i} 必须为对象")
+        f = e.get("from")
+        t = e.get("to")
+        if f not in ids:
+            raise OrchestrationError(f"边 #{i} 的 from={f} 指向不存在的节点")
+        if t not in ids:
+            raise OrchestrationError(f"边 #{i} 的 to={t} 指向不存在的节点")
+        if f == t:
+            raise OrchestrationError(f"边 #{i} 存在自环: {f}")
+
+
+def topological_layers(graph: dict) -> list:
+    """Kahn 算法返回分层(每层内节点可并行)；含环则抛错。"""
+    nodes = graph["nodes"]
+    edges = graph.get("edges", [])
+    indeg = {n["id"]: 0 for n in nodes}
+    adj = {n["id"]: [] for n in nodes}
+    for e in edges:
+        f, t = e["from"], e["to"]
+        adj[f].append(t)
+        indeg[t] += 1
+    order = {n["id"]: i for i, n in enumerate(nodes)}
+    layers: list = []
+    remaining = set(indeg)
+    while remaining:
+        layer = sorted([nid for nid in remaining if indeg[nid] == 0], key=lambda x: order[x])
+        if not layer:
+            cycle = sorted(remaining, key=lambda x: order[x])
+            raise OrchestrationError(f"检测到环路，无法拓扑排序: {cycle}")
+        layers.append(layer)
+        for nid in layer:
+            remaining.discard(nid)
+            for nxt in adj[nid]:
+                indeg[nxt] -= 1
+    return layers
+
+
+def _node_by_id(graph: dict, nid: str) -> dict:
+    for n in graph["nodes"]:
+        if n["id"] == nid:
+            return n
+    return {}
+
+
+def _predecessors(graph: dict, nid: str) -> list:
+    return [e["from"] for e in graph.get("edges", []) if e["to"] == nid]
+
+
+def _successors(graph: dict, nid: str) -> list:
+    return [e["to"] for e in graph.get("edges", []) if e["from"] == nid]
+
+
+def _find_output_node(graph: dict) -> str:
+    if graph.get("output_node"):
+        return graph["output_node"]
+    sinks = [n["id"] for n in graph["nodes"] if not _successors(graph, n["id"])]
+    if len(sinks) == 1:
+        return sinks[0]
+    for s in sinks:
+        if _node_by_id(graph, s).get("type") == "reduce":
+            return s
+    if sinks:
+        return sinks[0]
+    return graph["nodes"][-1]["id"]
+
+
+def _assemble_node_input(query: str, upstream: str, is_entry: bool) -> str:
+    """入口节点直接吃用户原问题；下游节点综合原问题 + 上游输出。"""
+    if is_entry:
+        return query or ""
+    if upstream:
+        return f"原始用户问题:\n{query}\n\n上游节点输出:\n{upstream}"
+    return query or ""
+
+
+# ---------------------------------------------------------------------------
+# 模型解析 + LLM 调用原语
+# ---------------------------------------------------------------------------
+async def _resolve_model_and_endpoint(model_id, db):
+    from app.models import AIModel, ModelProvider
+    from ..capability_registry import resolve_auto_model
+
+    ai_model = None
+    if model_id and model_id != "auto":
+        ai_model = db.query(AIModel).filter(AIModel.id == model_id).first()
+    if not ai_model:
+        ai_model = resolve_auto_model("text", db)
+    if not ai_model:
+        raise OrchestrationError("无法解析到可用模型")
+    provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first()
+    api_key = ai_model.api_key or (provider.api_key if provider else None)
+    api_url = ai_model.endpoint
+    if not api_url and provider and provider.api_base_url:
+        api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions"
+    if not api_url or not api_key:
+        raise OrchestrationError(f"节点模型 {ai_model.name} 缺少 endpoint 或 api_key")
+    return ai_model, api_url, api_key
+
+
+async def _llm_complete(system_prompt: str, user_content: str, model_id, db, user_id) -> str:
+    from app.api.chat_utils import _proxy_for_url
+    from app.core.observability import span
+
+    ai_model, api_url, api_key = await _resolve_model_and_endpoint(model_id, db)
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_content},
+    ]
+    request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
+    async with span("orchestrator.llm", model=ai_model.id):
+        async with httpx.AsyncClient(
+            timeout=ai_model.timeout or 120,
+            proxy=_proxy_for_url(api_url),
+            trust_env=False,
+        ) as client:
+            resp = await client.post(
+                api_url,
+                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+                json=request_body,
+            )
+        if resp.status_code != 200:
+            raise OrchestrationError(f"LLM 调用失败 HTTP {resp.status_code}: {resp.text[:200]}")
+        data = resp.json()
+        return data["choices"][0]["message"].get("content") or ""
+
+
+# ---------------------------------------------------------------------------
+# local 节点：复用单 Agent system prompt + 受限 tool round + 3.1 沙箱
+# ---------------------------------------------------------------------------
+async def _run_local_agent_turn(agent: dict, node_input: str, db, user_id, conv_id) -> dict:
+    from app.api.chat_skill import _resolve_market_skills, _execute_skill_tool
+    from app.api.chat_context import _build_system_prompt, _retrieve_kb_context
+    from app.api.chat_service import _resolve_agent_tool_policy, _check_tool_policy
+    from app.api.chat_utils import _proxy_for_url
+
+    skills = _resolve_market_skills(db, agent.get("skill_ids") or [], user_id)
+    kb_context = _retrieve_kb_context(db, agent.get("knowledge_base_ids") or [], node_input)
+    system_prompt, function_definitions = _build_system_prompt(
+        skills=skills,
+        agents=[agent],
+        kb_context=kb_context,
+        user_content=node_input,
+        auto_discovered=False,
+    )
+    ai_model, api_url, api_key = await _resolve_model_and_endpoint(agent.get("model_id"), db)
+    tool_policy = _resolve_agent_tool_policy([agent])
+
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": node_input},
+    ]
+    called_skills: list = []
+    tool_call_count = 0
+    max_tool_rounds = 5
+    last_content = ""
+    async with httpx.AsyncClient(
+        timeout=ai_model.timeout or 120,
+        proxy=_proxy_for_url(api_url),
+        trust_env=False,
+    ) as client:
+        for round_num in range(max_tool_rounds):
+            request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
+            if function_definitions:
+                request_body["tools"] = [
+                    {"type": "function", "function": fd} for fd in function_definitions
+                ]
+                request_body["tool_choice"] = "auto"
+            resp = await client.post(
+                api_url,
+                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+                json=request_body,
+            )
+            if resp.status_code != 200:
+                raise OrchestrationError(f"local 节点 LLM 失败 HTTP {resp.status_code}: {resp.text[:200]}")
+            data = resp.json()
+            message = data["choices"][0]["message"]
+            tool_calls = message.get("tool_calls") or []
+            content = message.get("content") or ""
+            last_content = content
+            if not tool_calls or not function_definitions:
+                return {"text": content, "used_tools": called_skills}
+            messages.append({"role": "assistant", "content": content, "tool_calls": tool_calls})
+            round_tool_count = 0
+            for tc in tool_calls:
+                func_name = tc.get("function", {}).get("name", "")
+                check = _check_tool_policy(
+                    func_name, None, skills, tool_policy, tool_call_count, round_tool_count
+                )
+                if not check["allowed"]:
+                    messages.append({
+                        "role": "tool",
+                        "tool_call_id": tc.get("id", f"call_{round_num}"),
+                        "content": f"[工具调用已被策略拦截] {check.get('message', check.get('reason'))}",
+                    })
+                    tool_call_count += 1
+                    round_tool_count += 1
+                    continue
+                tool_call_count += 1
+                round_tool_count += 1
+                try:
+                    func_args = json.loads(tc.get("function", {}).get("arguments", "{}") or "{}")
+                except Exception:
+                    func_args = {}
+                res = _execute_skill_tool(db, func_name, func_args, skills, node_input)
+                called_skills.append(func_name)
+                messages.append({
+                    "role": "tool",
+                    "tool_call_id": tc.get("id", f"call_{round_num}"),
+                    "content": res,
+                })
+    return {"text": last_content, "used_tools": called_skills}
+
+
+# ---------------------------------------------------------------------------
+# 单节点执行（含独立 DB 会话，保证并发安全）
+# ---------------------------------------------------------------------------
+async def _run_node(node, graph, blackboard, query, user_id, conv_id, trace_id) -> dict:
+    from app.api.chat_skill import _resolve_market_agents
+    from app.database import SessionLocal
+
+    nid = node["id"]
+    ntype = node["type"]
+    preds = _predecessors(graph, nid)
+    pred_results = [blackboard[p] for p in preds]
+    is_entry = not preds
+
+    if pred_results:
+        parts = []
+        for b in pred_results:
+            label = f"节点 {b.get('node_id')}/{b.get('node_type')}"
+            if b.get("ok"):
+                parts.append(f"【{label}】\n{b.get('output', '')}")
+            else:
+                parts.append(f"【{label} 执行失败】\n错误: {b.get('error', '')}")
+        upstream = "\n\n".join(parts)
+    else:
+        upstream = ""
+
+    with SessionLocal() as db:
+        if ntype in ("llm", "reduce"):
+            if ntype == "llm":
+                sys_prompt = node.get("system_prompt") or (
+                    "你是一个处理节点，请根据上游输入完成指定的文本变换、摘要或路由判断，"
+                    "只输出处理后的结果，不要多余解释。"
+                )
+            else:
+                sys_prompt = node.get("system_prompt") or (
+                    "你是一个结果汇总器，请将多个子任务的输出综合成连贯、完整、面向用户的最终回答。"
+                )
+            uc = _assemble_node_input(query, upstream, is_entry)
+            text = await _llm_complete(sys_prompt, uc, node.get("model_id"), db, user_id)
+            return {"ok": True, "node_id": nid, "node_type": ntype, "output": text}
+
+        if ntype == "local":
+            agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
+            if not agents:
+                raise OrchestrationError(f"local 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
+            ni = _assemble_node_input(query, upstream, is_entry)
+            res = await _run_local_agent_turn(agents[0], ni, db, user_id, conv_id)
+            return {
+                "ok": True,
+                "node_id": nid,
+                "node_type": "local",
+                "output": res["text"],
+                "used_tools": res.get("used_tools"),
+            }
+
+        if ntype == "external":
+            from app.api.chat_agents import _call_external_agent
+
+            agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
+            if not agents:
+                raise OrchestrationError(f"external 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
+            ni = _assemble_node_input(query, upstream, is_entry)
+            answer, _ = await _call_external_agent(
+                db, agents[0], query=ni, user_id=user_id, ext_conversation_id=None, history=[]
+            )
+            return {"ok": True, "node_id": nid, "node_type": "external", "output": answer or ""}
+
+    raise OrchestrationError(f"未知节点类型 {ntype}")
+
+
+# ---------------------------------------------------------------------------
+# 入口
+# ---------------------------------------------------------------------------
+async def run_graph(graph, query, db, user_id, conv_id, trace_id=None):
+    """执行编排图，返回 (answer: str, trace: dict)。
+
+    - 拓扑分层后逐层 asyncio.gather 并行执行。
+    - 节点异常按 on_error=halt|continue 处理（默认 continue，失败时下游拿到错误作为输入）。
+    """
+    validate_graph(graph)
+    layers = topological_layers(graph)
+    blackboard: dict = {}
+    trace = {"nodes": [], "trace_id": trace_id, "layers": layers}
+
+    for layer in layers:
+        tasks = [
+            _run_node(_node_by_id(graph, nid), graph, blackboard, query, user_id, conv_id, trace_id)
+            for nid in layer
+        ]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        for nid, res in zip(layer, results):
+            node = _node_by_id(graph, nid)
+            if isinstance(res, Exception):
+                on_error = node.get("on_error", "continue")
+                err_msg = str(res)
+                blackboard[nid] = {
+                    "ok": False,
+                    "node_id": nid,
+                    "node_type": node["type"],
+                    "error": err_msg,
+                }
+                trace["nodes"].append({"id": nid, "type": node["type"], "ok": False, "error": err_msg})
+                logger.warning(f"[orchestrator] 节点 {nid} 失败 on_error={on_error}: {err_msg}")
+                if on_error == "halt":
+                    raise OrchestrationError(f"节点 {nid} 执行失败(halt): {err_msg}")
+            else:
+                blackboard[nid] = res
+                trace["nodes"].append({
+                    "id": nid,
+                    "type": res.get("node_type"),
+                    "ok": True,
+                    "used_tools": res.get("used_tools"),
+                })
+
+    output_node = _find_output_node(graph)
+    final = blackboard.get(output_node, {})
+    if final.get("ok"):
+        answer = final.get("output", "")
+    else:
+        answer = f"⚠️ 编排输出节点 {output_node} 执行失败: {final.get('error', '未知错误')}"
+    trace["output_node"] = output_node
+    return answer, trace

backend/python/app/middleware/log_service.py

@@ -57,7 +57,7 @@ class LogService:
                         app_id: str = None, app_name: str = None,
                         conversation_id: str = None, prompt_tokens: int = 0,
                         completion_tokens: int = 0, total_tokens: int = 0,
-                        cost: float = 0):
+                        cost: float = 0, trace_id: str = None):
         """记录Token消耗日志"""
         db = SessionLocal()
         try:
@@ -73,6 +73,7 @@ class LogService:
                 completion_tokens=completion_tokens,
                 total_tokens=total_tokens,
                 cost=cost,
+                trace_id=trace_id,
             )
             db.add(log)
             db.commit()

backend/python/app/models.py

@@ -5,6 +5,7 @@ AIFlow 数据库模型定义 - SQLAlchemy ORM
 import uuid
 from datetime import datetime, date
 from sqlalchemy import (
+    Text,
     Column, String, Integer, BigInteger, Float, Boolean,
     DateTime, Date, Text, Enum, ForeignKey, Index, JSON,
     DECIMAL, UniqueConstraint
@@ -299,7 +300,9 @@ class MarketApp(Base):
     platform = Column(String(20), default=None, comment="Agent 接入平台：dify/n8n/coze")
     capability = Column(String(20), default=None, comment="能力分类：text/thinking/vision/tool/image/audio/asr/ocr/embedding/rerank/video")
     agent_endpoint = Column(String(500), default=None)
-    agent_api_key = Column(String(500), default=None, comment="单个Agent接入的API Key")
+    agent_api_key = Column(Text, default=None, comment="Agent 接入 API Key(3.2 起 Fernet 加密存储)")
+    tool_policy = Column(JSON, default=None, comment="Agent 工具/MCP 权限沙箱策略(3.1)")
+    orchestration = Column(JSON, default=None, comment="Agent 多Agent编排图(3.3)")
     created_at = Column(DateTime, default=datetime.now)
     updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
 
@@ -537,6 +540,7 @@ class TokenUsageLog(Base):
     app_id = Column(String(36), default=None)
     app_name = Column(String(200), default=None)
     conversation_id = Column(String(36), default=None)
+    trace_id = Column(String(64), default=None)
     prompt_tokens = Column(Integer, default=0)
     completion_tokens = Column(Integer, default=0)
     total_tokens = Column(Integer, default=0)
@@ -1124,3 +1128,12 @@ class SmsVerificationCode(Base):
     used = Column(Boolean, default=False)
     created_at = Column(DateTime, default=datetime.now)
 
+
+
+class ConversationSummary(Base):
+    __tablename__ = "conversation_summaries"
+
+    conversation_id = Column(String(64), primary_key=True)
+    summary_text = Column(Text, nullable=False, default="")
+    token_count = Column(Integer, default=0)
+    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)

backend/python/conftest.py

@@ -0,0 +1,12 @@
+"""Pytest 根配置：保证测试运行时 backend 包根目录在 sys.path 中。
+
+运行方式（容器内，使用项目 venv）：
+    cd /app/backend/python
+    /app/backend/python/.venv/bin/python -m pytest
+"""
+import os
+import sys
+
+ROOT = os.path.dirname(os.path.abspath(__file__))
+if ROOT not in sys.path:
+    sys.path.insert(0, ROOT)

backend/python/main.py

@@ -17,7 +17,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from app.api import auth, chat, knowledge, market, token_usage, payment, profile, notifications, assets, evolution
 from app.casdoor_auth import router as casdoor_router
-from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin, tenant_admin, billing_admin, analytics_admin
+from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin, tenant_admin, billing_admin, analytics_admin, observability_admin
 from app.api.im_bot_user import router as im_bot_user_router
 from app.api.im_bot_platform import router as im_bot_platform_router
 from app.api.portal import user_portal
@@ -25,6 +25,7 @@ from app.api.admin.developer_tools_admin import router as developer_tools_router
 from app.api.admin.mcp_market_admin import router as mcp_market_router
 from app.api.admin import asset_admin, achievement_admin
 from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD
+from app.core.observability import set_trace_id
 from app.database import SessionLocal
 from app.models import OperationLog as OperationLogModel, TokenUsageLog as TokenUsageLogModel
 
@@ -41,6 +42,7 @@ class APITrackingMiddleware(BaseHTTPMiddleware):
 
     async def dispatch(self, request: Request, call_next):
         request_id = str(uuid.uuid4())
+        set_trace_id(request_id)
         start_time = time.time()
         path = request.url.path
         method = request.method
@@ -53,6 +55,7 @@ class APITrackingMiddleware(BaseHTTPMiddleware):
             response = await call_next(request)
             elapsed = time.time() - start_time
             status_code = response.status_code
+            response.headers["X-Trace-Id"] = request_id
 
             if not should_skip:
                 access_logger.info(
@@ -236,6 +239,7 @@ app.include_router(promotion_admin.router, prefix="/app/api/admin/promotion", ta
 app.include_router(tenant_admin.router, prefix="/app/api/admin/tenant", tags=["tenant"])
 app.include_router(billing_admin.router, prefix="/app/api/admin/billing", tags=["billing"])
 app.include_router(analytics_admin.router, prefix="/app/api/admin/analytics", tags=["analytics"])
+app.include_router(observability_admin.router, prefix="/app/api/admin/observability", tags=["后台-调用可观测"])
 app.include_router(user_portal.router, prefix="/app/api/portal", tags=["portal"])
 app.include_router(achievement_admin.router, prefix="/app/api/admin/achievement", tags=["achievement"])
 app.include_router(casdoor_router, prefix="/app/api/auth", tags=["Casdoor SSO"])

backend/python/pytest.ini

@@ -0,0 +1,11 @@
+[pytest]
+testpaths = tests
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+asyncio_mode = auto
+asyncio_default_fixture_loop_scope = function
+addopts = -q --strict-markers
+filterwarnings =
+    ignore::DeprecationWarning
+    ignore::PendingDeprecationWarning

backend/python/requirements-dev.txt

@@ -0,0 +1,4 @@
+# Phase 0 测试基建 — 开发/测试依赖（不进入生产 requirements.txt）
+pytest==8.3.4
+pytest-asyncio==0.24.0
+httpx==0.27.2

backend/python/tests/test_context_engine.py

@@ -0,0 +1,114 @@
+"""Phase 2 context-engine regression: _build_chat_history budget/layering/summary.
+
+Pure unit test with a fake DB session — no network, no real model. Guards the
+layered-memory + compression + isolation behaviour against regressions.
+"""
+from app.api.chat_context import _estimate_tokens, _build_chat_history
+
+
+class _Msg:
+    def __init__(self, role, content, cid="c1"):
+        self.role = role
+        self.content = content
+        self.conversation_id = cid
+
+
+class _Summary:
+    def __init__(self, text):
+        self.summary_text = text
+
+
+class _Query:
+    def __init__(self, rows):
+        self._rows = list(rows)
+
+    def filter(self, *a, **k):
+        return self
+
+    def order_by(self, *a, **k):
+        # test pre-sorts newest-first to match desc(created_at)
+        return self
+
+    def limit(self, n=None):
+        return self
+
+    def all(self):
+        return self._rows
+
+    def first(self):
+        return self._rows[0] if self._rows else None
+
+
+class _DB:
+    def __init__(self, messages, summary=None):
+        self._messages = messages
+        self._summary = summary
+
+    def query(self, model):
+        if getattr(model, "__name__", "") == "ConversationSummary":
+            return _Query([self._summary] if self._summary else [])
+        return _Query(self._messages)
+
+
+def _newest_first(msgs):
+    # _build_chat_history expects desc(created_at) -> newest first.
+    # Test authors write the list already newest-first, so return as-is.
+    return list(msgs)
+
+
+def test_estimate_tokens():
+    assert _estimate_tokens("") == 0
+    assert _estimate_tokens(None) == 0
+    assert _estimate_tokens("abcd") == 2  # 4//2
+    assert _estimate_tokens("a" * 5) == 2  # max(1, 5//2)=2
+    # just ensure positive monotonic
+    assert _estimate_tokens("x" * 100) == 50
+
+
+def test_history_excludes_latest_user():
+    # newest-first: [current user, prev assistant, older user]
+    msgs = _newest_first([
+        _Msg("user", "current question"),
+        _Msg("assistant", "previous answer"),
+        _Msg("user", "older question"),
+    ])
+    db = _DB(msgs)
+    out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
+    roles = [m["role"] for m in out]
+    # current user dropped; chronological order = [older user, previous assistant]
+    assert roles == ["user", "assistant"]
+    assert out[0]["content"] == "older question"
+
+
+def test_summary_injected_as_system_first():
+    msgs = _newest_first([
+        _Msg("user", "hi"),
+        _Msg("assistant", "hello"),
+    ])
+    db = _DB(msgs, summary=_Summary("用户想做图片生成"))
+    out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
+    assert out[0]["role"] == "system"
+    assert out[0]["content"].startswith("【对话历史摘要")
+    assert "用户想做图片生成" in out[0]["content"]
+    # after summary: chronological history (assistant then user 'hi' excluded? exclude_latest_user drops 'hi')
+    assert out[1]["role"] == "assistant"
+
+
+def test_budget_truncation():
+    # 5 assistant messages each ~200 tokens; budget small -> truncated
+    msgs = _newest_first([_Msg("assistant", "x" * 400) for _ in range(5)])
+    db = _DB(msgs)
+    out = _build_chat_history(db, "c1", max_tokens=500, exclude_latest_user=False)
+    # each msg ~200 tokens; budget 500 -> keep newest ~2
+    assert len(out) <= 3
+    total = sum(_estimate_tokens(m["content"]) for m in out)
+    assert total <= 500 + 200  # within budget (+1 msg slack)
+
+
+def test_no_summary_when_absent():
+    msgs = _newest_first([_Msg("user", "hi"), _Msg("assistant", "hello")])
+    db = _DB(msgs)
+    out = _build_chat_history(db, "c1", max_tokens=10000, exclude_latest_user=True)
+    # 'hi' excluded; only assistant remains, no system summary
+    assert all(m["role"] != "system" for m in out)
+    assert [m["role"] for m in out] == ["assistant"]

backend/python/tests/test_market.py

@@ -0,0 +1,20 @@
+"""应用市场场景分类器回归（稳定模块，无外部依赖）。
+
+仅校验分类法结构本身不被破坏；不调用 _classify_scene（其内部会请求 LLM）。
+分类法是前端筛选/徽章的数据源，结构回归会导致全站分类错乱。
+"""
+from app.api.market import MARKET_SCENES
+
+
+def test_market_scenes_is_14_flat():
+    assert len(MARKET_SCENES) == 14
+
+
+def test_market_scenes_unique_ids_and_required_keys():
+    ids = [s["id"] for s in MARKET_SCENES]
+    assert len(set(ids)) == 14  # 无重复 id
+    for s in MARKET_SCENES:
+        assert "id" in s
+        assert "label" in s
+        assert "desc" in s
+        assert isinstance(s["label"], str) and s["label"].strip()

backend/python/tests/test_regression_r1r2r3.py

@@ -0,0 +1,87 @@
+"""Phase 1 回归用例 R1/R2/R3 — 固化三个反复出现的问题（适配 chat.py 拆分后）。
+
+R1: 技能列表不得被硬性截断为 15 条（微信 Bot 曾只返回 15 个技能）。
+R2: 显式选中技能/MCP 时，system prompt 必须注入「最高优先级覆盖规则」。
+R3: 媒体生成（图像/视频）必须有超时保护，超时后回写用户可懂的提示而非永久挂起。
+
+Phase 1 已将 chat.py 巨石拆分为 chat_{utils,media,agents,skill,mcp,context,knowledge,service}.py，
+因此本文件采用「扫描整个 app/api 包」做源码契约守卫，并对行为级用例从正确的子模块导入。
+"""
+import asyncio
+import os
+import glob
+from unittest.mock import MagicMock
+
+import pytest
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+API_DIR = os.path.abspath(os.path.join(_HERE, "..", "app", "api"))
+
+
+def _read_api_package():
+    parts = []
+    for p in sorted(glob.glob(os.path.join(API_DIR, "chat*.py"))):
+        with open(p, "r", encoding="utf-8") as f:
+            parts.append(f.read())
+    return "\n".join(parts)
+
+
+# ---------- R1 ----------
+def test_r1_skill_list_not_hard_capped_at_15():
+    """微信 Bot 技能列表曾因 limit(15) 只返回 15 个，必须防止回归。"""
+    src = _read_api_package()
+    assert "limit(15)" not in src, "检测到 limit(15)，技能列表可能被硬性截断"
+    assert "LIMIT 15" not in src, "检测到 LIMIT 15，技能列表可能被硬性截断"
+
+
+# ---------- R2 ----------
+def test_r2_explicit_selection_override_rule_present():
+    """显式选中技能/MCP 时，system prompt 必须包含最高优先级覆盖规则。
+
+    注：_build_system_prompt 已随 Phase 1 拆分迁入 chat_context.py，故扫描整个 api 包。
+    """
+    src = _read_api_package()
+    assert "最高优先级覆盖规则" in src, "显式选中工具的覆盖规则文本缺失"
+    assert "只基于上文提供的该工具具体资料作答" in src, "覆盖规则应要求只基于已选工具资料作答"
+    assert "平台知识库" in src and "不要引用" in src, "覆盖规则应明确禁止引用平台知识库通用描述"
+
+
+# ---------- R3 ----------
+@pytest.mark.asyncio
+async def test_r3_media_generation_timeout_returns_user_message(monkeypatch):
+    """媒体生成超时必须回写用户可懂的超时提示，而非让前端永远『生成中…』。
+
+    _generate_media_task 位于 chat_media（经 _send_message_core 的 asyncio.create_task 调用），
+    本身是 fire-and-forget、返回 None，超时保护体现在把提示回写到消息 content。
+    """
+    from app.api import chat_media as media_mod
+
+    class FakeModel:
+        id = "m_test"
+        capabilities = ["image"]  # 图像 -> 90s 超时
+
+    msg_mock = MagicMock()  # 捕获被回写的消息对象
+    fake_session = MagicMock()
+
+    def _query_side_effect(model_cls):
+        q = MagicMock()
+        if model_cls is media_mod.AIModel:
+            q.filter.return_value.first.return_value = FakeModel()
+        elif model_cls is media_mod.Message:
+            q.filter.return_value.first.return_value = msg_mock
+        return q
+
+    fake_session.query.side_effect = _query_side_effect
+    monkeypatch.setattr(media_mod, "SessionLocal", lambda: fake_session)
+
+    # 内部生成调用直接抛出 TimeoutError，等价模拟「生成耗时超过超时阈值」
+    async def _hang(db, ai_model, prompt, user_id):
+        raise asyncio.TimeoutError()
+
+    monkeypatch.setattr(media_mod, "_generate_media", _hang)
+
+    await media_mod._generate_media_task("msg1", "m_test", "画一只猫", "u1")
+
+    # 副作用：消息 content 被回写为用户可懂的超时提示
+    assert msg_mock.content is not None, "超时后必须回写消息内容，不能让前端永远『生成中…』"
+    assert "超时" in str(msg_mock.content), "媒体生成超时应回写用户可懂的超时提示"

backend/python/tests/test_smoke.py

@@ -0,0 +1,21 @@
+"""Phase 0 安全网 — 后端冒烟测试。
+
+核心价值：证明整个后端（含 chat.py 等巨型模块）在测试环境下能无错误导入并挂载路由。
+任何破坏 import 的脆弱补丁都会在此立刻变红，从而把"改了没生效 / 启动即崩"挡在部署前。
+"""
+
+
+def test_app_imports_and_has_routes():
+    from main import app
+
+    assert app is not None
+    # 真实路由数远超 0；若 chat.py 等模块 import 失败，本测试在 import 阶段即报错
+    assert len(app.routes) > 50
+
+
+def test_openapi_schema_builds():
+    from main import app
+
+    schema = app.openapi()
+    assert "paths" in schema
+    assert len(schema["paths"]) > 0

backend/python/tests/test_sms.py

@@ -0,0 +1,21 @@
+"""短信宝脚手架回归（稳定模块，纯函数）。
+
+验证登录密码 MD5 算法与短信宝约定一致（32 位小写十六进制）。
+这是短信验证码功能（注册/改密/充值）能连通短信宝的前提。
+"""
+import hashlib
+
+from app.sms_client import smsbao_md5
+
+
+def test_smsbao_md5_matches_hashlib():
+    samples = ["XHR521yhf", "abc123", "密码测试", ""]
+    for pw in samples:
+        assert smsbao_md5(pw) == hashlib.md5(pw.encode("utf-8")).hexdigest()
+
+
+def test_smsbao_md5_is_32_lower_hex():
+    out = smsbao_md5("XHR521yhf")
+    assert len(out) == 32
+    assert out == out.lower()
+    assert all(c in "0123456789abcdef" for c in out)

frontend/package.json

@@ -6,7 +6,8 @@
   "scripts": {
     "dev": "vite",
     "build": "vite build",
-    "preview": "vite preview"
+    "preview": "vite preview",
+    "test": "vitest run"
   },
   "keywords": [],
   "author": "",
@@ -28,6 +29,7 @@
     "@types/react-dom": "^18.3.7",
     "@vitejs/plugin-react": "^4.7.0",
     "typescript": "^5.9.3",
-    "vite": "^5.4.21"
+    "vite": "^5.4.21",
+    "vitest": "^4.1.10"
   }
 }

frontend/src/App.tsx

@@ -45,6 +45,7 @@ const Achievement = React.lazy(() => import('@/pages/admin/Achievement'));
 const Tenant = React.lazy(() => import('@/pages/admin/Tenant'));
 const Billing = React.lazy(() => import('@/pages/admin/Billing'));
 const Analytics = React.lazy(() => import('@/pages/admin/Analytics'));
+const Observability = React.lazy(() => import('@/pages/admin/Observability'));
 const AdminKnowledge = React.lazy(() => import('@/pages/admin/Knowledge'));
 const AdminWorkflow = React.lazy(() => import('@/pages/admin/Workflow'));
 const AdminStorage = React.lazy(() => import('@/pages/admin/Storage'));
@@ -151,6 +152,7 @@ const App: React.FC = () => {
             <Route path="tenant" element={<Tenant />} />
             <Route path="billing" element={<Billing />} />
             <Route path="analytics" element={<Analytics />} />
+      <Route path="observability" element={<Observability />} />
             <Route path="knowledge" element={<AdminKnowledge />} />
             <Route path="workflow" element={<AdminWorkflow />} />
             <Route path="storage" element={<AdminStorage />} />

frontend/src/api/index.ts

@@ -527,6 +527,7 @@ export const adminAPI = {
   getRecentAudit: () => get<any[]>('/admin/dashboard/recent-audit'),
 
     getDailyStats: (days?: number) => get<{ days: number; data: Array<{ date: string; tokens: number }> }>(`/admin/dashboard/daily-stats${days ? `?days=${days}` : ''}`),
+    getObservabilitySummary: (days?: number) => get<any>(`/admin/observability/summary${days ? `?days=${days}` : ''}`),
 
   // 用户管理
 
@@ -886,6 +887,10 @@ export const adminAPI = {
 
   getAdminAgentReviews: (id: string) => get<any>(`/admin/agents/${id}/reviews`),
 
+  // Phase 3.1: 工具/MCP 权限沙箱 (tool_policy)
+  getAgentToolPolicy: (id: string) => get<any>(`/admin/agents/${id}/tool-policy`),
+  setAgentToolPolicy: (id: string, toolPolicy: any) => put<any>(`/admin/agents/${id}/tool-policy`, { tool_policy: toolPolicy }),
+
   getAdminAvailableSkills: (params?: any) => {
 
     const query = params ? '?' + Object.entries(params).filter(([_, v]) => v !== undefined && v !== '').map(([k, v]) => `${k}=${v}`).join('&') : '';

frontend/src/components/layout/AdminLayout.tsx

@@ -97,6 +97,7 @@ const adminNavGroups: NavGroup[] = [
     icon: 'developer',
     items: [
       { path: '/admin/analytics', icon: 'analytics', label: '数据分析' },
+      { path: '/admin/observability', icon: 'observability', label: '调用可观测' },
       { path: '/admin/workflow', icon: 'workflow', label: '工作流编排' },
       { path: '/admin/apps', icon: 'apps', label: 'AI 应用管理' },
       { path: '/admin/code-exec', icon: 'code', label: '代码执行沙箱' },
@@ -154,6 +155,7 @@ const adminIconPaths: Record<string, JSX.Element> = {
   billing: <svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="1.8" strokeLinecap="round" strokeLinejoin="round"><rect x="2" y="5" width="20" height="14" rx="2"/><line x1="2" y1="10" x2="22" y2="10"/><line x1="6" y1="15" x2="10" y2="15"/></svg>,
 
   analytics: <svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="1.8" strokeLinecap="round" strokeLinejoin="round"><line x1="18" y1="20" x2="18" y2="10"/><line x1="12" y1="20" x2="12" y2="4"/><line x1="6" y1="20" x2="6" y2="14"/><line x1="3" y1="20" x2="21" y2="20"/></svg>,
+  observability: <svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="1.8" strokeLinecap="round" strokeLinejoin="round"><path d="M3 12h4l3 8 4-16 3 8h4"/></svg>,
 
   imbot: <svg width="18" height="18" viewBox="0 0 24 24" fill="none" stroke="currentColor" strokeWidth="1.8" strokeLinecap="round" strokeLinejoin="round"><path d="M21 11.5a8.38 8.38 0 0 1-.9 3.8 8.5 8.5 0 0 1-7.6 4.7 8.38 8.38 0 0 1-3.8-.9L3 21l1.9-5.7a8.38 8.38 0 0 1-.9-3.8 8.5 8.5 0 0 1 4.7-7.6 8.38 8.38 0 0 1 3.8-.9h.5a8.48 8.48 0 0 1 8 8v.5z"/></svg>,
 

frontend/src/pages/admin/Agents.tsx

@@ -105,6 +105,10 @@ const Agents: React.FC = () => {
   // Available skills for linking
   const [availableSkills, setAvailableSkills] = useState<any[]>([]);
   const [availableModels, setAvailableModels] = useState<AvailableModel[]>([]);
+  // Available MCP servers for tool_policy sandbox allowlist
+  const [availableMcps, setAvailableMcps] = useState<any[]>([]);
+  // Tool policy (sandbox) — null means disabled / no policy
+  const [toolPolicy, setToolPolicy] = useState<any>(null);
 
   // Toast
   const [toast, setToast] = useState<{ msg: string; type: 'success' | 'error' } | null>(null);
@@ -144,14 +148,17 @@ const Agents: React.FC = () => {
   // Load available skills & models
   const loadAvailable = async () => {
     try {
-      const [skillsResult, modelsResult] = await Promise.all([
+      const [skillsResult, modelsResult, mcpsResult] = await Promise.all([
         adminAPI.getAdminAvailableSkills({ page_size: 50 }) as any,
         adminAPI.getAdminAvailableModels() as any,
+        adminAPI.getAdminMcpServers({ page_size: 50 }) as any,
       ]);
       const skills = (skillsResult.data || skillsResult).items || [];
       const models = (modelsResult.data || modelsResult) || [];
+      const mcps = (mcpsResult.data || mcpsResult).items || [];
       setAvailableSkills(skills);
       setAvailableModels(models);
+      setAvailableMcps(mcps);
     } catch {}
   };
 
@@ -187,6 +194,7 @@ const Agents: React.FC = () => {
       memory_enabled: false,
       reasoning_enabled: false,
     });
+    setToolPolicy(null);
     setShowModal(true);
   };
 
@@ -194,6 +202,7 @@ const Agents: React.FC = () => {
   const openEdit = async (agent: Agent) => {
     setEditingAgent(agent);
     await loadAvailable();
+    await openEditLoadPolicy(agent.id);
 
     const cfg = agent.config || {};
     setForm({
@@ -213,7 +222,15 @@ const Agents: React.FC = () => {
     setShowModal(true);
   };
 
-  // Open detail
+  // Fetch existing tool_policy (sandbox) for this agent
+  const openEditLoadPolicy = async (agentId: string) => {
+    try {
+      const tpRes = await adminAPI.getAgentToolPolicy(agentId) as any;
+      setToolPolicy(tpRes?.data?.tool_policy || null);
+    } catch {
+      setToolPolicy(null);
+    }
+  };
   const openDetail = async (agent: Agent) => {
     setDetailAgent(agent);
     setShowDetail(true);
@@ -246,7 +263,13 @@ const Agents: React.FC = () => {
     try {
       if (editingAgent) {
         await adminAPI.updateAdminAgent(editingAgent.id, payload);
+        // Phase 3.1: persist tool_policy sandbox (null clears it)
+        try {
+          await adminAPI.setAgentToolPolicy(editingAgent.id, toolPolicy);
           showToast('Agent 已更新');
+        } catch (pe: any) {
+          showToast(`Agent 已更新，但策略保存失败: ${pe.message || pe}`, 'error');
+        }
       } else {
         await adminAPI.createAdminAgent(payload);
         showToast('Agent 创建成功');
@@ -612,6 +635,87 @@ const Agents: React.FC = () => {
                   </div>
                 )}
               </div>
+              {editingAgent && (
+                <div style={{ gridColumn: '1 / -1' }}>
+                  <hr style={{ border: 'none', borderTop: '1px solid #E5E7EB', margin: '16px 0' }} />
+                  <div style={{ fontWeight: 600, marginBottom: 12 }}>🛡️ 工具权限沙箱 (tool_policy)</div>
+                  <label style={{ display: 'flex', alignItems: 'center', gap: 8, cursor: 'pointer', marginBottom: 12 }}>
+                    <input type="checkbox" checked={toolPolicy !== null} onChange={e => {
+                      if (e.target.checked) {
+                        setToolPolicy({ allowed_tools: { skill_ids: [], mcp_ids: [] }, max_calls_per_conversation: null, per_round_cap: null, fallback_message: '' });
+                      } else {
+                        setToolPolicy(null);
+                      }
+                    }} />
+                    启用工具/MCP 权限沙箱（限制该 Agent 可调用的工具与调用频次）
+                  </label>
+                  {toolPolicy && (
+                    <div style={{ display: 'flex', flexDirection: 'column', gap: 16 }}>
+                      <div className="form-group" style={{ marginBottom: 0 }}>
+                        <label className="form-label">允许调用的技能（不选 = 禁止调用任何技能）</label>
+                        <div style={{ display: 'flex', flexWrap: 'wrap', gap: 6, maxHeight: 160, overflowY: 'auto', padding: 8, border: '1px solid #E5E7EB', borderRadius: 6 }}>
+                          {availableSkills.length === 0 ? (
+                            <span style={{ color: '#9CA3AF', fontSize: 13 }}>暂无技能</span>
+                          ) : availableSkills.map((skill: any) => {
+                            const checked = (toolPolicy.allowed_tools?.skill_ids || []).includes(skill.id);
+                            return (
+                              <label key={skill.id} style={{ display: 'flex', alignItems: 'center', gap: 4, padding: '4px 10px', borderRadius: 12, fontSize: 12, cursor: 'pointer', background: checked ? '#EEF2FF' : '#F9FAFB', border: `1px solid ${checked ? '#6366F1' : '#E5E7EB'}`, color: checked ? '#6366F1' : '#374151' }}>
+                                <input type="checkbox" style={{ display: 'none' }} checked={checked} onChange={() => {
+                                  const cur = toolPolicy.allowed_tools?.skill_ids || [];
+                                  const next = checked ? cur.filter((x: string) => x !== skill.id) : [...cur, skill.id];
+                                  setToolPolicy({ ...toolPolicy, allowed_tools: { ...(toolPolicy.allowed_tools || {}), skill_ids: next } });
+                                }} />
+                                {skill.skill_name}
+                              </label>
+                            );
+                          })}
+                        </div>
+                      </div>
+                      <div className="form-group" style={{ marginBottom: 0 }}>
+                        <label className="form-label">允许调用的 MCP 服务（不选 = 禁止调用任何 MCP）</label>
+                        <div style={{ display: 'flex', flexWrap: 'wrap', gap: 6, maxHeight: 160, overflowY: 'auto', padding: 8, border: '1px solid #E5E7EB', borderRadius: 6 }}>
+                          {availableMcps.length === 0 ? (
+                            <span style={{ color: '#9CA3AF', fontSize: 13 }}>暂无 MCP 服务</span>
+                          ) : availableMcps.map((m: any) => {
+                            const checked = (toolPolicy.allowed_tools?.mcp_ids || []).includes(m.id);
+                            return (
+                              <label key={m.id} style={{ display: 'flex', alignItems: 'center', gap: 4, padding: '4px 10px', borderRadius: 12, fontSize: 12, cursor: 'pointer', background: checked ? '#EEF2FF' : '#F9FAFB', border: `1px solid ${checked ? '#6366F1' : '#E5E7EB'}`, color: checked ? '#6366F1' : '#374151' }}>
+                                <input type="checkbox" style={{ display: 'none' }} checked={checked} onChange={() => {
+                                  const cur = toolPolicy.allowed_tools?.mcp_ids || [];
+                                  const next = checked ? cur.filter((x: string) => x !== m.id) : [...cur, m.id];
+                                  setToolPolicy({ ...toolPolicy, allowed_tools: { ...(toolPolicy.allowed_tools || {}), mcp_ids: next } });
+                                }} />
+                                {m.name}
+                              </label>
+                            );
+                          })}
+                        </div>
+                      </div>
+                      <div style={{ display: 'grid', gridTemplateColumns: '1fr 1fr', gap: 16 }}>
+                        <div className="form-group" style={{ marginBottom: 0 }}>
+                          <label className="form-label">单对话最大调用次数（留空=不限制）</label>
+                          <input className="admin-form-input" type="number" min="0"
+                            value={toolPolicy.max_calls_per_conversation ?? ''}
+                            onChange={e => setToolPolicy({ ...toolPolicy, max_calls_per_conversation: e.target.value === '' ? null : parseInt(e.target.value) })} />
+                        </div>
+                        <div className="form-group" style={{ marginBottom: 0 }}>
+                          <label className="form-label">单轮最大调用次数（留空=不限制）</label>
+                          <input className="admin-form-input" type="number" min="0"
+                            value={toolPolicy.per_round_cap ?? ''}
+                            onChange={e => setToolPolicy({ ...toolPolicy, per_round_cap: e.target.value === '' ? null : parseInt(e.target.value) })} />
+                        </div>
+                      </div>
+                      <div className="form-group" style={{ marginBottom: 0 }}>
+                        <label className="form-label">额度耗尽提示语（可选）</label>
+                        <textarea className="admin-form-input" rows={2}
+                          value={toolPolicy.fallback_message || ''}
+                          onChange={e => setToolPolicy({ ...toolPolicy, fallback_message: e.target.value })}
+                          placeholder="例如：该智能体已达到工具调用上限，请开启新对话或调整智能体设置。" />
+                      </div>
+                    </div>
+                  )}
+                </div>
+              )}
             </div>
             <div className="modal-footer">
               <button className="admin-btn" onClick={() => setShowModal(false)}>取消</button>

frontend/src/pages/admin/Observability.tsx

@@ -0,0 +1,165 @@
+import { useEffect, useState } from 'react';
+import { adminAPI } from '@/api';
+
+interface Summary {
+  totals: { calls: number; tokens: number; cost: number; users: number };
+  by_model: { model_name: string; calls: number; tokens: number; cost: number }[];
+  by_user: { user_id: string; username: string; calls: number; tokens: number; cost: number }[];
+  daily: { date: string; tokens: number; cost: number; calls: number }[];
+  recent: { trace_id: string | null; model_name: string; tokens: number; cost: number; created_at: string | null }[];
+}
+
+const ACCENT = '#5B5FED';
+const ACCENT_HOVER = '#4A4ECC';
+
+export default function Observability() {
+  const [days, setDays] = useState(30);
+  const [data, setData] = useState<Summary | null>(null);
+  const [loading, setLoading] = useState(false);
+  const [err, setErr] = useState('');
+
+  useEffect(() => {
+    let alive = true;
+    setLoading(true);
+    setErr('');
+    adminAPI.getObservabilitySummary(days)
+      .then((d: Summary) => { if (alive) setData(d); })
+      .catch((e: any) => { if (alive) setErr(e?.message || '加载失败'); })
+      .finally(() => { if (alive) setLoading(false); });
+    return () => { alive = false; };
+  }, [days]);
+
+  const maxTokens = data ? Math.max(1, ...data.daily.map(d => d.tokens)) : 1;
+  const maxModel = data ? Math.max(1, ...data.by_model.map(m => m.tokens)) : 1;
+  const fmt = (n: number) => n.toLocaleString('zh-CN');
+  const fmtCost = (n: number) => `¥${n.toFixed(4)}`;
+
+  const card: React.CSSProperties = {
+    background: '#fff', border: '1px solid #eceef5', borderRadius: 16,
+    padding: 20, boxShadow: '0 1px 3px rgba(20,30,60,0.04)', marginBottom: 18,
+  };
+  const stat: React.CSSProperties = {
+    flex: 1, minWidth: 150, background: '#fff', border: '1px solid #eceef5',
+    borderRadius: 16, padding: 18, boxShadow: '0 1px 3px rgba(20,30,60,0.04)',
+  };
+
+  return (
+    <div style={{ padding: 24, maxWidth: 1160, margin: '0 auto', color: '#1f2330' }}>
+      <div style={{ display: 'flex', alignItems: 'center', justifyContent: 'space-between', marginBottom: 20, flexWrap: 'wrap', gap: 12 }}>
+        <div>
+          <h2 style={{ margin: 0, fontSize: 22, fontWeight: 700 }}>调用可观测</h2>
+          <div style={{ color: '#8a90a2', fontSize: 13, marginTop: 4 }}>
+            媒体 / LLM 调用归因 · 基于 trace_id 关联请求链路（Phase 2.2）
+          </div>
+        </div>
+        <div style={{ display: 'flex', gap: 8 }}>
+          {[7, 30, 90].map(d => (
+            <button key={d} onClick={() => setDays(d)}
+              style={{
+                padding: '7px 14px', borderRadius: 10, cursor: 'pointer', fontSize: 13,
+                border: `1px solid ${days === d ? ACCENT : '#e2e5ee'}`,
+                background: days === d ? ACCENT : '#fff',
+                color: days === d ? '#fff' : '#5b6172', fontWeight: 600,
+                transition: 'all .15s',
+              }}>
+              {d} 天
+            </button>
+          ))}
+        </div>
+      </div>
+
+      {err && <div style={{ ...card, borderColor: '#f3c2c2', color: '#c0392b' }}>{err}</div>}
+      {loading && !data && <div style={{ color: '#8a90a2' }}>加载中…</div>}
+
+      {data && (
+        <>
+          <div style={{ display: 'flex', gap: 14, flexWrap: 'wrap', marginBottom: 18 }}>
+            <div style={stat}><div style={{ color: '#8a90a2', fontSize: 12 }}>调用次数</div><div style={{ fontSize: 24, fontWeight: 700, marginTop: 6 }}>{fmt(data.totals.calls)}</div></div>
+            <div style={stat}><div style={{ color: '#8a90a2', fontSize: 12 }}>Token 消耗</div><div style={{ fontSize: 24, fontWeight: 700, marginTop: 6 }}>{fmt(data.totals.tokens)}</div></div>
+            <div style={stat}><div style={{ color: '#8a90a2', fontSize: 12 }}>成本</div><div style={{ fontSize: 24, fontWeight: 700, marginTop: 6, color: ACCENT }}>{fmtCost(data.totals.cost)}</div></div>
+            <div style={stat}><div style={{ color: '#8a90a2', fontSize: 12 }}>活跃用户</div><div style={{ fontSize: 24, fontWeight: 700, marginTop: 6 }}>{fmt(data.totals.users)}</div></div>
+          </div>
+
+          <div style={card}>
+            <h3 style={{ margin: '0 0 14px', fontSize: 16 }}>按模型</h3>
+            <div style={{ display: 'flex', flexDirection: 'column', gap: 10 }}>
+              {data.by_model.length === 0 && <div style={{ color: '#8a90a2', fontSize: 13 }}>暂无数据</div>}
+              {data.by_model.map(m => (
+                <div key={m.model_name} style={{ display: 'flex', alignItems: 'center', gap: 12 }}>
+                  <div style={{ width: 150, fontSize: 13, fontWeight: 600, overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap' }}>{m.model_name}</div>
+                  <div style={{ flex: 1, height: 10, background: '#f1f3f9', borderRadius: 6, overflow: 'hidden' }}>
+                    <div style={{ width: `${(m.tokens / maxModel) * 100}%`, height: '100%', background: ACCENT, borderRadius: 6 }} />
+                  </div>
+                  <div style={{ width: 90, textAlign: 'right', fontSize: 13, color: '#5b6172' }}>{fmt(m.tokens)}</div>
+                  <div style={{ width: 90, textAlign: 'right', fontSize: 13, color: ACCENT }}>{fmtCost(m.cost)}</div>
+                </div>
+              ))}
+            </div>
+          </div>
+
+          <div style={card}>
+            <h3 style={{ margin: '0 0 14px', fontSize: 16 }}>每日趋势（Token）</h3>
+            <div style={{ display: 'flex', alignItems: 'flex-end', gap: 6, height: 140, overflowX: 'auto', paddingBottom: 4 }}>
+              {data.daily.length === 0 && <div style={{ color: '#8a90a2', fontSize: 13 }}>暂无数据</div>}
+              {data.daily.map(d => (
+                <div key={d.date} title={`${d.date} · ${fmt(d.tokens)} tokens · ${fmtCost(d.cost)}`}
+                  style={{ display: 'flex', flexDirection: 'column', alignItems: 'center', gap: 4, minWidth: 22 }}>
+                  <div style={{ width: 16, height: `${(d.tokens / maxTokens) * 110}px`, background: ACCENT, borderRadius: '4px 4px 0 0', opacity: 0.85 }} />
+                  <div style={{ fontSize: 9, color: '#aab', transform: 'rotate(-45deg)', whiteSpace: 'nowrap' }}>{d.date.slice(5)}</div>
+                </div>
+              ))}
+            </div>
+          </div>
+
+          <div style={{ display: 'flex', gap: 18, flexWrap: 'wrap' }}>
+            <div style={{ ...card, flex: 1, minWidth: 320 }}>
+              <h3 style={{ margin: '0 0 14px', fontSize: 16 }}>按用户（Top 20）</h3>
+              <table style={{ width: '100%', borderCollapse: 'collapse', fontSize: 13 }}>
+                <thead><tr style={{ color: '#8a90a2', textAlign: 'left' }}>
+                  <th style={th}>用户</th><th style={thR}>调用</th><th style={thR}>Token</th><th style={thR}>成本</th>
+                </tr></thead>
+                <tbody>
+                  {data.by_user.map(u => (
+                    <tr key={u.user_id} style={{ borderTop: '1px solid #f1f3f9' }}>
+                      <td style={td}>{u.username}</td>
+                      <td style={tdR}>{fmt(u.calls)}</td>
+                      <td style={tdR}>{fmt(u.tokens)}</td>
+                      <td style={{ ...tdR, color: ACCENT }}>{fmtCost(u.cost)}</td>
+                    </tr>
+                  ))}
+                  {data.by_user.length === 0 && <tr><td colSpan={4} style={{ ...td, color: '#8a90a2' }}>暂无数据</td></tr>}
+                </tbody>
+              </table>
+            </div>
+          </div>
+
+          <div style={card}>
+            <h3 style={{ margin: '0 0 14px', fontSize: 16 }}>最近调用（trace_id 可下钻）</h3>
+            <table style={{ width: '100%', borderCollapse: 'collapse', fontSize: 13 }}>
+              <thead><tr style={{ color: '#8a90a2', textAlign: 'left' }}>
+                <th style={th}>Trace ID</th><th style={th}>模型</th><th style={thR}>Token</th><th style={thR}>成本</th><th style={thR}>时间</th>
+              </tr></thead>
+              <tbody>
+                {data.recent.map((r, i) => (
+                  <tr key={(r.trace_id || '') + i} style={{ borderTop: '1px solid #f1f3f9' }}>
+                    <td style={{ ...td, fontFamily: 'monospace', fontSize: 12, maxWidth: 200, overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap' }}>{r.trace_id || '—'}</td>
+                    <td style={td}>{r.model_name}</td>
+                    <td style={tdR}>{fmt(r.tokens)}</td>
+                    <td style={{ ...tdR, color: ACCENT }}>{fmtCost(r.cost)}</td>
+                    <td style={tdR}>{r.created_at || '—'}</td>
+                  </tr>
+                ))}
+                {data.recent.length === 0 && <tr><td colSpan={5} style={{ ...td, color: '#8a90a2' }}>暂无数据</td></tr>}
+              </tbody>
+            </table>
+          </div>
+        </>
+      )}
+    </div>
+  );
+}
+
+const th: React.CSSProperties = { padding: '8px 6px', fontWeight: 600, fontSize: 12 };
+const thR: React.CSSProperties = { ...th, textAlign: 'right' };
+const td: React.CSSProperties = { padding: '8px 6px' };
+const tdR: React.CSSProperties = { ...td, textAlign: 'right', color: '#5b6172' };

frontend/src/utils/datetime.test.ts

@@ -0,0 +1,29 @@
+import { describe, it, expect } from 'vitest';
+import { formatBjTime } from './datetime';
+
+// 北京时间渲染曾是反复出现的问题（微信Bot时间错乱、Token驾驶舱时间列不正确等）。
+// 该测试固化「后端按北京时间存储、前端显式按 Asia/Shanghai 渲染」的契约。
+describe('formatBjTime', () => {
+  it('空输入返回占位符 -', () => {
+    expect(formatBjTime(undefined)).toBe('-');
+    expect(formatBjTime(null)).toBe('-');
+    expect(formatBjTime('')).toBe('-');
+  });
+
+  it('无时区标记的时间戳按 Asia/Shanghai (+08:00) 渲染', () => {
+    // 2026-07-18T12:00:00 视为北京时间，渲染应含 12:00:00
+    const out = formatBjTime('2026-07-18T12:00:00');
+    expect(out).toContain('2026');
+    expect(out).toContain('12:00:00');
+  });
+
+  it('UTC(Z) 时间戳正确转换为北京时间 (+8)', () => {
+    // 2026-07-18T04:00:00Z = 北京时间 12:00:00
+    const out = formatBjTime('2026-07-18T04:00:00Z');
+    expect(out).toContain('12:00:00');
+  });
+
+  it('非法日期原样返回', () => {
+    expect(formatBjTime('not-a-date')).toBe('not-a-date');
+  });
+});

frontend/vite.config.ts

@@ -1,4 +1,5 @@
-import { defineConfig } from 'vite';
+/// <reference types="vitest/config" />
+import { defineConfig } from 'vitest/config';
 import react from '@vitejs/plugin-react';
 import path from 'path';
 
@@ -17,8 +18,6 @@ export default defineConfig({
     chunkSizeWarningLimit: 1200,
     rollupOptions: {
       output: {
-        // 仅将 react 核心库单独打包（长期缓存）；
-        // 其余第三方库交给 Rollup 自动按需拆分，避免被强制合并成一个大 vendor 在首屏加载
         manualChunks(id) {
           if (id.includes('node_modules')) {
             if (
@@ -37,4 +36,9 @@ export default defineConfig({
   server: {
     port: 5173,
   },
+  // Phase 0 测试基建：vitest 配置（纯函数测试用 node 环境，组件测试可改 jsdom）
+  test: {
+    environment: 'node',
+    include: ['src/**/*.test.ts'],
+  },
 });

migrations/020_add_trace_id_token_usage_logs.sql

@@ -0,0 +1,7 @@
+-- Phase 2.2: add trace_id to token_usage_logs for cost attribution (idempotent)
+SET @db = DATABASE();
+SET @col_exists = (SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = @db AND table_name = 'token_usage_logs' AND column_name = 'trace_id');
+SET @sql = IF(@col_exists = 0, 'ALTER TABLE token_usage_logs ADD COLUMN trace_id VARCHAR(64) NULL DEFAULT NULL', 'SELECT 1');
+PREPARE stmt FROM @sql;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;

migrations/021_create_conversation_summaries.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS conversation_summaries (
+  conversation_id VARCHAR(64) NOT NULL,
+  summary_text TEXT NULL,
+  token_count INT NULL DEFAULT 0,
+  updated_at DATETIME(3) NULL,
+  PRIMARY KEY (conversation_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

migrations/022_add_tool_policy.sql

@@ -0,0 +1,15 @@
+-- Phase 3.1: MarketApp 增加 tool_policy（Agent 工具/MCP 权限沙箱策略）
+-- 幂等：仅当列不存在时才 ALTER
+SET @db = 'aiflow';
+SET @tbl = 'market_apps';
+SET @col = 'tool_policy';
+SET @exists = (
+    SELECT COUNT(*) FROM information_schema.columns
+    WHERE table_schema = @db AND table_name = @tbl AND column_name = @col
+);
+SET @sql = IF(@exists = 0,
+    'ALTER TABLE market_apps ADD COLUMN tool_policy JSON NULL',
+    'SELECT 1');
+PREPARE stmt FROM @sql;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;

migrations/023_widen_agent_api_key.sql

@@ -0,0 +1,11 @@
+-- Phase 3.2: 放宽 market_apps.agent_api_key 容量以容纳 Fernet 密文
+SET @db = DATABASE();
+SET @tbl = 'market_apps';
+SET @col = 'agent_api_key';
+SELECT COLUMN_TYPE INTO @ctype FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=@db AND TABLE_NAME=@tbl AND COLUMN_NAME=@col;
+SET @sql = IF(@ctype IS NULL OR @ctype NOT LIKE '%text%',
+  CONCAT('ALTER TABLE ', @tbl, ' MODIFY COLUMN ', @col, ' TEXT NULL COMMENT \'Agent 接入 API Key(3.2 起 Fernet 加密存储)\''),
+  'SELECT 1');
+PREPARE stmt FROM @sql;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;

migrations/024_add_orchestration.sql

@@ -0,0 +1,15 @@
+-- Phase 3.3: MarketApp 增加 orchestration（Agent 多 Agent 编排图，JSON）
+-- 幂等：仅当列不存在时才 ALTER
+SET @db = 'aiflow';
+SET @tbl = 'market_apps';
+SET @col = 'orchestration';
+SET @exists = (
+    SELECT COUNT(*) FROM information_schema.columns
+    WHERE table_schema = @db AND table_name = @tbl AND column_name = @col
+);
+SET @sql = IF(@exists = 0,
+    "ALTER TABLE market_apps ADD COLUMN orchestration JSON NULL COMMENT 'Agent 多Agent编排图(3.3)'",
+    'SELECT 1');
+PREPARE stmt FROM @sql;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;

scripts/ci_gate.py

@@ -0,0 +1,115 @@
+"""Minimal CI gate for qihuaiflow.
+
+This is the canonical, single-source-of-truth gate. It runs the two test
+suites that guard production behavior, INSIDE the long-lived qihu-aiflow
+container (our deployment model keeps the app running in a container, not
+built fresh in CI):
+
+  1. backend pytest  -> /app/backend/python/.venv/bin/python -m pytest -c pytest.ini
+  2. frontend vitest -> /app/frontend && npx vitest run
+
+Exit codes:
+  0  all suites green  -> deploy/push may proceed
+  1  a test suite failed -> BLOCK (do not deploy/push)
+  2  environment error  -> docker missing or container not running
+
+It auto-detects where it runs:
+  * If the qihu-aiflow container is reachable via LOCAL docker (i.e. on the
+    deploy host), it execs directly.
+  * Otherwise it falls back to ssh_helper (remote control from a laptop),
+    which reaches the host over SSH and execs into the container there.
+
+Usage:
+  python scripts/ci_gate.py
+"""
+import os
+import subprocess
+import sys
+
+CONTAINER = "qihu-aiflow"
+
+BACKEND_CMD = "cd /app/backend/python && .venv/bin/python -m pytest -c pytest.ini -q"
+FRONTEND_CMD = "cd /app/frontend && npx vitest run --reporter=dot"
+
+
+def shell_quote(s: str) -> str:
+    return "'" + s.replace("'", "'\\''") + "'"
+
+
+def local_runner(cmd: str):
+    """Run via local docker (deploy host)."""
+    p = subprocess.run(cmd, shell=True, capture_output=True, text=True)
+    return p.stdout, p.stderr, p.returncode
+
+
+def get_runner():
+    # Prefer local docker ONLY if the actual container is reachable here
+    # (avoids a Windows-docker edge case where `docker ps` works but the
+    # qihu-aiflow container lives on a different host).
+    try:
+        r = subprocess.run(
+            "docker ps --format '{{.Names}}'",
+            shell=True, capture_output=True, text=True, timeout=10,
+        )
+        if r.returncode == 0 and CONTAINER in r.stdout.split():
+            return local_runner
+    except Exception:
+        pass
+    # Fall back to SSH control of the real host.
+    try:
+        import ssh_helper as s  # type: ignore
+        return lambda cmd: s.run(cmd)
+    except Exception as e:  # pragma: no cover
+        print(f"[CI Gate] ERROR: docker unavailable locally and ssh_helper missing: {e}",
+              file=sys.stderr)
+        sys.exit(2)
+
+
+def run_in_container(runner, cmd: str):
+    return runner(f"docker exec {CONTAINER} bash -c {shell_quote(cmd)}")
+
+
+def container_up(runner) -> bool:
+    out, _, _ = runner(f"docker ps --format {shell_quote('{{.Names}}')}")
+    return CONTAINER in out.split()
+
+
+def main():
+    runner = get_runner()
+    if not container_up(runner):
+        print(f"[CI Gate] ERROR: container {CONTAINER} is not running", file=sys.stderr)
+        sys.exit(2)
+
+    failures = []
+
+    print("=== [CI Gate] backend pytest ===")
+    o, e, rc = run_in_container(runner, BACKEND_CMD)
+    sys.stdout.write(o)
+    if e.strip():
+        print("STDERR:", e.strip(), file=sys.stderr)
+    if rc != 0:
+        failures.append("backend pytest")
+        print(f"[CI Gate] backend pytest FAILED (rc={rc})", file=sys.stderr)
+    else:
+        print("[CI Gate] backend pytest OK")
+
+    print("=== [CI Gate] frontend vitest ===")
+    o, e, rc = run_in_container(runner, FRONTEND_CMD)
+    sys.stdout.write(o)
+    if e.strip():
+        print("STDERR:", e.strip(), file=sys.stderr)
+    if rc != 0:
+        failures.append("frontend vitest")
+        print(f"[CI Gate] frontend vitest FAILED (rc={rc})", file=sys.stderr)
+    else:
+        print("[CI Gate] frontend vitest OK")
+
+    if failures:
+        print(f"\n[CI Gate] BLOCKED - failed: {', '.join(failures)}", file=sys.stderr)
+        sys.exit(1)
+    print("\n[CI Gate] PASSED - all suites green")
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()

scripts/git-hooks/pre-push

@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+# qihuaiflow minimal CI gate - git pre-push hook.
+#
+# Runs backend pytest + frontend vitest INSIDE the long-lived qihu-aiflow
+# container before any `git push` leaves this host. If a suite fails the
+# push is blocked (exit non-zero), so a broken tree can never be published.
+#
+# Install (on the deploy host, inside the repo):
+#   cp scripts/git-hooks/pre-push .git/hooks/pre-push
+#   chmod +x .git/hooks/pre-push
+#
+# This is a self-contained bash gate (no python dependency on the host).
+# The canonical, documented version lives in scripts/ci_gate.py.
+set -uo pipefail
+
+CONTAINER="qihu-aiflow"
+
+if ! docker ps --format '{{.Names}}' | grep -qx "$CONTAINER"; then
+  echo "[CI Gate] ERROR: container $CONTAINER is not running - aborting push" >&2
+  exit 2
+fi
+
+echo "=== [CI Gate] backend pytest ==="
+docker exec "$CONTAINER" bash -c 'cd /app/backend/python && .venv/bin/python -m pytest -c pytest.ini -q'
+backend_rc=$?
+if [ "$backend_rc" -ne 0 ]; then
+  echo "[CI Gate] backend pytest FAILED (rc=$backend_rc) - push blocked" >&2
+  exit 1
+fi
+echo "[CI Gate] backend pytest OK"
+
+echo "=== [CI Gate] frontend vitest ==="
+docker exec "$CONTAINER" bash -c 'cd /app/frontend && npx vitest run --reporter=dot'
+frontend_rc=$?
+if [ "$frontend_rc" -ne 0 ]; then
+  echo "[CI Gate] frontend vitest FAILED (rc=$frontend_rc) - push blocked" >&2
+  exit 1
+fi
+echo "[CI Gate] frontend vitest OK"
+
+echo "[CI Gate] PASSED - all suites green - push allowed"
+exit 0