工作流 SSE / 事件协议 v1.0（冻结）

实现源：packages/harness/deerflow/workflows/events.py
错误体：packages/harness/deerflow/workflows/errors.py
事件清单与开发方案对齐：WORKFLOW_STUDIO_BACKEND_DEV_ZH.md §9

不变式

先持久化，后发布 — 用户可见事件先写入 workflow_run_events，提交成功后再 fan-out。
seq 严格单调 — 唯一键 (run_id, seq)。
可重放 — Last-Event-ID 优先于 after_seq，否则从 0。
增量合并 — 模型 token 建议 30–100ms 或约 1KB 合并落库。
终态停连 — run.completed / run.failed / run.cancelled 且库已补齐后关闭 SSE。
不泄密 — 无凭据、无完整认证头、无 DB 连接串、无隐藏思维链。

端点（阶段 2 已实现）

GET /api/workflows/runs/{run_id}/stream?after=128
Accept: text/event-stream
Last-Event-ID: 128

实现：app/gateway/routers/workflow_runs.py。

库是排序的权威来源：连接先按 Last-Event-ID（优先）或 after 从 workflow_run_events 回放，再挂 WorkflowLiveHub 订阅实时帧；seq 不大于已回放游标的实时帧被丢弃，下一轮 DB 轮询（0.8s）会带上它。断线重连因此不会看到空洞或乱序。
终态必闭：读到 run.completed / run.failed / run.cancelled 即关闭。若运行已是终态却没有终态事件（例如崩在写事件之前），补发一条 synthesised: true 的终态帧再关闭，客户端不会永久挂住。
响应头 X-Workflow-Run-Status 给出连接建立瞬间的状态。
Heartbeat：SSE comment : heartbeat（不写库），15s 静默后发送。
同一份事件也可用 GET /api/workflows/runs/{run_id}/events?after=&limit= 分页拉取（无长连接场景）。

Envelope

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

SSE 帧：

id: 42
event: node.output.delta
data: { ... envelope ... }

Python：WorkflowEventEnvelope.to_sse_frame()。

事件类型（封闭集合）

事件	`data` 要点
`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

统一错误体

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

完整错误码列表：ALL_WORKFLOW_ERROR_CODES（deerflow.workflows.errors）。

`resumeToken` 只走事件流

run.awaiting_input 的 data.resumeToken 是恢复运行的唯一凭证，只在事件流里出现——GET /api/workflows/runs/{run_id} 的运行视图不包含它。恢复时 POST /api/workflows/runs/{run_id}/resume 携带该 token，服务端以「消费即失效」的方式原子翻转状态（同一个 token 不能恢复两次）。

前后端契约测试

前后端共用 fixture：

workflows/samples/event_envelope_example.json
本文件事件名表

实现测试：tests/test_workflow_runs.py（回放游标、终态关闭、暂停/恢复的事件序列）。

4.0 KiB Raw Permalink Blame History Unescape Escape