Merge pull request #2 from blizzardzhang/dev

Dev
2026-06-23 17:25:45 +08:00
parent dfa1e163bf 6d839dbb30
commit fc1494b634
41 changed files with 1808 additions and 147 deletions
@@ -0,0 +1,70 @@
+name: CI
+
+on:
+  push:
+    branches: [main, dev]
+  pull_request:
+    branches: [main, dev]
+
+# 同一分支新 push 取消上一次未完成的运行，省 CI 时间。
+concurrency:
+  group: ci-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  go:
+    name: Go · build + vet + test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version: "1.25"
+          cache-dependency-path: "**/go.sum"
+      - name: build + vet + test（4 模块；bus 用内嵌 NATS，无需外部服务）
+        run: |
+          set -e
+          for m in sundynix-shared sundynix-gateway sundynix-dispatcher sundynix-mcp-go; do
+            echo "::group::$m"
+            (cd "$m" && go build ./... && go vet ./... && go test ./...)
+            echo "::endgroup::"
+          done
+
+  web:
+    name: Frontend · tsc
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        dir: [sundynix-desktop/frontend, sundynix-admin]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: "20"
+          cache: npm
+          cache-dependency-path: ${{ matrix.dir }}/package-lock.json
+      - name: install + typecheck
+        working-directory: ${{ matrix.dir }}
+        run: |
+          npm ci
+          npx tsc --noEmit
+
+  py:
+    name: mcp-py · sandbox guard
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: pip
+      - name: install + test（无 Docker 时测降级路径）
+        working-directory: sundynix-mcp-py
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install -e . pytest
+          PYTHONPATH=src:tests python -m pytest tests/ -q || \
+          PYTHONPATH=src:tests python -c "import test_sandbox as t; \
+            fns=[getattr(t,n) for n in dir(t) if n.startswith('test_')]; \
+            [ (f(),print('PASS',f.__name__)) for f in fns ]; print(f'{len(fns)} passed')"
@@ -0,0 +1,60 @@
+name: Release
+
+# 打 vX.Y.Z 标签即触发：各平台用 wails 构建桌面端安装包 → 发布到 GitHub Release。
+# 用户旧版 App 启动时查 /releases/latest，发现新版即提示下载（见 frontend/src/lib/version.ts）。
+on:
+  push:
+    tags: ["v*"]
+
+permissions:
+  contents: write # 创建 release + 上传产物需要
+
+jobs:
+  build:
+    name: 构建 ${{ matrix.label }}
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - os: macos-latest
+            label: macOS (universal)
+            platform: darwin/universal
+          - os: windows-latest
+            label: Windows (amd64)
+            platform: windows/amd64
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version: "1.25"
+      - uses: actions/setup-node@v4
+        with:
+          node-version: "20"
+      - name: 安装 wails CLI
+        run: go install github.com/wailsapp/wails/v2/cmd/wails@v2.12.0
+
+      - name: 构建桌面端（wails build；GOWORK=off 因 desktop 不在工作区）
+        working-directory: sundynix-desktop
+        env:
+          GOWORK: "off" # 必须加引号：YAML 裸 off 会被解析成布尔 false，使 workflow 校验不合法而不触发
+        run: wails build -clean -platform ${{ matrix.platform }}
+
+      # macOS：.app 是目录，打包成 zip 才能作为单文件资产分发。
+      - name: 打包 macOS 产物
+        if: runner.os == 'macOS'
+        working-directory: sundynix-desktop/build/bin
+        run: zip -r -y sundynix_desktop_macos_universal.zip sundynix_desktop.app
+
+      - name: 重命名 Windows 产物
+        if: runner.os == 'Windows'
+        working-directory: sundynix-desktop/build/bin
+        run: ren sundynix_desktop.exe sundynix_desktop_windows_amd64.exe
+
+      - name: 发布到 GitHub Release（按 tag 创建/追加资产）
+        uses: softprops/action-gh-release@v2
+        with:
+          fail_on_unmatched_files: false
+          files: |
+            sundynix-desktop/build/bin/sundynix_desktop_macos_universal.zip
+            sundynix-desktop/build/bin/sundynix_desktop_windows_amd64.exe
@@ -9,8 +9,12 @@

 - [x] **Phase A · 地基**：`llm.Pool` 换 Eino ChatModel 组件（commit d84b1ec，验收通过）
 - [x] **Phase B · 质变**：MCP 工具→`InvokableTool` + ReAct agent（模型自主调工具，验收 7/7 命中）
- [ ] **Phase C · 编排归一**：`Flow→compose.Graph` + callbacks 桥接
- [ ] **Phase D · 状态化执行**：任务生命周期 FSM / HITL 中断恢复 / 多智能体
+- [x] **Phase C · 编排归一**：✅ 全图 `DSL→compose.Graph` 编译器（全节点 + branch + DAG 并行调度）+ callbacks→ExecEvent 归一，等价回归通过（EINO_COMPOSE 灰度开关，默认关；过渡期后 graph.go 退役）
+- [~] **Phase D · 状态化执行**：✅ 任务生命周期 FSM（已完成）/ ⬜ HITL 中断恢复 / ⬜ 多智能体（按场景）
+
+**组件化补完（A）**：检索 → `ragRetriever`(`components/retriever.Retriever`，`eino_components.go`)；提示词 → `buildMessages` 改用 `prompt.FromMessages`+`MessagesPlaceholder`；工具 → `mcpTool`(`InvokableTool`，模型自主调用)。至此终态架构 8 层中 模型/工具/检索/提示词/编排/智能体(单)/可观测 均已 Eino 组件化；剩 人机交互(中断恢复，按场景)。
+**自主 agent 工具集动态化**：agent 工具集不再硬编码——mcp-go 注册表(单一事实源)每个工具声明 `agent/params/inject`，`list_tools` 上报，dispatcher `agentTools()` 动态发现并建 `InvokableTool`、运行时注入 `user_id/session_id/kb`（不暴露给模型）。**加工具只改 mcp-go 注册表，dispatcher 零改动。** 当前暴露 4 个(wiki_search/recall_user_memory/remember_user_fact/history_get)；实测模型自主调用新暴露的 remember_user_fact 成功（参数自生成、user_id 服务端注入）。
+**性能注记**：compose 每任务编译实测 ~13µs（基准 BenchmarkComposeCompile），相对 LLM 秒级可忽略 → 编译图缓存判定为 premature optimization，暂不做；真正的并行效率已由 Phase C 的 DAG 调度（`AllPredecessor`）拿到。

 ---

@@ -117,7 +121,28 @@ github.com/cloudwego/eino-ext/...             # ⚠️ 官方组件实现（open

 ---

-## Phase C · 编排归一：迁到 compose.Graph 〔P2〕
+## Phase C · 编排归一：迁到 compose.Graph 〔P2〕✅ 已完成（并存灰度，等价回归通过）
+
+> 全图编译器：`compose_compiler.go` 把整张 DSL 图编译为 `compose.Graph`——
+> - 每个节点 = 一个 Lambda，节点体复用现有逻辑（`execDSLNode`：input/memory/retriever/tool/agent/aggregate/render/map/output），黑板进 compose 本地状态（`WithGenLocalState` + `ProcessState`）。
+> - branch = `AddBranch` + 状态感知条件（复用 `branchNode` 选路）；边载荷用空 `flowSignal`（注册 no-op 合并支持 fan-in），真实数据全走黑板。
+> - `WithNodeTriggerMode(AllPredecessor)` DAG 模式：无依赖节点并行调度（效率）。
+> - `Handle → executeGraph` 按 `EINO_COMPOSE` 开关选 compose / graph.go；compose 编译失败自动降级回 graph.go（安全网）。
+> - **等价回归**：线性图、分支图经解释器与 compose 两路径产出逐字一致（单测）；live 多节点分支图 compose 路径 2800 字答复 eval 1.00、FSM done、0 幽灵。
+>
+> 待过渡期 soak 后把默认翻到 compose、退役 graph.go。**性能后续**：编译图按 DSL-hash 缓存（当前每任务编译一次）。
+
+
+
+> 落地（并存+等价回归策略）：对话主流程已可跑在 `compose.Graph` 上——
+> - `compose_graph.go`：`runConversation` 按 `EINO_COMPOSE` 开关分流；`runComposeConversation` 建图 `START→ChatModel→END`、`Compile`→`Stream`，token 回流；模型未就绪/编译失败降级回 `runAgent`。**默认关，graph.go 仍是默认且权威。**
+> - `compose_callbacks.go`：`composeTracer` 用 `utils/callbacks` 把 ChatModel/Tool 的 start/end/error 桥到 ExecEvent（可观测归一）。
+> - 测试：compose 图编译+运行、compose 对话流式回流、开关关→走 runAgent，三个单测；live 实测 compose 路径出 54 字答复 + eval 1.00，默认路径 eval 1.00。
+> - **顺带修真 bug**：`SubjectTaskStatus` 原为 `sundynix.tasks.status`，落在任务流捕获通配 `sundynix.tasks.>` 内 → 状态事件被当成"幽灵任务"自我放大（实测污染 2300+ 条）。已挪到 `sundynix.status.task` + dispatcher 加空任务护栏。
+>
+> 剩余：branch / map / render / retriever / prompt 等节点逐步迁 compose（同并存+等价回归），对齐后 graph.go 退役。
+
+

 **目标**：自研解释器退役，DSL 图编译为 `compose.Graph`，吃到流式 reduce/branch、类型化边、自动并发、callbacks。

@@ -139,11 +164,11 @@ github.com/cloudwego/eino-ext/...             # ⚠️ 官方组件实现（open

 主题：把"执行"从一次性 DAG 升级为**可持久化、可恢复的状态机**。三件事同一条线，一起做。

- **任务生命周期 FSM** 🆕：现在 `Task.Status` 只写死 `submitted`、全仓从不流转（`store/models.go` + `pgsql.go:96`），是个摆设——这正是"卡运行中看不出来"的根因。
-  - 设计：`submitted → running → done / failed / timeout` 显式状态机。
-  - dispatcher 开跑/跑完/出错 经 NATS 回写状态（新增 `sundynix.tasks.status` 或复用 exec 流），网关落 PG 并推给 UI。
-  - 收益：管理端「服务状态」/ 桌面端能看到任务真实进度，超时自动翻红，无需人工猜。
-  - 与下面同源：Eino compose 的 graph state + 节点级状态正好承载它。
+- **任务生命周期 FSM** ✅ 已完成：`submitted → running → done / failed / timeout` 显式状态机。
+  - dispatcher 经新主题 `sundynix.tasks.status` 回写（`TaskStatusEvent`）：进入执行→running、收尾→done/failed、整体超时上限 `taskExecTimeout=3min`→timeout；网关 `SubscribeTaskStatus` 落 PG（`Task.Status/Detail`）。
+  - UI 轮询：`GET /api/v1/tasks/:id` 返回 `{status, detail}`。
+  - 验收：submitted→running→done 实测流转、PG 持久化、3min 超时兜底——根治"卡运行中看不出来"。
+  - 桌面端轮询接线仍可补（当前后端 + 端点已就绪）。
 - **中断/恢复（HITL）**：审批型工业流程（生成中途人工确认）。需 checkpoint 持久化（PG/Redis）。等有具体审批用例再做。
 - **多智能体协同**（`flow/agent/multiagent`）：出现真实多角色编排需求时再上，现在无用例。

@@ -8,6 +8,9 @@

 ## 第 1 层 · CLIENT（sundynix-desktop）

+
+
+
 - [x] React 19 + TypeScript + Tailwind 工业级 UI（🟡 自建 UI primitives，未用 shadcn —— 与架构图有偏差）
 - [x] React Flow 编排画布 + JSON DSL 导出（含 branch 真/假边手柄）
 - [x] Wails 本地 Go 运行时 + TS/Go 强绑定 + 本地文件 I/O（另存为 / 系统打开 / 系统通知）
@@ -0,0 +1,62 @@
+# 发版清单（桌面端 Release）
+
+打 `vX.Y.Z` 标签即触发 [`.github/workflows/release.yml`](.github/workflows/release.yml)：
+GitHub 用 macOS/Windows runner `wails build` 出安装包并发布到 **Releases**；
+旧版用户的 App 启动时查 `releases/latest`，发现新版即弹横幅提示下载
+（逻辑见 [`version.ts`](sundynix-desktop/frontend/src/lib/version.ts)）。
+
+> 版本号一律用语义化 `主.次.补`（如 `0.1.1`）；git tag 带 `v` 前缀（`v0.1.1`），App 内不带。
+
+---
+
+## 一、发版前：改版本号（必须三处对齐）
+
+| 文件 | 字段 | 作用 |
+|---|---|---|
+| `sundynix-desktop/frontend/src/lib/version.ts` | `APP_VERSION` | **功能性**：App 拿它和 GitHub 最新版比，决定是否提示更新 |
+| `sundynix-desktop/frontend/package.json` | `version` | 前端包版本（保持一致，便于追溯） |
+
+> （`wails.json` 当前无版本字段；要在原生包元数据里带版本，可自行加 `info.productVersion`。）
+
+⚠️ **`APP_VERSION` 必须等于即将打的 tag（去掉 v）**，否则更新提示会错乱：
+新包内嵌的 APP_VERSION = 新版本；旧用户 App 内是旧版本，比对才会提示。
+
+## 二、发版步骤
+
+```bash
+# 1. 改完上面三处版本号，提交
+git add -A && git commit -m "release: v0.1.1"
+
+# 2. 先 push 代码（release 从被打标签的提交构建；同时触发 CI 回归）
+git push origin main          # 或当前分支 dev
+
+# 3. 打标签 + push 标签 → 触发 release 构建
+git tag -a v0.1.1 -m "v0.1.1：<一句话变更>"
+git push origin v0.1.1
+```
+
+## 三、验证
+
+1. GitHub → **Actions** → 看 `Release` 工作流 macOS / Windows 两个 job 是否绿。
+2. GitHub → **Releases** → 确认 `v0.1.1` 下有两个资产：
+   - `sundynix_desktop_macos_universal.zip`
+   - `sundynix_desktop_windows_amd64.exe`
+3. 用一台装了**旧版**的机器打开 App → 顶部应出现「新版本 v0.1.1 可用」横幅。
+
+## 四、出错回滚 / 重发
+
+```bash
+# 删标签（本地 + 远程），改完重打
+git tag -d v0.1.1
+git push origin :refs/tags/v0.1.1
+# 修复后重新执行「二、发版步骤」
+```
+
+## 五、注意事项
+
+- **仓库需 public**：Release 资产要让终端用户直接下载；private 仓库的资产下载需 token，不适合分发。
+- **代码签名未做**（待办）：
+  - macOS 未公证 → 用户首次打开被 Gatekeeper 拦，需「右键 → 打开」或系统设置放行。正式分发需 Apple 开发者证书 + `notarize`。
+  - Windows 未签名 → SmartScreen 警告。需代码签名证书。
+- **GitHub API 限流**：更新检查未认证 60 次/小时/IP，按"每用户启动查一次"完全够用。
+- CI（`ci.yml`）在 push 到 `main`/`dev` 或 PR 时自动跑回归；Release 仅在 push `v*` 标签时触发。
@@ -12,6 +12,7 @@ import { RunsView } from "./views/RunsView";
 import { Home } from "./views/Home";
 import { Placeholder } from "./views/Placeholder";
 import { CommandPalette, type Command } from "./components/CommandPalette";
+import { UpdateBanner } from "./components/UpdateBanner";
 import { Login } from "./views/Login";
 import { submitTask, streamTokens, streamExec, authMe, logout, type Identity, type AuthUser } from "./lib/api";
 import type { TaskDsl } from "./lib/dsl";
@@ -158,6 +159,7 @@ export default function App() {
        className="pointer-events-none absolute inset-x-0 top-0 h-64 opacity-60"
        style={{ background: "radial-gradient(60% 100% at 50% 0%, rgba(124,92,246,0.10), transparent 70%)" }}
      />
+      <UpdateBanner />
      <TopBar user={user} onLogout={onLogout} onCommand={() => setCmdOpen(true)} />
      <div className="relative flex min-h-0 flex-1">
        <LeftNav active={view} onSelect={setView} />
@@ -0,0 +1,38 @@
+import { useEffect, useState } from "react";
+import { Download, X } from "lucide-react";
+import { checkUpdate, openExternal, type ReleaseInfo } from "../lib/version";
+
+// 启动时查 GitHub Releases，有新版则在顶部显示一条可关闭的横幅，点击去下载页。
+export function UpdateBanner() {
+  const [rel, setRel] = useState<ReleaseInfo | null>(null);
+  const [dismissed, setDismissed] = useState(false);
+
+  useEffect(() => {
+    let alive = true;
+    void checkUpdate().then((r) => {
+      if (alive) setRel(r);
+    });
+    return () => {
+      alive = false;
+    };
+  }, []);
+
+  if (!rel || dismissed) return null;
+  return (
+    <div className="flex shrink-0 items-center gap-3 border-b border-brand/30 bg-brand/10 px-4 py-1.5 text-[12px] text-brand-200">
+      <Download className="h-3.5 w-3.5 text-brand-400" />
+      <span>
+        新版本 <span className="font-semibold">v{rel.version}</span> 可用
+      </span>
+      <button
+        onClick={() => openExternal(rel.url)}
+        className="rounded bg-brand/20 px-2 py-0.5 text-brand-200 hover:bg-brand/30"
+      >
+        前往下载
+      </button>
+      <button onClick={() => setDismissed(true)} className="ml-auto text-slate-500 hover:text-slate-300" title="忽略">
+        <X className="h-3.5 w-3.5" />
+      </button>
+    </div>
+  );
+}
@@ -0,0 +1,50 @@
+// 桌面端版本与"检查更新"——查 GitHub Releases 最新版，比当前版本新则提示下载。
+// 发版流程：bump 此处 APP_VERSION（= package.json version）→ 打 tag vX.Y.Z → release workflow 自动构建+发布。
+
+export const APP_VERSION = "0.1.0"; // 当前桌面端版本（与 git tag 去掉 v 前缀对齐）
+
+const REPO = "blizzardzhang/sundynix-agentix";
+
+export interface ReleaseInfo {
+  version: string; // 最新版本号（不含 v）
+  url: string; // release 页面（用户在此选 mac/win 安装包）
+  notes: string; // 更新日志
+}
+
+// checkUpdate 查 GitHub 最新 release；有比当前更新的版本则返回信息，否则 null（含出错/限流时静默）。
+export async function checkUpdate(): Promise<ReleaseInfo | null> {
+  try {
+    const res = await fetch(`https://api.github.com/repos/${REPO}/releases/latest`, {
+      headers: { Accept: "application/vnd.github+json" },
+    });
+    if (!res.ok) return null; // 404=还没发过 release / 403=限流 → 静默
+    const d = (await res.json()) as { tag_name?: string; html_url?: string; body?: string };
+    const latest = String(d.tag_name ?? "").replace(/^v/, "");
+    if (!latest || !isNewer(latest, APP_VERSION)) return null;
+    return { version: latest, url: d.html_url ?? `https://github.com/${REPO}/releases/latest`, notes: d.body ?? "" };
+  } catch {
+    return null; // 离线等 → 不打扰
+  }
+}
+
+// openExternal 在系统浏览器打开链接（Wails 用 runtime.BrowserOpenURL，浏览器模式回退 window.open）。
+export function openExternal(url: string): void {
+  const w = window as unknown as { runtime?: { BrowserOpenURL?: (u: string) => void } };
+  if (w.runtime?.BrowserOpenURL) {
+    w.runtime.BrowserOpenURL(url);
+  } else {
+    window.open(url, "_blank");
+  }
+}
+
+// isNewer 语义化版本比较：a 是否比 b 新（按 major.minor.patch）。
+function isNewer(a: string, b: string): boolean {
+  const pa = a.split(".").map((n) => parseInt(n, 10) || 0);
+  const pb = b.split(".").map((n) => parseInt(n, 10) || 0);
+  for (let i = 0; i < 3; i++) {
+    const x = pa[i] ?? 0;
+    const y = pb[i] ?? 0;
+    if (x !== y) return x > y;
+  }
+  return false;
+}
@@ -58,8 +58,9 @@ export const NODE_KINDS: Record<string, NodeKind> = {
      { key: "model", label: "模型", type: "select", options: ["占位 Pool", "ollama:qwen", "vllm:custom"], required: true },
      { key: "system", label: "系统提示词", type: "textarea", placeholder: "你是…" },
      { key: "temperature", label: "温度", type: "number" },
+      { key: "autonomous", label: "自主工具 (ReAct)", type: "checkbox" },
    ],
-    defaults: { model: "占位 Pool", system: "", temperature: 0.7 },
+    defaults: { model: "占位 Pool", system: "", temperature: 0.7, autonomous: false },
  },
  tool: {
    kind: "tool",
@@ -41,8 +41,8 @@ func main() {
 		log.Printf("[dispatcher] subscribe model config: %v", err)
 	}

-	// sub 同时作为 Token 回流出口（TokenSink）、MCP 工具调用出口（ToolCaller）与执行事件出口（ExecSink）。
-	orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub)
+	// sub 同时作为 Token 回流(TokenSink)、MCP 工具调用(ToolCaller)、执行事件(ExecSink)与任务状态回写(StatusSink)出口。
+	orch, err := eino.NewOrchestrator(pool, breaker, eval, sub, sub, sub, sub)
 	if err != nil {
 		log.Fatalf("[dispatcher] build eino graph: %v", err)
 	}
@@ -0,0 +1,46 @@
+package eino
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/sundynix/sundynix-dispatcher/internal/harness"
+	"github.com/sundynix/sundynix-dispatcher/internal/llm"
+	"github.com/sundynix/sundynix-shared/contract"
+)
+
+// TestAgentCollaborationPassesOutput 验证编排式多智能体接力：
+// agent1 的产出注入 agent2 的上下文，agent2 基于它继续；最终成稿 = 下游 agent 的产出。
+func TestAgentCollaborationPassesOutput(t *testing.T) {
+	graph := `{"version":"1","nodes":[
+		{"id":"a1","kind":"agent","config":{"system":"研究"}},
+		{"id":"a2","kind":"agent","config":{"system":"撰写"}}
+	],"edges":[{"source":"a1","target":"a2"}]}`
+
+	saw := false
+	ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string {
+		var sys string
+		for _, x := range m {
+			if x.Role == "system" {
+				sys = x.Content
+			}
+		}
+		if strings.Contains(sys, "研究产出XYZ") { // agent1 的产出出现在 agent2 的 system → 接力成功
+			saw = true
+			return "最终报告：基于上游研究撰写"
+		}
+		return "研究产出XYZ"
+	}}
+	o := &Orchestrator{pool: ll, breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
+	ans, err := o.runGraph(context.Background(), &contract.Task{ID: "tc", Graph: []byte(graph)}, &execTracer{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !saw {
+		t.Fatal("下游 agent 未看到上游 agent 的产出——协作没接力")
+	}
+	if !strings.Contains(ans, "最终报告") {
+		t.Fatalf("最终成稿应为下游 agent 产出，得 %q", ans)
+	}
+}
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"strings"

+	"github.com/cloudwego/eino/components/prompt"
 	"github.com/cloudwego/eino/schema"
 )

@@ -18,10 +19,20 @@ type RunCtx struct {
 	Profile   string            // 召回的画像
 	History   []*schema.Message // 短期历史
 	ToolOut   []string          // 工具/检索节点产出（含参考资料）
+	Upstream  []string          // 前序协作 agent 的产出（多 agent 接力时注入，让下游基于上游继续）
 }

+// chatTemplate 是会话消息模板：系统提示词 + 历史占位 + 用户输入。
+// 用 Eino components/prompt.ChatTemplate（FString 仅解析模板串、值原样注入，故 JSON 花括号安全）。
+var chatTemplate = prompt.FromMessages(schema.FString,
+	schema.SystemMessage("{system}"),
+	schema.MessagesPlaceholder("history", true),
+	schema.UserMessage("{query}"),
+)
+
 // buildMessages 把上下文组装为发给模型的消息序列（系统提示词 + 画像 + 工具产出 + 历史 + 用户输入）。
-func buildMessages(_ context.Context, rc *RunCtx) ([]*schema.Message, error) {
+// 系统串的动态拼装（按需注入画像/参考资料）留在 Go 侧；最终经 ChatTemplate 成型。
+func buildMessages(ctx context.Context, rc *RunCtx) ([]*schema.Message, error) {
 	var sys strings.Builder
 	sys.WriteString(rc.System)
 	if rc.Profile != "" {
@@ -33,11 +44,15 @@ func buildMessages(_ context.Context, rc *RunCtx) ([]*schema.Message, error) {
 		sys.WriteString("\n\n以下是工具/检索得到的参考资料：\n")
 		sys.WriteString(strings.Join(rc.ToolOut, "\n---\n"))
 	}
-	msgs := make([]*schema.Message, 0, len(rc.History)+2)
-	msgs = append(msgs, schema.SystemMessage(sys.String()))
-	msgs = append(msgs, rc.History...)
-	msgs = append(msgs, schema.UserMessage(rc.Query))
-	return msgs, nil
+	if len(rc.Upstream) > 0 {
+		sys.WriteString("\n\n以下是前序协作 agent 的产出，请在此基础上继续完成你的部分（不要重头再来）：\n")
+		sys.WriteString(strings.Join(rc.Upstream, "\n---\n"))
+	}
+	return chatTemplate.Format(ctx, map[string]any{
+		"system":  sys.String(),
+		"history": rc.History,
+		"query":   rc.Query,
+	})
 }

 // previewArgs 把工具入参压成一行短预览。
@@ -0,0 +1,68 @@
+package eino
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cloudwego/eino/callbacks"
+	"github.com/cloudwego/eino/components/model"
+	"github.com/cloudwego/eino/components/tool"
+	"github.com/cloudwego/eino/schema"
+	ucallbacks "github.com/cloudwego/eino/utils/callbacks"
+)
+
+// composeTracer 把 Eino compose 运行时的回调（ChatModel / Tool 的 start/end/error）
+// 翻译成我们现有的 ExecEvent 轨迹——可观测"归一"：用框架原生回调，而非各处手写 emit。
+// node 为该次运行在"运行·观测"里的归属节点 id（如 agent:xxx）。
+func composeTracer(tr *execTracer, node string) callbacks.Handler {
+	return ucallbacks.NewHandlerHelper().
+		ChatModel(&ucallbacks.ModelCallbackHandler{
+			OnStart: func(ctx context.Context, _ *callbacks.RunInfo, in *model.CallbackInput) context.Context {
+				tr.emit(node, "model", "start", "compose·ChatModel", inputMsgsPreview(in), 0)
+				return ctx
+			},
+			OnEnd: func(ctx context.Context, _ *callbacks.RunInfo, out *model.CallbackOutput) context.Context {
+				detail := ""
+				if out != nil && out.Message != nil {
+					detail = truncate(out.Message.Content, 120)
+				}
+				tr.emit(node, "model", "end", "compose·ChatModel", detail, 0)
+				return ctx
+			},
+			OnEndWithStreamOutput: func(ctx context.Context, _ *callbacks.RunInfo, out *schema.StreamReader[*model.CallbackOutput]) context.Context {
+				out.Close() // 仅观测：正文经主输出流消费，这里只标记结束
+				tr.emit(node, "model", "end", "compose·ChatModel", "流式完成", 0)
+				return ctx
+			},
+			OnError: func(ctx context.Context, _ *callbacks.RunInfo, err error) context.Context {
+				tr.emit(node, "model", "error", "compose·ChatModel", err.Error(), 0)
+				return ctx
+			},
+		}).
+		Tool(&ucallbacks.ToolCallbackHandler{
+			OnStart: func(ctx context.Context, _ *callbacks.RunInfo, in *tool.CallbackInput) context.Context {
+				if in != nil {
+					tr.emit(node, "tool", "start", "compose·工具", truncate(in.ArgumentsInJSON, 120), 0)
+				}
+				return ctx
+			},
+			OnEnd: func(ctx context.Context, _ *callbacks.RunInfo, out *tool.CallbackOutput) context.Context {
+				if out != nil {
+					tr.emit(node, "tool", "end", "compose·工具", truncate(out.Response, 160), 0)
+				}
+				return ctx
+			},
+			OnError: func(ctx context.Context, _ *callbacks.RunInfo, err error) context.Context {
+				tr.emit(node, "tool", "error", "compose·工具", err.Error(), 0)
+				return ctx
+			},
+		}).
+		Handler()
+}
+
+func inputMsgsPreview(in *model.CallbackInput) string {
+	if in == nil {
+		return ""
+	}
+	return fmt.Sprintf("%d 条消息", len(in.Messages))
+}
@@ -0,0 +1,217 @@
+package eino
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/cloudwego/eino/compose"
+
+	"github.com/sundynix/sundynix-dispatcher/internal/dsl"
+	"github.com/sundynix/sundynix-shared/contract"
+)
+
+// flowSignal 是 compose 编排图的边载荷（占位）：真实数据全走 compose 本地状态(*board)，
+// 边只传"该走了"的信号。注册 no-op 合并以支持 fan-in（多分支汇聚到一个节点）。
+type flowSignal struct{}
+
+var registerMergeOnce sync.Once
+
+func registerFlowMerge() {
+	registerMergeOnce.Do(func() {
+		compose.RegisterValuesMergeFunc(func([]flowSignal) (flowSignal, error) { return flowSignal{}, nil })
+	})
+}
+
+// executeGraph 按灰度开关选编排实现：compose.Graph（Phase C）或自研 graph.go（默认/权威）。
+func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, error) {
+	if composeEnabled() {
+		return o.runComposeGraph(ctx, t, tr)
+	}
+	return o.runGraph(ctx, t, tr)
+}
+
+// runComposeGraph 把 DSL 图编译为 Eino compose.Graph 并执行（Phase C 编排归一）：
+// 节点体复用现有 execDSLNode；黑板进 compose 本地状态；branch 走 AddBranch；
+// DAG 触发模式让无依赖节点并行调度（效率）。编译失败即降级回自研 graph.go（安全网）。
+func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, error) {
+	registerFlowMerge()
+	flow, ferr := dsl.Parse(t.Graph)
+	plan := dsl.Compile(t.Graph)
+	b := &board{
+		uid:   meta(t, contract.MetaUserID),
+		sid:   meta(t, contract.MetaSessionID),
+		query: plan.Query,
+	}
+
+	// 无图/空图：退化为 compose 单轮对话。
+	if ferr != nil || flow == nil || len(flow.Nodes) == 0 {
+		tr.info("task", "system", "无结构化图", "按单轮对话执行（compose）")
+		b.profile = o.fetchMemory(ctx, b.uid, b.query)
+		b.history = o.fetchHistory(ctx, b.sid)
+		o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent")
+		return b.answer, nil
+	}
+
+	// 邻接 + 入度（只认两端都存在的边）。
+	nodeByID := make(map[string]dsl.Node, len(flow.Nodes))
+	outE := make(map[string][]dsl.Edge)
+	indeg := make(map[string]int, len(flow.Nodes))
+	for _, n := range flow.Nodes {
+		nodeByID[n.ID] = n
+		indeg[n.ID] = 0
+	}
+	for _, e := range flow.Edges {
+		if _, ok := nodeByID[e.Source]; !ok {
+			continue
+		}
+		if _, ok := nodeByID[e.Target]; !ok {
+			continue
+		}
+		outE[e.Source] = append(outE[e.Source], e)
+		indeg[e.Target]++
+	}
+
+	// 图里无 memory 节点 → 沿用默认：注入画像+历史（与 graph.go 对齐，避免回归）。
+	hasMemory := false
+	for _, n := range flow.Nodes {
+		if n.Kind == "memory" {
+			hasMemory = true
+			break
+		}
+	}
+	if !hasMemory {
+		b.profile = o.fetchMemory(ctx, b.uid, b.query)
+		b.history = o.fetchHistory(ctx, b.sid)
+	}
+
+	// 建 compose 图：黑板进本地状态（GenLocalState 闭包持有本任务的 b）。
+	g := compose.NewGraph[flowSignal, flowSignal](
+		compose.WithGenLocalState(func(context.Context) *board { return b }),
+	)
+	key := func(id string) string { return "n_" + id } // 节点 key 加前缀，避开 START/END 保留字
+
+	// 1) 加节点（branch 为 passthrough，路由交给 AddBranch）。
+	for _, n := range flow.Nodes {
+		node := n
+		if node.Kind == "branch" {
+			_ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda(
+				func(context.Context, flowSignal) (flowSignal, error) { return flowSignal{}, nil }))
+			continue
+		}
+		_ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda(
+			func(c context.Context, _ flowSignal) (flowSignal, error) {
+				perr := compose.ProcessState(c, func(sc context.Context, bd *board) error {
+					o.execDSLNode(sc, t, node, bd, plan, tr)
+					return nil
+				})
+				return flowSignal{}, perr
+			}))
+	}
+
+	// 2) 连边。branch 用 AddBranch（条件读 board 选下游）；其余直连；终端节点连 END。
+	for _, n := range flow.Nodes {
+		node := n
+		outs := outE[node.ID]
+		if node.Kind == "branch" {
+			endNodes := map[string]bool{compose.END: true}
+			for _, e := range outs {
+				if _, ok := nodeByID[e.Target]; ok {
+					endNodes[key(e.Target)] = true
+				}
+			}
+			brn := node
+			cond := func(c context.Context, _ flowSignal) (map[string]bool, error) {
+				chosen := map[string]bool{}
+				_ = compose.ProcessState(c, func(sc context.Context, bd *board) error {
+					for _, tgt := range o.branchNode(brn, bd, outE[brn.ID], nodeByID, tr) {
+						chosen[key(tgt)] = true
+					}
+					return nil
+				})
+				if len(chosen) == 0 {
+					chosen[compose.END] = true // 没选中任何下游 → 收口到 END，避免悬挂
+				}
+				return chosen, nil
+			}
+			_ = g.AddBranch(key(node.ID), compose.NewGraphMultiBranch(cond, endNodes))
+			continue
+		}
+		if len(outs) == 0 {
+			_ = g.AddEdge(key(node.ID), compose.END)
+			continue
+		}
+		for _, e := range outs {
+			if _, ok := nodeByID[e.Target]; ok {
+				_ = g.AddEdge(key(node.ID), key(e.Target))
+			}
+		}
+	}
+
+	// 3) 入口节点（入度 0）连 START。
+	for _, n := range flow.Nodes {
+		if indeg[n.ID] == 0 {
+			_ = g.AddEdge(compose.START, key(n.ID))
+		}
+	}
+
+	// 4) 编译（DAG 模式：无依赖节点并行调度）。编译失败 → 降级回自研 graph.go（安全网）。
+	r, cerr := g.Compile(ctx, compose.WithNodeTriggerMode(compose.AllPredecessor))
+	if cerr != nil {
+		tr.info("task", "system", "compose 编译失败", "退回自研 graph.go："+cerr.Error())
+		return o.runGraph(ctx, t, tr)
+	}
+	if _, ierr := r.Invoke(ctx, flowSignal{}); ierr != nil {
+		tr.info("task", "system", "compose 执行告警", ierr.Error()) // 副作用已落 board；下方按需补一段答复
+	}
+
+	// 图里无 agent 节点（纯工具/检索图）也要出一段答复。
+	if b.answer == "" {
+		o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent")
+	}
+	return b.answer, nil
+}
+
+// execDSLNode 执行一个非 branch 的 DSL 节点（compose 编译器用；节点体与 graph.go 一致，
+// 区别仅 agent 直接走 runAgent/runReactAgent——compose 已是编排层，不再二次套 compose）。
+func (o *Orchestrator) execDSLNode(ctx context.Context, t *contract.Task, n dsl.Node, b *board, plan dsl.Plan, tr *execTracer) {
+	switch n.Kind {
+	case "input":
+		if txt := cstr(n.Config, "text"); txt != "" {
+			b.query = txt
+		}
+		tr.info("input:"+n.ID, "system", labelOf(n, "输入"), truncate(b.query, 80))
+	case "memory":
+		if cbool(n.Config, "profile") {
+			b.profile = o.fetchMemory(ctx, b.uid, b.query)
+		}
+		if cbool(n.Config, "history") {
+			b.history = o.fetchHistory(ctx, b.sid)
+		}
+		tr.info("memory:"+n.ID, "memory", labelOf(n, "记忆"),
+			fmt.Sprintf("画像 %d 字 · 历史 %d 条", len([]rune(b.profile)), len(b.history)))
+	case "retriever":
+		o.retrieverNode(ctx, n, b, tr)
+	case "tool":
+		o.execToolNode(ctx, t.ID, n, b, tr)
+	case "agent":
+		sys := firstNonEmpty(cstr(n.Config, "system"), plan.System)
+		if cbool(n.Config, "autonomous") {
+			o.runReactAgent(ctx, t.ID, b, sys, n, tr, "agent:"+n.ID)
+		} else {
+			o.runAgent(ctx, t.ID, b, sys, tr, "agent:"+n.ID)
+		}
+	case "aggregate":
+		merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...))
+		b.refs, b.toolOut = merged, nil
+		tr.info("aggregate:"+n.ID, "system", labelOf(n, "汇聚"), "策略："+firstNonEmpty(cstr(n.Config, "strategy"), "拼接"))
+	case "render":
+		o.renderNode(ctx, t.ID, n, b, tr)
+	case "map":
+		o.mapNode(ctx, t.ID, n, b, tr)
+	case "output":
+		tr.info("output:"+n.ID, "system", labelOf(n, "输出"), "目标："+firstNonEmpty(cstr(n.Config, "target"), "屏幕"))
+	default:
+		tr.info(n.Kind+":"+n.ID, "system", labelOf(n, n.Kind), "未识别节点，跳过")
+	}
+}
@@ -0,0 +1,76 @@
+package eino
+
+import (
+	"context"
+	"testing"
+
+	"github.com/sundynix/sundynix-dispatcher/internal/harness"
+	"github.com/sundynix/sundynix-dispatcher/internal/llm"
+	"github.com/sundynix/sundynix-shared/contract"
+)
+
+// echoLLM 回显最后一条 user 消息内容（确定性），便于两条执行路径逐字对比。
+func echoLLM() *fakeLLM {
+	return &fakeLLM{
+		ready: true,
+		stream: func(m []llm.ChatMessage) string {
+			for i := len(m) - 1; i >= 0; i-- {
+				if m[i].Role == "user" {
+					return "ANS:" + m[i].Content
+				}
+			}
+			return "ANS:"
+		},
+	}
+}
+
+func runBoth(t *testing.T, graph string) (interp, comp string) {
+	t.Helper()
+	task := &contract.Task{ID: "t_eq", Graph: []byte(graph)}
+
+	o1 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
+	a1, err := o1.runGraph(context.Background(), task, &execTracer{})
+	if err != nil {
+		t.Fatalf("runGraph: %v", err)
+	}
+	o2 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}}
+	a2, err := o2.runComposeGraph(context.Background(), task, &execTracer{})
+	if err != nil {
+		t.Fatalf("runComposeGraph: %v", err)
+	}
+	return a1, a2
+}
+
+// TestComposeEquivalentLinear 多节点线性图：input→memory→agent→output，两路径成稿应逐字一致。
+func TestComposeEquivalentLinear(t *testing.T) {
+	graph := `{"version":"1","nodes":[
+		{"id":"in","kind":"input","config":{"text":"什么是图编排"}},
+		{"id":"m","kind":"memory","config":{}},
+		{"id":"a","kind":"agent","config":{"system":"你是助手"}},
+		{"id":"out","kind":"output","config":{}}
+	],"edges":[
+		{"source":"in","target":"m"},{"source":"m","target":"a"},{"source":"a","target":"out"}
+	]}`
+	interp, comp := runBoth(t, graph)
+	if interp == "" || interp != comp {
+		t.Fatalf("线性图不等价: interp=%q compose=%q", interp, comp)
+	}
+}
+
+// TestComposeEquivalentBranch 分支图：input→branch→(真)A/(假)B，条件恒真应都走 A，两路径一致。
+func TestComposeEquivalentBranch(t *testing.T) {
+	graph := `{"version":"1","nodes":[
+		{"id":"in","kind":"input","config":{"text":"hi"}},
+		{"id":"br","kind":"branch","config":{"condition":""}},
+		{"id":"a","kind":"agent","config":{"system":"A"}},
+		{"id":"b","kind":"agent","config":{"system":"B"}}
+	],"edges":[
+		{"source":"in","target":"br"},
+		{"source":"br","target":"a","sourceHandle":"true"},
+		{"source":"br","target":"b","sourceHandle":"false"}
+	]}`
+	interp, comp := runBoth(t, graph)
+	if interp == "" || interp != comp {
+		t.Fatalf("分支图不等价: interp=%q compose=%q", interp, comp)
+	}
+}
@@ -0,0 +1,97 @@
+package eino
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/cloudwego/eino/compose"
+	"github.com/cloudwego/eino/schema"
+
+	"github.com/sundynix/sundynix-dispatcher/internal/harness"
+)
+
+// composeEnabled 报告是否启用 compose.Graph 编排路径（Phase C 灰度开关，默认关 → 走自研 graph.go）。
+// 并存策略：EINO_COMPOSE=1 时对话主流程改走 Eino compose 运行时，行为对齐后再逐步退役 graph.go。
+func composeEnabled() bool { return os.Getenv("EINO_COMPOSE") == "1" }
+
+// runConversation 是对话/模型节点的统一入口：按灰度开关选 compose.Graph 或自研 runAgent。
+// 二者对外行为一致（据黑板拼消息 → 流式回流 token → 累计成稿），便于等价回归。
+func (o *Orchestrator) runConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) {
+	if composeEnabled() {
+		o.runComposeConversation(ctx, taskID, b, system, tr, node)
+		return
+	}
+	o.runAgent(ctx, taskID, b, system, tr, node)
+}
+
+// runComposeConversation 用 Eino compose.Graph 跑对话主流程：
+// START → ChatModel 节点 → END，编译为 Runnable 后流式执行；可观测经 callbacks 桥到 ExecEvent。
+// 模型未就绪 / 编译失败时降级回自研 runAgent，保证不回归。
+func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) {
+	cm := o.pool.ChatModel()
+	if cm == nil {
+		o.runAgent(ctx, taskID, b, system, tr, node) // 无模型 → 走自研路径的降级桩
+		return
+	}
+
+	g := compose.NewGraph[[]*schema.Message, *schema.Message]()
+	if err := g.AddChatModelNode("model", cm); err != nil {
+		tr.info(node, "system", "compose 降级", "建图失败，退回自研路径："+err.Error())
+		o.runAgent(ctx, taskID, b, system, tr, node)
+		return
+	}
+	_ = g.AddEdge(compose.START, "model")
+	_ = g.AddEdge("model", compose.END)
+	r, err := g.Compile(ctx)
+	if err != nil {
+		tr.info(node, "system", "compose 降级", "编译失败，退回自研路径："+err.Error())
+		o.runAgent(ctx, taskID, b, system, tr, node)
+		return
+	}
+
+	rc := &RunCtx{
+		UserID: b.uid, SessionID: b.sid,
+		System:   firstNonEmpty(system, defaultAgentSystem),
+		Query:    b.query,
+		Profile:  b.profile,
+		History:  b.history,
+		ToolOut:  append(append([]string{}, b.toolOut...), b.refs...),
+		Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力
+	}
+	msgs, _ := buildMessages(ctx, rc)
+
+	t0 := time.Now()
+	// ChatModel 的 start/end 由 composeTracer（callbacks）落轨迹，这里不再手写 emit（归一）。
+	sr, err := r.Stream(ctx, msgs, compose.WithCallbacks(composeTracer(tr, node)))
+	if err != nil {
+		tr.emit(node, "model", "error", "compose 图执行", err.Error(), time.Since(t0).Milliseconds())
+		return
+	}
+	defer sr.Close()
+
+	chunks := 0
+	var produced strings.Builder // 本节点产出（供下游 agent 接力）
+	for {
+		chunk, rerr := sr.Recv()
+		if rerr == io.EOF {
+			break
+		}
+		if rerr != nil {
+			tr.emit(node, "model", "error", "compose 图执行", rerr.Error(), time.Since(t0).Milliseconds())
+			return
+		}
+		if chunk.Content == "" {
+			continue
+		}
+		safe, _ := harness.RedactSecrets(chunk.Content) // 输出护栏：逐片脱敏
+		_ = o.sink.PublishToken(taskID, []byte(safe))
+		produced.WriteString(safe)
+		chunks++
+	}
+	o.recordAgentOutput(b, produced.String())
+	tr.info(node, "system", "compose 图", fmt.Sprintf("%d 段输出 / %d 字（Eino compose 运行时）", chunks, len([]rune(produced.String()))))
+}
@@ -0,0 +1,89 @@
+package eino
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/cloudwego/eino/components/model"
+	"github.com/cloudwego/eino/compose"
+	"github.com/cloudwego/eino/schema"
+
+	"github.com/sundynix/sundynix-dispatcher/internal/harness"
+	"github.com/sundynix/sundynix-dispatcher/internal/llm"
+)
+
+// stubModel 是实现 Eino model.BaseChatModel 的测试桩，固定回一段文本（确定性）。
+type stubModel struct{ reply string }
+
+func (s *stubModel) Generate(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.Message, error) {
+	return schema.AssistantMessage(s.reply, nil), nil
+}
+
+func (s *stubModel) Stream(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.StreamReader[*schema.Message], error) {
+	sr, sw := schema.Pipe[*schema.Message](1)
+	go func() {
+		sw.Send(schema.AssistantMessage(s.reply, nil), nil)
+		sw.Close()
+	}()
+	return sr, nil
+}
+
+// TestComposeGraphRuns 直接验证 compose.Graph（START→ChatModel→END）能编译并流式产出。
+func TestComposeGraphRuns(t *testing.T) {
+	g := compose.NewGraph[[]*schema.Message, *schema.Message]()
+	if err := g.AddChatModelNode("model", &stubModel{reply: "你好世界"}); err != nil {
+		t.Fatal(err)
+	}
+	_ = g.AddEdge(compose.START, "model")
+	_ = g.AddEdge("model", compose.END)
+	r, err := g.Compile(context.Background())
+	if err != nil {
+		t.Fatalf("compile: %v", err)
+	}
+	out, err := r.Invoke(context.Background(), []*schema.Message{schema.UserMessage("hi")})
+	if err != nil {
+		t.Fatalf("invoke: %v", err)
+	}
+	if out.Content != "你好世界" {
+		t.Fatalf("got %q", out.Content)
+	}
+}
+
+// TestComposeConversationEquivalent 验证 EINO_COMPOSE=1 时对话主流程走 compose 路径，
+// 把模型输出流式回流到 sink 并累计成稿——与自研 runAgent 行为等价（都逐片转发模型输出）。
+func TestComposeConversationEquivalent(t *testing.T) {
+	t.Setenv("EINO_COMPOSE", "1")
+	fs := &fakeSink{}
+	o := &Orchestrator{
+		pool:    &fakeLLM{ready: true, cm: &stubModel{reply: "我是 compose 路径的回答"}},
+		breaker: harness.NewCircuitBreaker(),
+		sink:    fs,
+	}
+	b := &board{query: "你好"}
+	tr := &execTracer{} // sink 为 nil → 轨迹发射空操作
+	o.runConversation(context.Background(), "task_compose", b, "", tr, "agent")
+
+	if !strings.Contains(b.answer, "compose 路径") {
+		t.Fatalf("成稿未含模型输出: %q", b.answer)
+	}
+	if !strings.Contains(fs.text(), "compose 路径") {
+		t.Fatalf("sink 未收到流式 token: %q", fs.text())
+	}
+}
+
+// TestComposeDisabledUsesRunAgent 验证开关关闭时仍走自研 runAgent（默认行为不变）。
+func TestComposeDisabledUsesRunAgent(t *testing.T) {
+	t.Setenv("EINO_COMPOSE", "0")
+	fs := &fakeSink{}
+	o := &Orchestrator{
+		pool:    &fakeLLM{ready: true, stream: func([]llm.ChatMessage) string { return "自研路径回答" }},
+		breaker: harness.NewCircuitBreaker(),
+		sink:    fs,
+	}
+	b := &board{query: "你好"}
+	o.runConversation(context.Background(), "task_legacy", b, "", &execTracer{}, "agent")
+	if !strings.Contains(b.answer, "自研路径") {
+		t.Fatalf("未走自研路径: %q", b.answer)
+	}
+}
@@ -0,0 +1,62 @@
+package eino
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+
+	"github.com/cloudwego/eino/components/retriever"
+	"github.com/cloudwego/eino/schema"
+
+	"github.com/sundynix/sundynix-shared/contract"
+)
+
+const defaultRetrieveTopK = 4
+
+// ragRetriever 把 mcp-go 的混合检索（NATS kb_search）包成 Eino components/retriever.Retriever。
+// kb 在构造时绑定（owner 作用域库名）；查询经 NATS 调 mcp-go，命中转 *schema.Document。
+type ragRetriever struct {
+	caller ToolCaller
+	kb     string
+	topK   int
+}
+
+// newRetriever 构造一个绑定到指定库的 Eino Retriever 组件。
+func (o *Orchestrator) newRetriever(kb string) retriever.Retriever {
+	return &ragRetriever{caller: o.tools, kb: kb, topK: defaultRetrieveTopK}
+}
+
+// Retrieve 实现 retriever.Retriever：经 NATS kb_search 检索，返回命中文档（不可用时降级空）。
+func (r *ragRetriever) Retrieve(ctx context.Context, query string, _ ...retriever.Option) ([]*schema.Document, error) {
+	if r.caller == nil || r.kb == "" {
+		return nil, nil
+	}
+	topK := r.topK
+	if topK <= 0 {
+		topK = defaultRetrieveTopK
+	}
+	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
+	defer cancel()
+	res, err := r.caller.CallTool(cctx, contract.ToolSubjectGo("kb_search"), &contract.ToolCall{
+		Tool: "kb_search", Args: map[string]any{"kb": r.kb, "q": query, "topK": topK},
+	})
+	if err != nil || res == nil || !res.OK || res.Content == "" || res.Content == "[]" {
+		return nil, nil
+	}
+	var hits []struct {
+		Text  string  `json:"text"`
+		Score float64 `json:"score"`
+	}
+	if json.Unmarshal([]byte(res.Content), &hits) != nil {
+		// 非结构化命中：整体作为单篇文档返回（与旧行为对齐）。
+		return []*schema.Document{{Content: res.Content}}, nil
+	}
+	docs := make([]*schema.Document, 0, len(hits))
+	for _, h := range hits {
+		docs = append(docs, &schema.Document{
+			Content:  strings.TrimSpace(h.Text),
+			MetaData: map[string]any{"score": h.Score},
+		})
+	}
+	return docs, nil
+}
@@ -0,0 +1,56 @@
+package eino
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/cloudwego/eino/schema"
+
+	"github.com/sundynix/sundynix-shared/contract"
+)
+
+// TestBuildMessagesShape 锁定 ChatTemplate 产出的消息序列与旧手拼逻辑等价：
+// [system(含画像+参考资料)] + [历史...] + [user(query)]，且 JSON 花括号不被误解析。
+func TestBuildMessagesShape(t *testing.T) {
+	rc := &RunCtx{
+		System:  "你是助手",
+		Query:   "今天几号",
+		Profile: "称呼=Dexter",
+		History: []*schema.Message{schema.UserMessage("上一句"), schema.AssistantMessage("上一答", nil)},
+		ToolOut: []string{`{"k":"v"}`}, // 含花括号，验证 FString 不误解析
+	}
+	msgs, err := buildMessages(context.Background(), rc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(msgs) != 4 {
+		t.Fatalf("应为 system+2历史+user=4 条，实际 %d", len(msgs))
+	}
+	if msgs[0].Role != schema.System || !strings.Contains(msgs[0].Content, "你是助手") ||
+		!strings.Contains(msgs[0].Content, "称呼=Dexter") || !strings.Contains(msgs[0].Content, `{"k":"v"}`) {
+		t.Fatalf("system 消息不对: %q", msgs[0].Content)
+	}
+	if msgs[3].Role != schema.User || msgs[3].Content != "今天几号" {
+		t.Fatalf("user 消息不对: %q", msgs[3].Content)
+	}
+}
+
+// fakeCaller 是 ToolCaller 测试替身，固定返回 kb_search 命中。
+type fakeCaller struct{ content string }
+
+func (f *fakeCaller) CallTool(_ context.Context, _ string, _ *contract.ToolCall) (*contract.ToolResult, error) {
+	return &contract.ToolResult{OK: true, Content: f.content}, nil
+}
+
+// TestRagRetrieverComponent 验证 Retriever 组件把 kb_search 命中转成 *schema.Document。
+func TestRagRetrieverComponent(t *testing.T) {
+	o := &Orchestrator{tools: &fakeCaller{content: `[{"text":"片段A","score":0.9},{"text":"片段B","score":0.8}]`}}
+	docs, err := o.newRetriever("u/kb").Retrieve(context.Background(), "q")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(docs) != 2 || docs[0].Content != "片段A" || docs[1].Content != "片段B" {
+		t.Fatalf("文档解析不对: %+v", docs)
+	}
+}
@@ -27,7 +27,8 @@ type board struct {
 	refs     []string        // 检索 / 聚合得到的参考资料
 	toolOut  []string        // 工具节点产出
 	sections []reportSection // map 并行 fan-out 产出的分项成稿（供 render 多章渲染）
-	answer   string          // 终端 agent / map 的成稿（流式累计）
+	answer   string          // 当前成稿（多 agent 协作时 = 最近一个 agent 的产出 = 成品）
+	agentOut []string        // 各上游 agent 的产出（按序），注入下游 agent 上下文以实现接力协作
 }

 // runGraph 按 DSL 图的真实拓扑与连线执行（替代旧的线性拍平 compileFlow）。
@@ -50,7 +51,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
 		tr.info("task", "system", "无结构化图", "按单轮对话执行")
 		b.profile = o.fetchMemory(ctx, b.uid, b.query)
 		b.history = o.fetchHistory(ctx, b.sid)
-		o.runAgent(ctx, t.ID, b, plan.System, tr, "agent")
+		o.runConversation(ctx, t.ID, b, plan.System, tr, "agent")
 		return b.answer, nil
 	}

@@ -123,7 +124,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
 			if cbool(n.Config, "autonomous") { // 开启自主工具 → ReAct（模型自己选工具）
 				o.runReactAgent(ctx, t.ID, b, sys, n, tr, "agent:"+n.ID)
 			} else {
-				o.runAgent(ctx, t.ID, b, sys, tr, "agent:"+n.ID)
+				o.runConversation(ctx, t.ID, b, sys, tr, "agent:"+n.ID)
 			}
 		case "aggregate":
 			merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...))
@@ -147,7 +148,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT

 	// 图里无 agent 节点（纯工具/检索图）也要出一段模型答复，否则没有输出。
 	if b.answer == "" {
-		o.runAgent(ctx, t.ID, b, plan.System, tr, "agent")
+		o.runConversation(ctx, t.ID, b, plan.System, tr, "agent")
 	}
 	return b.answer, nil
 }
@@ -230,16 +231,18 @@ func (o *Orchestrator) execToolNode(ctx context.Context, taskID string, n dsl.No
 // runAgent 执行 agent/模型节点：据黑板拼消息 → 流式回流 token → 累计成稿。
 func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) {
 	rc := &RunCtx{
-		System:  firstNonEmpty(system, defaultAgentSystem),
-		Query:   b.query,
-		Profile: b.profile,
-		History: b.history,
-		ToolOut: append(append([]string{}, b.toolOut...), b.refs...),
+		System:   firstNonEmpty(system, defaultAgentSystem),
+		Query:    b.query,
+		Profile:  b.profile,
+		History:  b.history,
+		ToolOut:  append(append([]string{}, b.toolOut...), b.refs...),
+		Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力
 	}
 	msgs, _ := buildMessages(ctx, rc)
 	tr.emit(node, "model", "start", "模型流式推理", "", 0)
 	t0 := time.Now()
 	n, redacted := 0, 0
+	var produced strings.Builder // 本节点自身产出（用于沿图向下游传递）
 	send := func(s string) {
 		if s == "" {
 			return
@@ -248,7 +251,7 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy
 		safe, hit := harness.RedactSecrets(s)
 		redacted += hit
 		_ = o.sink.PublishToken(taskID, []byte(safe))
-		b.answer += safe
+		produced.WriteString(safe)
 		n++
 	}
 	var err error
@@ -264,8 +267,19 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy
 	if redacted > 0 {
 		tr.info(node, "system", "输出护栏", fmt.Sprintf("已脱敏 %d 处疑似密钥/令牌", redacted))
 	}
+	o.recordAgentOutput(b, produced.String()) // 产出入黑板：成当前成稿 + 供下游接力
 	tr.emit(node, "model", "end", "模型流式推理",
-		fmt.Sprintf("%d tokens / %d 字", n, len([]rune(b.answer))), time.Since(t0).Milliseconds())
+		fmt.Sprintf("%d tokens / %d 字", n, len([]rune(produced.String()))), time.Since(t0).Milliseconds())
+}
+
+// recordAgentOutput 把一个 agent 节点的产出记入黑板：append 到 agentOut（供下游 agent 注入接力），
+// 并设为当前成稿 answer（多 agent 协作时，最后一个 agent 的产出即最终成品）。空产出忽略。
+func (o *Orchestrator) recordAgentOutput(b *board, out string) {
+	if strings.TrimSpace(out) == "" {
+		return
+	}
+	b.agentOut = append(b.agentOut, out)
+	b.answer = out
 }

 // renderNode 执行渲染节点：把当前成稿渲染成 Word（经 mcp-go report_render）。
@@ -20,6 +20,7 @@ type fakeLLM struct {
 	ready  bool
 	stream func(msgs []llm.ChatMessage) string          // ChatStream 要回流的整段文本
 	chat   func(msgs []llm.ChatMessage) (string, error) // Chat 返回
+	cm     model.BaseChatModel                          // compose 路径用的 Eino 模型（可为 nil）
 }

 func (f *fakeLLM) Ready() bool { return f.ready }
@@ -43,6 +44,9 @@ func (f *fakeLLM) Chat(_ context.Context, msgs []llm.ChatMessage) (string, error
 // ToolCallingModel：假模型不支持函数调用 → ReAct 路径会降级回普通对话。
 func (f *fakeLLM) ToolCallingModel() model.ToolCallingChatModel { return nil }

+// ChatModel：compose 路径用；nil 时 runComposeConversation 降级回 runAgent。
+func (f *fakeLLM) ChatModel() model.BaseChatModel { return f.cm }
+
 type fakeSink struct {
 	mu     sync.Mutex
 	tokens []string
@@ -4,6 +4,7 @@ package eino
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log"
 	"sync"
@@ -29,6 +30,11 @@ type ToolCaller interface {
 	CallTool(ctx context.Context, subject string, call *contract.ToolCall) (*contract.ToolResult, error)
 }

+// StatusSink 回写任务生命周期状态（由 NATS bus 实现；可为 nil → 不回写）。
+type StatusSink interface {
+	PublishTaskStatus(taskID, status, detail string) error
+}
+
 // LLM 是编排所需的语言模型能力（生产由 *llm.Pool 实现）。抽成接口便于测试注入假模型。
 type LLM interface {
 	Ready() bool
@@ -37,11 +43,16 @@ type LLM interface {
 	Chat(ctx context.Context, msgs []llm.ChatMessage) (string, error)
 	// ToolCallingModel 返回支持函数调用的模型（ReAct agent 用）；不支持则返回 nil。
 	ToolCallingModel() model.ToolCallingChatModel
+	// ChatModel 返回 Eino ChatModel 组件（compose.Graph 编排用）；未就绪则 nil。
+	ChatModel() model.BaseChatModel
 }

 // 工具调用超时；超时即降级（不带工具上下文继续推理）。
 const toolCallTimeout = 3 * time.Second

+// taskExecTimeout 是单个任务整体执行上限；超时即判 timeout（状态机），避免无限期"运行中"。
+const taskExecTimeout = 3 * time.Minute
+
 // Orchestrator 把每个 DSL 任务动态编译为 Eino 图并执行（记忆召回 → 工具节点 → 注入 → 流式）。
 type Orchestrator struct {
 	pool    LLM
@@ -50,19 +61,48 @@ type Orchestrator struct {
 	sink    TokenSink
 	tools   ToolCaller
 	exec    ExecSink
+	status  StatusSink // 任务生命周期状态回写（可为 nil）

 	turnMu sync.Mutex     // 保护 turns（攒批计数，多任务 goroutine 共享）
 	turns  map[string]int // sessionID → 累计轮次，用于每 N 轮触发 consolidate
 }

 // NewOrchestrator 持有依赖；图按任务的 DSL 在 Handle 内动态编译。
-// exec 为执行可视化事件出口（可为 nil，则不发轨迹事件）；eval 为自动化评测（可为 nil）。
-func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink) (*Orchestrator, error) {
-	return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec}, nil
+// exec 为执行可视化事件出口（可为 nil，则不发轨迹事件）；eval 为自动化评测（可为 nil）；
+// status 为任务生命周期状态回写出口（可为 nil）。
+func NewOrchestrator(pool LLM, breaker *harness.CircuitBreaker, eval *harness.Evaluator, sink TokenSink, tools ToolCaller, exec ExecSink, status StatusSink) (*Orchestrator, error) {
+	return &Orchestrator{pool: pool, breaker: breaker, eval: eval, sink: sink, tools: tools, exec: exec, status: status}, nil
+}
+
+// setStatus 回写一次任务状态流转（status 为 nil 时静默跳过）。
+func (o *Orchestrator) setStatus(taskID, status, detail string) {
+	if o.status == nil {
+		return
+	}
+	if err := o.status.PublishTaskStatus(taskID, status, detail); err != nil {
+		log.Printf("[eino] 回写任务状态 %s=%s 失败: %v", taskID, status, err)
+	}
+}
+
+// finishStatus 据收尾错误把任务置为 done / timeout / failed。
+func (o *Orchestrator) finishStatus(taskID string, err error) {
+	switch {
+	case err == nil:
+		o.setStatus(taskID, contract.TaskDone, "")
+	case errors.Is(err, context.DeadlineExceeded):
+		o.setStatus(taskID, contract.TaskTimeout, "执行超时")
+	default:
+		o.setStatus(taskID, contract.TaskFailed, truncate(err.Error(), 200))
+	}
 }

 // Handle 消费一个任务：按 DSL 编译 Eino 图并执行，把 Token 流回流到 sundynix.streams.<id>。
 func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
+	// 护栏：丢弃空任务（无 id），避免误投/历史脏数据被当真任务处理并触发状态回写放大。
+	if t.ID == "" {
+		log.Printf("[eino] 跳过空任务（无 id）")
+		return nil
+	}
 	tr := o.tracer(t.ID)
 	defer tr.done()

@@ -72,22 +112,31 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
 		tr.info("task", "system", "服务熔断", "后端连续失败，暂时拒绝新任务，请稍后重试")
 		_ = o.sink.PublishToken(t.ID, []byte("⚠️ 服务繁忙（已触发熔断保护），请稍后重试。"))
 		_ = o.sink.CompleteStream(t.ID)
+		o.setStatus(t.ID, contract.TaskFailed, "服务熔断")
 		return nil
 	}

+	// 任务状态机：进入执行 → running；整体加超时上限，超时判 timeout（杜绝无限期"运行中"）。
+	o.setStatus(t.ID, contract.TaskRunning, "")
+	tctx, cancel := context.WithTimeout(ctx, taskExecTimeout)
+	defer cancel()
+
 	// 报告生成走专用多步编排（规划→分章并行检索撰写→汇聚→渲染 Word），而非通用对话图。
 	if intent, _ := t.Meta[contract.MetaIntent].(string); intent == contract.IntentReport {
-		return o.handleReport(ctx, t, tr)
+		err := o.handleReport(tctx, t, tr)
+		o.finishStatus(t.ID, err)
+		return err
 	}
 	log.Printf("[eino] task %s received (graph=%d bytes), 按图执行（拓扑+连线+分支）...", t.ID, len(t.Graph))
 	tr.info("task", "system", "任务受理", fmt.Sprintf("DSL %d 字节，按图执行", len(t.Graph)))

-	// 按 DSL 图的真实拓扑/连线/分支执行（graph.go 解释器），agent 节点流式回流 token。
-	answer, err := o.runGraph(ctx, t, tr)
+	// 按 DSL 图执行：compose.Graph（EINO_COMPOSE=1）或自研 graph.go（默认）；agent 节点流式回流 token。
+	answer, err := o.executeGraph(tctx, t, tr)
 	if err != nil {
 		log.Printf("[eino] task %s graph error: %v", t.ID, err)
 		_ = o.sink.CompleteStream(t.ID)
 		o.breaker.Report(false)
+		o.finishStatus(t.ID, err)
 		return err
 	}

@@ -96,6 +145,7 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
 	}
 	log.Printf("[eino] task %s done (%d 字答复)", t.ID, len([]rune(answer)))
 	o.breaker.Report(true)
+	o.finishStatus(t.ID, nil)

 	// 写回阶段：离开热路径、异步落历史 + （TODO）抽取记忆。
 	go o.memorize(t, answer)
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"strings"
 	"time"

 	"github.com/cloudwego/eino/components/tool"
@@ -85,42 +86,84 @@ func (m *mcpTool) InvokableRun(ctx context.Context, argsJSON string, _ ...tool.O
 	return res.Content, nil
 }

-// agentTools 构建 ReAct 可用的工具集（按当前任务上下文绑定 uid/kb）。
-// 工具名是给"模型看"的语义名；mcpName 是实际 NATS 调用名。context 参数（user_id）服务端注入，不暴露给模型。
+// toolCatalogEntry 是 MCP list_tools 上报的一条工具元信息（与 mcp-go listTools 输出对齐）。
+type toolCatalogEntry struct {
+	Name      string `json:"name"`
+	CN        string `json:"cn"`
+	Desc      string `json:"desc"`
+	Agent     bool   `json:"agent_exposed"`
+	AgentName string `json:"agent_name"`
+	Params    []struct {
+		Name     string `json:"name"`
+		Type     string `json:"type"`
+		Desc     string `json:"desc"`
+		Required bool   `json:"required"`
+	} `json:"params"`
+	Inject []string `json:"inject"`
+}
+
+// agentTools 动态构建 ReAct 可用的工具集：分别向 mcp-go / mcp-py 探 list_tools 自描述目录，
+// 取 agent_exposed 的工具按上报参数 schema 建 InvokableTool；inject 参数（user_id/session_id/
+// kb/task_id）服务端运行时绑定、不暴露给模型。某台 MCP 离线即跳过（降级）。
+// 新增工具只需在对应 MCP 注册表标 agent，无需改这里（杜绝硬编码）。
 func (o *Orchestrator) agentTools(b *board, taskID string, tr *execTracer) []tool.BaseTool {
 	if o.tools == nil {
 		return nil
 	}
-	kbBind := map[string]any{}
-	if b.kb != "" {
-		kbBind["kb"] = b.kb
+	var out []tool.BaseTool
+	out = append(out, o.discoverTools(contract.ToolSubjectGo, b, taskID, tr)...)
+	out = append(out, o.discoverTools(contract.ToolSubjectPy, b, taskID, tr)...)
+	return out
+}
+
+// discoverTools 向某台 MCP（subject 前缀决定 go/py）探 list_tools，把 agent_exposed 的工具
+// 转成 Eino InvokableTool。该 MCP 不可用 / 无应答时返回空（不阻断）。
+func (o *Orchestrator) discoverTools(subject func(string) string, b *board, taskID string, tr *execTracer) []tool.BaseTool {
+	cctx, cancel := context.WithTimeout(context.Background(), toolCallTimeout)
+	defer cancel()
+	res, err := o.tools.CallTool(cctx, subject("list_tools"), &contract.ToolCall{Tool: "list_tools"})
+	if err != nil || res == nil || !res.OK {
+		return nil
 	}
-	return []tool.BaseTool{
-		&mcpTool{
-			mcpName: "wiki_search",
-			subject: contract.ToolSubjectGo,
-			caller:  o.tools, taskID: taskID, tr: tr,
-			bind: kbBind,
-			info: &schema.ToolInfo{
-				Name: "wiki_search",
-				Desc: "检索知识库，返回与查询最相关的资料片段。需要外部知识/事实依据时调用。",
-				ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{
-					"q": {Type: schema.String, Desc: "检索查询语句", Required: true},
-				}),
-			},
-		},
-		&mcpTool{
-			mcpName: "memory_get",
-			subject: contract.ToolSubjectGo,
-			caller:  o.tools, taskID: taskID, tr: tr,
-			bind: map[string]any{"user_id": b.uid},
-			info: &schema.ToolInfo{
-				Name: "recall_user_memory",
-				Desc: "召回当前用户的长期画像与偏好（称呼/职业/回答偏好等）。需要个性化、了解“我是谁”时调用。",
-				ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{}),
-			},
-		},
+	var cat struct {
+		Tools []toolCatalogEntry `json:"tools"`
 	}
+	if json.Unmarshal([]byte(res.Content), &cat) != nil {
+		return nil
+	}
+	injectVal := map[string]any{"user_id": b.uid, "session_id": b.sid, "task_id": taskID, "kb": b.kb}
+
+	var out []tool.BaseTool
+	for _, e := range cat.Tools {
+		if !e.Agent {
+			continue
+		}
+		params := map[string]*schema.ParameterInfo{}
+		for _, p := range e.Params {
+			params[p.Name] = &schema.ParameterInfo{Type: schema.DataType(p.Type), Desc: p.Desc, Required: p.Required}
+		}
+		bind := map[string]any{}
+		for _, inj := range e.Inject {
+			if v, ok := injectVal[inj]; ok && v != "" {
+				bind[inj] = v
+			}
+		}
+		name := e.AgentName
+		if name == "" {
+			name = e.Name
+		}
+		out = append(out, &mcpTool{
+			mcpName: e.Name,
+			subject: subject,
+			caller:  o.tools, taskID: taskID, tr: tr,
+			bind: bind,
+			info: &schema.ToolInfo{
+				Name: name, Desc: e.Desc,
+				ParamsOneOf: schema.NewParamsOneOfByParams(params),
+			},
+		})
+	}
+	return out
 }

 // runReactAgent 执行带"自主工具"的 agent 节点：模型在 ReAct 循环里自行决定调哪些 MCP 工具。
@@ -153,9 +196,10 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar
 		System: firstNonEmpty(system, defaultAgentSystem),
 		Query:  b.query,
 		// 自主 agent 不预注入画像：让它经 recall_user_memory 工具按需自取（否则模型直接答、不调工具）。
-		Profile: "",
-		History: b.history,
-		ToolOut: append(append([]string{}, b.toolOut...), b.refs...),
+		Profile:  "",
+		History:  b.history,
+		ToolOut:  append(append([]string{}, b.toolOut...), b.refs...),
+		Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力
 	}
 	msgs, _ := buildMessages(ctx, rc)

@@ -169,6 +213,7 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar
 	defer sr.Close()

 	chunks := 0
+	var produced strings.Builder // 本节点产出（供下游 agent 接力）
 	for {
 		chunk, rerr := sr.Recv()
 		if rerr == io.EOF {
@@ -183,9 +228,10 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar
 		}
 		safe, _ := harness.RedactSecrets(chunk.Content)
 		_ = o.sink.PublishToken(taskID, []byte(safe))
-		b.answer += safe
+		produced.WriteString(safe)
 		chunks++
 	}
+	o.recordAgentOutput(b, produced.String())
 	tr.emit(node, "model", "end", "ReAct 智能体",
-		fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(b.answer))), time.Since(t0).Milliseconds())
+		fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(produced.String()))), time.Since(t0).Milliseconds())
 }
@@ -214,29 +214,15 @@ func (o *Orchestrator) writeSection(ctx context.Context, topic, kb, heading stri
 	return strings.TrimSpace(txt)
 }

-// retrieve 经 mcp-go kb_search 工具检索知识库，整理为可读参考资料。kb 为空或无召回则返回空。
+// retrieve 经 Eino Retriever 组件（包 mcp-go kb_search）检索知识库，整理为可读参考资料。
 func (o *Orchestrator) retrieve(ctx context.Context, kb, query string) string {
-	if o.tools == nil || kb == "" {
+	docs, err := o.newRetriever(kb).Retrieve(ctx, query)
+	if err != nil || len(docs) == 0 {
 		return ""
 	}
-	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
-	defer cancel()
-	res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("kb_search"), &contract.ToolCall{
-		Tool: "kb_search", Args: map[string]any{"kb": kb, "q": query, "topK": 4},
-	})
-	if err != nil || res == nil || !res.OK || res.Content == "" || res.Content == "[]" {
-		return ""
-	}
-	var hits []struct {
-		Text  string  `json:"text"`
-		Score float64 `json:"score"`
-	}
-	if json.Unmarshal([]byte(res.Content), &hits) != nil {
-		return res.Content
-	}
 	var b strings.Builder
-	for i, h := range hits {
-		fmt.Fprintf(&b, "%d. %s\n", i+1, strings.TrimSpace(h.Text))
+	for i, d := range docs {
+		fmt.Fprintf(&b, "%d. %s\n", i+1, strings.TrimSpace(d.Content))
 	}
 	return b.String()
 }
@@ -76,6 +76,9 @@ func (p *Pool) model() model.BaseChatModel {
 // Ready 报告是否已配置可用后端（且 ChatModel 构建成功）。
 func (p *Pool) Ready() bool { return p.model() != nil }

+// ChatModel 返回当前 Eino ChatModel 组件（用于 compose.Graph 编排）；未就绪则 nil。
+func (p *Pool) ChatModel() model.BaseChatModel { return p.model() }
+
 // ToolCallingModel 返回支持函数调用的模型（用于 ReAct agent）；未就绪 / 不支持则 nil。
 func (p *Pool) ToolCallingModel() model.ToolCallingChatModel {
 	if tcm, ok := p.model().(model.ToolCallingChatModel); ok {
@@ -4,6 +4,7 @@ package nats
 import (
 	"context"
 	"log"
+	"time"

 	sharedbus "github.com/sundynix/sundynix-shared/bus"
 	"github.com/sundynix/sundynix-shared/contract"
@@ -73,6 +74,13 @@ func (s *Subscriber) ServeHealth(provide func() []byte) (func() error, error) {
 	return s.inner.ServeHealth(contract.SubjectHealthDispatcher, provide)
 }

+// PublishTaskStatus 让 Subscriber 满足 eino.StatusSink，把任务状态流转回写给网关。
+func (s *Subscriber) PublishTaskStatus(taskID, status, detail string) error {
+	return s.inner.PublishTaskStatus(&contract.TaskStatusEvent{
+		TaskID: taskID, Status: status, Detail: detail, TS: time.Now().UnixMilli(),
+	})
+}
+
 // RequestModelConfig 向控制面（Gateway）取当前激活的对话模型配置。
 func (s *Subscriber) RequestModelConfig(ctx context.Context) (*contract.ModelConfig, error) {
 	return s.inner.RequestConfig(ctx, contract.ConfigKindChat)
@@ -46,6 +46,15 @@ func main() {
 		}
 	}

+	// 任务生命周期：订阅 dispatcher 回写的状态流转（running/done/failed/timeout），落 PG 供 UI 查询。
+	if _, err := bus.SubscribeTaskStatus(func(ev *contract.TaskStatusEvent) {
+		if err := db.UpdateTaskStatus(context.Background(), ev.TaskID, ev.Status, ev.Detail); err != nil {
+			log.Printf("[gateway] 更新任务状态 %s=%s 失败: %v", ev.TaskID, ev.Status, err)
+		}
+	}); err != nil {
+		log.Printf("[gateway] subscribe task status: %v", err)
+	}
+
 	r := router.New(db, cache, bus, blobStore)
 	addr := envOr("GATEWAY_ADDR", ":8080")
 	log.Printf("[gateway] listening on %s", addr)
@@ -4,6 +4,7 @@ go 1.25.0

 require (
 	github.com/bwmarrin/snowflake v0.3.0
+	github.com/gin-contrib/sse v0.1.0
 	github.com/gin-gonic/gin v1.10.0
 	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/minio/minio-go/v7 v7.2.0
@@ -26,7 +27,6 @@ require (
 	github.com/cloudwego/base64x v0.1.6 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
-	github.com/gin-contrib/sse v0.1.0 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.20.0 // indirect
@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"time"

+	"github.com/gin-contrib/sse"
 	"github.com/gin-gonic/gin"

 	"github.com/sundynix/sundynix-gateway/internal/blob"
@@ -53,23 +54,86 @@ func (h *Handler) SubmitTask(c *gin.Context) {
 		c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
 		return
 	}
+	// 从提交即开始把 token 流录进 Redis Stream（订阅早于 dispatcher 产 token）→
+	// SSE 可从中回放/断点续传，根治"连晚/重连丢 token"。
+	h.startTokenRecorder(task.ID)
 	c.JSON(http.StatusAccepted, gin.H{"task_id": task.ID})
 }

-// StreamTask: 订阅 sundynix.streams.<task_id>，以 SSE 把零拷贝 Token Stream 推给客户端。
+// startTokenRecorder 后台订阅 token 流并落 Redis Stream，与 SSE 客户端是否在线无关。
+// Redis 降级时为空操作（SSE 自动回退到 live NATS 路径）。
+func (h *Handler) startTokenRecorder(taskID string) {
+	if !h.cache.Enabled() {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 6*time.Minute) // 兜底防泄漏
+	unsub, err := h.bus.SubscribeTokens(taskID,
+		func(tok []byte) { _ = h.cache.StreamAppend(ctx, taskID, "token", string(tok)) },
+		func() { _ = h.cache.StreamAppend(ctx, taskID, "done", ""); cancel() },
+	)
+	if err != nil {
+		cancel()
+		return
+	}
+	go func() { <-ctx.Done(); _ = unsub() }()
+}
+
+// TaskStatus: GET /api/v1/tasks/:id —— 返回任务生命周期状态（供 UI 轮询，根治"卡运行中看不出来"）。
+func (h *Handler) TaskStatus(c *gin.Context) {
+	id := c.Param("id")
+	status, detail := h.db.GetTaskStatus(c.Request.Context(), id)
+	if status == "" {
+		c.JSON(http.StatusNotFound, gin.H{"error": "task not found"})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"task_id": id, "status": status, "detail": detail})
+}
+
+// StreamTask: 以 SSE 把 Token Stream 推给客户端。
+// 优先从 Redis Stream 读（可回放 + 断点续传，根治连晚/重连丢 token）；Redis 降级时回退 live NATS。
 func (h *Handler) StreamTask(c *gin.Context) {
 	taskID := c.Param("id")
 	c.Writer.Header().Set("Content-Type", "text/event-stream")
 	c.Writer.Header().Set("Cache-Control", "no-cache")
 	c.Writer.Header().Set("Connection", "keep-alive")

+	if !h.cache.Enabled() {
+		h.streamTaskLive(c, taskID) // Redis 不可用 → 旧的 live-NATS 路径兜底
+		return
+	}
+
+	// 断点续传：浏览器 EventSource 重连会带 Last-Event-ID；缺省 "0" 从头回放。
+	lastID := c.GetHeader("Last-Event-ID")
+	if lastID == "" {
+		lastID = "0"
+	}
+	ctx := c.Request.Context()
+	c.Stream(func(w io.Writer) bool {
+		entries, nl, err := h.cache.StreamRead(ctx, taskID, lastID, 20*time.Second)
+		if err != nil {
+			return false // ctx 取消（客户端断开）或后端故障
+		}
+		lastID = nl
+		for _, e := range entries {
+			if e.Kind == "done" {
+				_ = sse.Encode(w, sse.Event{Id: e.ID, Event: "done", Data: taskID})
+				return false
+			}
+			_ = sse.Encode(w, sse.Event{Id: e.ID, Event: "token", Data: e.Data})
+		}
+		return true // 继续阻塞读取后续 token
+	})
+}
+
+// streamTaskLive 是 Redis 降级时的兜底：直接订阅 NATS token 流转 SSE（无回放/续传能力）。
+func (h *Handler) streamTaskLive(c *gin.Context, taskID string) {
 	tokens := make(chan []byte, 256)
 	done := make(chan struct{})
 	unsub, err := h.bus.SubscribeTokens(taskID,
 		func(tok []byte) {
 			select {
 			case tokens <- tok:
-			default: // 背压保护：客户端过慢则丢弃，避免阻塞 NATS 回调
+			default:
 			}
 		},
 		func() { close(done) },
@@ -79,8 +143,6 @@ func (h *Handler) StreamTask(c *gin.Context) {
 		return
 	}
 	defer func() { _ = unsub() }()
-
-	// gin 的流式写：返回 false 即结束响应。
 	c.Stream(func(w io.Writer) bool {
 		select {
 		case tok := <-tokens:
@@ -59,6 +59,11 @@ func (b *Bus) Ping(ctx context.Context, subject string) ([]byte, error) {
 	return b.inner.Ping(ctx, subject)
 }

+// SubscribeTaskStatus 订阅 dispatcher 回写的任务生命周期状态（落 PG）。
+func (b *Bus) SubscribeTaskStatus(onEvent func(*contract.TaskStatusEvent)) (func() error, error) {
+	return b.inner.SubscribeTaskStatus(onEvent)
+}
+
 // ServeConfig 让网关作为配置控制面，响应某 kind 的配置请求。
 func (b *Bus) ServeConfig(kind string, provide func() *contract.ModelConfig) (func() error, error) {
 	return b.inner.ServeConfig(kind, provide)
@@ -49,6 +49,7 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blobStore *blob.
 		p := api.Group("", middleware.RequireAuth())
 		{
 			p.POST("/tasks", h.SubmitTask)        // 解析 DSL 并 Publish 到 NATS（带已验证 uid）
+			p.GET("/tasks/:id", h.TaskStatus)     // 任务生命周期状态（UI 轮询 submitted/running/done/failed/timeout）
 			p.PUT("/memory", h.SetMemory)         // 偏好记忆登记（→ mcp-go memory_upsert）
 			p.GET("/memory", h.ListMemory)        // 列出当前用户偏好（记忆面板）
 			p.DELETE("/memory", h.DeleteMemory)   // 软删一条偏好（?key=）
@@ -19,5 +19,6 @@ type Task struct {
 	BaseModel
 	TaskID string `gorm:"uniqueIndex;size:64"` // task_xxx
 	Graph  string `gorm:"type:jsonb"`          // React Flow 导出的 DSL 原文
-	Status string `gorm:"size:32"`             // submitted / done / failed
+	Status string `gorm:"size:32"`             // submitted / running / done / failed / timeout
+	Detail string `gorm:"type:text"`           // 失败/超时原因等（状态机回写）
 }
@@ -9,6 +9,8 @@ import (
 	"gorm.io/driver/postgres"
 	"gorm.io/gorm"
 	"gorm.io/gorm/schema"
+
+	"github.com/sundynix/sundynix-shared/contract"
 )

 // errStoreDisabled 表示 Postgres 处于降级（未连接）模式，写操作无法进行。
@@ -93,7 +95,29 @@ func (p *Postgres) SaveTask(ctx context.Context, id, graph string) error {
 	if p.db == nil {
 		return nil
 	}
-	return p.db.WithContext(ctx).Create(&Task{TaskID: id, Graph: graph, Status: "submitted"}).Error
+	return p.db.WithContext(ctx).Create(&Task{TaskID: id, Graph: graph, Status: contract.TaskSubmitted}).Error
+}
+
+// UpdateTaskStatus 流转任务状态（running/done/failed/timeout），由 dispatcher 经 NATS 回写驱动。
+func (p *Postgres) UpdateTaskStatus(ctx context.Context, id, status, detail string) error {
+	if p.db == nil {
+		return nil
+	}
+	return p.db.WithContext(ctx).Model(&Task{}).
+		Where("task_id = ?", id).
+		Updates(map[string]any{"status": status, "detail": detail}).Error
+}
+
+// GetTaskStatus 取一条任务的当前状态（供 UI 轮询；不存在返回空串）。
+func (p *Postgres) GetTaskStatus(ctx context.Context, id string) (status, detail string) {
+	if p.db == nil {
+		return "", ""
+	}
+	var t Task
+	if err := p.db.WithContext(ctx).Select("status", "detail").Where("task_id = ?", id).First(&t).Error; err != nil {
+		return "", ""
+	}
+	return t.Status, t.Detail
 }

 // CountTasks 返回已提交任务数（降级模式返回 0）。
@@ -48,6 +48,64 @@ func (r *Redis) Allow(ctx context.Context, key string, limit int64, window time.
 	return n <= limit, nil
 }

+// ---- Token 流持久化（Redis Stream：可回放的追加日志，根治 SSE 连晚/重连丢 token）----
+
+const streamTTL = 10 * time.Minute
+
+func streamKey(taskID string) string { return "sundynix:stream:" + taskID }
+
+// StreamEntry 是 token 流里的一条记录（ID 用于 SSE 的 Last-Event-ID 断点续传）。
+type StreamEntry struct {
+	ID   string
+	Kind string // token / done
+	Data string
+}
+
+// StreamAppend 把一条 token / 结束标记追加到任务的 Redis Stream（带 TTL 自动清理）。
+func (r *Redis) StreamAppend(ctx context.Context, taskID, kind, data string) error {
+	if r.rdb == nil {
+		return nil
+	}
+	k := streamKey(taskID)
+	if err := r.rdb.XAdd(ctx, &redis.XAddArgs{Stream: k, Values: map[string]any{"kind": kind, "data": data}}).Err(); err != nil {
+		return err
+	}
+	return r.rdb.Expire(ctx, k, streamTTL).Err()
+}
+
+// StreamRead 从 lastID 之后阻塞读取新条目（XREAD BLOCK）。lastID="0" 表示从头回放。
+// 阻塞超时无新数据时返回空切片 + 原 lastID（调用方据此继续轮询）。
+func (r *Redis) StreamRead(ctx context.Context, taskID, lastID string, block time.Duration) ([]StreamEntry, string, error) {
+	if r.rdb == nil {
+		return nil, lastID, nil
+	}
+	res, err := r.rdb.XRead(ctx, &redis.XReadArgs{
+		Streams: []string{streamKey(taskID), lastID}, Block: block, Count: 256,
+	}).Result()
+	if err == redis.Nil {
+		return nil, lastID, nil // 阻塞超时、无新条目
+	}
+	if err != nil {
+		return nil, lastID, err
+	}
+	var out []StreamEntry
+	nl := lastID
+	for _, st := range res {
+		for _, m := range st.Messages {
+			out = append(out, StreamEntry{ID: m.ID, Kind: asString(m.Values["kind"]), Data: asString(m.Values["data"])})
+			nl = m.ID
+		}
+	}
+	return out, nl, nil
+}
+
+func asString(v any) string {
+	if s, ok := v.(string); ok {
+		return s
+	}
+	return ""
+}
+
 // Close 释放底层连接。
 func (r *Redis) Close() {
 	if r.rdb != nil {
@@ -32,11 +32,25 @@ type Gateway struct {
 	tools   map[string]toolDef // 工具注册表：唯一事实源，dispatch 与 list_tools 共用，杜绝漂移
 }

-// toolDef 是一个注册工具的元信息（中文名 / 作用）+ 处理函数。
+// paramSpec 是一个工具参数的声明（供自主 agent 据此生成调用入参）。
+type paramSpec struct {
+	Name     string `json:"name"`
+	Type     string `json:"type"` // string / number / integer / boolean / object / array
+	Desc     string `json:"desc"`
+	Required bool   `json:"required"`
+}
+
+// toolDef 是一个注册工具的元信息 + 处理函数。新增 agent 暴露元信息，让自主 agent 能动态发现工具：
+// agent=是否给模型自主调用；agentName=模型可见名（空=注册键）；params=模型可填参数；
+// inject=服务端运行时注入、不暴露给模型的参数名（如 user_id / session_id / kb / task_id）。
 type toolDef struct {
-	cn      string // 中文名
-	desc    string // 作用简述
-	handler func(context.Context, *contract.ToolCall) *contract.ToolResult
+	cn        string
+	desc      string
+	agent     bool
+	agentName string
+	params    []paramSpec
+	inject    []string
+	handler   func(context.Context, *contract.ToolCall) *contract.ToolResult
 }

 func NewGateway(b *sharedbus.Bus, s *search.Hybrid, m *memory.Store, h *history.Store, r *rag.Engine) *Gateway {
@@ -60,29 +74,53 @@ func (g *Gateway) Serve(ctx context.Context) error {

 // buildRegistry 注册 mcp-go 全部工具：名称 → (中文名, 作用, 处理函数)。
 // 这是工具的唯一事实源——dispatch 据此路由、list_tools 据此上报，二者永不漂移。
+// 想让某工具能被自主 agent 调用：把 agent 设 true，写清 params（模型可填）与 inject（服务端注入）。
+// 加新工具只改这一处——dispatcher 经 list_tools 动态发现，无需改调度代码。
 func (g *Gateway) buildRegistry() map[string]toolDef {
 	return map[string]toolDef{
-		"wiki_search":    {"知识检索", "向量检索知识库（Milvus），返回最相关片段", g.wikiSearch},
-		"kb_ingest":      {"知识入库", "文本切块 → 向量化 → 写入 Milvus / Bleve", g.kbIngest},
-		"kb_search":      {"检索台查询", "结构化返回命中内容与相似度分数", g.kbSearch},
-		"kb_graph":       {"知识图谱", "取某库的实体关系三元组（Neo4j）", g.kbGraph},
-		"report_render":  {"报告渲染", "把结构化报告渲染为 Word（.docx）", g.reportRender},
-		"report_store":   {"报告存源", "暂存报告源数据，供导出时按需渲染", g.reportStore},
-		"report_export":  {"报告导出", "按需把已存报告导出为 Word / Markdown", g.reportExport},
-		"external_api":   {"外部接口", "受控调用第三方 HTTP API（带 SSRF 校验）", g.externalAPI},
-		"memory_get":     {"记忆召回", "取用户长期画像（已按打分排序）", g.memoryGet},
-		"memory_upsert":  {"记忆写入", "新增 / 更新一条用户偏好（带重要度）", g.memoryUpsert},
-		"memory_delete":  {"记忆删除", "软删一条偏好（对账判定过时 / 矛盾时）", g.memoryDelete},
-		"memory_list":    {"记忆列表", "列出用户全部偏好（供管理面板查看）", g.memoryList},
-		"history_get":    {"历史召回", "取会话最近多轮对话", g.historyGet},
-		"history_append": {"历史追加", "往会话写入一条消息", g.historyAppend},
-		"health": {"健康检查", "上报 Milvus / Neo4j / embedding 就绪情况",
-			func(_ context.Context, _ *contract.ToolCall) *contract.ToolResult {
+		// —— 暴露给自主 agent 的工具（带参数 schema / 注入声明）——
+		"wiki_search": {
+			cn: "知识检索", desc: "检索知识库，返回与查询最相关的资料片段。需要外部知识/事实依据时调用。",
+			agent: true,
+			params: []paramSpec{{Name: "q", Type: "string", Desc: "检索查询语句", Required: true}},
+			inject: []string{"kb"}, handler: g.wikiSearch,
+		},
+		"memory_get": {
+			cn: "记忆召回", desc: "召回当前用户的长期画像与偏好（称呼/职业/回答偏好等）。需要个性化、了解“我是谁”时调用。",
+			agent: true, agentName: "recall_user_memory", inject: []string{"user_id"}, handler: g.memoryGet,
+		},
+		"memory_upsert": {
+			cn: "记忆写入", desc: "把关于用户的一条事实/偏好长期记住（如称呼、职业、回答偏好）。",
+			agent: true, agentName: "remember_user_fact",
+			params: []paramSpec{
+				{Name: "key", Type: "string", Desc: "记忆条目的键，如 称呼/职业/回答偏好", Required: true},
+				{Name: "value", Type: "string", Desc: "记忆条目的值", Required: true},
+			},
+			inject: []string{"user_id"}, handler: g.memoryUpsert,
+		},
+		"history_get": {
+			cn: "历史召回", desc: "取当前会话最近多轮对话，用于理解上下文。",
+			agent: true, inject: []string{"session_id"}, handler: g.historyGet,
+		},
+
+		// —— 仅内部/流水线/管理用，不暴露给自主 agent ——
+		"kb_ingest":      {cn: "知识入库", desc: "文本切块 → 向量化 → 写入 Milvus / Bleve", handler: g.kbIngest},
+		"kb_search":      {cn: "检索台查询", desc: "结构化返回命中内容与相似度分数", handler: g.kbSearch},
+		"kb_graph":       {cn: "知识图谱", desc: "取某库的实体关系三元组（Neo4j）", handler: g.kbGraph},
+		"report_render":  {cn: "报告渲染", desc: "把结构化报告渲染为 Word（.docx）", handler: g.reportRender},
+		"report_store":   {cn: "报告存源", desc: "暂存报告源数据，供导出时按需渲染", handler: g.reportStore},
+		"report_export":  {cn: "报告导出", desc: "按需把已存报告导出为 Word / Markdown", handler: g.reportExport},
+		"external_api":   {cn: "外部接口", desc: "受控调用第三方 HTTP API（带 SSRF 校验）", handler: g.externalAPI},
+		"memory_delete":  {cn: "记忆删除", desc: "软删一条偏好（对账判定过时 / 矛盾时）", handler: g.memoryDelete},
+		"memory_list":    {cn: "记忆列表", desc: "列出用户全部偏好（供管理面板查看）", handler: g.memoryList},
+		"history_append": {cn: "历史追加", desc: "往会话写入一条消息", handler: g.historyAppend},
+		"health": {cn: "健康检查", desc: "上报 Milvus / Neo4j / embedding 就绪情况",
+			handler: func(_ context.Context, _ *contract.ToolCall) *contract.ToolResult {
 				data, _ := json.Marshal(g.rag.Status())
 				return &contract.ToolResult{OK: true, Content: string(data)}
 			}},
-		"echo": {"回显", "原样返回入参（调试用）",
-			func(_ context.Context, call *contract.ToolCall) *contract.ToolResult {
+		"echo": {cn: "回显", desc: "原样返回入参（调试用）",
+			handler: func(_ context.Context, call *contract.ToolCall) *contract.ToolResult {
 				return &contract.ToolResult{OK: true, Content: fmt.Sprint(call.Args["text"])}
 			}},
 	}
@@ -102,16 +140,24 @@ func (g *Gateway) dispatch(ctx context.Context, call *contract.ToolCall) *contra
 	return td.handler(ctx, call)
 }

-// listTools 自省：上报本服务注册的工具清单（名称 + 中文名 + 作用），供管理端展示。
+// listTools 自省：上报本服务注册的工具清单（名称 + 中文名 + 作用 + agent 暴露元信息），
+// 供管理端展示 & dispatcher 动态构建自主 agent 工具集（加工具只改注册表，无需改调度代码）。
 func (g *Gateway) listTools() *contract.ToolResult {
 	type info struct {
-		Name string `json:"name"`
-		CN   string `json:"cn"`
-		Desc string `json:"desc"`
+		Name      string      `json:"name"`
+		CN        string      `json:"cn"`
+		Desc      string      `json:"desc"`
+		Agent     bool        `json:"agent_exposed"`       // 是否给自主 agent
+		AgentName string      `json:"agent_name,omitempty"`// 模型可见名（空=name）
+		Params    []paramSpec `json:"params,omitempty"`    // 模型可填参数
+		Inject    []string    `json:"inject,omitempty"`    // 服务端注入参数（不暴露给模型）
 	}
 	out := make([]info, 0, len(g.tools))
 	for name, td := range g.tools {
-		out = append(out, info{Name: name, CN: td.cn, Desc: td.desc})
+		out = append(out, info{
+			Name: name, CN: td.cn, Desc: td.desc,
+			Agent: td.agent, AgentName: td.agentName, Params: td.params, Inject: td.inject,
+		})
 	}
 	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) // map 无序 → 稳定输出
 	data, _ := json.Marshal(map[string]any{"service": "mcp-go", "tools": out})
@@ -0,0 +1,137 @@
+package rag
+
+import (
+	"strings"
+)
+
+// 切块参数（针对中文 + embedding token 上限调过；后续可配置化）。
+const (
+	chunkTargetRunes  = 500  // 目标大小：贪心打包到接近此值就在句界收口
+	chunkOverlapRunes = 80   // 块间重叠：保跨块上下文连续
+	chunkMinRunes     = 100  // 最小块：低于则并入相邻，避免碎块稀释向量
+	chunkMaxRunes     = 1000 // 原子硬上限：超大无标点段落兜底窗口切
+)
+
+// chunk 把文本切成检索友好的语义块：按段落/句界切成原子 → 贪心打包到目标大小（句末收口）
+// → 块间加重叠。全程按 rune 操作，杜绝中文 UTF-8 被字节切碎。
+func chunk(text string) []string {
+	atoms := splitToAtoms(text)
+	packed := packAtoms(atoms, chunkTargetRunes, chunkMinRunes)
+	return addOverlap(packed, chunkOverlapRunes)
+}
+
+// splitToAtoms 把文本切成"原子"（句子/行）：在换行与句末标点处断开，去空白；
+// 超大无标点原子按 rune 窗口兜底切到 ≤ 目标大小。原子是打包的最小不可分单元。
+func splitToAtoms(text string) []string {
+	runes := []rune(strings.ReplaceAll(text, "\r\n", "\n"))
+	var atoms []string
+	start := 0
+	flush := func(end int) {
+		if s := strings.TrimSpace(string(runes[start:end])); s != "" {
+			atoms = append(atoms, s)
+		}
+		start = end
+	}
+	for i := 0; i < len(runes); i++ {
+		r := runes[i]
+		switch {
+		case r == '\n' || isCJKEnd(r):
+			flush(i + 1)
+		case r == '.' || r == '!' || r == '?' || r == ';':
+			// ASCII 句末：仅当其后为空白/行尾才断（避开缩写、小数点）。
+			if i+1 >= len(runes) || runes[i+1] == ' ' || runes[i+1] == '\n' {
+				flush(i + 1)
+			}
+		}
+	}
+	flush(len(runes))
+
+	out := make([]string, 0, len(atoms))
+	for _, a := range atoms {
+		if runeLen(a) <= chunkMaxRunes {
+			out = append(out, a)
+		} else {
+			out = append(out, splitByRuneWindow(a, chunkTargetRunes)...)
+		}
+	}
+	return out
+}
+
+// packAtoms 贪心打包：把原子拼进当前块，直到再加会超 target 就收口（块在句界结束）。
+// 末尾过小的块（< min）并入前一块，避免碎块。
+func packAtoms(atoms []string, target, min int) []string {
+	var chunks []string
+	var cur strings.Builder
+	curLen := 0
+	closeCur := func() {
+		if curLen > 0 {
+			chunks = append(chunks, cur.String())
+			cur.Reset()
+			curLen = 0
+		}
+	}
+	for _, a := range atoms {
+		al := runeLen(a)
+		if curLen > 0 && curLen+al > target {
+			closeCur()
+		}
+		if curLen > 0 {
+			cur.WriteByte('\n')
+			curLen++
+		}
+		cur.WriteString(a)
+		curLen += al
+	}
+	if curLen > 0 {
+		if n := len(chunks); n > 0 && curLen < min {
+			chunks[n-1] = chunks[n-1] + "\n" + cur.String() // 尾块太小 → 并入上一块
+		} else {
+			chunks = append(chunks, cur.String())
+		}
+	}
+	return chunks
+}
+
+// addOverlap 给每块前缀上一块尾部的 overlap 个 rune，保跨块上下文连续（首块不加）。
+func addOverlap(chunks []string, overlap int) []string {
+	if overlap <= 0 || len(chunks) <= 1 {
+		return chunks
+	}
+	out := make([]string, len(chunks))
+	out[0] = chunks[0]
+	for i := 1; i < len(chunks); i++ {
+		prev := []rune(chunks[i-1])
+		tail := prev
+		if len(prev) > overlap {
+			tail = prev[len(prev)-overlap:]
+		}
+		out[i] = strings.TrimSpace(string(tail)) + "\n" + chunks[i]
+	}
+	return out
+}
+
+func isCJKEnd(r rune) bool {
+	switch r {
+	case '。', '！', '？', '；', '…':
+		return true
+	}
+	return false
+}
+
+// splitByRuneWindow 按 rune 窗口硬切（兜底：超大无标点原子），保证不切碎多字节字符。
+func splitByRuneWindow(s string, size int) []string {
+	r := []rune(s)
+	var out []string
+	for len(r) > size {
+		if t := strings.TrimSpace(string(r[:size])); t != "" {
+			out = append(out, t)
+		}
+		r = r[size:]
+	}
+	if t := strings.TrimSpace(string(r)); t != "" {
+		out = append(out, t)
+	}
+	return out
+}
+
+func runeLen(s string) int { return len([]rune(s)) }
@@ -0,0 +1,122 @@
+package rag
+
+import (
+	"strings"
+	"testing"
+	"unicode/utf8"
+)
+
+func TestChunkEmpty(t *testing.T) {
+	if got := chunk("   \n\n  "); len(got) != 0 {
+		t.Fatalf("空白文本应切出 0 块，得 %d", len(got))
+	}
+}
+
+func TestChunkShort(t *testing.T) {
+	got := chunk("这是一句很短的话。")
+	if len(got) != 1 || !strings.Contains(got[0], "很短") {
+		t.Fatalf("短文本应为单块，得 %v", got)
+	}
+}
+
+// TestChunkRuneSafe 是核心：中文必须按 rune 切，绝不能切碎出乱码（旧版按字节切的 bug）。
+func TestChunkRuneSafe(t *testing.T) {
+	// 1500 个汉字、无标点 → 触发窗口兜底切；每块必须是合法 UTF-8。
+	text := strings.Repeat("中", 1500)
+	for _, c := range chunk(text) {
+		if !utf8.ValidString(c) {
+			t.Fatalf("切出非法 UTF-8（中文被字节切碎）：%q", c)
+		}
+		for _, r := range c {
+			if r != '中' && !strings.ContainsRune("\n ", r) {
+				t.Fatalf("出现意外字符 %q，疑似切碎", r)
+			}
+		}
+	}
+}
+
+func TestChunkSizeBounds(t *testing.T) {
+	// 多段中文，每段以句号结尾。
+	var sb strings.Builder
+	for i := 0; i < 60; i++ {
+		sb.WriteString("这是用于测试切块大小上界的一个中文句子片段。")
+	}
+	for i, c := range chunk(sb.String()) {
+		if n := runeLen(c); n > chunkTargetRunes+chunkOverlapRunes+1 {
+			t.Fatalf("第 %d 块 %d 字，超过 target+overlap=%d", i, n, chunkTargetRunes+chunkOverlapRunes+1)
+		}
+	}
+}
+
+func TestChunkSentenceBoundary(t *testing.T) {
+	var sb strings.Builder
+	for i := 0; i < 80; i++ {
+		sb.WriteString("第一句话在这里。第二句话也在这里。")
+	}
+	chunks := chunk(sb.String())
+	if len(chunks) < 2 {
+		t.Fatalf("长文本应切出多块，得 %d", len(chunks))
+	}
+	// 多数块应以句号收口（允许重叠导致的少量例外）。
+	endsWell := 0
+	for _, c := range chunks {
+		if strings.HasSuffix(strings.TrimSpace(c), "。") {
+			endsWell++
+		}
+	}
+	if endsWell < len(chunks)/2 {
+		t.Fatalf("多数块应在句末收口，仅 %d/%d", endsWell, len(chunks))
+	}
+}
+
+func TestChunkOverlap(t *testing.T) {
+	var sb strings.Builder
+	for i := 0; i < 60; i++ {
+		sb.WriteString("用于验证相邻块之间存在上下文重叠的中文句子。")
+	}
+	chunks := chunk(sb.String())
+	if len(chunks) < 2 {
+		t.Skip("未切出多块，跳过重叠校验")
+	}
+	// 第 2 块开头应包含第 1 块尾部的一小段（重叠）。
+	prevTail := []rune(chunks[0])
+	tail := string(prevTail[max0(len(prevTail)-chunkOverlapRunes):])
+	// 取尾部一小片做包含判断（去掉可能的换行）。
+	probe := strings.TrimSpace(tail)
+	if len(probe) > 10 {
+		probe = probe[len(probe)-10:]
+	}
+	if probe != "" && !strings.Contains(chunks[1], strings.TrimSpace(string([]rune(probe)))) {
+		// 重叠是按 rune 尾部，probe 是字节尾部，宽松校验：第二块前缀应与首块尾部有交集
+		if !strings.HasPrefix(strings.TrimSpace(chunks[1]), strings.TrimSpace(tail)) {
+			t.Logf("重叠片段:%q\n块2前缀:%q", tail, []rune(chunks[1])[:min0(40, runeLen(chunks[1]))])
+			t.Fatalf("相邻块未见重叠")
+		}
+	}
+}
+
+func TestChunkOversizedNoPunct(t *testing.T) {
+	text := strings.Repeat("x", 3000) // 无标点超大块
+	chunks := chunk(text)
+	if len(chunks) < 3 {
+		t.Fatalf("3000 字无标点应窗口切成多块，得 %d", len(chunks))
+	}
+	for _, c := range chunks {
+		if runeLen(c) > chunkTargetRunes+chunkOverlapRunes+1 {
+			t.Fatalf("窗口切块超界：%d", runeLen(c))
+		}
+	}
+}
+
+func max0(n int) int {
+	if n < 0 {
+		return 0
+	}
+	return n
+}
+func min0(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}
@@ -6,7 +6,6 @@ import (
 	"context"
 	"errors"
 	"log"
-	"strings"
 	"sync"

 	"github.com/sundynix/sundynix-shared/contract"
@@ -252,19 +251,4 @@ func (e *Engine) Close() {
 	e.graph.close(context.Background())
 }

-// chunk 朴素切块：按行切，去空白；过长再按长度切。真实系统应做版面/语义切块。
-func chunk(text string) []string {
-	var out []string
-	for _, line := range strings.Split(text, "\n") {
-		s := strings.TrimSpace(line)
-		if s == "" {
-			continue
-		}
-		for len(s) > 2000 {
-			out = append(out, s[:2000])
-			s = s[2000:]
-		}
-		out = append(out, s)
-	}
-	return out
-}
+// chunk 的实现已移到 chunk.go（递归 + 句界 + 重叠 + rune 安全的语义切块）。
@@ -25,12 +25,18 @@ log = logging.getLogger("mcp_py")
 SUBJECT_PY_ALL = "sundynix.tools.py.>"
 QUEUE_PY = "mcp-py-workers"

-# 工具元信息：名称 → (中文名, 作用简述)。list_tools 据此上报给管理端展示。
+# 工具元信息（与 mcp-go 注册表同形）：list_tools 上报给管理端展示 + dispatcher 动态构建自主 agent 工具集。
+# agent=是否暴露给自主 agent；params=模型可填参数；inject=服务端注入参数（不暴露给模型）。
 TOOL_META = {
-    "echo": ("回显", "原样返回入参（调试用）"),
-    "run_code": ("代码执行", "静态守卫 + Docker 隔离沙箱运行代码（标准档 256m/10s）"),
-    "parse_document": ("文档解析", "文件 → 纯文本（MinerU / PaddleOCR）"),
-    "secure_sandbox": ("安全沙箱", "更严资源档（128m/5s）的隔离执行，用于高风险代码"),
+    "echo": {"cn": "回显", "desc": "原样返回入参（调试用）"},
+    "run_code": {
+        "cn": "代码执行",
+        "desc": "在隔离沙箱里执行 Python 代码（静态守卫 + Docker，256m/10s）。用于计算、数据处理、逻辑验证。",
+        "agent": True,
+        "params": [{"name": "code", "type": "string", "desc": "要执行的 Python 代码", "required": True}],
+    },
+    "parse_document": {"cn": "文档解析", "desc": "文件 → 纯文本（MinerU / PaddleOCR）"},
+    "secure_sandbox": {"cn": "安全沙箱", "desc": "更严资源档（128m/5s）的隔离执行，用于高风险代码"},
 }


@@ -102,10 +108,18 @@ class McpGateway:
        return str(args.get("text", ""))

    async def _list_tools(self, args: dict) -> str:
-        """自省：上报业务工具清单（名称 + 中文名 + 作用），供管理端探活 + 展示。"""
+        """自省：上报业务工具清单（名称/中文名/作用 + agent 暴露元信息），供管理端展示 + dispatcher 动态发现。"""
        tools = [
-            {"name": n, "cn": cn, "desc": d}
-            for n, (cn, d) in TOOL_META.items()
+            {
+                "name": n,
+                "cn": m["cn"],
+                "desc": m["desc"],
+                "agent_exposed": m.get("agent", False),
+                "agent_name": m.get("agent_name", ""),
+                "params": m.get("params", []),
+                "inject": m.get("inject", []),
+            }
+            for n, m in TOOL_META.items()
            if n in self._tools  # 仅上报真正注册的业务工具（list_tools 自身不计入）
        ]
        return json.dumps({"service": "mcp-py", "tools": tools})
@@ -237,6 +237,31 @@ func (b *Bus) Ping(ctx context.Context, subject string) ([]byte, error) {
 	return msg.Data, nil
 }

+// ---- 任务生命周期状态回写（core NATS pub-sub）----
+
+// PublishTaskStatus 广播一次任务状态流转（dispatcher 调用）。
+func (b *Bus) PublishTaskStatus(ev *contract.TaskStatusEvent) error {
+	data, err := json.Marshal(ev)
+	if err != nil {
+		return err
+	}
+	return b.nc.Publish(contract.SubjectTaskStatus, data)
+}
+
+// SubscribeTaskStatus 订阅任务状态流转（网关调用，落 PG + 推 UI）。
+func (b *Bus) SubscribeTaskStatus(onEvent func(*contract.TaskStatusEvent)) (unsub func() error, err error) {
+	sub, err := b.nc.Subscribe(contract.SubjectTaskStatus, func(m *nats.Msg) {
+		var ev contract.TaskStatusEvent
+		if json.Unmarshal(m.Data, &ev) == nil {
+			onEvent(&ev)
+		}
+	})
+	if err != nil {
+		return nil, fmt.Errorf("subscribe task status: %w", err)
+	}
+	return sub.Unsubscribe, nil
+}
+
 // ---- 配置控制面（core NATS request-reply + broadcast）----

 // RequestConfig 向控制面（Gateway）请求某 kind 当前激活配置（chat/embedding）。
@@ -29,6 +29,31 @@ const (
 	// request-reply 心跳主题让控制面（管理端「服务状态」）能判定它在不在线。
 	SubjectHealthDispatcher = "sundynix.health.dispatcher"

+	// 任务生命周期状态回写：dispatcher 开跑/跑完/出错经此主题广播，网关订阅落 PG 并推 UI。
+	// core NATS pub-sub（状态是幂等覆盖，丢一条由下一条纠正，无需持久化）。
+	// 注意：必须在 sundynix.tasks.> 之外，否则会被任务流捕获成"幽灵任务"自我放大。
+	SubjectTaskStatus = "sundynix.status.task"
+)
+
+// 任务生命周期状态机：submitted（网关建任务）→ running（dispatcher 开跑）
+// → done / failed / timeout（dispatcher 收尾）。
+const (
+	TaskSubmitted = "submitted"
+	TaskRunning   = "running"
+	TaskDone      = "done"
+	TaskFailed    = "failed"
+	TaskTimeout   = "timeout"
+)
+
+// TaskStatusEvent 是一次任务状态流转事件（经 SubjectTaskStatus 回流给网关）。
+type TaskStatusEvent struct {
+	TaskID string `json:"task_id"`
+	Status string `json:"status"`           // running / done / failed / timeout
+	Detail string `json:"detail,omitempty"` // 失败原因等
+	TS     int64  `json:"ts"`               // unix 毫秒
+}
+
+const (
 	// MetaUserID 是 Task.Meta 中承载已登录用户标识的键（用于偏好记忆召回）。
 	MetaUserID = "user_id"
 	// MetaSessionID 是 Task.Meta 中承载会话标识的键（用于短期多轮历史）。