Agent DAG Workflow

August 30, 2026 · View on GitHub

@gm-hz/agent-dag-workflow 是一个轻量、Host-neutral 的 DAG Workflow Runtime。它让 Codex、DSH 或其他 Agent 直接复用已有的 Tool、Skill 与 MCP，把离散调用组织成可保存、校验、恢复、审计和重放的 WorkflowTemplate JSON。

它不是另一个 Coze/Dify 平台，也不内建模型 Provider、凭据中心或 Tool 市场。Host 继续负责已有的 Agent/Tool/Skill/MCP 生态；本项目只负责把离散能力变成稳定流程。

为什么这样设计

flowchart LR
  A["Codex / Terminal Agent"] --> S["On-demand Skill"] --> C0["CLI"]
  M["MCP-only Agent"] --> G["Fixed MCP Gateway"]
  D["DSH / Embedded Host"] --> H["Native Adapter / SDK"]
  C0 --> X["WorkflowAgentAccess"] --> R["WorkflowRuntime"]
  G --> X
  H --> R
  T["Cron / Webhook / Channel"] --> I["Trigger Ingress"] --> R
  R --> C["Catalog + Compiler"]
  R --> E["DAG Engine"]
  E --> HG["Host Tool / Agent Gateway"]
  E --> J["Journal + Checkpoint"]
  J --> V["Trace / Canvas / Replay"]

核心约束：

一份 JSON：SDK、Agent、CLI、MCP、DSH 和 Canvas 使用同一 WorkflowTemplate，不维护第二套 DSL。
两级扩展：普通外部能力注册为 Host Tool，由 tool.call@1 调用；只有暂停恢复、长任务 checkpoint、补偿等生命周期语义才实现自定义 Node。
没有 Provider 层：MCP Tool、本地受控命令、DMS、HTTP、数据库和消息能力都由 Host Gateway 适配。
权限只会收窄：模板先声明 requires，节点再声明固定依赖；最终能力是模板声明、节点声明、Authority 和 Host policy 的交集。
Agent Access 默认只允许同一 authorityRef 读取、追踪、重放或恢复持久化 Run；多租户管理员访问必须通过显式 authorize policy 授权。
Host 通过 WorkflowDeploymentLimits 持有不可提升的并发、时长、节点次数、单次输出、累计 Checkpoint 和子流程深度 ceiling。
编译器执行分支路径支配检查，拒绝发布在某条激活路径上必然缺少数据的 Workflow。
外部动态结果只有通过 lossless JSON、schema、expects、端口和大小检查后，才会进入 Artifact、Journal 和 Checkpoint。
Script 只做纯 JSON：core.script@1 没有网络、文件、环境变量、密钥或 eval。含外部副作用的循环必须使用 core.foreach@1。
运行可复现：run 固化模板、发布修订、依赖闭包、Engine 版本和 NodeDefinition set hash；Journal 与 Checkpoint 原子提交。
Trigger 不进入 DAG：Cron、Webhook、钉钉等只产生可信 Envelope，再通过固定 Binding 启动发布修订。
默认由当前 Agent、CLI 或 Host 直接调用 Runtime；Queue/Runner 只是不可靠进程或分布式部署需要时才启用的可选适配器。

代码按职责分成五层，但仍作为一个 npm 包发布：

层	负责	不负责
Template / Catalog	v1 JSON、校验、草稿 CAS、不可变发布修订	执行外部能力
Runtime / Engine	编译 DAG、调度、恢复、Replay、资源上限	发现凭据或绕过 Host 权限
Journal / SQLite	Run、Event、Checkpoint、Artifact、Ingress 与 Delivery 事实	猜测旧格式语义
Access / Adapter	SDK、CLI、固定 MCP Gateway、Skill、DSH、Trigger、Canvas	创建第二套执行引擎或 DSL
Host Gateway	Tool、Agent、Approval、Authority 和自定义 Node 实现	修改 Workflow 的调度事实

当前架构见总体架构，安全与恢复不变量见 Core hardening 和 Core Verification Harness，模板字段见 Workflow Template v1。

完整的使用手册发布在 Agent DAG Workflow 文档站。仓库内文档是站点唯一内容源，随 main 分支自动部署，不维护另一份 Wiki 副本。

安装

要求 Node.js 22.19+：

npm install @gm-hz/agent-dag-workflow

这是唯一公开包。不同宿主通过 subpath export 按需引用：

import { WorkflowRuntime } from '@gm-hz/agent-dag-workflow'
import { SqliteWorkflowRunStore } from '@gm-hz/agent-dag-workflow/sqlite'
import { createMcpGateway } from '@gm-hz/agent-dag-workflow/mcp'
import * as DshWorkflow from '@gm-hz/agent-dag-workflow/dsh'

未导入的 DSH、Canvas、MCP 或 Trigger Adapter 不会自动启动。

最小 SDK 用例

下面的流程不需要任何外部 Provider，只执行确定性 JSON 变换：

import {
  InMemoryWorkflowCatalogRepository,
  InMemoryWorkflowRunStore,
  WorkflowNodeRegistry,
  WorkflowRuntime,
  WorkflowTemplateCatalog,
  registerCoreNodes,
} from '@gm-hz/agent-dag-workflow'

const nodes = new WorkflowNodeRegistry()
registerCoreNodes(nodes)

const catalog = new WorkflowTemplateCatalog(
  new InMemoryWorkflowCatalogRepository(),
  nodes,
)

const runtime = new WorkflowRuntime({
  nodes,
  catalog,
  runStore: new InMemoryWorkflowRunStore(),
})

const template = {
  apiVersion: 'workflow.gm-hz.dev/v1',
  kind: 'WorkflowTemplate',
  metadata: { id: 'hello', name: 'Hello' },
  spec: {
    inputSchema: {
      type: 'object',
      required: ['name'],
      properties: { name: { type: 'string' } },
    },
    outputSchema: {
      type: 'object',
      required: ['message'],
      properties: { message: { type: 'string' } },
    },
    requires: [{ kind: 'script-runtime', uses: 'json.expr@1' }],
    nodes: [
      { id: 'start', uses: 'core.start@1', with: {}, inputs: {} },
      {
        id: 'format',
        uses: 'core.script@1',
        with: { language: 'json.expr@1', source: '{ message: "Hello, " + input.name }' },
        inputs: { name: { input: { path: ['name'] } } },
      },
      {
        id: 'end',
        uses: 'core.end@1',
        with: {},
        inputs: { message: { output: { nodeId: 'format', path: ['message'] } } },
      },
    ],
    edges: [
      { id: 'start-format', source: 'start', target: 'format' },
      { id: 'format-end', source: 'format', target: 'end' },
    ],
    outputs: { message: { output: { nodeId: 'end', path: ['message'] } } },
  },
}

const handle = await runtime.launch({
  target: { type: 'inline', template },
  inputs: { name: 'Workflow' },
  authorityRef: 'sdk:local',
  authority: {},
  origin: { type: 'sdk' },
})

console.log(await handle.result)

生产调用应先创建 draft、校验并发布，再使用固定 revision：

const draft = await runtime.createDraft(template)
const published = await runtime.publish(draft.id, draft.revision)

const handle = await runtime.launch({
  target: { type: 'published', id: published.id, revision: published.revision },
  inputs: { name: 'Workflow' },
  authorityRef: 'user:42',
  authority: currentUser,
  origin: { type: 'sdk' },
  idempotencyKey: requestId,
})

接入 Host Tool 与 Agent

模板中的外部调用只经过显式 Gateway：

const runtime = new WorkflowRuntime({
  nodes,
  catalog,
  runStore,
  services: {
    tools: {
      async execute(request) {
        // 在这里执行 Host 自己的 scope、guard、审批、凭据和审计策略。
        return hostTools.execute(request.uses, request.inputs, {
          authority: request.authority,
          invocationId: request.invocationId,
          signal: request.signal,
        })
      },
    },
    agents: hostAgentGateway,
  },
})

tool.call@1 的 with.uses 必须是固定能力名，并同时出现在 spec.requires。模板不能传入任意 shell、动态 Tool 名或明文 Secret；connectionRef/credentialRef 只是不透明引用，最终由 Host 解析。

Script、Condition 与 Foreach

三者不是重复能力：

场景	节点	原因
JSON map/filter/reduce/sort	`core.script@1`	无副作用，可作为一个原子节点重算
选择静态 DAG 端口	`core.condition@1`	Scheduler 必须记录 taken/skipped edge
对每个 item 调用 Tool/Agent/子流程	`core.foreach@1`	需要并发上限、逐项 checkpoint、稳定 invocationId 和恢复

不支持无界 while，也不允许 Script 返回动态节点后让 Engine 隐式执行。

Journal、恢复与 Replay

Runtime 提供三种不同语义：

inspect：只读取历史事实，不执行任何节点；
recorded：创建新 run，使用已提交的外部节点结果，重新计算确定性下游；
live：创建新 run，并重新调用外部能力。

const page = await runtime.readEvents(runId, { afterSeq: 0, limit: 100 })
const replay = await runtime.replay({ runId, mode: 'recorded' })

Recorded Replay 不声称重放模型隐藏思维链。它只使用显式输入、公开内容、结构化输出和按部署 Capture Policy 保存的 Artifact。默认 Memory/SQLite Artifact Store 不伪装提供静态加密或自动过期；启用对应策略时必须换成声明了 encryptionAtRest/retentionPolicy capability 的 Store，否则 Runtime 会拒绝启动。

CLI

CLI 提供简写 adw 和完整命令 agent-workflow，二者完全等价。日常交互推荐使用 adw；脚本可以继续使用语义更明确的完整命令。CLI 默认使用当前目录的 .agent-dag-workflow.db，也可以用 --db 指定 SQLite 文件：

adw validate examples/script-transform.workflow.json
adw draft put examples/script-transform.workflow.json --db workflows.db
adw publish script-transform-demo --expected 1 --db workflows.db
adw search "transform" --db workflows.db
adw describe script-transform-demo@1 --view schema --db workflows.db
adw run script-transform-demo@1 --input input.json --db workflows.db
adw run-get <runId> --db workflows.db
adw trace <runId> --events --db workflows.db
adw trace <runId> --follow --format jsonl --db workflows.db
adw replay <runId> --mode recorded --db workflows.db
adw resume <runId> --db workflows.db
adw cancel <runId> --reason "operator stop" --db workflows.db

所有非流式命令都返回单个 agent-workflow.cli/v1 JSON Envelope；--input - 从 stdin 读取 JSON，不需要把大型输入塞进 shell 参数。CLI 对每个命令使用严格参数契约，未知、重复或多余参数会在打开数据库前 fail closed。包含 Tool/Agent 节点时，必须显式传入 --host ./host.mjs。该模块导出 Gateway、Authority 和可选自定义 Node；CLI 不会隐式读取环境变量来猜测能力或凭据。

Host 不需要 Provider 层。最小 Tool Adapter、加载时契约校验、Authority 边界和错误排查方式见 Host Adapter 接入。

后台调用使用 run ... --detach，并由 agent-workflow worker --once claim/resume。Host 必须提供可恢复的 Authority Resolver，否则 Runtime 会拒绝后台启动。1.0 的 Worker 是单进程、单次 claim/resume 的参考执行器；Core 不内置 worker_threads、进程池或分布式调度。需要水平扩展时，由 Host 在共享 Store 上补充 fencing token 与部署级调度约束。

Codex、Skill 与 MCP

具备终端能力的 Codex 类 Agent 默认使用仓库内的 workflow-builder Skill 和 CLI。Skill 只在 Workflow 任务命中时加载，不包含执行逻辑。Codex Plugin 位于 integrations/codex/agent-dag-workflow，已按官方 manifest 结构打包同一 Skill。

从源码安装 Codex Plugin 时，把该目录作为一个本地 marketplace；安装后新建会话即可按需发现 Skill，且不会常驻启动 MCP：

codex plugin marketplace add "$PWD/integrations/codex"
codex plugin add agent-dag-workflow@agent-dag-workflow-local

Plugin 的 wrapper 只发现并调用同一个 agent-workflow CLI。卸载 Plugin 不会删除 Workflow SQLite 数据；数据路径仍由 CLI/Host 配置决定。

没有本地命令能力的 Agent 可以启动一个固定 Tool 数量的 MCP Gateway：

agent-workflow-mcp --db workflows.db --profile invoke
agent-workflow-mcp --db workflows.db --profile author

invoke profile 永远只有 workflow_search、workflow_describe、workflow_run、workflow_run_get、workflow_cancel 和 workflow_trace 六个 Tool。author 额外提供六个有界的节点、校验、草稿、diff 和发布 Tool。Catalog 中有多少 Workflow 都不会改变 Tool 数量；Agent 只按需读取被选中 Workflow 的 Schema。搜索由 Repository 在已发布 revision 上有界执行，不会读取未发布 Draft 元数据。

DSH 与 Canvas

DeepSeek Harness 是一个 Adapter，不是 Core 前提。安装同一个包即可加载 DSH Tool/Agent/Skill、SQLite 和 Canvas：

dsh plugin --profile web add @gm-hz/agent-dag-workflow

从当前源码验证时只链接仓库根目录：

pnpm install
pnpm build
dsh plugin --profile web add "$PWD"
dsh web

插件向 DSH 注册 workflow-builder Skill，以及查询节点、创建/更新/校验 draft、发布和运行的受保护工具。Canvas 编辑的是同一份 WorkflowTemplate，Trace 来自同一份 Journal。 Canvas 的“触发与投递”页面还能查看 Binding、重复 Ingress、run 关联和状态不确定的 Delivery，并从入口直接打开权威 Trace。

根 bundle 会把持久 Run 的 authorityRef 绑定到稳定的 DSH Session.id，并通过 agents 服务在重启后恢复当前 Agent；它不会读取旧的 Session 字段，也不会把 Agent object 或凭据写进 SQLite。外部 Tool 在未知副作用边界上恢复时仍会进入 paused，需要操作者显式选择 retry/fail。

Trigger

Trigger 通过不可变 Binding 把可信入口映射到固定发布修订：

验签 → 生成可信 Envelope → Ingress 去重 → Binding 映射
     → 幂等 launch → WorkflowRun → Result Delivery

外部 payload 不能指定最终 Authority、幂等键或 Workflow revision。Cron、Webhook 和钉钉只提供 reference adapter；生产部署仍需按平台协议实现可靠 HTTP/消息接收、加密凭据、持久队列和运维告警。

后台 Worker 在 Run 进入终态后会按 deliveryRef 自动调用 Result Delivery。投递与 Workflow 终态分离：失败或状态未知不会把已完成 Workflow 改成失败，而是写入可重试的 delivery attention；运维方法与 SQLite 备份、导出和清理见运行与存储运维。

Trigger Adapter 只需注册自己的 uses 与配置 Schema；通用 Binding Catalog 负责目标、映射和 CAS，不需要 Provider 层：

import {
  SqliteWorkflowBindingRepository,
  WorkflowBindingCatalog,
  WorkflowTriggerDefinitionRegistry,
} from '@gm-hz/agent-dag-workflow'

const triggers = new WorkflowTriggerDefinitionRegistry()
triggers.register({
  uses: 'acme.message@1',
  configSchema: { type: 'object', additionalProperties: false },
})

const bindings = new WorkflowBindingCatalog(
  new SqliteWorkflowBindingRepository({ path: 'workflows.db' }),
  catalog,
  triggers,
)

await bindings.publish({
  apiVersion: 'workflow.gm-hz.dev/v1',
  kind: 'WorkflowBinding',
  metadata: { id: 'weekly-from-acme' },
  spec: {
    workflow: { id: 'weekly-ai-model-news', revision: 1 },
    trigger: { uses: 'acme.message@1', with: {} },
    inputMapping: { from: { payload: { path: ['from'] } }, to: { payload: { path: ['to'] } } },
    authorityRef: 'service:acme-channel',
  },
}, 0)

CLI、固定 MCP Gateway、DSH Plugin、SDK 和 Trigger 最终都调用同一个 Runtime。入口不会改变固定 revision、输入输出 Schema、Authority、Journal、Checkpoint 或 Replay 语义。

版本与兼容边界

Template 只接受 workflow.gm-hz.dev/v1 和当前节点 uses@major，没有旧 API Version、旧节点别名或双解析器。
SQLite 只初始化空数据库，或打开 application id 与 schema version 都精确匹配当前实现的数据库；旧、未知或被篡改的数据库会在启动时拒绝。
包不导出迁移 API，CLI 也不提供隐式转换命令。升级协议时应先导出当前模板/审计数据，再由明确的独立工具生成并人工校验新模板。
发布修订和历史 Run 永不原地改写。破坏性节点语义使用新的 uses@major，并发布新的 Workflow revision。

这一边界是 1.0 的刻意约束：Runtime 只执行一种事实模型，避免兼容分支进入调度、恢复和权限路径。

示例与验证

仓库包含以下长期基准：

script-transform.workflow.json：纯 JSON 变换；
approval-gate.workflow.json：条件分支；
batch-contract-review.workflow.yaml：foreach 与子工作流；
weekly-ai-model-news.workflow.json：多路检索、Agent 结构化、确定性排序和 Top 10；
Example 回归清单：9 个模板的固定输入、期望输出与 Codex Plugin 全链路回归；
showcase 说明：复杂场景的依赖与运行方式。

源码验证：

pnpm install
pnpm check
pnpm exec playwright-cli install-browser chromium # 首次运行或 CI 镜像中执行
pnpm verify:canvas-browser
pnpm verify:pack
pnpm demo
pnpm examples:codex
pnpm example:weekly

pnpm examples:codex 会通过真实 Codex Plugin wrapper 对清单中的 9 个模板逐一执行 validate、draft、publish、search、describe、run、run-get 和 trace；确定性 Host 让契约回归可在本地与 CI 重复。pnpm example:weekly 则单独执行 21 节点的“AI 模型周报”：13 路 Tool 调用、4 次 Agent 结构化处理、确定性合并排序、Top 10 输出和完整 Journal Trace。替换为真实 Host 的方式见 Showcase 说明。

项目使用 MIT License。验证命令和发布门禁见 Core Verification Harness 与 1.0 发布流程。