dsh-state-graph
把 StateGraph 有向状态图编排引擎移植为 DeepSeek Harness (dsh) 插件：用声明式的节点 / 静态边 / 条件边把 Harness 单向线性的 ReAct 调度提升为支持多分支条件路由、循环重试与迭代熔断的图编排运行时（子图嵌套可经 addSubgraph 或节点内调用 ctx.graph.create() 组合实现；并行分支 Fan-out / Fan-in 已内置）。
addNode / addEdge / addConditionalEdge → 纯函数增量补丁 → 迭代熔断 → 轨迹事件流
这是什么
dsh-state-graph 是一个 dsh bundle 包：导出一个 cordis 插件（state-graph），提供 ctx.graph 服务。插件的核心是设计文档《状态图（StateGraph）编排引擎技术方案与架构设计》中的有向状态图运行时内核：
| 能力 | 落点 |
|---|
| 声明式拓扑构造（Fluent API：addNode / addEdge / addConditionalEdge） | StateGraph |
声明式图定义（GraphDefinitionSpec，配置驱动 / 命令集成） | ctx.graph.fromDefinition() |
纯函数增量补丁：节点只返回 Partial<State>，引擎不可变合并 | StateGraph.run |
| 防死循环熔断（默认 25 次迭代，超限抛错） | StateGraph 内置 |
| 并行分支 Fan-out / Fan-in（条件路由返回目标数组，join 汇聚） | StateGraph.run |
双向事件流观测（graph/* 事件，追加式轨迹） | ctx.emit 事件总线 |
| 节点级 checkpoint 回调（宿主可接 dsh 持久化） | StateGraph.run |
审批门节点（执行前请求 ctx.approval 一次性授权） | StateGraph.addApprovalGate |
| dsh slash 命令集成（图直接注册为命令） | ctx.graph.registerCommand() |
| 隔离的状态图实例工厂 | ctx.graph.create() |
| 子代理节点（节点委托给真实 dsh 子代理） | StateGraph.addSubagent |
核心接口
| 接口 / 类型 | 签名 | 说明 |
|---|
NodeHandler<T> | (state: T, ctx: Context, signal?: AbortSignal) => Promise<Partial<T>> | Partial<T> | 节点业务执行体，返回需合并的状态增量；第二参数仍为 ctx，第三参数接收本次运行的取消信号 |
ConditionHandler<T> | (state: T, ctx: Context, signal?: AbortSignal) => string | string[] | Promise<string | string[]> | 动态路由，返回下一个 NodeName、并行目标数组或 "__END__"；第二参数仍为 ctx，第三参数接收本次运行的取消信号 |
GraphExecutionResult<T> | { graphId: string; finalState: T; trajectory: string[]; iterations: number } | 每次执行的 graphId（区分重复/并发轨迹）、最终状态、全量跳转轨迹、迭代次数 |
ctx.graph.create<T>(maxIterations?) | → StateGraph<T> | 创建隔离的图实例 |
ctx.graph.fromDefinition<T>(spec) | → StateGraph<T> | 从声明式定义构建图（与链式 API 等价） |
ctx.graph.registerCommand<T>(def) | → () => void | 把图注册为 dsh slash 命令（需 ctx.commands），返回注销函数 |
StateGraph.addApprovalGate(name, options) | → this | 注册审批门：执行前请求 ctx.approval 一次性授权 |
StateGraph.addSubagent(name, options) | → this | 注册子代理节点：执行时经 ctx.subagents.start(provider, …) 委托给真实 dsh 子代理，输出映射回父状态 |
StateGraph.setSubagentProvider(provider) | → this | 设置本图子代理节点的默认 provider 名（未在 addSubagent 显式指定时生效） |
GraphRunOptions | { signal?, agent?, checkpoint? } | agent 供审批门透传与子代理节点作为父 Agent；checkpoint 每次节点合并后回调 |
一次运行可通过 run(initialState, { signal }) 传入 AbortSignal；不传第二参数时保持原有调用方式。引擎会在节点执行、节点结果合并、条件路由以及跳转边界检查取消状态，并把同一个 signal 作为 NodeHandler / ConditionHandler 的第三参数传入。
checkpoint 回调在每次节点补丁合并后调用（载荷 { graphId, node, state, iteration }），适合把状态快照写入 dsh 持久化（session log / ctx.fs 由宿主决定）；回调不应抛错。agent 仅在存在审批门节点时需要（透传给 ctx.approval.request）。
事件（ctx.on("graph/…") 订阅）
| 事件 | 载荷 | 时机 |
|---|
graph/start | { graphId, initialState, entryPoint } | 图开始执行 |
graph/node-start | { graphId, node, state, iteration } | 每个节点执行前 |
graph/node-end | { graphId, node, state } | 节点补丁合并后 |
graph/node-error | { graphId, node, error } | 节点抛错或被取消中断（随后上抛；取消还会补发 graph/error） |
graph/error | { graphId, error, state, lastNode } | 迭代熔断 / 目标节点缺失 / 路由抛错 / 取消（graph/start 之后） |
graph/end | { graphId, finalState, trajectory, iterations } | 仅正常终止（END 或无出边） |
graphId 标识每次执行（同一图实例的重复/并发运行各自独立生成），用于区分并发图之间可能重名的节点轨迹。graph/end 只在正常终止时发出；所有异常终止统一发 graph/error 后上抛——包括迭代熔断、条件路由返回未注册节点名/非字符串/混用哨兵的并行数组、路由函数抛错，以及 graph/start 之后的取消（节点执行中的取消先发 graph/node-error 过程诊断、再发 graph/error 终态；并行分支的取消会为每个失败分支各发一次 graph/node-error 过程诊断，但 graph/error 终态仅一次；graph/start 之前的预取消无任何事件）。节点自身业务抛错仅发 graph/node-error 后上抛（并行分支同规则）。graph/* 监听器同步执行且异常会传播进引擎（cordis emit 语义），订阅方不应在监听器中抛错。
logTrajectory 开启时，插件订阅 node-start（debug 级）与 end（info 级）把轨迹写入 ctx.logger("graph")。
安装
插件已发布到 npm registry（dsh-state-graph，随 v* tag 由 CI 自动发布）。在 DeepSeek Harness 中通过 npm 包路径安装——dsh plugin add 会安装依赖并自动把包名追加到 profile 的 dsh.profile.bundles：
dsh plugin add --profile web dsh-state-graph
dsh --profile web --dump-config # 确认 state-graph 行已组合
profile 的 dsh.profile.bundles 需要包含 dsh-state-graph（与 @deepseek-ai/dsh-base 一起）。插件无其他注入依赖，任何 profile 均可加载。升级到新版本：dsh plugin add --profile web dsh-state-graph@latest。
配置（cordis.patch.yml 的 config）
- insert:
- id: state-graph
name: 'dsh-state-graph'
config:
defaultMaxIterations: 25 # 防死循环单次执行最大迭代上限
logTrajectory: true # 输出路由轨迹到日志总线
defaultSubagentProvider: spawn # 子代理节点未显式指定 provider 时的默认值
Graph Studio（可视化配置与执行）
安装到 web profile 后，浏览器 half 随包发布（dsh.client 声明 + lib/client.js），提供两个入口：
- 全屏抽屉（推荐，画布最开阔）：访问
http://<dsh-host>/#graph-studio-open（或 /graph-studio-open）即铺满主界面——左侧 480px 配置列，右侧 SVG DAG 独占剩余画布；点「✕ 关闭」收起。
- 设置分区：右上角 设置（齿轮）→ Graph Studio，窄容器快捷查看/编辑，与「Agent 预设」同层。
设计器与单个会话无关，因此不占会话视图 tab。
- 表单配置：节点（
patch / counter / subagent / gate 四类）、静态边、条件边（规则按序匹配 + 兜底目标）、入口与迭代上限；左侧编辑，右侧实时渲染 SVG DAG（静态边实线、条件边虚线、回环绕行）。
- 可视化执行：
▶ 运行 在浏览器内用与宿主同一份 StateGraph 引擎执行（引擎零 Node 依赖，单源内联进 client bundle）：当前节点高亮、已执行节点计数、逐步轨迹（耗时 + 状态增量 patch）、终态与错误详情，■ 停止 随时取消（AbortSignal）。
- host-only 节点：
subagent / gate 节点带 ▲ 徽标——需要 dsh-subagent / dsh-user-approval 运行时，浏览器内执行到该节点会得到教学错误；导出 JSON 后在宿主侧用 studioBuildGraph 运行即可。
- 图定义存 localStorage，
导入 JSON / 导出 JSON 双向互通；内置三个示例（代码生成质量门回环 / 并行展开汇聚 / 子代理委托）。
导出的图定义 JSON（DSL）形状：
{
"entryPoint": "generate_code",
"maxIterations": 25,
"nodes": [
{ "name": "generate_code", "kind": "patch",
"patch": { "rev": { "$inc": 1 }, "code": "fn_draft" } },
{ "name": "static_analyze", "kind": "patch",
"patch": { "lintOk": { "$test": { "field": "rev", "op": "gte", "value": 2 } } } }
],
"edges": [{ "from": "generate_code", "to": "static_analyze" }],
"conditionalEdges": [
{ "from": "static_analyze",
"rules": [{ "field": "lintOk", "op": "eq", "value": true, "to": "__END__" }],
"fallback": "generate_code" }
]
}
- patch 值支持
{"$inc": n}（数值自增）与 {"$test": {field, op, value}}（布尔判定，eq/ne/gt/gte/lt/lte/exists），均基于节点执行前的状态快照；其余为字面量。
counter 节点：{ field, limit, then? } 自增字段，新值达到 limit 时额外合并 then。- 条件规则的目标支持数组（
"to": ["a", "b"] 即 Fan-out，不可与 "__END__" 混用）；缺省 field 的规则恒命中；全部未命中走 fallback（默认 "__END__"）。
- 宿主侧复用：
import { studioValidate, studioBuildGraph, studioExamples } from "dsh-state-graph"。
用法示例：代码生成与双向对齐检查图
import { Context } from "@deepseek-ai/cordis";
// 在其他插件 / 工具 / 命令中：
export function buildGraph(ctx: Context) {
return ctx.graph
.create() // 隔离实例，maxIterations 默认取插件配置
.addNode("generate_code", async (s, ctx, signal) => {
const code = await llmGenerate(ctx, s.task, signal); // 业务函数自行遵守 signal
return { code, rev: s.rev + 1 };
})
.addNode("static_analyze", (s) => ({ lintOk: lint(s.code).ok }))
.addNode("run_unit_test", async (s, ctx) => ({
testOk: (await runTests(ctx, s.code)).passed,
}))
.addEdge("generate_code", "static_analyze")
.addEdge("static_analyze", "run_unit_test")
.addConditionalEdge("static_analyze", (s, _ctx, signal) => {
signal?.throwIfAborted();
return s.lintOk ? "run_unit_test" : "generate_code";
})
.addConditionalEdge("run_unit_test", (s) =>
s.testOk ? "__END__" : "generate_code",
)
.setEntryPoint("generate_code");
}
// 使用：
const graph = buildGraph(ctx);
const controller = new AbortController();
const { finalState, trajectory, iterations } = await graph.run(
{ rev: 0, task },
{ signal: controller.signal },
);
ctx.logger("app").info(`route: ${trajectory.join(" -> ")} (${iterations} steps)`);
条件边优先于静态边；无出边或条件路由返回 "__END__" 即终止。addEdge / addConditionalEdge 对同一 from 重复注册会抛错（与 addNode 的重名检查一致）；静态边与条件边可挂在同一节点上，条件边优先生效。条件路由返回未注册节点名、非字符串或 undefined（如忘写 return）时抛错并发 graph/error（熔断语义同上，非静默终止）。
条件路由返回字符串数组时并行执行各目标（Fan-out），所有分支补丁合并后汇聚：取分支中定义静态边的最后一个分支的出边作为 Join 节点，无出边则终止。并行数组不能混用 "__END__" 与节点名（混用抛错并发 graph/error）。并行路径下 iterations 为执行轮数（一轮可执行多个分支节点），trajectory 为节点级轨迹（含全部分支），二者长度可不一致。
与 dsh harness 集成
图注册为 slash 命令（ctx.graph.registerCommand）
需要 dsh-commands 插件（ctx.commands）。命令的 rawInput 默认按 JSON 解析为初始状态，执行后以轨迹文本作为 CommandResult 返回；parseInput 可自定义解析：
export function registerPipelineCommand(ctx: Context) {
ctx.graph.registerCommand({
name: "pipeline",
description: "运行代码生成质量门流水线",
inputHint: '{"task":"..."}',
graph: {
entryPoint: "generate_code",
nodes: {
generate_code: async (s, ctx) => ({ code: await llmGenerate(ctx, s.task) }),
static_analyze: (s) => ({ lintOk: lint(s.code).ok }),
run_unit_test: async (s, ctx) => ({ testOk: (await runTests(ctx, s.code)).passed }),
},
edges: [
{ from: "generate_code", to: "static_analyze" },
{ from: "static_analyze", to: "run_unit_test" },
],
conditionalEdges: [
{ from: "static_analyze", condition: (s) => (s.lintOk ? "run_unit_test" : "generate_code") },
{ from: "run_unit_test", condition: (s) => (s.testOk ? "__END__" : "generate_code") },
],
},
});
}
持久化：节点级 checkpoint
run(initialState, { checkpoint }) 在每次节点补丁合并后回调，宿主把 { graphId, node, state, iteration } 写入自己的持久化层（session log、ctx.fs、SQLite 等）。图引擎本身保持内存态——dsh 的"模型可见即落日志"不变量由宿主的 checkpoint 实现决定。
审批门（addApprovalGate）
需要 dsh-user-approval 插件（ctx.approval）。节点执行前请求一次性授权，allowed-once 放行，拒绝 / 取消 / 无应答按节点错误路径上报：
graph
.addNode("deploy", deployHandler)
.addApprovalGate("deploy", {
toolName: "graph.deploy",
reason: "部署到生产环境需要人工确认",
})
.setEntryPoint("deploy");
await graph.run(state, { agent }); // agent 为审批上下文（如命令 invocation.agent）
限制：ctx.approval.request 要求会话处于 open turn 内（dsh 持久化日志的提交边界），因此审批门应在 turn 上下文（agent 工具 / 事件处理器）中使用；无 turn 的 slash 命令场景审批请求会被 dsh 拒绝——需要审批的命令图应改用 agent 工具或 agent/inject 路径执行。
子代理节点（addSubagent）：子图 = 真实 dsh 子代理
addSubgraph 是把一张进程内状态机嵌套进父图（复用同一 ctx，节点 handler 直接跑）。addSubagent 则把节点委托给一个真实的 dsh 子代理——独立的 Agent、独立会话、可走 LLM 与工具，通过 ctx.subagents.start(provider, …)（需要 dsh-subagent 运行时，如 dsh-subagent-spawn-in-process 提供的 spawn / dsh-subagent-fork-in-process 提供的 fork）。
graph
.addSubagent("research", {
provider: "spawn", // 默认取插件配置 defaultSubagentProvider
prompt: "研究任务：{{state}}，返回 JSON：{ \"result\": { \"summary\": \"…\" } }",
inputMapper: (s) => ({ task: s.task }),
// outputMapper 默认 fromSubagentPatch：
// structured.result → 直接作为对象增量合并；否则尝试 JSON.parse(文本)；
// 仍失败则落到 state[resultKey]（默认 "result"）字段
})
.addEdge("research", "write_report")
.setEntryPoint("research");
// run 需提供父 Agent 上下文（子代理的 parent）：
await graph.run({ task }, { agent: invocation.agent, signal });
要点：
- 父 Agent：
run(initialState, { agent }) 注入的 agent 作为子代理的 parent（dsh 需要真实父 Agent 推导会话、谱系与委托深度）。必须在 turn 上下文（agent 工具 / 事件处理器）中执行——无 agent 或 ctx.subagents 缺失时节点抛错，按节点错误路径（graph/node-error）上报。
- 输入映射：
inputMapper 把父状态投影为子代理初始输入，默认透传整个 state；prompt 支持 {{node}}（节点名）与 {{state}}（JSON 序列化后的输入）模板变量。
- 模型路由（自动继承，可覆盖）：子代理默认自动继承父 Agent 的模型路由（dsh
resolveChildAgentOptions 从 parent.options 继承 provider/model/maxTokens）。若父 Agent 的 options 缺失 provider 或 model（例如父会话实际走 agent/request waterfall 或每步选模型），插件会用 ctx.agentDefaultModel.currentSelection() 补齐缺失字段（保留父已有的）。显式传 agentOptions: { provider, model, maxTokens } 则完全覆盖自动解析结果，优先级：显式 agentOptions > 父 parent.options > agentDefaultModel 兜底。实测教训：子代理继承的模型可能路由失败（如 PI_AI_ERROR: all model candidates failed），此时用 agentOptions 显式指定可用模型即可。
- 结构化输出：节点默认请求
outputSchema（{ type: "object", properties: { [resultKey]: {} }, required: [resultKey], additionalProperties: false }），子代理返回 structured[resultKey] 即作为父状态增量合并；不请求结构化时文本输出先试 JSON.parse，失败则落到 state[resultKey]。
- 失败与取消：
stopReason !== "completed"（aborted / error / max-tokens / refusal）抛错并附 diagnostic 与部分输出；父图 signal 透传给子代理请求（SubagentStartRequest.signal），取消按节点错误路径上报（先 graph/node-error，若取消则补发 graph/error 终态）。结果收集后 run.dispose() 释放子代理。
- 透传：
agentOptions（model / maxTokens）、maxDepth（需 provider 支持 depthLimit）、toolFilter、persona 均可透传；provider 缺能力时 dsh 会 fail loud 拒绝。
- provider 默认值：节点级
options.provider → 图级 setSubagentProvider → 插件配置 defaultSubagentProvider（默认 spawn）。
- 与并行 Fan-out 组合：条件路由返回数组时各分支可含子代理节点，并发委托；取消为每个失败分支各发一次
graph/node-error，graph/error 终态仅一次。
运维与性能考量
- 状态合并策略：默认浅拷贝 + 结构补丁合并（
{ ...state, ...patch }）。大对象（文件内容、仓库快照）建议经引用路径或沙箱虚拟文件系统管理，避免全量复制造成 GC 压力。
- 取消与异步超时控制：
run(initialState, { signal }) 是合作式取消。引擎只在节点、路由和图遍历边界检查 signal，不会强杀忽略 signal 的普通 Promise；NodeHandler / ConditionHandler 也应把第三参数传给所调用的 LLM、工具或外部服务。取消的终态事件语义见上文事件表：graph/start 之后的取消一律补发 graph/error。引擎不内置单节点超时，单节点超时仍由调用方或 handler 自行组合 AbortSignal 与定时器。
- 观测：节点级耗时分析 = 同一节点相邻
node-start / node-end 事件时间差；执行流回放 = 按序消费 graph/* 事件。
开发
npm install
npm run build # tsc → lib/ + 引擎 CJS（.client-build/）+ 组装 lib/client.js（Graph Studio 浏览器 bundle）
npm run typecheck
npm run smoke # build + 冒烟测试（设计文档 §5 工作流 + 熔断/入口校验/事件流）
npm test # build + node --test（StateGraph.run 全流程/事件契约/Studio DSL/客户端 bundle 装载）
许可
MIT。实现遵循《DeepSeek Harness (dsh) 状态图（StateGraph）编排引擎技术方案与架构设计》。
