# 工作流编排与流式执行后端开发方案

> 文档状态：开发基线方案  
> 适用项目：`offline-backend-20260512/backend`  
> 对接前端：Coze Studio 独立工作流页面 `frontend/apps/coze-studio/src/workflow-standalone/`  
> 更新时间：2026-08-29  
> **阶段 0 状态：已完成** — 协议代码 `packages/harness/deerflow/workflows/`；文档见 `WORKFLOW_SCHEMA_V1_ZH.md` / `WORKFLOW_SSE_V1_ZH.md` / `WORKFLOW_OPENAPI_DRAFT_V1.yaml` / `adr/0001-workflow-studio-runtime.md`；契约测试 `tests/test_workflow_schemas.py`  
> **阶段 1 状态：已完成** — 定义/版本持久化、草稿乐观锁、validate/publish、节点目录、Coze canvas/save/create/node_template 最小适配；路由 `/api/workflows*` + `/api/workflow_api*`；测试 `tests/test_workflow_definitions.py`（尚不执行运行）  
> **阶段 2 状态：已完成** — run/node_run/event/artifact 表 + 迁移 `20260828_02`；幂等启动、租约调度、重启恢复、持久事件日志、SSE 回放/追尾。**编排内核改为自建 DAG 调度器，不再编译成 LangGraph**（理由见 `adr/0001` 阶段 2 修订）  
> **阶段 3 状态：已完成** — start/output/transform/condition/merge + 安全表达式引擎；http 节点 SSRF 默认拒内网；sql_read 只读策略 + 数据源登记（密钥加密、只写不读，迁移 `20260828_03`）；code 节点加固子进程执行器（**默认关闭**）  
> **阶段 4 状态：已完成** — agent / skill 节点走 DeerFlow lead agent，模型增量与工具步骤映射为 `node.output.delta` / `node.tool.*`  
> **阶段 5 状态：已完成** — `human_input` 暂停 + 一次性 resume token、子工作流（深度上限 3）、显式循环与 `max_loop_iterations` / `max_steps` 上限  
> **阶段 6 状态：已完成** — iframe 票据（origin 白名单 + 60s 一次性 + `frameAncestors`）、按 owner/admin 的运行与数据源鉴权、每用户并发运行上限、结构化审计日志、事件/运行保留期配置与后台清理作业（`WorkflowRetentionCleaner`，每小时扫一次）、失败运行 `POST /retry`（复制已完成节点从失败点续跑，新 run 标 `retry_of_run_id`）、SSE `Last-Event-ID` 优先于 `after`。制品落盘文件仍跟随现有 sandbox/artifact 生命周期，不做工作流专属文件回收。多实例租约竞争 + SSE 重连有单机并发测试（`test_two_dispatchers_only_one_claims_a_run` / `test_sse_reconnect_storm_is_ordered_and_complete`），不是外部分布式压测。  
> **会话规划补充（2026-08-31）** — 普通对话先由独立内置智能体 `workflow-planner`（工作流总控）读取用户可见业务智能体目录、理解意图并选择角色，返回 2–3 条候选策略；它与 `roundtable-coordinator` 完全隔离，只做规划，不调用工具、不执行研究或正式运行。运行时强制 `disable_tools=True` 与 `thinking_enabled=False`，因此只产生一次轻量规划模型调用。模型仅能挑选已验证的 `agentId` 与既定策略，服务端才负责构造和校验可编辑 DAG；控制器不可用时显式报错，绝不退回关键词/静态模板。`POST /api/workflows/{workflow_id}/planning-sessions/stream` 复用 DeerFlow 控制器的真实流式回调，按目录读取、总控解析、角色选择、候选图组装、校验和完成依次发 SSE 里程碑，最后才发送候选会话；连接建立即写出一个真实的「已接收」传输里程碑，并以 SSE comment 保活/跨越代理缓冲阈值，comment 不会进入业务进度或持久化事件。前端默认收起该步骤条，只展示最新阶段，不能以本地定时器伪造进度。
> **审计整改（2026-08-29）状态：功能健壮性项已完成** — 按 `WORKFLOW_STUDIO_BACKEND_REMEDIATION_ZH.md`「0. 整改决策记录」：①人工恢复原子化（`resume_with_payload` 单条件更新 + `set_awaiting_input` 一次写入 checkpoint）；②草稿保存 CAS 条件更新（409 带 `currentRevision`）+ 发布版本号计数器 `workflow_definitions.next_version_number`；③事件 seq 计数器 `workflow_runs.next_event_seq`（同事务取号，冲突自愈重试），事件落库失败**丢弃不发布**（删除负序号幽灵事件）；④输入/输出完整 JSON Schema 校验（`workflows/schema_validation.py`，路由 400 + 节点错误均带字段路径）；⑤测试纳入 Git（`.gitignore` 白名单）+ 新增 SQL 并发回归 `tests/test_workflow_concurrency.py`。迁移 `20260828_04_workflow_counter_columns.py` 增加两个计数器列（存量库由启动列同步自动补齐）。P0-01/02（内网可信环境，不实施）、P1-04/05（安全隔离，暂缓）见整改文档决策记录。  
> 运行时测试：`tests/test_workflow_engine.py`（引擎、表达式、节点、安全策略）+ `tests/test_workflow_runs.py`（HTTP 生命周期、SSE、暂停恢复、租约回收、数据源、票据）+ `tests/test_workflow_concurrency.py`（整改并发回归：内存 + 真实 SQLite）

## 1. 目标与结论

本次建设的目标不是把 Coze Studio 或 ChatDev 2.0 整体并入 DeerFlow，而是在 DeerFlow 后端中增加一套可持久化、可恢复、可审计的工作流运行时，并让 Coze Studio 的工作流画布成为它的编辑器。

首期需要支持下面这类业务链：

```text
开始/表单输入
  → SQL 只读查询 / HTTP 接口 / 数据采集智能体 / 技能
  → 数据清洗与整理智能体
  → 报告撰写智能体
  → 审核智能体
  → 条件分支（通过 / 退回修改）
  → 输出格式校验
  → Markdown、Word 或其他制品
```

最终技术选择：

1. **画布与节点配置 UI 复用 Coze Studio**，但不复用其完整后端协议和内置试运行系统。
2. **工作流执行内核建设在 DeerFlow 中**，编译为 LangGraph 图；智能体、技能、沙箱、模型、制品均复用 DeerFlow 的现有能力。
3. **ChatDev 2.0 仅作为 DAG、并行、循环、动态图和会话执行语义的参考**，不整体搬入其运行时。
4. **执行过程采用“数据库持久事件日志 + SSE 实时推送”**。前端断线后使用 `Last-Event-ID` 或 `after_seq` 补放，不依赖内存中的单次连接。
5. SQL 节点只允许服务器端已登记的数据源，并强制只读账号、语句检查、超时、行数/字节上限、表白名单和审计。
6. 代码节点必须运行在 DeerFlow 沙箱中，不能照搬 ChatDev 的宿主机 `subprocess.run`。

## 2. 范围与非目标

### 2.1 首期范围

- 工作流草稿、发布版本、复制、归档和权限校验。
- 画布 JSON 的保存、校验、发布和不可变版本快照。
- 节点：开始、输出、智能体、技能、HTTP、SQL 只读、代码、模板/转换、条件、合并、人工介入、子工作流、受限循环。
- 同一层无依赖节点并行执行。
- 运行创建、后台执行、取消、失败重试、人工介入后恢复。
- 流式返回节点状态、模型输出增量、工具步骤、制品和终态。
- 运行历史、事件重放、节点输入/输出审计和错误定位。
- Coze 画布所需的最小兼容接口。
- iframe 场景的短时票据、来源校验和权限隔离。

### 2.2 首期非目标

- 不复制 Coze Studio 全部 Bot、插件市场、数据库产品和发布渠道。
- 不实现 Coze 全量 `workflow_api`、试运行、Trace 和调试协议。
- 不把 ChatDev 的消息类、模型提供商、WebSocket 服务或 SQLite 存储直接接入 DeerFlow。
- 不允许用户把数据库密码、Bearer Token 等密钥写进画布 JSON。
- 不允许 SQL 节点执行写语句、DDL、多语句或存储过程。
- 不允许代码节点在 Gateway 宿主机直接启动任意进程。
- 首期不承诺任意环图；循环必须是显式循环节点，并带次数、时长和成本上限。

## 3. 总体架构

```text
Coze workflow-standalone（iframe）
          │
          │ HTTPS + Bearer/短时 iframe ticket
          ▼
FastAPI Gateway /api/workflows/*
  ├─ 定义、版本、资源目录、数据源 API
  ├─ 运行、取消、恢复、历史 API
  └─ SSE：持久事件补放 + 实时事件追尾
          │
          ▼
WorkflowApplicationService（app 层组合）
  ├─ WorkflowStore / RunStore / EventStore
  ├─ Dispatcher（扫描、抢占、租约、恢复）
  └─ Executor（后台任务）
          │
          ▼
deerflow.workflows（harness 层）
  ├─ Schema + Validator
  ├─ Safe expression resolver
  ├─ Node registry
  ├─ LangGraph compiler
  └─ Node executors
       ├─ DeerFlow agent
       ├─ DeerFlow skill/agent
       ├─ HTTP policy client
       ├─ read-only SQL
       └─ DeerFlow sandbox
```

必须继续遵守当前仓库的分层约束：

- `packages/harness/deerflow/**` 可以定义通用工作流模型、编译器和节点执行协议。
- `packages/harness/deerflow/**` **不得导入** `app.*`。
- FastAPI、鉴权、`app.state`、数据库 Store 组合、后台 dispatcher/executor 启停放在 `app/**`。
- 工作流智能体节点通过依赖注入使用现有 agent/runtime 能力，不新建另一套模型和工具体系。

## 4. 为什么不直接集成 ChatDev 后端

ChatDev 与本项目虽然同为 Python，但运行时边界不同。

| 对比项 | DeerFlow | ChatDev 2.0 | 处理方式 |
|---|---|---|---|
| 图运行时 | LangGraph + DeerFlow middleware | 自建 DAG executor | 保留 DeerFlow，参考 ChatDev 调度语义 |
| 智能体 | Lead Agent、定制 Agent、工具、MCP、技能 | ChatDev 自有 Agent runtime | 只调用 DeerFlow Agent |
| 流式协议 | 已有 run/event/stream 基础设施 | WebSocket 为主 | 建立 DeerFlow SSE 工作流协议 |
| 沙箱 | DeerFlow Sandbox 抽象 | Python executor 可直接 subprocess | 禁止搬入 subprocess，实现 SandboxNode |
| 持久化 | 项目统一 SQLAlchemy/数据库 | 自有 SQLite/存储服务 | 使用 DeerFlow persistence 层 |
| 技能 | `SKILL.md` 指令包，由智能体加载 | skills manager 也偏指令加载 | 明确区分 agent_skill 与 callable_skill |
| 数据库节点 | 需要新增 | 未提供通用业务库查询节点 | 自己实现严格只读 SQL 节点 |

直接整包移植会产生两套图状态、两套消息协议、两套模型调用、两套会话和两套错误恢复。后续每增加一个 DeerFlow 能力都需要做两边适配，短期看似快，长期成本反而更高。

ChatDev 适合参考的源码路径见第 15 节。

## 5. 核心领域模型

### 5.1 定义、草稿与发布版本

建议采用“工作流元数据 + 可变草稿 + 不可变发布版本”的模型：

- `workflow_definitions`
  - `id`
  - `name`
  - `description`
  - `owner_id`
  - `status`: `draft | active | archived`
  - `draft_revision`: 乐观锁版本号
  - `draft_graph_json`: 当前画布草稿
  - `created_at / updated_at`
- `workflow_versions`
  - `id`
  - `workflow_id`
  - `version_number`
  - `graph_json`: 发布时完整不可变快照
  - `graph_hash`: 规范化 JSON 的 SHA-256
  - `input_schema / output_schema`
  - `published_by / published_at`
  - `change_note`

运行必须绑定 `workflow_version_id`。运行开始后，即使用户继续编辑草稿，已有运行也不能改变。

### 5.2 运行、节点运行和事件

- `workflow_runs`
  - `id`
  - `workflow_id / workflow_version_id`
  - `status`: `queued | running | awaiting_input | completed | failed | cancel_requested | cancelled`
  - `owner_id`
  - `idempotency_key`
  - `input_json`
  - `output_json`
  - `error_json`
  - `context_json`: 非敏感运行上下文
  - `lease_owner / lease_until / heartbeat_at`
  - `attempt / max_attempts`
  - `version`: CAS 乐观锁字段
  - `started_at / finished_at / created_at / updated_at`
- `workflow_node_runs`
  - `id`
  - `run_id`
  - `node_id / node_type`
  - `status`
  - `attempt`
  - `input_snapshot / output_snapshot`
  - `error_json`
  - `started_at / finished_at`
  - `parent_node_run_id`: 循环、子工作流或重试层级
- `workflow_run_events`
  - `run_id`
  - `seq`: 每个 run 内严格递增，唯一键 `(run_id, seq)`
  - `event_type`
  - `node_id / node_run_id`
  - `payload_json`
  - `created_at`

事件日志是前端重建执行过程的事实来源；`workflow_runs` 和 `workflow_node_runs` 是便于列表查询的当前状态投影。

### 5.3 数据源与凭据

- `workflow_data_sources`
  - `id / name / kind`
  - `owner_scope`: 用户、组织或全局
  - `connection_config`: 主机、端口、库名、SSL 等非密钥字段
  - `credential_ref`: 指向服务端密钥存储，不存明文
  - `allowed_schemas / allowed_tables`
  - `statement_timeout_ms / max_rows / max_bytes`
  - `enabled`
  - `created_by / created_at / updated_at`

画布只保存 `data_source_id`。前端永远拿不到数据库密码，运行事件也不能回传连接串或认证头。

HTTP 接口资源使用同一张表：kind=http 时，base_url 和 allowed_methods 是可展示的非密钥元数据；请求头仍加密放在 credential blob，列表、详情、资源目录和运行事件均不得回显。创建时至少提供一个允许方法，地址只接受不含账户密码、查询参数或片段的 http(s) URL。迁移：20260829_01_workflow_http_source_metadata.py。

### 5.4 iframe 短时票据

如 Coze 页面与 DeerFlow 不同域，新增短时、一次性的嵌入票据：

- `workflow_embed_tickets`
  - `ticket_hash`
  - `user_id`
  - `workflow_id`
  - `allowed_actions`
  - `allowed_origin`
  - `expires_at`
  - `consumed_at`

只保存票据哈希。父页面请求票据后通过 `postMessage` 交给 iframe，iframe 再兑换短期访问令牌。不得把正式 JWT、数据库密钥或 API 密钥放在 iframe URL 查询参数中。

## 6. 画布 JSON 与运行数据契约

### 6.1 推荐的标准图结构

Coze 画布 DTO 进入 Gateway 后要转换成内部标准结构，运行时不得直接依赖 Coze 私有字段：

```json
{
  "schemaVersion": "1.0",
  "id": "wf_report_001",
  "name": "专题报告生成",
  "inputSchema": {
    "type": "object",
    "required": ["topic"],
    "properties": {
      "topic": { "type": "string" },
      "taskId": { "type": "string" }
    }
  },
  "outputSchema": {
    "type": "object",
    "required": ["reportMarkdown"],
    "properties": {
      "reportMarkdown": { "type": "string" },
      "artifacts": { "type": "array" }
    }
  },
  "nodes": [
    {
      "id": "start_1",
      "type": "start",
      "name": "开始",
      "config": {}
    },
    {
      "id": "sql_1",
      "type": "sql_read",
      "name": "查询业务数据",
      "config": {
        "dataSourceId": "ds_001",
        "statement": "SELECT id, title FROM report_source WHERE task_id = :task_id",
        "parameters": {
          "task_id": "{{ inputs.taskId }}"
        },
        "maxRows": 1000
      }
    }
  ],
  "edges": [
    { "id": "e1", "source": "start_1", "target": "sql_1" }
  ],
  "settings": {
    "runTimeoutSeconds": 1800,
    "nodeTimeoutSeconds": 300,
    "maxSteps": 200,
    "maxLoopIterations": 5,
    "maxParallelism": 4
  }
}
```

### 6.2 统一节点结果

所有节点统一返回 `NodeResult`，避免节点之间直接传递某个 LLM SDK 的对象：

```json
{
  "data": {},
  "messages": [],
  "artifacts": [
    {
      "artifactId": "artifact_01",
      "name": "专题报告.md",
      "mimeType": "text/markdown",
      "path": "/mnt/data/专题报告.md"
    }
  ],
  "metadata": {
    "rowCount": 25,
    "model": "configured-model-name"
  },
  "warnings": []
}
```

大文本、二进制文件和超大查询结果不直接塞进事件或 `output_json`，而是保存为 artifact，再传引用和有上限的预览。

### 6.3 安全表达式

支持以下只读引用：

- `{{ inputs.topic }}`
- `{{ nodes.sql_1.data.rows }}`
- `{{ nodes.agent_1.data.answer }}`
- `{{ run.id }}`

实现要求：

- 使用受限 JSON Path/JMESPath/自研字段访问器；禁止 Python `eval`、JavaScript 执行和任意属性反射。
- 发布时解析全部表达式，检查引用节点是否在当前节点上游。
- 条件节点只支持白名单运算符：相等、包含、数值比较、空值、布尔组合。
- 模板渲染默认 HTML/文本转义；HTTP 和 SQL 参数分别走参数绑定，不能字符串拼接。
- 运行时记录“引用不存在”和“类型不匹配”的结构化错误。

## 7. 节点类型与实现边界

### 7.1 开始节点 `start`

- 按工作流 `inputSchema` 校验调用参数。
- 注入用户、运行 ID、时间等安全上下文。
- 不注入访问令牌、数据库密码等密钥。

### 7.2 输出节点 `output`

- 从上游结果中映射最终字段。
- 按 `outputSchema` 做 JSON Schema 校验。
- 校验失败时工作流失败，错误中返回字段路径和期望类型。
- 可选“LLM 修复格式”必须是明确配置，并保留原始结果与修复记录；首期建议默认关闭。

### 7.3 智能体节点 `agent`

配置建议：

```json
{
  "agentId": "agent_xxx",
  "promptTemplate": "根据以下数据撰写报告：{{ nodes.merge_1.data }}",
  "responseMode": "text|json",
  "responseSchema": {},
  "threadMode": "isolated_per_node|shared_run_thread",
  "timeoutSeconds": 600
}
```

实现原则：

- 通过现有 `make_lead_agent` / 定制 Agent 配置加载机制执行。
- 不复制模型 provider、MCP、工具、memory 或 middleware。
- 每个 node run 使用稳定的 child thread 或命名空间，保证重试不把半次输出混入下一次。
- 将模型文字增量转成 `node.output.delta`；工具调用转成 `node.tool.started/finished`。
- 不向工作流事件写入模型隐藏思维链；只记录可展示的回答、状态和工具摘要。
- 节点取消要传播到 agent run 和沙箱任务。

### 7.4 技能节点 `skill`

现有 DeerFlow 技能主要是 `SKILL.md` 指令包，并不是每个技能都具有可直接调用的 Python 函数。因此首期要区分两种模式：

1. `agent_skill`：选择一个 DeerFlow 智能体和技能列表，由智能体加载指定技能完成任务。可立即落地。
2. `callable_skill`：技能必须显式声明入口、输入 JSON Schema、输出 JSON Schema、权限和超时。此模式需要新增技能 manifest 后再开放。

不能把“技能文件存在”直接等同于“可以当函数节点执行”。

### 7.5 HTTP 节点 `http`

- 方法白名单：`GET/POST/PUT/PATCH/DELETE`，生产环境可按角色收紧。
- URL 模板和请求体可引用上游数据。
- 认证使用服务端 credential 引用，画布不保存密钥。
- 支持 JSON、文本和有限大小的文件响应。
- 默认禁止重定向；如允许，每一跳都重新执行安全检查。
- 阻止 loopback、link-local、内网保留地址、云元数据地址和 DNS rebinding。
- 配置域名白名单、连接/读取/总超时、最大响应字节数。
- 日志和事件对 `Authorization`、Cookie、API Key 和配置的敏感字段脱敏。
- 返回非 2xx 时按节点策略决定失败、重试或进入错误分支。

### 7.6 SQL 只读节点 `sql_read`

这是新增能力，不能把 DeerFlow 自身业务数据库连接直接暴露给节点。

强制策略：

1. 数据库必须由管理员在服务端登记。
2. 使用数据库原生只读账号；账号不拥有 INSERT/UPDATE/DELETE/DDL/EXECUTE 权限。
3. 应用层再做 SQL AST 检查，只允许一个只读查询语句。
4. 参数全部使用驱动绑定变量，不做字符串替换。
5. 禁止多语句、注释逃逸、锁表/锁行、写入型 CTE、`SELECT ... INTO`、文件读写、危险函数和存储过程。
6. 设置 statement timeout、连接超时、最大行数、最大字段长度和最大总字节数。
7. 数据源配置可限制 schema、表和列。
8. 结果中按策略遮蔽敏感列；审计记录操作者、数据源、规范化 SQL hash、参数名、行数、耗时和状态，不记录密钥。
9. 发布校验时可做 `EXPLAIN` 或语法预检，但不能在保存画布时自动执行用户 SQL。
10. 生产环境建议从只读副本查询，避免报表工作流影响主业务库。

首期驱动建议只做项目实际使用的 MySQL 和/或 PostgreSQL，不要一开始抽象所有数据库。

### 7.7 代码节点 `code`

- 使用 `deerflow.sandbox` 抽象创建隔离执行环境。
- 输入通过 JSON 文件或 stdin 传入，输出必须符合约定 JSON。
- 设置 CPU、内存、磁盘、进程数、网络和执行时长上限。
- 默认禁网；如业务需要访问网络，按域名能力授权。
- 代码、依赖声明和运行镜像版本进入发布版本快照或可追溯引用。
- 输出文件登记为 artifact。
- 禁止调用 ChatDev `python_executor.py` 中的宿主机 `subprocess.run` 方案。

### 7.8 模板/转换节点 `transform`

- 负责字段映射、数组过滤、模板渲染、Markdown 拼装。
- 优先提供声明式操作，减少为了简单数据整理而启动智能体或代码沙箱。
- 操作符与函数采用白名单。

### 7.9 条件和合并节点

- `condition`：根据安全表达式选择一个或多个具名出口。
- `merge`：支持 `all`、`first_success`、`any_success` 等明确策略。
- 并行分支写入独立命名空间，合并节点使用确定性 reducer，不能依赖任务完成先后顺序。

### 7.10 人工介入节点 `human_input`

- 发出 `run.awaiting_input`，持久化表单 schema、提示和可执行动作。
- 将 run 状态原子更新为 `awaiting_input`，释放 worker 租约。
- `resume` API 必须校验 run 当前状态、用户权限和输入 schema。
- 同一个 `resume_token` 只消费一次，重复提交幂等返回当前状态。

### 7.11 子工作流与循环

- 子工作流绑定一个已发布的不可变版本。
- 记录 parent run/node run，事件中保留层级路径。
- 防止直接或间接递归依赖。
- 循环只能通过显式 loop 节点，必须配置最大次数；同时受全局 `maxSteps`、超时和预算限制。

## 8. 编译与执行流程

### 8.1 发布时校验

发布必须完成：

1. JSON Schema 和内部 DTO 校验。
2. 节点 ID、边 ID 唯一性。
3. 开始/输出节点数量和可达性检查。
4. 普通边拓扑检查；环只能存在于受控 loop 结构。
5. 节点输入引用必须来自上游。
6. 条件出口与边名称一致。
7. Agent、Skill、数据源、子工作流版本存在且调用者有权限。
8. 资源限制在系统上限内。
9. 输出映射能够生成声明的 output schema。
10. 规范化后生成 `graph_hash`，创建不可变版本。

### 8.2 LangGraph 编译

内部编译器将标准节点转换成 `StateGraph`：

- `WorkflowState` 保存 inputs、各节点 `NodeResult`、运行控制字段和 artifact 引用。
- `NodeRegistry` 根据 `node.type` 返回 executor factory。
- 条件节点编译为 conditional edges。
- 无依赖分支由 LangGraph 并行调度。
- 合并字段必须注册确定性 reducer。
- interrupt 用于 `human_input`，恢复时从持久 checkpoint 继续。

编译结果可按 `graph_hash` 做有界缓存；缓存对象不能包含用户级密钥或运行态数据。

### 8.3 后台运行状态机

```text
queued
  → running
      → awaiting_input → queued/running
      → completed
      → failed
      → cancel_requested → cancelled
```

实现参考 roundtable job 的成熟模式：

- 创建运行时用 `idempotency_key` 去重。
- dispatcher 只抢占租约过期或 queued 的任务。
- worker 定期 heartbeat 延长租约。
- 所有状态变更使用版本字段/CAS，防止多个实例重复完成。
- `cancel` 先设置 `cancel_requested`，worker 在安全点确认后写 `cancelled`。
- Gateway 重启后扫描 `queued` 和租约过期的 `running` 任务。
- 非幂等 HTTP 节点默认不自动重试；节点必须声明重试策略和幂等性。

## 9. SSE 流式协议

### 9.1 原则

1. **先持久化，后发布**：用户可见事件先写 `workflow_run_events`，提交成功后再推送到 live bridge。
2. **事件严格有序**：同一 run 的 `seq` 单调递增。
3. **可重放**：断线后从最后一个 seq 继续。
4. **增量事件有上限**：高频 token 可在 30–100ms 窗口内合并，避免一 token 一条数据库记录。
5. **终态明确**：`completed/failed/cancelled` 后停止重连。
6. **不泄密**：事件中不得出现凭据、完整认证头、数据库连接串和隐藏思维链。

### 9.2 端点

```http
GET /api/workflows/runs/{run_id}/stream?after_seq=128
Accept: text/event-stream
Authorization: Bearer ...
Last-Event-ID: 128
```

服务端游标优先级建议：

1. 合法的 `Last-Event-ID`
2. `after_seq`
3. `0`

连接流程：

1. 校验 run 访问权限。
2. 从数据库按 `seq > cursor` 分页补放。
3. 订阅 live bridge。
4. 再查一次数据库关闭“补放到订阅”之间的竞态窗口。
5. 实时追尾；发现 seq 跳号时回数据库补齐。
6. 无事件时每 15–25 秒发送 heartbeat/comment。
7. 收到终态且数据库已补齐后关闭连接。

### 9.3 帧格式

```text
id: 42
event: node.output.delta
data: {"schemaVersion":"1.0","runId":"run_01","workflowId":"wf_01","versionId":"wv_03","seq":42,"nodeId":"agent_writer","nodeRunId":"nr_09","timestamp":"2026-08-26T10:00:00+08:00","data":{"channel":"answer","delta":"第一段内容"}}

```

通用 envelope：

```json
{
  "schemaVersion": "1.0",
  "runId": "run_01",
  "workflowId": "wf_01",
  "versionId": "wv_03",
  "seq": 42,
  "nodeId": "agent_writer",
  "nodeRunId": "nr_09",
  "timestamp": "2026-08-26T10:00:00+08:00",
  "data": {}
}
```

### 9.4 事件清单

| 事件 | 关键数据 | 前端用途 |
|---|---|---|
| `run.created` | input 摘要、version | 创建历史记录 |
| `run.queued` | queue 信息 | 显示排队 |
| `run.started` | startedAt | 运行计时 |
| `node.queued` | node 信息 | 画布等待态 |
| `node.started` | attempt、输入摘要 | 高亮节点、创建步骤卡 |
| `node.progress` | phase、message、percent | 进度条/说明 |
| `node.output.delta` | channel、delta | 流式文本 |
| `node.tool.started` | toolCallId、name、input 摘要 | 工具步骤进行中 |
| `node.tool.finished` | toolCallId、status、output 摘要 | 完成工具步骤 |
| `artifact.created` | artifact 元数据 | 文件卡和预览 |
| `node.completed` | output 摘要、duration | 节点完成 |
| `node.failed` | error code、message、retryable | 节点失败 |
| `run.awaiting_input` | formSchema、actions、resumeToken | 人工介入卡片 |
| `run.resumed` | action | 恢复时间线 |
| `run.cancel_requested` | requestedBy | 取消中 |
| `run.cancelled` | finishedAt | 终态 |
| `run.completed` | output、artifacts | 终态与结果 |
| `run.failed` | error、failedNodeId | 终态与定位 |

heartbeat 建议使用 SSE comment `: heartbeat`，不写入数据库，也不进入前端业务 reducer。

### 9.5 事件粒度与存储控制

- 模型输出增量按时间窗口或字符数批量落库，例如 50ms/1KB 合并一次。
- 工具输入输出只存有界摘要；完整文件写 artifact。
- SQL 结果默认只在事件里放列名、行数和前 N 行预览；完整结果按节点设置进入运行快照或 artifact。
- 单事件 payload 设置字节上限；超限时写 `{truncated: true, artifactId: ...}`。
- 事件保留周期应配置化；终态 run 可异步压缩快照，但在保留期内不能破坏 seq 重放。

## 10. HTTP API 设计

### 10.1 工作流定义

```text
GET    /api/workflows
POST   /api/workflows
GET    /api/workflows/{workflow_id}
PATCH  /api/workflows/{workflow_id}
DELETE /api/workflows/{workflow_id}             # 软删除/归档
PUT    /api/workflows/{workflow_id}/draft        # draft_revision 乐观锁
POST   /api/workflows/{workflow_id}/validate
POST   /api/workflows/{workflow_id}/publish
POST   /api/workflows/{workflow_id}/copy
GET    /api/workflows/{workflow_id}/versions
GET    /api/workflows/{workflow_id}/versions/{version_id}
```

`PUT draft` 请求必须包含 `expectedRevision`，冲突返回 `409` 和服务端当前 revision，前端提示刷新或另存副本。

### 10.2 节点资源目录

```text
GET /api/workflows/node-types
GET /api/workflows/resources/agents
GET /api/workflows/resources/skills
GET /api/workflows/resources/data-sources
GET /api/workflows/resources/subworkflows
```

资源目录返回工作流节点需要的精简 DTO，不直接暴露后端内部对象。

### 10.3 数据源

```text
GET    /api/workflows/data-sources
POST   /api/workflows/data-sources
GET    /api/workflows/data-sources/{id}
PUT    /api/workflows/data-sources/{id}
DELETE /api/workflows/data-sources/{id}
POST   /api/workflows/data-sources/{id}/introspect
POST   /api/workflows/data-sources/{id}/validate-query
```

`introspect` 只返回白名单内 schema/table/column 元数据，不返回样例敏感数据。

HTTP 接口资源的创建体使用 kind=http、baseUrl、allowedMethods 和可选 headers；headers 是只写字段，服务端加密后不会出现在响应。资源目录 GET /api/workflows/resources/data-sources 返回前端所需的 dataSourceId、baseUrl、allowedMethods、description 等精简字段。

### 10.4 运行

```text
POST /api/workflows/{workflow_id}/runs
GET  /api/workflows/{workflow_id}/runs
GET  /api/workflows/runs/{run_id}
GET  /api/workflows/runs/{run_id}/events?after_seq=0&limit=500
GET  /api/workflows/runs/{run_id}/stream
POST /api/workflows/runs/{run_id}/cancel
POST /api/workflows/runs/{run_id}/resume
POST /api/workflows/runs/{run_id}/retry
GET  /api/workflows/runs/{run_id}/artifacts
```

启动示例：

```json
{
  "versionId": "wv_03",
  "inputs": { "topic": "市场分析", "taskId": "123" },
  "idempotencyKey": "client-generated-uuid",
  "executionMode": "normal"
}
```

### 10.5 Coze 最小兼容层

独立页面当前画布仍可能调用 Coze 生成 API。建议建立薄适配路由，而不是让内部 service 使用 Coze DTO：

```text
POST /api/workflow_api/canvas
POST /api/workflow_api/save
POST /api/workflow_api/create
POST /api/workflow_api/node_template_list
```

兼容层职责只有：

- Coze request/response DTO 与内部 draft DTO 互转。
- 返回节点模板和图标元数据。
- 调用内部 workflow service。

Coze 自带的 test-run、trace、workflow debug 接口首期不实现；前端关闭内置测试运行入口，改用第 10.4 节的运行 API。

### 10.6 iframe 会话

```text
POST /api/workflow-embed/tickets
POST /api/workflow-embed/tickets/exchange
```

- 创建票据需 DeerFlow 正常登录。
- ticket 绑定 `workflow_id + user_id + origin + permissions`，有效期建议 60 秒。
- exchange 一次后即失效，返回短期访问令牌和用户/工作流精简信息。
- 响应带 `Cache-Control: no-store`。

## 11. 安全、权限与审计

### 11.1 权限

工作流至少区分：

- `view`：查看定义和运行。
- `edit`：修改草稿。
- `publish`：发布不可变版本。
- `run`：启动运行。
- `manage_data_source`：登记或修改数据源。
- `view_sensitive_output`：查看未脱敏的节点输出。

所有列表、详情、SSE、artifact 下载和 resume/cancel API 都必须重新校验权限，不能因为知道 run ID 就可访问。

### 11.2 凭据处理

- 数据源、HTTP 认证和第三方 Token 使用服务端 credential reference。
- 画布和版本快照仅保存引用 ID。
- 进入日志、异常、SSE 和 node snapshot 前统一 redaction。
- 发布者必须有权引用相应 credential；运行者是否可使用由策略决定。
- credential 轮换不修改已发布图，但要记录实际使用的 credential version/audit id。

### 11.3 审计

记录：

- 草稿保存、发布、归档、权限变更。
- 运行创建、取消、恢复、重试。
- 数据源测试与 SQL 执行摘要。
- HTTP 目标域名、状态码和耗时。
- 人工介入提交者和动作。
- artifact 创建、下载和删除。

审计中保存 hash/摘要而不是密钥和完整敏感数据。

### 11.4 资源上限

系统级上限必须覆盖画布配置：

- 每用户并发 run 数。
- 单 run 最大时长、步骤数、循环次数、并行节点数。
- 单节点重试次数。
- LLM token/费用预算。
- HTTP/SQL 响应大小。
- 单 run artifact 数量与总大小。

## 12. 失败、重试、取消和恢复

### 12.1 错误结构

统一错误：

```json
{
  "code": "WORKFLOW_SQL_POLICY_DENIED",
  "message": "SQL 节点只允许单条只读查询",
  "retryable": false,
  "nodeId": "sql_1",
  "details": {
    "rule": "single_read_statement"
  }
}
```

用户可见 message 不包含堆栈、SQL 密码、完整响应体或内部路径；完整异常进入受控服务端日志。

### 12.2 重试

- 智能体节点：仅对明确的临时网络/限流错误重试，沿用稳定 node run lineage。
- SQL 只读：连接临时失败可有限重试；语法/权限错误不重试。
- HTTP：GET/HEAD 可按策略重试；POST 默认不重试，除非配置幂等键。
- 代码：资源不足或代码错误默认不重试。
- 工作流级 retry：`POST /api/workflows/runs/{run_id}/retry` 仅接受 `failed` 运行。新 run 复制已完成节点输出并从失败节点续跑；没有任何可回放 checkpoint 时则整图重跑，并标记 `retry_of_run_id`。

### 12.3 取消

- API 原子设置 `cancel_requested` 并发事件。
- worker 在节点边界、模型流、HTTP、SQL、沙箱等待点检查取消令牌。
- 尽力取消外部操作；无法中止的请求完成后也不能再提交后续节点结果。
- 终态只允许一次，不能出现 completed 后又 cancelled。

### 12.4 服务重启

- 事件已持久化，因此 SSE 可恢复。
- dispatcher 根据租约扫描未完成任务。
- awaiting_input 不占 worker。
- 对可能产生外部副作用的节点，恢复前检查幂等记录；状态不确定时标记人工处理，不能盲目重放。

## 13. 可观测性

日志统一带：

- `workflow_id`
- `workflow_version_id`
- `run_id`
- `node_id`
- `node_run_id`
- `user_id`
- `trace_id`

建议指标：

- queued/running/awaiting_input 数量。
- run 和节点成功率、P50/P95/P99 耗时。
- SSE 活跃连接、重连次数、补放事件数、游标落后量。
- dispatcher 抢占、租约过期、重复执行阻止次数。
- 各类节点错误率。
- SQL 行数/超时、HTTP 状态码、sandbox 超时。
- LLM token 与费用。

## 14. DeerFlow 现有代码参考路径

以下路径相对于 `offline-backend-20260512/backend/`。

### 14.1 运行与流式事件

| 路径 | 参考内容 | 复用方式 |
|---|---|---|
| `packages/harness/deerflow/runtime/runs/manager.py` | run 生命周期、去重、取消、持久化组合 | 复用模式，不把 thread run 数据模型硬套到 workflow run |
| `packages/harness/deerflow/runtime/runs/worker.py` | `agent.astream`、流模式、结束/异常处理 | 智能体节点适配时参考 |
| `packages/harness/deerflow/runtime/runs/schemas.py` | run DTO | 参考命名和序列化 |
| `packages/harness/deerflow/runtime/runs/store/base.py` | store 接口边界 | 为 workflow store 建立同风格接口 |
| `packages/harness/deerflow/runtime/stream_bridge/base.py` | 发布/订阅协议 | 可复用接口思想 |
| `packages/harness/deerflow/runtime/stream_bridge/memory.py` | 内存 replay、heartbeat/end sentinel | 作为 live bridge；不能代替 DB 事件日志 |
| `packages/harness/deerflow/runtime/events/store/base.py` | 事件 store 抽象 | 工作流事件仓库参考 |
| `packages/harness/deerflow/runtime/events/store/db.py` | DB 批量事件写入、按 seq 读取 | 优先抽取可复用机制或照此实现 workflow event store |
| `packages/harness/deerflow/runtime/journal.py` | 事件 journal | 参考批量与刷新策略 |
| `app/gateway/routers/thread_runs.py` | run/SSE Gateway 写法 | 参考鉴权、响应头、断连处理 |
| `docs/STREAMING.md` | 当前流协议约定 | 新协议必须兼容仓库通用规则 |

### 14.2 后台任务与租约

| 路径 | 参考内容 |
|---|---|
| `app/gateway/routers/roundtable_jobs.py` | start/get/stream/resume/cancel API |
| `app/gateway/roundtable_job_executor.py` | 后台执行、状态投影、恢复 |
| `app/gateway/roundtable_job_dispatcher.py` | 扫描、唤醒、租约 |
| `packages/harness/deerflow/persistence/roundtable_jobs/model.py` | job 表和状态字段 |
| `packages/harness/deerflow/persistence/roundtable_jobs/sql.py` | claim、heartbeat、CAS、取消 |

应复用租约和状态机经验，但不要继续采用“前端轮询完整 job snapshot”作为工作流主流协议；工作流使用增量事件 SSE。

### 14.3 智能体、技能、沙箱与制品

| 路径 | 参考内容 |
|---|---|
| `packages/harness/deerflow/agents/lead_agent/agent.py` | `make_lead_agent` 与 middleware 组装 |
| `packages/harness/deerflow/config/agents_config.py` | 定制 Agent 配置加载 |
| `app/gateway/routers/agents.py` | 智能体资源列表与权限 |
| `app/gateway/routers/skills.py` | 技能列表/读取 API |
| `packages/harness/deerflow/skills/storage/local_skill_storage.py` | 技能存储模型 |
| `packages/harness/deerflow/tools/builtins/skill_tools.py` | Agent 如何查看/加载技能 |
| `packages/harness/deerflow/sandbox/sandbox.py` | 沙箱能力接口 |
| `packages/harness/deerflow/sandbox/tools.py` | 沙箱工具调用 |
| `packages/harness/deerflow/sandbox/middleware.py` | Agent 与沙箱生命周期 |
| `packages/harness/deerflow/sandbox/local/local_sandbox.py` | 本地沙箱实现及风险边界 |
| `app/gateway/routers/artifacts.py` | artifact 查询/读取 API |

### 14.4 Gateway 和持久化接线

| 路径 | 参考内容 |
|---|---|
| `app/gateway/deps.py` | `app.state` 初始化 store/service/runtime |
| `app/gateway/app.py` | router 注册、生命周期、CORS |
| `packages/harness/deerflow/persistence/models.py` | ORM model 汇总导入 |
| `packages/harness/deerflow/persistence/engine.py` | 数据库 session/engine |
| `packages/harness/deerflow/persistence/types.py` | MySQL/SQLite 可移植 JSON、LongText、时间类型 |
| `tests/test_harness_boundary.py` | harness → app 导入防火墙 |

### 14.5 iframe 参考但不要直接复用

| 路径 | 可参考 | 不可照搬原因 |
|---|---|---|
| `app/gateway/routers/public_embed.py` | session 映射、嵌入接口形态 | public embed 偏匿名/公开读取，不满足工作流编辑与执行权限 |
| `packages/harness/deerflow/persistence/embed_sessions/model.py` | 嵌入 session 表结构 | 工作流需要一次性票据、origin 和 actions 绑定 |
| `packages/harness/deerflow/persistence/embed_sessions/sql.py` | session store 写法 | 需新增独立安全模型 |

## 15. ChatDev 2.0 参考与明确不复用的路径

以下路径相对于 `F:/react01/chatDev/`。

### 15.1 可参考

| 路径 | 参考点 |
|---|---|
| `workflow/topology_builder.py` | 从配置构建拓扑 |
| `workflow/graph.py` | 图对象职责 |
| `workflow/graph_context.py` | 图运行上下文 |
| `workflow/graph_manager.py` | 图生命周期管理 |
| `workflow/cycle_manager.py` | 显式循环控制 |
| `workflow/subgraph_loader.py` | 子图加载 |
| `workflow/executor/dag_executor.py` | DAG 调度语义 |
| `workflow/executor/parallel_executor.py` | 并行分支 |
| `workflow/executor/cycle_executor.py` | 循环执行 |
| `workflow/executor/dynamic_edge_executor.py` | 动态边选择 |
| `workflow/executor/resource_manager.py` | 并发资源控制 |
| `workflow/runtime/execution_strategy.py` | 执行策略抽象 |
| `workflow/runtime/runtime_builder.py` | runtime 组装 |
| `workflow/runtime/runtime_context.py` | 运行上下文 |
| `entity/configs/graph.py` | 图配置 schema |
| `entity/configs/node/` | 节点配置拆分方式 |
| `entity/configs/edge/` | 边配置拆分方式 |
| `server/services/workflow_run_service.py` | run service 边界 |
| `server/services/session_execution.py` | 会话执行编排 |
| `server/routes/execute.py` | 异步执行 API 形态 |
| `server/routes/execute_sync.py` | 同步执行边界；本项目不建议面向长任务开放 |
| `server/routes/websocket.py` | 实时消息类型参考；传输改为 SSE |

### 15.2 不复用

| 路径 | 原因 |
|---|---|
| `runtime/node/executor/python_executor.py` | 直接宿主机 subprocess，无法满足本项目沙箱边界 |
| `server/services/websocket_executor.py` | 本项目统一 SSE，避免并存两套重连协议 |
| `server/services/websocket_manager.py` | 同上 |
| `server/services/workflow_storage.py` | 使用 ChatDev 自身存储模型，会与 DeerFlow persistence 重叠 |
| `runtime/node/agent/skills/manager.py` | 只说明技能加载，不等于通用可调用技能节点 |
| ChatDev 模型 provider、消息对象、Agent runtime | 与 DeerFlow middleware/MCP/skill/memory 重复 |

## 16. 建议新增和修改的后端代码路径

以下均相对于 `offline-backend-20260512/backend/`。路径是开发拆分基线；实现时可在不破坏分层的前提下微调文件名。

### 16.1 harness：通用工作流内核

```text
packages/harness/deerflow/workflows/
├── __init__.py
├── schemas.py                    # 图、节点、边、NodeResult、运行参数 DTO
├── state.py                      # WorkflowState 与确定性 reducer
├── errors.py                     # 稳定错误码和异常类型
├── expressions.py                # 安全变量引用与模板求值
├── validator.py                  # 发布前结构/引用/资源限制校验
├── registry.py                   # NodeRegistry 与 executor factory
├── compiler.py                   # 标准图 → LangGraph StateGraph
├── events.py                     # 事件类型、envelope、EventSink 协议
├── context.py                    # RunContext、取消令牌、依赖接口
├── nodes/
│   ├── __init__.py
│   ├── base.py                   # NodeExecutor 协议
│   ├── start.py
│   ├── output.py
│   ├── agent.py
│   ├── skill.py
│   ├── http.py
│   ├── sql_read.py
│   ├── code.py
│   ├── transform.py
│   ├── condition.py
│   ├── merge.py
│   ├── human_input.py
│   └── subworkflow.py
├── security/
│   ├── __init__.py
│   ├── http_policy.py            # SSRF、域名、重定向、响应大小
│   ├── sql_policy.py             # AST、只读规则、表白名单
│   └── redaction.py              # 事件/日志脱敏
└── execution/
    ├── __init__.py
    ├── service.py                # 单 run 执行用例
    ├── checkpoint.py             # interrupt/resume checkpoint 适配
    └── limits.py                 # 超时、步骤、循环、预算
```

### 16.2 persistence：定义、运行、事件和数据源

```text
packages/harness/deerflow/persistence/workflows/
├── __init__.py
├── model.py                      # definition/version ORM
├── base.py                       # repository 协议
└── sql.py                        # SQLAlchemy 实现

packages/harness/deerflow/persistence/workflow_runs/
├── __init__.py
├── model.py                      # run/node_run ORM
├── base.py
└── sql.py                        # claim/lease/CAS/idempotency

packages/harness/deerflow/persistence/workflow_events/
├── __init__.py
├── model.py
├── base.py
└── sql.py                        # append/list/max_seq/批量写入

packages/harness/deerflow/persistence/workflow_data_sources/
├── __init__.py
├── model.py
├── base.py
└── sql.py

packages/harness/deerflow/persistence/workflow_embed/
├── __init__.py
├── model.py
├── base.py
└── sql.py
```

需要修改：

- `packages/harness/deerflow/persistence/models.py`：汇总新 ORM model，确保 `create_all` 和迁移发现。
- `packages/harness/deerflow/persistence/types.py`：仅在现有 portable 类型不足时扩展，禁止模型直接使用仅某一数据库支持的类型。
- `alembic/versions/<revision>_add_workflow_studio_tables.py`：正式迁移。
- 仓库现有 MySQL 增量脚本目录：按当前约定补齐等价 DDL，不能只依赖 `create_all`。

### 16.3 app：FastAPI、后台任务和资源适配

```text
app/gateway/routers/workflows.py                 # 定义/草稿/发布/版本
app/gateway/routers/workflow_runs.py             # start/get/events/SSE/cancel/resume
app/gateway/routers/workflow_resources.py        # agents/skills/subworkflows/node-types
app/gateway/routers/workflow_data_sources.py     # 数据源管理/测试/元数据
app/gateway/routers/workflow_embed.py            # iframe ticket/exchange
app/gateway/routers/workflows_coze_compat.py      # 最小 Coze DTO 兼容层

app/gateway/workflow_executor.py                 # app 依赖注入后调用 harness runtime
app/gateway/workflow_dispatcher.py               # scan/claim/lease/wakeup/shutdown
app/gateway/workflow_event_hub.py                # 单进程实时 fan-out；DB 仍为事实来源
app/gateway/workflow_dependencies.py             # Agent/Sandbox/HTTP/SQL adapter 组合
```

需要修改：

- `app/gateway/deps.py`：创建 repositories、event hub、executor、dispatcher，放入 `app.state`。
- `app/gateway/app.py`：注册 routers，生命周期中启动/停止 dispatcher。
- `config.yaml`：增加工作流运行上限、SSE、数据源和 iframe origin 配置；密钥继续使用环境变量/凭据引用。
- `packages/harness/deerflow/config/`：如需强类型配置，在此增加 `workflow_config.py`，不要让 harness 读取 `app.state`。

### 16.4 测试文件

```text
tests/test_workflow_schemas.py
tests/test_workflow_validator.py
tests/test_workflow_expressions.py
tests/test_workflow_compiler.py
tests/test_workflow_runtime.py
tests/test_workflow_parallel_merge.py
tests/test_workflow_loop_limits.py
tests/test_workflow_agent_node.py
tests/test_workflow_skill_node.py
tests/test_workflow_http_policy.py
tests/test_workflow_sql_policy.py
tests/test_workflow_code_sandbox.py
tests/test_workflow_run_repository.py
tests/test_workflow_event_repository.py
tests/test_workflow_stream.py
tests/test_workflow_stream_replay.py
tests/test_workflow_cancel_resume.py
tests/test_workflow_dispatcher.py
tests/test_workflow_embed_auth.py
tests/test_workflow_coze_compat.py
tests/test_workflow_migrations.py
```

`tests/test_harness_boundary.py` 必须继续通过。

## 17. 分阶段开发计划

### 阶段 0：协议冻结与样例图（2–3 人日）

- 冻结内部 graph schema、NodeResult、错误码和 SSE envelope v1。
- 建立“专题报告生成”标准样例图及每类节点最小样例。
- 确认 MySQL/PostgreSQL 首期支持范围、凭据存储方式和 iframe 部署域名。
- 用 ADR 记录“Coze 只负责画布、ChatDev 只参考、LangGraph 为执行内核”。

交付：schema 文档、事件表、样例 JSON、接口 OpenAPI 草案。

### 阶段 1：定义、版本与 Coze 保存兼容（5–8 人日）

- 新建 workflow definition/version repository 和迁移。
- 实现草稿 revision 乐观锁、validate、publish。
- 实现节点目录和 Coze canvas/save/node_template 最小适配。
- 前端可以真实加载、编辑、保存、发布工作流，但尚不执行。

### 阶段 2：运行骨架与持久 SSE（7–10 人日）

- 新建 run/node_run/event 表。
- 实现 run start/get/list/cancel 和 idempotency。
- 建立 PersistingEventSink、event hub 和 SSE replay/tail。
- 实现 dispatcher 租约、重启恢复和空图/start→output 运行。
- 先完成事件顺序、断线重连、终态一致性测试。

### 阶段 3：基础数据处理节点（7–10 人日）

- start/output/transform/condition/merge。
- HTTP 节点及 SSRF/大小/超时策略。
- SQL 数据源管理、元数据和只读查询节点。
- code 节点接 DeerFlow sandbox。
- 完成并行、错误分支和 artifact 引用。

### 阶段 4：智能体、技能和流式工具步骤（7–12 人日）

- AgentNode 接定制 DeerFlow Agent。
- `agent_skill` 模式。
- 模型 delta、工具 started/finished、artifact 事件映射。
- JSON 输出 schema、超时、取消和有限重试。
- 以“采集→整理→撰写→审核”跑通端到端。

### 阶段 5：人工介入、子工作流和受控循环（5–8 人日）

- awaiting_input/resume 与一次性 resume token。
- 子工作流版本绑定和层级事件。
- 显式循环、max iteration/steps/time/budget。
- 审核不通过退回撰写节点的业务用例。

### 阶段 6：iframe 安全与生产加固（5–8 人日）

- ticket/exchange、strict origin、CSP `frame-ancestors`。
- 权限、审计、限流、配额、凭据脱敏。
- 多 Gateway 实例、租约竞争、数据库短暂失败、SSE 大量重连压测。
- 事件保留、清理和 artifact 生命周期。

阶段 6 已落地：`WorkflowRetentionCleaner` 消费 `workflows.retention.*`；`POST /api/workflows/runs/{run_id}/retry`；SSE `Last-Event-ID`；单机租约竞争与 SSE 重连风暴测试。制品文件仍走 sandbox/artifact，不另做工作流专属文件回收。

### 阶段 7：试点与扩展（持续）

- 选择 1–2 个报告业务真实试点。
- 根据执行数据增加 callable skill manifest、更多数据库驱动或专用业务节点。
- 暂不满足稳定性前，不开放任意用户自建外网 HTTP 和数据库数据源。

## 18. 端到端报告工作流验收样例

建议以如下图作为首个验收基线：

```text
开始(topic, taskId, dateRange)
  ├─ SQL：查询内部结构化数据
  ├─ HTTP：调用只读业务接口
  └─ Agent：使用检索技能搜集外部材料
          ↓
Merge：合并材料并保留来源
          ↓
Agent：数据整理与事实去重（JSON Schema 输出）
          ↓
Agent：撰写 Markdown 报告
          ↓
Agent：审核，输出 {passed, issues, revisedAdvice}
          ↓
Condition
  ├─ passed=true  → Output + Word/Markdown artifact
  └─ passed=false → Writer（最多回退 2 次）
```

验收必须覆盖：

- 用户点击运行后 1 秒内看到 `run.created/queued` 或明确排队状态。
- 节点开始、模型文字、工具步骤和 artifact 实时可见。
- 刷新 iframe 后能从数据库恢复完整时间线并继续追尾。
- 网络中断后无重复文字、无丢步骤、无跨 run 串流。
- SQL 写语句、多语句和越权表被发布校验或运行策略拒绝。
- HTTP 内网探测和云元数据地址被拒绝。
- 代码超时后沙箱任务终止，Gateway 不受影响。
- 审核退回次数达到上限时明确失败，不形成无限循环。
- 取消后不再启动新节点。
- 最终输出符合 JSON Schema，报告 artifact 可下载。

## 19. Definition of Done

一个阶段只有同时满足以下条件才算完成：

- API、数据库迁移、配置和权限实现齐全。
- 单元、repository、router、流重放和端到端测试通过。
- 多实例/重启场景不会重复提交终态或丢事件。
- 所有用户可见事件均可重放且不包含密钥和隐藏思维链。
- 新代码通过 Ruff、类型检查和 `test_harness_boundary.py`。
- 更新后端 `README.md`、`CLAUDE.md`/相关架构说明和 OpenAPI 示例。
- 前后端基于同一个事件 schema fixture 做契约测试。

## 20. 实施时优先顺序

建议严格按下面顺序推进：

1. 先冻结标准图和事件协议。
2. 再做持久化、运行状态机和断线重放。
3. 先接 start/output/transform，确认基础链路可靠。
4. 再加 SQL、HTTP、Sandbox 等高风险节点。
5. 最后接 Agent/Skill、人工介入、循环和复杂分支。

这样 Coze 画布、DeerFlow 后端和 iframe 时间线不会各自形成一套难以兼容的临时协议。