From 6e7ec56f9a2bf8dc017397f1e393b84250eedf9c Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 11:26:35 +0800 Subject: [PATCH] =?UTF-8?q?feat(loop,gateway):=20=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E6=89=B9=20=E2=80=94=20tool=5Fexecution=20=E4=B8=89=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=20+=20streamId=20+=20turn.start/end=EF=BC=88=E8=A3=81?= =?UTF-8?q?=E5=86=B3=2001.4.1/01.4.2/01.4.3/01.4.5/03.4.4/1.12=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 执行方案阶段 2 · P2-3（`DevDocs/audit/_plan.md`）。契约 `Docs/api-client.md` → **v0.50**。 - **01.4.1 `tool_execution` 拆三事件**：`model/stream.go` 删旧 `StreamKindToolExecution`，新增 `tool_execution.start/update/end` + `ToolPhaseRunning/Done/Error`（与 design/19 status 同值）； `StreamEvent` 增 `Tool`/`Phase`；`loop/tools.go` 的 `emitToolExecution` 重写为结构化字段、 **Delta 恒空**（断言锁死），`ID` 复用 provider tool call id（可与 `tool_call.*` 关联）。 ⚠️ **诚实边界**：start/end 有真实生产者；`update` 的 wire 映射已实现并单测锁定，但**工具分发器 无中途进度上报源 ⇒ loop 暂不产出 update 帧**（已在 api-client §3.1 如实登记）。 - **03.4.4 phase 驱动图节点状态**：`invokeOne` 用**单一 phase 决策**同时决定 end 帧 phase、 span 的 `phase` prop 与终态；`trace/graph.go` 新增 `FlowStatusForPhase`，`graph_build.go` 的 `AddSpan` 对 tool 节点**优先消费 phase prop**（缺省回落 span 终态 ⇒ 两路不可能矛盾）。 - **01.4.2 `streamId`**：`StreamEvent` 增 `StreamID`；`loop/stream.go` 每次 ChatStream 生成 `stream-`（crypto/rand，降级时间戳），该流所有帧共享、多步工具轮取新 id；`ws_bridge` 把它并进**模型流帧** `data`（`tool_execution.*`/`turn.*` 不带）。语义区分：`turnId` = 哪轮回答， `streamId` = 这条流式消息。 - **01.4.3 `turn.start`/`turn.end`**：`loop/run.go` 入口发 start、defer 发 end，**覆盖全部出口** （最终答案/安全网终止/预算耗尽/步骤错误）；帧**无 data 载荷**；静音的内部 run（backthink/子代理） 不发。顺带把 91 行的 `Run` 拆成 `driveTurn` + `settleAndReturn`（清掉一处既有 >80 行违规）。 - **01.4.5**：仅 `Docs/api-client.md` §3.5 补 `graph_update` 三范围（`web`/`web:`/`sess-backthink`）， 与 `wire_runtime.go` 的 allow 谓词一致，**零代码变更**。 验证：`go build ./...` 绿；本批四包（loop/model/trace/gateway）vet + 测试绿（含 `-count=1`）； linux/amd64 + darwin/arm64 交叉编译绿；本批 16 文件 ≤500 行、函数 ≤80 行。 **真 WS 帧断言 8 条**：帧序 `turn.start → message.delta → tool_execution.start → tool_execution.end → message.delta → turn.end`（+`stream_end`）、turn 边界无载荷且整轮 turnId 一致、 两条模型流 `streamId` 互异、tool 帧 `{tool,phase,id}` 且无 `delta`/`streamId` 键、 `turn.end` 在步骤出错时也发出、内部静音 run 不发边界帧。 **变异验证**：把 `ev.StreamID = streamID` 改为赋空 → 两条断言 FAIL（`streamId` 为空/两流同空）， 还原后复绿（直接删行会先被编译期 `declared and not used` 拦截）。 未验证：前端未消费这些帧（本轮未动 `frontend/**`，「前端能渲染」待前端批）；无真实模型端到端。 --- Docs/api-client.md | 26 +- internal/runtime/loop/loop_extra_test.go | 4 +- internal/runtime/loop/loop_test.go | 11 +- internal/runtime/loop/run.go | 37 ++- internal/runtime/loop/stream.go | 37 +++ internal/runtime/loop/stream_events_test.go | 275 +++++++++++++++++ internal/runtime/loop/stream_scope_test.go | 20 +- internal/runtime/loop/thinking_gate_test.go | 8 +- internal/runtime/loop/tools.go | 163 +++++----- internal/runtime/model/stream.go | 57 +++- internal/runtime/trace/graph.go | 18 ++ internal/runtime/trace/graph_build.go | 18 +- internal/runtime/trace/graph_phase_test.go | 80 +++++ internal/services/gateway/ws_bridge.go | 60 +++- .../services/gateway/ws_frame_scope_test.go | 6 +- .../services/gateway/ws_stream_events_test.go | 288 ++++++++++++++++++ 16 files changed, 982 insertions(+), 126 deletions(-) create mode 100644 internal/runtime/loop/stream_events_test.go create mode 100644 internal/runtime/trace/graph_phase_test.go create mode 100644 internal/services/gateway/ws_stream_events_test.go diff --git a/Docs/api-client.md b/Docs/api-client.md index 2168913..c18bbff 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -110,7 +110,7 @@ - Web 端镜像实现在 `frontend/web/src/lib/enums.ts`（`labelFor(kind, value)` 对未知值返回原值），权威仍以本端点为准。 - **反漂移保证（后端）**：后端有测试断言「每个内置适配器的 `Name()` 都在 `platforms` 表内」「`moods` 与 persona 情绪状态机的内置状态一致」「`modes` 与前端 `MODES` 九模式一致」——漏登记即 CI FAIL，客户端拿到的表不会滞后于代码。 -> **契约版本 `v0.49`**（**WS 握手回放与上行帧的角色门（design/11 §7.2）**：补上 §7.1 下行门之外的**同一类缺口的另两处**，对 `viewer` **继续收紧**，`owner` / `admin` 行为不变，**无端点、无字段形状变更**——① **握手 `history` 回放改为仅 `owner` + `admin`**：此前任意角色一连上 `/ws/chat` 就会收到主会话最近 50 条消息正文（`thinking` 已剥离，但**正文本身**就是主人的对话），`viewer` 从此**不收**；② **上行 JSON 帧按角色授权（fail-closed）**：非 `owner`/`admin` 连接发来的**任何**上行 JSON 帧一律拒绝，**定向**（只回发送者）回 `{type:"error",data:{code:"forbidden"}}`，**不广播、不驱动模型、不落库、不返回任何内容**——被拒帧含 `chat`（此前 `viewer` 可**真的驱动模型并落库**）、`interrupt`、`subagent_ask` / `task_output_query`（其应答是**定向**帧、能绕过 §7.1 下行门把子代理上下文/任务输出直接回给 `viewer`）、`voice_input`、`mode_switch`，且**未登记的新上行帧默认拒**；**协议级帧（ping/pong、close）不受影响**。③ **合口径**：`viewer` = **可连接、可读 12 条只读监控 REST（v0.40）、可收 7 个状态机信号帧（v0.48）**；**除此以外 WS 上拿不到任何内容、也发不出任何指令**。见 §3 开头的「上行帧与握手回放的角色门」与 v0.49 变更记录。**上一版 `v0.48`**：（**WS 下行帧的角色门（design/11 §7.1）**：hub 的下行现在是**按帧类型分档**的——只有 7 个**状态机信号**（`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full` / `session_read`，**均不含任何内容**）继续发给**所有已认证连接**（含 `viewer`），**其余全部下行帧**（流式内容 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` / `stream_chunk` / `stream_end` / `thinking`；持久化 `message_persisted` / `message_edited` / `message_deleted` / `channel_persisted`；系统内部 `graph_update` / `subagent_event` / `task_output` / `task_done` / `tool_used` / `capacity_awaken` / `mode_switch`；私有状态 `mood` / `device_update` / `reminder`）**改为仅发 `owner` + `admin`**）。**对 `viewer` 是收紧**：此前它连上 `/ws/chat` 就能收到角色回复正文与推理内容，现在收不到；**`owner` / `admin` 行为不变**；**无端点、无字段形状变更**——`viewer` 的 12 条只读监控 REST 端点（`viewerAuth`，v0.40）与 `/ws/chat` 的连接鉴权都**照旧**；**新帧类型默认只发 owner/admin**（fail-closed）。见 §3 开头的「下行帧的角色门」与 v0.48 变更记录。**上一版 `v0.47`**：（**`graph_update` 子代理泳道根节点新增可选 `props.persona`**：派发子代理时若指定了目标角色，该分支根节点的 `props` 就多一个 `persona` 键，标注它**正在扮演的角色**——用途是让流程图能显示「这个子代理是以谁的身份在替班」，而不只是子代理 id；**纯增量**：不带角色派发时该键不出现，老客户端忽略未知键即可，见 §3.5 与 v0.47 变更记录）。**上一版 `v0.46`**：（**`PATCH /api/v1/sessions/{id}` 的 `personaId` 由「可重新绑定该会话的角色」改为「出现即 `400`」**：**会话角色在创建时确定、之后不可变**，要换角色请新开会话（design/22 §8-1）——**明确拒绝**而非静默忽略/200 空操作；**键出现**（任何取值：`"yuki"` / `""` / `null` / 未知 id / 空白串）一律 `400 bad_request` 且消息点名 `personaId`，该检查**先于**会话存在性检查（未知会话 + `personaId` 也回 `400` 而不是 `404`）。**破坏性**：按 v0.42–v0.45 旧文实现、依赖「PATCH 换角色 / 显式 `""` 解绑」的客户端必须改为**新建会话**选角色；`title`/`pinned` 行为与 `POST /sessions {personaId}`（创建时绑定）**完全不变**，不再有 `501/200 {personaId}` 的 PATCH 路径）。上一版 `v0.45`：`PUT /admin/channels/{platform}/config` 的 `personaId` 由「省略/`""` 都 = 解绑」修订为三态（**省略（或 `null`）= 保留已存绑定**、**显式 `""` = 解绑**、**非空 id = 绑定**，未知角色仍 `400`）。 +> **契约版本 `v0.50`**（**事件批：流式帧的结构化补齐**，依据 design/04 §1.1，裁决 01.4.1 / 01.4.2 / 01.4.3 / 03.4.4，2026-10-04 落实）：① **`tool_execution` 拆三事件**——旧单帧 `tool_execution.update {id,index,delta}`（phase 与工具名拼在 `delta` 字符串里）被 **`tool_execution.start`**（进入执行，`phase:"running"`）/ **`tool_execution.update`**（运行中进度/局部输出，可多帧）/ **`tool_execution.end`**（完成或失败，`phase:"done"|"error"`）替换，载荷改为结构化 `{tool, phase, id?}`（`tool` = 工具名；`phase` = 细粒度执行阶段，与 §3.5 流程图节点 status 同一套 `running/done/error` 枚举；`id` = provider 工具调用 id，可与 `tool_call.*` 帧关联）——**phase 不再塞进 `delta`**；loop 现已真实发出 start/end（update 的 wire 映射已实现并验证，但工具分发器尚无中途进度上报源，**暂无生产者**，见 §3.1）。② **`streamId`（模型流帧 `data` 内新增可选字段）**：同一次模型流（一次 ChatStream 调用）的多帧共享同一 `streamId`，多步工具轮的第二次模型调用取新 id；与帧壳的 `turnId`（整个回合恒定）分工——`turnId` 标记「这轮回答」，`streamId` 标记「这轮里的哪一条流式消息」。③ **`turn.start` / `turn.end`（新帧）**：loop 在回合边界发出、经 ws_bridge 下行；**无 `data` 载荷**（帧壳的 `sessionId`/`turnId`/`seq` 即边界信号；回合终态信息仍由既有 `stream_end` 携带）；`turn.end` 覆盖回合的所有出口（最终答案 / 工具循环安全网终止 / 预算耗尽 / 步骤出错），被静音的内部流程（backthink / 子代理）不发出。④ **向后兼容**：`turn.start/end` 与 `streamId` 为纯新增，老客户端可安全忽略；**破坏性**在旧 `tool_execution.update` 单帧——loop 不再产生它，按旧形状（解析 `delta` 里的 `phase:工具名` 拼串）消费的客户端必须迁移到三事件；现有 Web 前端未消费这些帧，行为不变。⑤ 本批新帧全部落在 v0.48 下行帧门的「仅 `owner` + `admin`」档（fail-closed 默认，无需扩白名单）。⑥ **`graph_update` 会话范围口径修正**（裁决 01.4.5，随实现事实补记）：`web` / `web:` / `sess-backthink` 三范围（见 §3.5）。见 §3.1 与 v0.50 变更记录。**上一版 `v0.49`**：（**WS 握手回放与上行帧的角色门（design/11 §7.2）**：补上 §7.1 下行门之外的**同一类缺口的另两处**，对 `viewer` **继续收紧**，`owner` / `admin` 行为不变，**无端点、无字段形状变更**——① **握手 `history` 回放改为仅 `owner` + `admin`**：此前任意角色一连上 `/ws/chat` 就会收到主会话最近 50 条消息正文（`thinking` 已剥离，但**正文本身**就是主人的对话），`viewer` 从此**不收**；② **上行 JSON 帧按角色授权（fail-closed）**：非 `owner`/`admin` 连接发来的**任何**上行 JSON 帧一律拒绝，**定向**（只回发送者）回 `{type:"error",data:{code:"forbidden"}}`，**不广播、不驱动模型、不落库、不返回任何内容**——被拒帧含 `chat`（此前 `viewer` 可**真的驱动模型并落库**）、`interrupt`、`subagent_ask` / `task_output_query`（其应答是**定向**帧、能绕过 §7.1 下行门把子代理上下文/任务输出直接回给 `viewer`）、`voice_input`、`mode_switch`，且**未登记的新上行帧默认拒**；**协议级帧（ping/pong、close）不受影响**。③ **合口径**：`viewer` = **可连接、可读 12 条只读监控 REST（v0.40）、可收 7 个状态机信号帧（v0.48）**；**除此以外 WS 上拿不到任何内容、也发不出任何指令**。见 §3 开头的「上行帧与握手回放的角色门」与 v0.49 变更记录。**上一版 `v0.48`**：（**WS 下行帧的角色门（design/11 §7.1）**：hub 的下行现在是**按帧类型分档**的——只有 7 个**状态机信号**（`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full` / `session_read`，**均不含任何内容**）继续发给**所有已认证连接**（含 `viewer`），**其余全部下行帧**（流式内容 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` / `stream_chunk` / `stream_end` / `thinking`；持久化 `message_persisted` / `message_edited` / `message_deleted` / `channel_persisted`；系统内部 `graph_update` / `subagent_event` / `task_output` / `task_done` / `tool_used` / `capacity_awaken` / `mode_switch`；私有状态 `mood` / `device_update` / `reminder`）**改为仅发 `owner` + `admin`**）。**对 `viewer` 是收紧**：此前它连上 `/ws/chat` 就能收到角色回复正文与推理内容，现在收不到；**`owner` / `admin` 行为不变**；**无端点、无字段形状变更**——`viewer` 的 12 条只读监控 REST 端点（`viewerAuth`，v0.40）与 `/ws/chat` 的连接鉴权都**照旧**；**新帧类型默认只发 owner/admin**（fail-closed）。见 §3 开头的「下行帧的角色门」与 v0.48 变更记录。**上一版 `v0.47`**：（**`graph_update` 子代理泳道根节点新增可选 `props.persona`**：派发子代理时若指定了目标角色，该分支根节点的 `props` 就多一个 `persona` 键，标注它**正在扮演的角色**——用途是让流程图能显示「这个子代理是以谁的身份在替班」，而不只是子代理 id；**纯增量**：不带角色派发时该键不出现，老客户端忽略未知键即可，见 §3.5 与 v0.47 变更记录）。**上一版 `v0.46`**：（**`PATCH /api/v1/sessions/{id}` 的 `personaId` 由「可重新绑定该会话的角色」改为「出现即 `400`」**：**会话角色在创建时确定、之后不可变**，要换角色请新开会话（design/22 §8-1）——**明确拒绝**而非静默忽略/200 空操作；**键出现**（任何取值：`"yuki"` / `""` / `null` / 未知 id / 空白串）一律 `400 bad_request` 且消息点名 `personaId`，该检查**先于**会话存在性检查（未知会话 + `personaId` 也回 `400` 而不是 `404`）。**破坏性**：按 v0.42–v0.45 旧文实现、依赖「PATCH 换角色 / 显式 `""` 解绑」的客户端必须改为**新建会话**选角色；`title`/`pinned` 行为与 `POST /sessions {personaId}`（创建时绑定）**完全不变**，不再有 `501/200 {personaId}` 的 PATCH 路径）。上一版 `v0.45`：`PUT /admin/channels/{platform}/config` 的 `personaId` 由「省略/`""` 都 = 解绑」修订为三态（**省略（或 `null`）= 保留已存绑定**、**显式 `""` = 解绑**、**非空 id = 绑定**，未知角色仍 `400`）。 ## 2. REST 端点（管理，`/api/v1/admin/*`） @@ -243,20 +243,25 @@ ## 3. WebSocket `/ws/chat`（实时） - **握手**: `ws://:8080/ws/chat?token=` 或 `Authorization: Bearer`；未认证在握手前回 `401` JSON。 -- **下行（服务端→客户端）**: 经 bus 广播 `{type:"", data:..., turnId?, seq?, sessionId?}`（`turnId`/`seq` 为流式关联字段，见 §3.1；`sessionId` 为**会话持久化（design/16）新增**，能确定会话的事件带上，供客户端按激活会话过滤；**v0.36 起结构化流帧 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` 也带 `sessionId` + `turnId`**——多会话并发流式时客户端据此把 delta 路由到激活会话，见 §3.1）——`message.start/delta/end` / `thinking.start/delta/end` / `tool_call.start/delta/end` / `tool_execution.update`（结构化流，design/04 §1.1），以及向后兼容的 `stream_chunk` / `stream_end` / `thinking`(思维链) / `mood`(情绪变化，见 §3.7) / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update` / **`message_persisted`**（会话持久化增量，见 §3.3）/ **`message_edited`**、**`message_deleted`**（消息编辑/删除增量，见 §3.8）/ **`session_read`**（已读游标推进，见 §3.9）。**会话编排状态（design/15，均为广播，`data.sessionId` 区分会话）**：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（见 §3.2）。语音回复走 `voice_output`（定向回请求连接，非广播）。 -- **下行帧的角色门（v0.48，design/11 §7.1）**: 下行帧按**类型**分两类，由后端**唯一一处**收口判定（**不在白名单里的帧默认只发 `owner` + `admin`**，fail-closed）：① **状态机信号**（白名单，**不含任何内容**）→ **所有已认证连接**（含 `viewer`）：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（§3.2）、`session_read`（§3.9）；② **其余全部** → **仅 `owner` + `admin`**：流式内容 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` / `stream_chunk` / `stream_end` / `thinking`（§3.1）、持久化 `message_persisted`（§3.3）/ `channel_persisted`（§3.4）/ `message_edited` / `message_deleted`（§3.8）、系统内部 `graph_update`（§3.5）/ `subagent_event` / `task_output` / `task_done` / `tool_used` / `capacity_awaken` / `mode_switch`、私有状态 `mood`（§3.7）/ `device_update` / `reminder`。 +- **下行（服务端→客户端）**: 经 bus 广播 `{type:"", data:..., turnId?, seq?, sessionId?}`（`turnId`/`seq` 为流式关联字段，见 §3.1；`sessionId` 为**会话持久化（design/16）新增**，能确定会话的事件带上，供客户端按激活会话过滤；**v0.36 起结构化流帧 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.*` 也带 `sessionId` + `turnId`**——多会话并发流式时客户端据此把 delta 路由到激活会话，见 §3.1）——`message.start/delta/end` / `thinking.start/delta/end` / `tool_call.start/delta/end`（以上模型流帧的 `data` 另带 `streamId`，v0.50）/ `tool_execution.start/update/end`（工具执行三事件，v0.50）/ `turn.start` / `turn.end`（回合边界帧，v0.50）（结构化流，design/04 §1.1），以及向后兼容的 `stream_chunk` / `stream_end` / `thinking`(思维链) / `mood`(情绪变化，见 §3.7) / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update` / **`message_persisted`**（会话持久化增量，见 §3.3）/ **`message_edited`**、**`message_deleted`**（消息编辑/删除增量，见 §3.8）/ **`session_read`**（已读游标推进，见 §3.9）。**会话编排状态（design/15，均为广播，`data.sessionId` 区分会话）**：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（见 §3.2）。语音回复走 `voice_output`（定向回请求连接，非广播）。 +- **下行帧的角色门（v0.48，design/11 §7.1）**: 下行帧按**类型**分两类，由后端**唯一一处**收口判定（**不在白名单里的帧默认只发 `owner` + `admin`**，fail-closed）：① **状态机信号**（白名单，**不含任何内容**）→ **所有已认证连接**（含 `viewer`）：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（§3.2）、`session_read`（§3.9）；② **其余全部** → **仅 `owner` + `admin`**：流式内容 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.start/update/end` / `turn.start` / `turn.end`（**v0.50 新增的这几种同属本档**——工具进度与回合边界对 `viewer` 也是内容相邻信息，fail-closed 默认即只发 owner+admin，未扩白名单）/ `stream_chunk` / `stream_end` / `thinking`（§3.1）、持久化 `message_persisted`（§3.3）/ `channel_persisted`（§3.4）/ `message_edited` / `message_deleted`（§3.8）、系统内部 `graph_update`（§3.5）/ `subagent_event` / `task_output` / `task_done` / `tool_used` / `capacity_awaken` / `mode_switch`、私有状态 `mood`（§3.7）/ `device_update` / `reminder`。 - **上行帧与握手回放的角色门（v0.49，design/11 §7.2）**: 与上面那条下行门**成对**，收的是同一个缺口的两处残余——① **客户端→服务端（上行）JSON 帧**：只有 `owner` / `admin` 连接可以发；非 `owner`/`admin`（即 `viewer`）连接发来的**任何**上行 JSON 帧都**默认被拒**，回复是**定向**（只回发送者）的 `{type:"error", data:{code:"forbidden"}}`，**不广播、不驱动模型、不落库、不返回任何内容**。被拒帧：`chat` / `interrupt` / `subagent_ask` / `task_output_query` / `voice_input` / `mode_switch`——**未登记的新上行帧同样默认拒**，要放行必须先在 design/11 §7.2 登记白名单（实现上那把「默认拒」也必须同步改为登记表，不能直接删）。② **握手 `history` 回放**：仅 `owner` + `admin` 连接收到；`viewer` 连接建立后**不收**回放。**协议级帧不受影响**：WS ping/pong 与正常 close 照常（否则连接会掉），被拒的上行帧也不会关掉连接。 > **这是「谁被主动推送」的门，不是数据可见性**：`viewer` 的 12 条只读监控端点（`viewerAuth`，v0.40）**照旧**（含 `GET /personas` 的角色级 `mood`）；`viewer` **仍可连** `/ws/chat`（连接鉴权不变），只是收不到第 ② 类帧。定向回请求连接的帧（`voice_output` 等）不受影响。**上一条的门只管「推给谁」；「谁能发、谁能拿回放」见上面 v0.49 那条**——`viewer` 连上后**发不出任何指令**、也**收不到 `history` 回放**。 > **`history` 回放事件（design/16，新增）**：WS 握手升级成功后、该连接下一帧即收到 `{type:"history", sessionId:"web", data:{messages:[{id,sessionId,role,content,mode?,mood?,image?,platform?,deviceId?,personaId?,createdAt}]}}`——回放主会话最新 50 条（`thinking` 已剥离；默认主会话，次级会话由客户端按激活会话走 `GET /sessions/{id}/messages` 拉取）。客户端据此在重连/新开连接后补全上下文。消息体含 `senderName?`（第三方昵称，design/16 §4.6，主人私聊为空）与 `personaId?`（v0.42，说话时的角色，design/22 §7-5，空 = 默认角色）。**角色门（v0.49，design/11 §7.2 A）**：这条回放**只发 `owner` + `admin` 连接**；`viewer` 连接建立后**不会**收到 `history` 帧（它若需要只读数据，走 §1 的只读 REST 端点）。 -### 3.1 结构化流事件（`message.*` / `thinking.*` / `tool_call.*` / `tool_execution.*`） -- **事件壳**：`{type, sessionId?, turnId?, seq?, data}`, `data` 为**类型化结构**（非 `any`）。`seq` 为 hub 级单调递增序号，供客户端将一次突发的流式事件排序关联。**`sessionId` 为事件所属会话、`turnId` 为该轮助手回合 id（均 v0.36 起带上）**：`sessionId` 由 loop 在回合边界注入（模型适配器在 loop 之下、不感知会话，恒留零值；loop 外无会话上下文的事件该字段为空），`turnId` 由网关每个回合新生成并挂进运行上下文——同一会话同时只有一个 in-flight 回合，故**同轮内所有帧的 `turnId` 一致**。客户端据此把 `message.delta` / `thinking.delta` / `tool_call.*` / `tool_execution.update` 过滤到 `sessionId === activeSessionId` 的会话，忽略非激活会话的帧（多会话并发流式不再串话）。 -- **`message.start`**(`{index}`) / **`message.delta`**(`{delta, index}`) / **`message.end`**(`{index}`)：正文段。`delta` 为正文 token 增量；客户端对 `message.delta` 累加到当前消息正文；`message.start/end` 划定正文段边界，**不**结束整轮（结束由 `stream_end` 负责）。 -- **`thinking.start`**(`{index}`) / **`thinking.delta`**(`{delta, index}`) / **`thinking.end`**(`{index}`)：思维链段（前端折叠块）。`thinking.delta` 累加增量。 -- **`tool_call.start`**(`{id, index}`) / **`tool_call.delta`**(`{id, index, delta}`) / **`tool_call.end`**(`{id, index}`)：工具调用中间态；`tool_call.delta` 携带参数增量（如 Anthropic `input_json_delta`）。 -- **`tool_execution.update`**(`{id, index, delta}`)：工具执行进度/局部输出（保留，工具循环 §1.2 后置实现）。 +### 3.1 结构化流事件（`message.*` / `thinking.*` / `tool_call.*` / `tool_execution.*` / `turn.start/end`） +- **事件壳**：`{type, sessionId?, turnId?, seq?, data}`, `data` 为**类型化结构**（非 `any`）。`seq` 为 hub 级单调递增序号，供客户端将一次突发的流式事件排序关联。**`sessionId` 为事件所属会话、`turnId` 为该轮助手回合 id（均 v0.36 起带上）**：`sessionId` 由 loop 在回合边界注入（模型适配器在 loop 之下、不感知会话，恒留零值；loop 外无会话上下文的事件该字段为空），`turnId` 由网关每个回合新生成并挂进运行上下文——同一会话同时只有一个 in-flight 回合，故**同轮内所有帧的 `turnId` 一致**。客户端据此把 `message.delta` / `thinking.delta` / `tool_call.*` / `tool_execution.*` 过滤到 `sessionId === activeSessionId` 的会话，忽略非激活会话的帧（多会话并发流式不再串话）。 +- **`streamId`（v0.50，模型流帧 `data` 内可选字段）**：`message.*` / `thinking.*` / `tool_call.*` 帧的 `data` 带 `streamId`（如 `{"delta":"…","index":0,"streamId":"stream-a1b2…"}`）——**同一次模型流的多帧共享同一 id**，loop 在每次模型流（一次 ChatStream 调用）开始时生成；多步工具轮的第二次模型调用取**新** id。`turnId` 标记「这轮回答」（整轮恒定），`streamId` 标记「这轮里的哪一条流式消息」；一个回合含工具调用时会有多个 `streamId`。`tool_execution.*` 与 `turn.*` 帧不属于任何模型流，**不带** `streamId`。 +- **`message.start`**(`{index, streamId?}`) / **`message.delta`**(`{delta, index, streamId?}`) / **`message.end`**(`{index, streamId?}`)：正文段。`delta` 为正文 token 增量；客户端对 `message.delta` 累加到当前消息正文；`message.start/end` 划定正文段边界，**不**结束整轮（结束由 `stream_end` 负责）。 +- **`thinking.start`**(`{index, streamId?}`) / **`thinking.delta`**(`{delta, index, streamId?}`) / **`thinking.end`**(`{index, streamId?}`)：思维链段（前端折叠块）。`thinking.delta` 累加增量。 +- **`tool_call.start`**(`{id, index, streamId?}`) / **`tool_call.delta`**(`{id, index, delta, streamId?}`) / **`tool_call.end`**(`{id, index, streamId?}`)：工具调用中间态；`tool_call.delta` 携带参数增量（如 Anthropic `input_json_delta`）。 +- **`tool_execution.start`**(`{tool, phase:"running", id?}`)：工具**进入执行**（v0.50）。`tool` 为工具名；`phase` 为细粒度执行阶段（枚举与 §3.5 流程图节点 status 同一套 `running/done/error`）；`id` 为 provider 工具调用 id（可关联同轮的 `tool_call.*` 帧，provider 未给 id 时省略）。 +- **`tool_execution.update`**(`{tool, phase:"running", delta?, id?}`)：工具**运行中**的进度/局部输出，一次执行可多帧（v0.50）。**现状**：wire 映射已实现并验证，但工具分发器尚无中途进度上报源，loop 目前不产出该帧（start/end 有真实生产者）——出现生产者后无需再改契约。 +- **`tool_execution.end`**(`{tool, phase:"done"|"error", id?}`)：工具执行**完成或失败**（v0.50）；`phase:"error"` 即失败（design/19 流程图的 tool 节点据此翻 `error` 态；错误详情不进该帧，走流程图节点日志的脱敏摘要）。 +- **`turn.start` / `turn.end`**（无 `data` 载荷）：回合边界帧（v0.50），loop 在 Run 的起止发出、经 ws_bridge 下行；帧壳的 `sessionId`/`turnId`/`seq` 即边界信号，客户端用它括起「一轮回答」的完整区间（含其全部模型流与工具执行）。`turn.end` 覆盖回合所有出口（最终答案 / 工具循环安全网终止 / 预算耗尽 / 步骤出错）；**回合终态信息（`stopReason`/`steps`）仍由既有 `stream_end` 携带**，两帧分工不重叠。内部自主流程（backthink / 子代理）被静音，不发出这两帧。 +- **向后兼容（v0.50 对 `tool_execution` 的替换说明）**：v0.49 及之前的单帧 `tool_execution.update {id,index,delta}`（`delta` 形如 `"start:工具名"` / `"end:error:工具名"` 的拼串）**不再由 loop 产生**——按旧拼串解析的客户端必须迁移到三事件（`tool_call.*` 帧形状不变）。`turn.start/end` 与各帧 `data` 里的 `streamId` 为纯新增键，老客户端按未知帧/未知键忽略即可。现有 Web 前端未消费这些新帧，行为不变。 - **向后兼容**：未映射/无法确定的事件仍走 `stream_chunk` + `stream_end`（非流式回退路径）。 - **上行（客户端→服务端）**: `mode_switch` / `chat`（跑 loop 并广播流式） / `voice_input` / `subagent_ask` / `task_output_query` / `interrupt`。 - **`interrupt`**（上行，客户端→服务端）：`{sessionId}` → 会话编排器中断（design/15 §3）——取消该会话当前在跑的任务（context cancel）、清空其待处理队列、回 `session_interrupted`；用户说「停下/算了/换个话题」时发。命中空闲会话为 no-op。 @@ -293,7 +298,7 @@ ### 3.5 实时全链路流程事件（`graph_update`，design/19） > **角色门（v0.48，design/11 §7.1）**：`graph_update` 属「系统内部」帧 → **仅 `owner` + `admin`**（`viewer` 收不到，与本节「仅主人 Web 会话发布」及 design/19 §1「flow 可视化仅 owner/admin」一致）。REST 侧可见性不变。 -- **触发**：主会话 Web 每次会话轮次中，loop 的每个 trace span **开始/结束时**由 `loop.FlowPublisher` 经 bus `ws.graph_update` 广播增量 **FlowGraph**。**仅主人 Web 会话**（`sessionId == "web"` 或其 `web:` 前缀）发布——IM / 群聊 / 陌生人 / 子代理轮次**不**发布，防敏感信息泄向面板。管理员/主人 Web 端订阅后实时渲染「输入 → 思考(LLM) → 工具 → 输出」主干图。 +- **触发**：每次会话轮次中，loop 的每个 trace span **开始/结束时**由 `loop.FlowPublisher` 经 bus `ws.graph_update` 广播增量 **FlowGraph**。**发布范围为三类会话（v0.50 口径，裁决 01.4.5）**：`web`（Web 主会话） / `web:`（Web 子会话，含 IM 主人私聊镜像落主会话的形态） / `sess-backthink`（后台思考的固定内部会话，design/20 §2 的「思考链路图」由此取数）——**其余隔离会话（IM 群聊/陌生人的 `sess-xxx`）与子代理轮次不发布**，防敏感信息泄向面板。管理员/主人 Web 端订阅后实时渲染「输入 → 思考(LLM) → 工具 → 输出」主干图。**tool 节点的 `status` 由细粒度 phase 驱动**（v0.50，design/04 §1.1 + design/19 §1）：工具执行期间为 `running`，执行结束按成败翻 `done` / `error`——与 §3.1 `tool_execution.*` 帧的 `phase` 同一套枚举、同一处决策。 - **载荷**：`{type:"graph_update", data:{id, nodes:[{id,label,kind?,status?,startedAt?,endedAt?,props?,log?}], edges:[{from,to}], sessionId, startedAt, endedAt}}`。 - **子代理分支节点的 `props.persona`（v0.47，design/22 §8-2）**：`subagent_spawn` 指定了目标角色派发时，该子代理在父会话图里的**分支根节点** `props` 会多一个 `persona` 键（值 = 目标角色 id），标注它**正在扮演的角色**；**不指定角色派发时该键不出现**（与旧版逐字一致），客户端按未知键忽略即可。 - `id` 为 `sessionId-round`（每轮一图）；`sessionId` 为会话 id；`startedAt/endedAt` 为 RFC3339。 @@ -378,6 +383,7 @@ - 语音：上行 `voice_input`（base64 音频或已转写 text），下行 `voice_output`（合成音频 base64 + text）。 ## 5. 变更记录 +- `v0.50`（**事件批：流式帧的结构化补齐**——`tool_execution` 三事件 + `streamId` + `turn.start/end` + `graph_update` 三范围口径；依据 design/04 §1.1（裁决 01.4.1 / 01.4.2 / 01.4.3 / 03.4.4）与裁决 01.4.5（design/20 §2），2026-10-04 落实）：① **`tool_execution` 拆三事件（破坏性替换）**：旧单帧 `tool_execution.update {id,index,delta}`（`delta` 为 `"start:工具名"` / `"end:error:工具名"` 拼串）**不再由 loop 产生**，替换为 `tool_execution.start`（`{tool, phase:"running", id?}`，进入执行）/ `tool_execution.update`（`{tool, phase:"running", delta?, id?}`，运行中进度，可多帧）/ `tool_execution.end`（`{tool, phase:"done"|"error", id?}`，完成或失败）——**phase 与工具名是结构化字段，不再塞进 `delta`**；`id` 复用 provider 工具调用 id（可关联 `tool_call.*` 帧，无 id 时省略该键）。**迁移**：按旧 `delta` 拼串解析的客户端必须改读 `phase`/`tool` 字段；`tool_call.*` 帧形状不变。**诚实边界**：start/end 由 loop 在每次工具执行的首尾真实发出；`update` 的 wire 映射已实现并有单测锁定，但工具分发器尚无中途进度上报源，**loop 暂不产出 update 帧**（未来接入进度上报无需再改契约）。② **`streamId`（新增可选字段，位于模型流帧的 `data` 内）**：`message.*` / `thinking.*` / `tool_call.*` 帧的 `data` 带 `streamId`——loop 在**每次模型流**（一次 ChatStream 调用）开始时生成，同流多帧共享，多步工具轮的下一个模型调用取新 id；与帧壳 `turnId`（整轮恒定）分工：`turnId` = 哪轮回答，`streamId` = 这轮里的哪条流式消息。`tool_execution.*` / `turn.*` 帧不带。③ **`turn.start` / `turn.end`（新帧，无 `data` 载荷）**：loop 在 Run 起止发出、经 ws_bridge 下行（`model.StreamEvent` 新 kind → hub `broadcastStream` 映射）；帧壳 `sessionId`/`turnId`/`seq` 即边界信号；`turn.end` 覆盖所有出口（最终答案 / 工具循环安全网终止 / 预算耗尽 / 步骤出错）；回合终态仍由 `stream_end` 携带（分工不重叠）；被静音的内部流程（backthink / 子代理）不发出。④ **`graph_update` 三范围口径**（01.4.5，随实现事实补记）：发布范围为 `web` / `web:` / `sess-backthink` 三类（此前文档只写「web 及其子会话」，漏了 backthink 思考链路图这一实现事实；实现侧 `cmd/cyrene/wire_runtime.go` 的 allow 谓词本就如此，本批零代码变更）；并补记 **tool 节点 `status` 由 phase 驱动**（与 §3.1 `tool_execution.*` 的 `phase` 同枚举同决策：`trace.FlowStatusForPhase` 消费 tool span 上的 phase，缺省回落 span 终态）。⑤ **角色门**：本批新帧（`tool_execution.start/end`、`turn.start/end`）全部落 v0.48 下行门的「仅 `owner` + `admin`」档（fail-closed 默认，白名单未扩）；`viewer` 行为不变。⑥ **向后兼容总结**：新增帧与新增键（`streamId`、`data.tool`、`data.phase`）可被老客户端安全忽略；唯一破坏点是旧 `tool_execution.update` 单帧消失（见 ①）；**现有 Web 前端未消费 `tool_execution.*` / `turn.*` / `streamId`，本批对它零影响**。⑦ **验证**：loop 包断言三事件顺序/phase/工具名/id/Delta 为空、同流共享 streamId 跨流相异、turn 边界成对且覆盖错误出口、静音 run 零事件、flow 图 tool 节点 running→done/error 与 phase 一致；gateway 包真 WS 帧端到端（真 loop + 真 dispatcher + bus → relay → broadcastStream）：按 seq 排序断言完整帧序 `turn.start → message.delta → tool_execution.start → tool_execution.end → message.delta → turn.end`（+`stream_end`）、`streamId` 每流独立、`tool_execution` 帧无 `delta`/`streamId` 键、turn 帧无载荷；帧门矩阵补 5 行（三事件 + turn 两帧 × 三角色）。**变异验证**：去掉 loop 的 `ev.StreamID = streamID` 透传 → 「streamId 每流独立/前缀」断言 FAIL（已跑，还原后复绿）。 - `v0.49`（**WS 握手回放 + 上行帧的角色门**：把 design/11 §7.1 下行门落地时查出的**同一类缺口的另两处**一并收口——**对 `viewer` 继续收紧**，**`owner` / `admin` 行为不变**，**无端点、无字段形状变更**；依据 design/11 §7.2，2026-10-03 落实）：① **问题**：本质与 §7.1 同一个——`viewer` 是**只读监控档**，却能通过 WS 拿到内容、甚至发起写入。(a) 握手 `history` 回放**对任意角色照发**（回放主会话最近 50 条**消息正文**，`thinking` 已剥离但正文本身即主人的对话），`viewer` 一连上就把对话拿走了；(b) `route` 的**上行分发完全没有角色检查**，`viewer` 发 `chat` 能**真的驱动模型并把消息落库**（只读档拿到写权限），`subagent_ask` / `task_output_query` 的应答是**定向帧**（`enqueueJSON`）因此**绕过 §7.1 的下行门**，把子代理上下文 / 任务输出直接回给 `viewer`。② **规则（fail-closed，默认拒）**：**A** 握手 `history` 回放**仅 `owner` + `admin`**（`ownerAdminClaims`，与主动推送同一个判据）；**B** 非 `owner`/`admin` 连接发来的**任何**上行 JSON 帧一律拒绝——**定向**（只回发送者）回 `{type:"error",data:{code:"forbidden"}}`，**不广播、不驱动模型、不落库、不返回任何内容**；被拒帧：`chat` / `interrupt` / `subagent_ask` / `task_output_query` / `voice_input` / `mode_switch`（**默认拒**，未登记的一律拒）。**协议级帧不受影响**：WS ping/pong 与正常 close 照常（否则连接会掉），被拒帧也不关连接。③ **实现**：A 收口在**唯一**发出回放的 `hub.replayHistory` 顶部（**不放调用点**，没有第二个调用者能绕过）；B 收口在 `wsConn.route` 的**第一条语句**，且写成「**默认拒**」而**不是被拒类型黑名单**——新加的上行帧类型自动被拒，漏一处也不会放行。④ **合口径（与 §7.1 一起读）**：`viewer` = **可以连接、可以读 12 条只读监控 REST（v0.40）、可以收 7 个状态机信号帧（v0.48）**；**除此以外 WS 上拿不到任何内容，也发不出任何指令**。⑤ **未来放宽的要求**：将来若真需要 `viewer` 上行（如手机端以 viewer 凭据做只读订阅之外的交互），**必须先在 design/11 §7.2 给它开口并登记白名单**；代码里那把「默认拒」必须**改造成登记表**，**不得直接删掉**。⑥ **测试与变异验证**：真 WS 连接 + 真 JWT（复用 v0.48 的测试装置）。A：`viewer` 连接**收不到** `history` 回放，`owner` / `admin` **收得到**（且回放里 `thinking` 仍为空）；B：`viewer` 依次发 `chat` / `interrupt` / `subagent_ask` / `task_output_query` / `voice_input` / `mode_switch`，**每次都收到定向 `forbidden` error**，且其后**无任何内容帧**（对 `subagent_ask` / `task_output_query` 尤其断言「不得收到定向的 `subagent_event` / `task_output`」），全程**计数 runner = 0 且落库 = 0**；另一条 `viewer` 连接作为观察者**不收**这些 error（证明是**定向**而非广播）；回归：`owner` 发同一个 `chat` 仍然**真的驱动 runner（= 1）**并把 user + assistant 两条**落库**；协议级：被拒后 ping/pong 往返与正常 close（服务端注销连接）**均正常**。**变异验证**：① 去掉 A 的门 → `viewer` 立刻收到 `history`（A 断言 FAIL，已跑）；② 去掉 B 的门 → `viewer` 的 `chat` 驱动 runner **1 次**并**落库 2 条**、`subagent_ask` 直接回 `subagent_event`（B 断言 FAIL，已跑）。⑦ **影响面**：`owner` / `admin` **逐字不变**；`/ws/chat` **连接鉴权不变**（`viewer` 仍可连）；无新端点、无字段形状变更；今天**没有 `viewer` 客户端在跑**（`/auth/login` 仍写死签发 `RoleOwner`，除测试外无 viewer 凭据），故线上实际影响为 0。 - `v0.48`（**WS 下行帧的角色门**：`/ws/chat` 的下行帧改为**按帧类型分档**——**对 `viewer` 是收紧**（此前它连上就能收到角色回复正文与推理内容，现在收不到），**`owner` / `admin` 行为不变**，**无端点、无字段形状变更**；依据 design/11 §7.1，主人 2026-10-03 拍板「按建议收紧」）：① **问题**：hub 的下行是**全连接广播**，而 `/ws/chat` 只验 JWT（任意角色可连），于是 `viewer` 连接会收到 `message.*` / `thinking.*` / `tool_call.*`（**含角色回复正文与推理内容**）、`message_persisted` / `message_edited` / `message_deleted`、`graph_update`、`mood`、`subagent_event` / `task_output` / `device_update` / `reminder` 等——与 design/19 §1/§3「flow 可视化仅 owner/admin」直接冲突，也与 `GET /sessions/{id}/messages` 的 owner/admin 档位不一致（**同一批内容从 WS 旁路漏出**）。② **规则（fail-closed，默认拒）**：**只有白名单里的 7 个「状态机信号」**（`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full` / `session_read`，**均不含任何内容**）发给**所有已认证连接**（含 `viewer`）；**其余全部**（`stream_chunk` / `stream_end` / `thinking`、`message_persisted` / `message_edited` / `message_deleted` / `channel_persisted`、`graph_update` / `subagent_event` / `subagent.result` / `task_output` / `task_done` / `tool_used` / `capacity_awaken` / `mode_switch`、`mood` / `device_update` / `reminder`）**只发 `owner` + `admin`**；**未登记的新帧类型默认只发 owner/admin**（要进白名单须先在 design/11 §7.1 登记理由「不含内容」）。③ **实现**：收口为**唯一一处**「按帧类型分类后投递」的入口（`internal/services/gateway/ws_frame_scope.go` + `hub.broadcast`），**不给每个调用点加开关**（那正是「漏一处即泄漏」的写法）——流式 `message.*` / `thinking.*` / `tool_call.*`（`broadcastStream`）、`stream_chunk` / `stream_end`（`runTurn` / `streamChat`）、`message_persisted` / `message_edited` / `message_deleted`、`mood` / `capacity_awaken`、编排状态信号（`StateSink`）、`graph_update` / `subagent_event` / `device_update` / `reminder`（`relay` 归一化）**全部经同一入口**，不存在第二条投递路径。④ **语义澄清（不是数据从此不可见）**：这条门管「**谁该被主动推送**」——`viewer` 的 **12 条只读监控端点**（v0.40 `viewerAuth`，含 `GET /personas` 的角色级 `mood`）**逐条照旧**；`/ws/chat` 的**连接鉴权不变**（`viewer` 仍可连，只是收不到上列帧）；**定向回请求连接的帧**（`voice_output` / `error` / `subagent_ask`、`task_output_query` 的应答）不进这道门、行为不变。⑤ **测试与变异验证**：真 WS 连接 + 真角色 token 的**逐帧表驱动**断言（design/11 §7.1 表里**每一个**帧类型 × 三种角色：`owner` / `admin` 两类都收得到，`viewer` 只收白名单）+ 帧类型 × 角色的直接矩阵 + 「白名单精确等于 7 个状态信号」与「未登记类型默认拒」的锁定；变异验证：把 `graph_update` 改回全广播 → 「viewer 收不到 `graph_update`」必须 FAIL（已跑，见提交说明），把 `session_idle` 改走受限路径 → 「viewer 收得到 `session_idle`」必须 FAIL（已跑）。⑥ **已知残留（本次未动，留待主人裁决）**：`/ws/chat` 的**上行**仍未按角色授权（`viewer` 仍可发 `chat` / `interrupt` / `subagent_ask` / `task_output_query` / `voice_input`，其应答是**定向**帧、不经过本门），且握手时的 `history` 回放仍对任意角色照发（`thinking` 已剥离）——二者都在 design/11 §7.1 本次门的作用域之外。**（本残留已由 `v0.49` 收口：上行帧默认拒、握手回放仅 `owner`/`admin`，见上一条变更记录与 §3 顶部第二把角色门。）** - `v0.47`（**`graph_update` 子代理泳道根节点新增可选 `props.persona`**，对客户端**纯增量**；实现依据 design/22 §8-2「跨角色派发 = 把其他角色以子代理形式派出去」）：① **`subagent_spawn` 指定目标角色派发时**，该子代理在父会话图里的**分支根节点** `props` 多一个 `persona` 键（值 = 目标角色 id），供流程图显示"这个子代理是以谁的身份在替班"，而不只是子代理 id。② **不指定角色派发时该键不出现**：帧形状与 v0.46 逐字一致，老客户端忽略未知键即可。③ **无新端点、无字段改名、无鉴权变更**；`subagent_event` 帧**未新增字段**（其 `role` 会等于目标角色——既有字段的取值语义变化，非形状变更）。④ **`subagent_spawn` 的 `persona` 参数不在本文档契约内**：那是给**模型**用的工具 schema（工具层以「进程能否按该角色运行」为判据，未知角色整批拒绝且不产生任何子代理），不是给客户端的 HTTP/WS 契约。⑤ **验证**：进程内 e2e 断言「子代理实际收到的 system prompt = 目标角色 instructions」+「其记忆写入落在**目标角色**域」+「未知角色整批拒绝」+「**省略 `persona` 时结果 JSON 键集与旧版完全相同**」；变异验证（把目标角色换回父/默认角色、关掉角色校验）对应断言均 FAIL。 diff --git a/internal/runtime/loop/loop_extra_test.go b/internal/runtime/loop/loop_extra_test.go index e52f196..9a75ba8 100644 --- a/internal/runtime/loop/loop_extra_test.go +++ b/internal/runtime/loop/loop_extra_test.go @@ -319,12 +319,12 @@ func TestRun_StreamToolExecution_EmitsProgress(t *testing.T) { defer mu.Unlock() saw := false for _, ev := range evs { - if ev.Kind == model.StreamKindToolExecution { + if ev.Kind == model.StreamKindToolExecutionStart || ev.Kind == model.StreamKindToolExecutionEnd { saw = true break } } if !saw { - t.Fatalf("expected a tool_execution_update event on the streaming path, got %+v", evs) + t.Fatalf("expected a tool_execution start/end event on the streaming path, got %+v", evs) } } diff --git a/internal/runtime/loop/loop_test.go b/internal/runtime/loop/loop_test.go index 808d659..6d7277a 100644 --- a/internal/runtime/loop/loop_test.go +++ b/internal/runtime/loop/loop_test.go @@ -283,8 +283,15 @@ func TestRun_StreamDeltas(t *testing.T) { var mu sync.Mutex var deltas []string l := New(Options{ - Agent: a, - OnStream: func(ev model.StreamEvent) { mu.Lock(); deltas = append(deltas, ev.Delta); mu.Unlock() }, + Agent: a, + OnStream: func(ev model.StreamEvent) { + if ev.Kind != model.StreamKindTextDelta { + return + } + mu.Lock() + deltas = append(deltas, ev.Delta) + mu.Unlock() + }, MaxRounds: 8, }) diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index a9a092b..789fab5 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -23,6 +23,10 @@ import ( // goroutine 上，因此提取绝不延迟回复），错误记录日志 // 而非致命。 // +// 回合边界发出 turn.start / turn.end 流式帧（design/04 §1.1）： +// start 在任何工作之前，end 经 defer 覆盖 Run 的每个出口 +// （最终答案 / 安全网终止 / 预算耗尽 / 步骤错误）。 +// // 当 turn 带着 Result 落定时，OnTurnEnd 钩子对该 // turn 调用一次（绝不按步调用），其情绪标签记录在 Result.Mood 上， // 使循环外围的分层能推进人格状态并标记持久化的回复。 @@ -46,6 +50,9 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa ctx = withCapState(ctx, &capStateAccum{state: capacity.StateQuiet}) ctx = withLoopSession(ctx, sessionID) + l.emitTurnBoundary(ctx, model.StreamKindTurnStart) + defer l.emitTurnBoundary(ctx, model.StreamKindTurnEnd) + // 恢复被中断的 run（design §1 durable execution）：该会话有 // 可恢复的 checkpoint 时，重建其上下文并从下一步继续， // 而不是从全新对话重启。不同的 @@ -53,7 +60,6 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa // 全新开始（LoadLatest 未命中，或 stop reason 不可恢复）。 msgs, startStep, _ := l.loadCheckpoint(ctx, sessionID, messages) msgs = l.resolveImages(ctx, msgs) - toolState := &toolLoopState{prevFP: make(map[string]bool)} res := &Result{} // 分支 Run（子代理）把其 span 并入父会话的实时 // 图；只有所属的 Run 开始与结束轮次。没有它，子代理 @@ -62,6 +68,14 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa l.flow.StartRound(sessionID, lastUserText(msgs)) defer func() { l.flow.EndRound(sessionID, res.Text) }() } + return l.driveTurn(ctx, sessionID, msgs, startStep, res) +} + +// driveTurn 运行至多 maxRounds 个 step 的主循环：每步组装上下文、 +// 调用模型、执行被请求的工具并回灌。它在任何一种落定路径上返回 +// 已 settle 的 Result；step 或工具批次的错误以 %w 包装返回。 +func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Message, startStep int, res *Result) (*Result, error) { + toolState := &toolLoopState{prevFP: make(map[string]bool)} for step := startStep; step <= l.maxRounds; step++ { if err := ctx.Err(); err != nil { return nil, fmt.Errorf("loop: run: step %d: %w", step, err) @@ -88,10 +102,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa if outcome.final { res.Text = outcome.text res.StopReason = outcome.stopReason - l.saveCheckpoint(ctx, sessionID, step, msgs, outcome.stopReason) - l.extractMemoryAsync(ctx, msgs, step) - l.settleTurn(ctx, sessionID, msgs, res) - return res, nil + return l.settleAndReturn(ctx, sessionID, msgs, step, outcome.stopReason, res) } batch, err := l.runToolCalls(rc, msgs, outcome.toolCalls, toolState) @@ -107,18 +118,22 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa if batch.terminate { res.Text = criticalToolLoopTerminalReason res.StopReason = StopReasonToolLoopTerminated - l.saveCheckpoint(ctx, sessionID, step, msgs, StopReasonToolLoopTerminated) - l.extractMemoryAsync(ctx, msgs, step) - l.settleTurn(ctx, sessionID, msgs, res) - return res, nil + return l.settleAndReturn(ctx, sessionID, msgs, step, StopReasonToolLoopTerminated, res) } } // 预算耗尽而无最终答案。 res.Steps = l.maxRounds res.StopReason = StopReasonMaxRounds - l.saveCheckpoint(ctx, sessionID, l.maxRounds, msgs, StopReasonMaxRounds) - l.extractMemoryAsync(ctx, msgs, l.maxRounds) + return l.settleAndReturn(ctx, sessionID, msgs, l.maxRounds, StopReasonMaxRounds, res) +} + +// settleAndReturn 落定一个已得出 Result 的回合：以该路径的 +// stop reason 持久化终态 checkpoint、触发后台记忆提取、 +// 上报 turn 钩子（情绪标签写回 Result.Mood），然后把结果交还调用方。 +func (l *Loop) settleAndReturn(ctx context.Context, sessionID string, msgs []model.Message, step int, stopReason string, res *Result) (*Result, error) { + l.saveCheckpoint(ctx, sessionID, step, msgs, stopReason) + l.extractMemoryAsync(ctx, msgs, step) l.settleTurn(ctx, sessionID, msgs, res) return res, nil } diff --git a/internal/runtime/loop/stream.go b/internal/runtime/loop/stream.go index 76a942d..71669f1 100644 --- a/internal/runtime/loop/stream.go +++ b/internal/runtime/loop/stream.go @@ -2,7 +2,10 @@ package loop import ( "context" + "crypto/rand" + "encoding/hex" "fmt" + "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" @@ -36,6 +39,11 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO } } + // 一次模型流（一次 ChatStream 调用）取一个 streamId：该流的 + // 每一帧共享它，而多步工具轮的后续模型调用取新的 id + // （design/04 §1.1）。它与 TurnID 分工：turnId 标记整个回合， + // streamId 标记回合内的一条流式消息。 + streamID := newStreamID() resp, err := adapter.ChatStream(ctx, req, func(ev model.StreamEvent) { // 在循环边界打上 session/turn 身份：适配器运行在 // 循环之下，无法知道该流属于哪个会话， @@ -43,6 +51,7 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO // 前端便能把增量路由到活跃会话。 ev.SessionID = loopSession(ctx) ev.TurnID = loopTurn(ctx) + ev.StreamID = streamID // 为内部自主流程（backthink）抑制 UI 广播， // 其中间推理不得泄露进聊天流；当该 turn 的角色 // 关闭思考帧时（showToUser: false）也抑制。两处门 @@ -95,3 +104,31 @@ func isThinkingKind(kind model.StreamKind) bool { return false } } + +// newStreamID 返回新的随机十六进制标识，用于把一串流帧关联到 +// 同一次模型流（design/04 §1.1 streamId）。它消耗 crypto/rand 源， +// 源不可用时回退到纳秒时间戳，因此总能产出 id（空的 streamId 会 +// 静默破坏客户端关联）。 +func newStreamID() string { + var b [8]byte + if _, err := rand.Read(b[:]); err == nil { + return "stream-" + hex.EncodeToString(b[:]) + } + return "stream-" + time.Now().UTC().Format("20060102150405.000000000") +} + +// emitTurnBoundary 在循环的回合边界发出 turn.start / turn.end 帧 +// （design/04 §1.1）。它复用流式路径的广播门：未配置 OnStream 或 +// 被静音的内部流程（backthink / 子代理）不发出。两帧不带载荷 —— +// 帧壳上的 session/turn 身份就是边界信号，回合的终态信息仍由 +// hub 在 Run 返回后发出的 stream_end 携带。 +func (l *Loop) emitTurnBoundary(ctx context.Context, kind model.StreamKind) { + if l.onStream == nil || isMutedBroadcast(ctx) { + return + } + l.onStream(model.StreamEvent{ + Kind: kind, + SessionID: loopSession(ctx), + TurnID: loopTurn(ctx), + }) +} diff --git a/internal/runtime/loop/stream_events_test.go b/internal/runtime/loop/stream_events_test.go new file mode 100644 index 0000000..786cb9f --- /dev/null +++ b/internal/runtime/loop/stream_events_test.go @@ -0,0 +1,275 @@ +package loop + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// captureFlow 返回一个 FlowPublisher 与读取其已发布快照的 reader， +// 使测试能断言工具节点的 status 如何随 phase 演进（design/19 §1）。 +func captureFlow() (*FlowPublisher, func() []*trace.FlowGraph) { + var mu sync.Mutex + var seen []*trace.FlowGraph + flow := NewFlowPublisher(func(subject string, payload any) { + if subject != graphUpdateSubject { + return + } + if g, ok := payload.(*trace.FlowGraph); ok { + mu.Lock() + seen = append(seen, g) + mu.Unlock() + } + }, func(string) bool { return true }) + return flow, func() []*trace.FlowGraph { + mu.Lock() + defer mu.Unlock() + return append([]*trace.FlowGraph(nil), seen...) + } +} + +// toolExecEvents 过滤出 tool_execution 三事件，保持发出顺序。 +func toolExecEvents(evs []model.StreamEvent) []model.StreamEvent { + var out []model.StreamEvent + for _, ev := range evs { + switch ev.Kind { + case model.StreamKindToolExecutionStart, model.StreamKindToolExecutionUpdate, model.StreamKindToolExecutionEnd: + out = append(out, ev) + } + } + return out +} + +// toolNodeStatus 返回图中所有 tool 节点的当前 status。 +func toolNodeStatus(g *trace.FlowGraph) []string { + var out []string + for _, n := range g.Nodes { + if n.Kind == "tool" { + out = append(out, n.Status) + } + } + return out +} + +// runToolTurn 用「模型请求 datetime → 工具执行 → 模型收尾」的两步流 +// 驱动一个带流式回调与 flow 图的 Run，返回收集到的事件与图快照。 +func runToolTurn(t *testing.T, toolResult any, toolErr error) ([]model.StreamEvent, []*trace.FlowGraph) { + t.Helper() + ad := &mockAdapter{stream: []mgResp{ + {resp: nativeDatetimeCall()}, + {resp: model.Response{Model: "m", Content: "done"}}, + }} + td := dispatcherWith(t, &mockTool{name: "datetime", result: toolResult, err: toolErr}) + a := newTestAgent(t, ad, td) + onStream, streams := collectStream() + flow, graphs := captureFlow() + l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8, Flow: flow, Tracer: trace.New(nil)}) + + if _, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "time?")}); err != nil { + t.Fatalf("Run: %v", err) + } + return streams(), graphs() +} + +// TestRun_ToolExecutionThreeEvents 锁住裁决 01.4.1：一次工具执行发出 +// start 与 end 两帧（顺序固定），phase 是 running/done 的结构化字段， +// 绝不再塞进 Delta；工具名走 Tool 字段，ID 复用 provider 的 +// tool call id，使执行帧可与 tool_call.* 帧关联。 +func TestRun_ToolExecutionThreeEvents(t *testing.T) { + t.Parallel() + evs, _ := runToolTurn(t, map[string]any{"now": "2026-09-08 22:34:00"}, nil) + ex := toolExecEvents(evs) + if len(ex) != 2 { + t.Fatalf("tool_execution events = %d, want 2 (start+end): %+v", len(ex), ex) + } + start, end := ex[0], ex[1] + if start.Kind != model.StreamKindToolExecutionStart || start.Phase != model.ToolPhaseRunning { + t.Fatalf("start event = %+v, want kind start phase running", start) + } + if end.Kind != model.StreamKindToolExecutionEnd || end.Phase != model.ToolPhaseDone { + t.Fatalf("end event = %+v, want kind end phase done", end) + } + for _, ev := range ex { + if ev.Tool != "datetime" { + t.Fatalf("event tool = %q, want datetime (%+v)", ev.Tool, ev) + } + if ev.ID != "call_native_1" { + t.Fatalf("event id = %q, want the provider tool call id (%+v)", ev.ID, ev) + } + if ev.Delta != "" { + t.Fatalf("event delta = %q, want empty (phase must not ride in Delta)", ev.Delta) + } + if ev.StreamID != "" { + t.Fatalf("event streamId = %q, want empty (tool execution is not a model stream)", ev.StreamID) + } + } +} + +// TestRun_ToolExecutionErrorPhase 覆盖失败路径：end 帧的 phase 为 +// error，且 flow 图中该 tool 节点最终也是 error —— 同一 phase 驱动 +// 两处（裁决 03.4.4）。 +func TestRun_ToolExecutionErrorPhase(t *testing.T) { + t.Parallel() + evs, graphs := runToolTurn(t, nil, errors.New("boom")) + ex := toolExecEvents(evs) + if len(ex) != 2 { + t.Fatalf("tool_execution events = %d, want 2: %+v", len(ex), ex) + } + if ex[1].Kind != model.StreamKindToolExecutionEnd || ex[1].Phase != model.ToolPhaseError { + t.Fatalf("end event = %+v, want kind end phase error", ex[1]) + } + last := graphs[len(graphs)-1] + statuses := toolNodeStatus(last) + if len(statuses) == 0 || statuses[0] != trace.FlowStatusError { + t.Fatalf("final tool node status = %v, want [error] (phase-driven)", statuses) + } +} + +// TestRun_ToolPhaseDrivesFlowNodeStatus 锁住裁决 03.4.4：工具执行期间 +// flow 图出现 running 的 tool 节点，执行结束后翻转为 done —— 节点状态 +// 由细粒度 phase 驱动，取值与事件 phase 完全一致。 +func TestRun_ToolPhaseDrivesFlowNodeStatus(t *testing.T) { + t.Parallel() + evs, graphs := runToolTurn(t, map[string]any{"now": "x"}, nil) + sawRunning, sawDone := false, false + for _, g := range graphs { + for _, s := range toolNodeStatus(g) { + switch s { + case trace.FlowStatusRunning: + sawRunning = true + case trace.FlowStatusDone: + sawDone = true + } + } + } + if !sawRunning || !sawDone { + t.Fatalf("tool node statuses over the round: running=%v done=%v, want both (phase-driven)", sawRunning, sawDone) + } + ex := toolExecEvents(evs) + if len(ex) != 2 || ex[0].Phase != model.ToolPhaseRunning || ex[1].Phase != model.ToolPhaseDone { + t.Fatalf("phases = %+v, want [running done] matching the node statuses", ex) + } +} + +// TestRun_StreamIDPerModelStream 锁住裁决 01.4.2：同一次模型流的多帧 +// 共享同一 streamId；多步工具轮的第二次模型调用取新的 streamId； +// turn 与 tool_execution 帧不属于任何模型流，不带 streamId。 +func TestRun_StreamIDPerModelStream(t *testing.T) { + t.Parallel() + ad := &mockAdapter{stream: []mgResp{ + {resp: nativeDatetimeCall()}, + {resp: model.Response{Model: "m", Content: "done"}}, + }, chunks: [][]string{{"Hel"}, {"lo"}}} + td := dispatcherWith(t, &mockTool{name: "datetime", result: "now"}) + a := newTestAgent(t, ad, td) + onStream, streams := collectStream() + l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8}) + + if _, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "time?")}); err != nil { + t.Fatalf("Run: %v", err) + } + var firstStream, secondStream string + for _, ev := range streams() { + if ev.Kind != model.StreamKindTextDelta { + continue + } + if !strings.HasPrefix(ev.StreamID, "stream-") { + t.Fatalf("delta streamId = %q, want a stream-* id", ev.StreamID) + } + switch ev.Delta { + case "Hel": + firstStream = ev.StreamID + case "lo": + secondStream = ev.StreamID + } + } + if firstStream == "" || secondStream == "" { + t.Fatalf("missing deltas for the two model streams (first=%q second=%q)", firstStream, secondStream) + } + if firstStream == secondStream { + t.Fatalf("the two model streams share streamId %q; each ChatStream call must mint a fresh one", firstStream) + } + for _, ev := range streams() { + switch ev.Kind { + case model.StreamKindTurnStart, model.StreamKindTurnEnd, + model.StreamKindToolExecutionStart, model.StreamKindToolExecutionEnd: + if ev.StreamID != "" { + t.Fatalf("%s carries streamId %q; only model-stream frames do", ev.Kind, ev.StreamID) + } + } + } +} + +// TestRun_TurnBoundaryFrames 锁住裁决 01.4.3：回合以 turn.start 开始、 +// turn.end 结束（无论正常落定还是预算耗尽），两帧携带会话与 turn +// 作用域、无载荷。 +func TestRun_TurnBoundaryFrames(t *testing.T) { + t.Parallel() + ad := &mockAdapter{stream: []mgResp{{resp: model.Response{Content: "ok"}}}, chunks: [][]string{{"ok"}}} + a := newTestAgent(t, ad, nil) + onStream, streams := collectStream() + l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8}) + + ctx := WithTurnID(context.Background(), "turn-b") + if _, err := l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}); err != nil { + t.Fatalf("Run: %v", err) + } + evs := streams() + if len(evs) < 3 { + t.Fatalf("events = %+v, want at least turn.start + one delta + turn.end", evs) + } + if evs[0].Kind != model.StreamKindTurnStart || evs[len(evs)-1].Kind != model.StreamKindTurnEnd { + t.Fatalf("boundary kinds = %q…%q, want turn_start…turn_end", evs[0].Kind, evs[len(evs)-1].Kind) + } + for _, boundary := range []model.StreamEvent{evs[0], evs[len(evs)-1]} { + if boundary.SessionID != "web" || boundary.TurnID != "turn-b" { + t.Fatalf("boundary scope = session %q turn %q, want web turn-b", boundary.SessionID, boundary.TurnID) + } + if boundary.Delta != "" || boundary.Phase != "" || boundary.Tool != "" { + t.Fatalf("boundary carries payload: %+v", boundary) + } + } +} + +// TestRun_TurnEndEmittedOnStepError 覆盖错误出口：模型步失败时 Run +// 返回错误，但回合已经以 turn.start 开过，因此 turn.end 必须照样 +// 发出，客户端不会留下一个永远开着的回合。 +func TestRun_TurnEndEmittedOnStepError(t *testing.T) { + t.Parallel() + ad := &mockAdapter{stream: []mgResp{{err: errors.New("model down")}}} + a := newTestAgent(t, ad, nil) + onStream, streams := collectStream() + l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8}) + + if _, err := l.Run(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}); err == nil { + t.Fatal("Run: expected the step error to surface") + } + evs := streams() + if len(evs) != 2 || evs[0].Kind != model.StreamKindTurnStart || evs[1].Kind != model.StreamKindTurnEnd { + t.Fatalf("events = %+v, want exactly turn.start + turn.end around the failed run", evs) + } +} + +// TestRun_TurnBoundaryMutedForInternalRuns 内部自主流程（backthink / +// 子代理）经 WithMutedBroadcast 驱动同一个循环：回合边界帧与流帧 +// 一样被静音，中间过程绝不泄露进面向用户的聊天流。 +func TestRun_TurnBoundaryMutedForInternalRuns(t *testing.T) { + t.Parallel() + ad := &mockAdapter{stream: []mgResp{{resp: model.Response{Content: "ok"}}}, chunks: [][]string{{"o"}}} + a := newTestAgent(t, ad, nil) + onStream, streams := collectStream() + l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8}) + + ctx := WithMutedBroadcast(context.Background()) + if _, err := l.Run(ctx, "sess-backthink", []model.Message{model.TextMessage(model.RoleUser, "reflect")}); err != nil { + t.Fatalf("Run: %v", err) + } + if evs := streams(); len(evs) != 0 { + t.Fatalf("muted run emitted %d events, want none: %+v", len(evs), evs) + } +} diff --git a/internal/runtime/loop/stream_scope_test.go b/internal/runtime/loop/stream_scope_test.go index 4bfc3fe..5ae84d0 100644 --- a/internal/runtime/loop/stream_scope_test.go +++ b/internal/runtime/loop/stream_scope_test.go @@ -29,7 +29,8 @@ func collectStream() (func(model.StreamEvent), func() []model.StreamEvent) { // TestRun_StreamCarriesSessionAndTurn 钉住 A1 契约：循环给每个 // 流式事件打上它被驱动的会话，以及经 WithTurnID 挂载的 turn id， // 使 WS hub 能把 message.delta/thinking.delta/tool_call.* -// 路由到正确的会话与 turn（design/16 §4.2）。 +// 路由到正确的会话与 turn（design/16 §4.2）。回合边界帧 +// （turn.start/end，v0.50）同样带作用域，因此计入总数。 func TestRun_StreamCarriesSessionAndTurn(t *testing.T) { t.Parallel() ad := &mockAdapter{ @@ -45,16 +46,19 @@ func TestRun_StreamCarriesSessionAndTurn(t *testing.T) { t.Fatalf("Run: %v", err) } events := streams() - if len(events) != 2 { - t.Fatalf("stream events = %d, want 2", len(events)) + if len(events) != 4 { + t.Fatalf("stream events = %d, want 4 (turn.start + 2 deltas + turn.end)", len(events)) } for i, ev := range events { if ev.SessionID != "web" || ev.TurnID != "turn-xyz" { t.Fatalf("event[%d] scope = session %q turn %q, want session web turn turn-xyz", i, ev.SessionID, ev.TurnID) } } - if events[0].Delta != "Hel" || events[1].Delta != "lo" { - t.Fatalf("deltas = [%q %q], want [Hel lo]", events[0].Delta, events[1].Delta) + if events[0].Kind != model.StreamKindTurnStart || events[3].Kind != model.StreamKindTurnEnd { + t.Fatalf("boundary kinds = %q…%q, want turn_start…turn_end", events[0].Kind, events[3].Kind) + } + if events[1].Delta != "Hel" || events[2].Delta != "lo" { + t.Fatalf("deltas = [%q %q], want [Hel lo]", events[1].Delta, events[2].Delta) } } @@ -86,7 +90,7 @@ func TestRun_StreamWithoutTurnIDLeavesItEmpty(t *testing.T) { } // TestRun_ToolExecutionCarriesScope 覆盖循环的另一个流式发出点： -// tool_execution_update 进度帧也必须带作用域，否则它就会成为 +// tool_execution.start/end 进度帧也必须带作用域，否则它就会成为 // 多会话客户端唯一无法路由的帧。 func TestRun_ToolExecutionCarriesScope(t *testing.T) { t.Parallel() @@ -105,7 +109,7 @@ func TestRun_ToolExecutionCarriesScope(t *testing.T) { } found := 0 for _, ev := range streams() { - if ev.Kind != model.StreamKindToolExecution { + if ev.Kind != model.StreamKindToolExecutionStart && ev.Kind != model.StreamKindToolExecutionEnd { continue } found++ @@ -114,6 +118,6 @@ func TestRun_ToolExecutionCarriesScope(t *testing.T) { } } if found == 0 { - t.Fatal("no tool_execution_update event was emitted") + t.Fatal("no tool_execution event was emitted") } } diff --git a/internal/runtime/loop/thinking_gate_test.go b/internal/runtime/loop/thinking_gate_test.go index a90cd00..d88d6c9 100644 --- a/internal/runtime/loop/thinking_gate_test.go +++ b/internal/runtime/loop/thinking_gate_test.go @@ -85,8 +85,8 @@ func TestRun_ThinkingShowToUserFalseWithholdsThinkingFrames(t *testing.T) { t.Fatalf("event kind %q emitted %d time(s), want 1 (body must be unaffected)", k, kinds[k]) } } - if len(events) != 3 { - t.Fatalf("events = %d, want only the 3 text frames", len(events)) + if len(events) != 5 { + t.Fatalf("events = %d, want the 3 text frames plus the 2 turn boundaries", len(events)) } } @@ -100,8 +100,8 @@ func TestRun_ThinkingShowToUserDefaultBroadcastsThinking(t *testing.T) { if kinds[model.StreamKindThinkingDelta] != 1 || kinds[model.StreamKindThinkingStart] != 1 || kinds[model.StreamKindThinkingEnd] != 1 { t.Fatalf("thinking frames = %v, want start/delta/end each once with the option unset", kinds) } - if len(events) != 6 { - t.Fatalf("events = %d, want all 6 frames", len(events)) + if len(events) != 8 { + t.Fatalf("events = %d, want all 8 frames (6 segment frames + 2 turn boundaries)", len(events)) } } diff --git a/internal/runtime/loop/tools.go b/internal/runtime/loop/tools.go index 8011676..d0472ac 100644 --- a/internal/runtime/loop/tools.go +++ b/internal/runtime/loop/tools.go @@ -114,42 +114,8 @@ func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls res.terminate = true } - // 安全门（design/11 §3）：群聊/陌生人会话只能调用 - // 其可编辑标记允许该环境的工具（默认关闭）。不被允许的 - // 调用绝不执行；它以 not_allowed 软结果回灌 - // 并记录到审计 sink。它是该批的最终权威 —— - // 后续的审计钩子可以收紧，但绝不重新放行被拒的工具。 - if d != nil { - l.applyToolSecurityGate(ctx, d, calls, actions) - } - - // BatchAdmit 钩子（审计 / 政治审查）：拦截整批。 - if l.hooks.BatchAdmit != nil { - if ok, reason := l.hooks.BatchAdmit(ctx, calls); !ok { - if reason == "" { - reason = "tool batch blocked by policy" - } - for i := range calls { - actions[i] = toolCallAction{run: false, reason: reason} - } - } - } - - // CallAdmit 钩子：对我们打算运行的调用逐个否决。 - if l.hooks.CallAdmit != nil { - for i := range calls { - if !actions[i].run { - continue - } - if veto := l.hooks.CallAdmit(ctx, calls[i]); veto != nil && veto.Block { - reason := veto.Reason - if reason == "" { - reason = "tool call blocked by policy" - } - actions[i] = toolCallAction{run: false, reason: reason} - } - } - } + // 执行前的三道门（安全门 + 批次/逐调用钩子）见 applyToolAdmissions。 + l.applyToolAdmissions(ctx, d, calls, actions) // 决定串行还是并行：显式串行模式，或任一调用的工具 // 被声明为串行（单个串行标记即让整组 @@ -187,26 +153,8 @@ func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls res.executed += len(idxs) } - // AfterCall 钩子（仅对实际执行过的调用）。 - if l.hooks.AfterCall != nil { - for i := range calls { - if outcomes[i].Skipped { - continue - } - if after := l.hooks.AfterCall(ctx, calls[i], outcomes[i]); after != nil { - outcomes[i] = *after - } - } - } - - // AfterOutcome 钩子（对每个最终结果运行，无论是否跳过）。 - if l.hooks.AfterOutcome != nil { - for i := range calls { - if after := l.hooks.AfterOutcome(ctx, calls[i], outcomes[i]); after != nil { - outcomes[i] = *after - } - } - } + // 结果落定后的两类观察/改写钩子见 applyToolOutcomeHooks。 + l.applyToolOutcomeHooks(ctx, calls, outcomes) // 钩子可以把每个调用都标记为 terminate 来要求终止本轮 // （OpenClaw 的 shouldTerminateToolBatch）。 @@ -277,48 +225,125 @@ func classifyToolBatch(st *toolLoopState, calls []agent.ToolCall) ([]toolCallAct return actions, terminate } +// applyToolAdmissions 对一批调用应用执行前的三道门，把不放行的 +// 调用改写为携带原因的跳过决策：① 安全门（design/11 §3）—— +// 群聊/陌生人会话只能调用其标记允许该环境的工具（默认关闭）， +// 不被允许的调用绝不执行，以 not_allowed 软结果回灌并记录审计； +// 它是该批的最终权威，后续钩子可以收紧但绝不重新放行。② +// BatchAdmit 批次钩子（审计 / 政治审查）拦截整批。③ CallAdmit +// 逐调用否决。dispatcher 缺席时安全门跳过（runToolCalls 已保证 +// 此时整批以错误回灌）。 +func (l *Loop) applyToolAdmissions(ctx context.Context, d *tool.Dispatcher, calls []agent.ToolCall, actions []toolCallAction) { + if d != nil { + l.applyToolSecurityGate(ctx, d, calls, actions) + } + if l.hooks.BatchAdmit != nil { + if ok, reason := l.hooks.BatchAdmit(ctx, calls); !ok { + if reason == "" { + reason = "tool batch blocked by policy" + } + for i := range calls { + actions[i] = toolCallAction{run: false, reason: reason} + } + } + } + if l.hooks.CallAdmit != nil { + for i := range calls { + if !actions[i].run { + continue + } + if veto := l.hooks.CallAdmit(ctx, calls[i]); veto != nil && veto.Block { + reason := veto.Reason + if reason == "" { + reason = "tool call blocked by policy" + } + actions[i] = toolCallAction{run: false, reason: reason} + } + } + } +} + +// applyToolOutcomeHooks 运行结果落定后的两类观察/改写钩子， +// 就地改写 outcomes：AfterCall 仅对实际执行过的调用（在 +// AfterOutcome 之前），AfterOutcome 对每个最终结果（无论是否 +// 跳过），都在结果回灌给模型之前运行。 +func (l *Loop) applyToolOutcomeHooks(ctx context.Context, calls []agent.ToolCall, outcomes []ToolOutcome) { + if l.hooks.AfterCall != nil { + for i := range calls { + if outcomes[i].Skipped { + continue + } + if after := l.hooks.AfterCall(ctx, calls[i], outcomes[i]); after != nil { + outcomes[i] = *after + } + } + } + if l.hooks.AfterOutcome != nil { + for i := range calls { + if after := l.hooks.AfterOutcome(ctx, calls[i], outcomes[i]); after != nil { + outcomes[i] = *after + } + } + } +} + // invokeOne 在其每工具截止时间下执行单个工具调用，发出一个 -// tool span、流式输出 tool_execution_update 进度并归一化 -// 结果。 +// tool span、以 tool_execution.start/end 事件流式输出执行进度 +// 并归一化结果。执行全程由单一 phase 决策驱动：start 帧带 +// running，Invoke 返回后按成败取 done/error，同一取值既决定 +// end 帧的 phase，也决定 span 终态（design/19 的 tool 节点 +// running/done/error 状态即由此驱动）。 func (l *Loop) invokeOne(ctx context.Context, d *tool.Dispatcher, tc agent.ToolCall) ToolOutcome { if d == nil { return ToolOutcome{Err: fmt.Errorf("no tool dispatcher configured for %q", tc.Name), ErrKind: tool.KindExecution} } - l.emitToolExecution(ctx, tc.Name, "start") _, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool) + l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionStart, model.ToolPhaseRunning) result, err := d.Invoke(ctx, tc.Name, tc.Args) - phase := "end" + phase := model.ToolPhaseDone if err != nil { - phase = "end:error" + phase = model.ToolPhaseError } - l.emitToolExecution(ctx, tc.Name, phase) + l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionEnd, phase) out := ToolOutcome{Result: result, Err: err, ErrKind: errorKind(err)} + l.spanProp(toolSpan, "phase", phase) l.recordToolDone(toolSpan, tc, out) if err != nil { l.spanProp(toolSpan, "error", err.Error()) l.spanProp(toolSpan, "error_kind", out.ErrKind) - l.finishSpan(toolSpan, trace.SpanStatusError) } else { l.spanProp(toolSpan, "result", result) - l.finishSpan(toolSpan, trace.SpanStatusOK) } + l.finishSpan(toolSpan, spanStatusForToolPhase(phase)) return out } // emitToolExecution 在循环处于流式路径（配置了 OnStream）时把 -// tool_execution_update 进度事件转发给前端；否则为空操作 -// （被静音的内部流程也会跳过它）。phase 是一个短标签， -// 如 "start" 或 "end:error"。 -func (l *Loop) emitToolExecution(ctx context.Context, name, phase string) { +// tool_execution 三事件（start / update / end）转发给前端；否则为 +// 空操作（被静音的内部流程也会跳过它）。phase 是结构化字段 +// （running/done/error），绝不塞进 Delta；工具名走 Tool 字段， +// ID 复用 provider 的 tool call id，使执行帧能与其 tool_call.* 帧 +// 关联。 +func (l *Loop) emitToolExecution(ctx context.Context, tc agent.ToolCall, kind model.StreamKind, phase string) { if l.onStream == nil || isMutedBroadcast(ctx) { return } - ev := model.StreamEvent{Kind: model.StreamKindToolExecution, Delta: phase + ":" + name} + ev := model.StreamEvent{Kind: kind, Phase: phase, Tool: tc.Name, ID: tc.ID} ev.SessionID = loopSession(ctx) ev.TurnID = loopTurn(ctx) l.onStream(ev) } +// spanStatusForToolPhase 把工具执行的终态 phase 映射为 span 终态， +// 使 trace/流程图与 WS 帧从同一个 phase 取值得出一致结论。 +// 非终态 phase（running）不会走到这里 —— span 只在执行结束时收尾。 +func spanStatusForToolPhase(phase string) trace.SpanStatus { + if phase == model.ToolPhaseError { + return trace.SpanStatusError + } + return trace.SpanStatusOK +} + // invokeParallel 并发运行 idxs 处的调用，上限为循环的 // 并行度，并把每个结果写回 out 的对应调用索引处，使调用方 // 能按原始顺序渲染结果。信号量限制并发，在途 diff --git a/internal/runtime/model/stream.go b/internal/runtime/model/stream.go index b95f96f..f80cbdb 100644 --- a/internal/runtime/model/stream.go +++ b/internal/runtime/model/stream.go @@ -33,11 +33,40 @@ const ( // StreamKindToolCallEnd closes a tool invocation block. StreamKindToolCallEnd StreamKind = "tool_call_end" - // StreamKindToolExecution reports intermediate progress while the loop runs - // a tool that the model requested (design/04 §1.1 tool_execution_update). It - // is emitted by the tool loop, not the model adapter; reserved for a later - // milestone, but the wire mapping is defined so the frontend can react. - StreamKindToolExecution StreamKind = "tool_execution_update" + // StreamKindToolExecutionStart opens the execution of a tool the model + // requested (design/04 §1.1: entering execution). It is emitted by the tool + // loop, not the model adapter; Phase carries the running label. + StreamKindToolExecutionStart StreamKind = "tool_execution_start" + // StreamKindToolExecutionUpdate reports intermediate progress or partial + // output while a requested tool runs (design/04 §1.1). Multiple frames per + // execution are allowed; the wire mapping is defined, and a producer appears + // once tools can report mid-execution progress. + StreamKindToolExecutionUpdate StreamKind = "tool_execution_update" + // StreamKindToolExecutionEnd closes a tool execution with its terminal + // phase (done or error). It is emitted by the tool loop after Invoke + // returns. + StreamKindToolExecutionEnd StreamKind = "tool_execution_end" + + // StreamKindTurnStart opens an assistant turn at the loop's run boundary + // (design/04 §1.1 turn.start). It carries no payload: the frame shell's + // session/turn identity brackets the turn on the client. + StreamKindTurnStart StreamKind = "turn_start" + // StreamKindTurnEnd closes an assistant turn at the loop's run boundary, + // however the run settled (final answer, guardrail stop, terminated tool + // loop, exhausted rounds or an error). + StreamKindTurnEnd StreamKind = "turn_end" +) + +// 工具执行 phase 标签（design/04 §1.1 细粒度 phase）。取值与 design/19 §1 +// 的节点 status 枚举（running/done/error）一致：start/update 帧带 running， +// end 帧带 done 或 error，前端与实时流程图据此驱动工具节点的状态。 +const ( + // ToolPhaseRunning 标记工具进入执行或仍在执行中。 + ToolPhaseRunning = "running" + // ToolPhaseDone 标记工具执行成功结束。 + ToolPhaseDone = "done" + // ToolPhaseError 标记工具执行失败结束。 + ToolPhaseError = "error" ) // StreamEvent is a single structured event in a streamed completion. It replaces @@ -49,14 +78,30 @@ type StreamEvent struct { Kind StreamKind // Delta carries the incremental payload: body text for TextDelta, reasoning // text for ThinkingDelta, or a partial_argument fragment for ToolCallDelta. + // Tool-execution events never pack their phase or tool name in here — + // those live in the structured Phase/Tool fields. Delta string // Index is the zero-based content-block index the event belongs to, when the // provider reports one. The frontend uses it to correlate a segment's // start/delta/end events. Index int // ID is the provider block/index identifier for a tool call (e.g. Anthropic's - // tool_use block id). Empty for text segments. + // tool_use block id). Empty for text segments. Tool-execution events reuse it + // so the client can correlate an execution with its tool_call.* frames. ID string + // Tool is the invoked tool's name on a tool_execution.* event. Empty on + // every other kind. + Tool string + // Phase is the fine-grained execution label (ToolPhase*) on a + // tool_execution.* event: running on start/update, done or error on end. + // Empty on every other kind. + Phase string + // StreamID identifies the single model stream (one ChatStream call) a frame + // belongs to. The loop stamps it at the stream boundary so every frame of + // the same streamed message shares it, while a multi-step turn's later + // model calls get fresh ids; consumers distinguish it from TurnID, which + // stays constant across the whole turn. + StreamID string // SessionID is the conversation the streamed event belongs to. It is stamped // by the loop at the turn boundary (loop.streamStep), which reads the session // the Run was started for; a model adapter, running below the loop and with no diff --git a/internal/runtime/trace/graph.go b/internal/runtime/trace/graph.go index 45cf482..e000ec3 100644 --- a/internal/runtime/trace/graph.go +++ b/internal/runtime/trace/graph.go @@ -23,6 +23,24 @@ const ( FlowStatusError = "error" ) +// FlowStatusForPhase 把工具执行的细粒度 phase（design/04 §1.1， +// loop 随 tool_execution.end 帧一同记在 tool span 上）映射为流程图 +// 节点状态：phase 取值与本包状态枚举（design/19 §1 统一枚举） +// 同构（running/done/error）。未知 phase 返回空串，调用方回落到 +// span 终态推导，因此缺 phase 的旧 span 不受影响。 +func FlowStatusForPhase(phase string) string { + switch phase { + case FlowStatusRunning: + return FlowStatusRunning + case FlowStatusDone: + return FlowStatusDone + case FlowStatusError: + return FlowStatusError + default: + return "" + } +} + // FlowNode is one node in a flow graph. Status/SpanKind mirror the frontier. type FlowNode struct { ID string `json:"id"` diff --git a/internal/runtime/trace/graph_build.go b/internal/runtime/trace/graph_build.go index c7c3ad1..50b0deb 100644 --- a/internal/runtime/trace/graph_build.go +++ b/internal/runtime/trace/graph_build.go @@ -143,17 +143,26 @@ func (g *RunGraph) addEdge(from, to string) { // AddSpan folds a finished span record into the graph as a node with its final // status. State-transition spans are the step container and are omitted — but // their parent link is remembered so lane resolution can walk through them. +// tool 节点的终态优先由其携带的细粒度 phase 驱动（design/04 §1.1 / +// design/19 §1：running/done/error），缺失时回落到 span 终态推导， +// 因此两套来源对同一节点绝不可能得出不同状态。 func (g *RunGraph) AddSpan(rec SpanRecord) { kind := kindForSpan(rec.Kind) if kind == "" { g.rememberParent(rec.ID, rec.ParentID) return } + status := statusForFinished(rec.Status) + if kind == nodeKindTool { + if ph := FlowStatusForPhase(spanPhase(rec)); ph != "" { + status = ph + } + } g.Upsert(NodeSpec{ ID: rec.ID, Label: labelForSpan(rec), Kind: kind, - Status: statusForFinished(rec.Status), + Status: status, StartedAt: rec.StartedAt, EndedAt: rec.EndedAt, Props: safeSpanProps(rec), @@ -162,6 +171,13 @@ func (g *RunGraph) AddSpan(rec SpanRecord) { }) } +// spanPhase 读取 tool span 上记录的执行 phase；缺失或类型不符时返回 +// 空串（调用方据此回落），不区分「没有」与「读不到」。 +func spanPhase(rec SpanRecord) string { + ph, _ := rec.Props["phase"].(string) + return ph +} + // AddSpanStart registers a running placeholder for a span that just began, so a // node appears as "running" before it finishes. State-transition spans are // omitted (their parent link is still remembered for lane resolution), matching diff --git a/internal/runtime/trace/graph_phase_test.go b/internal/runtime/trace/graph_phase_test.go new file mode 100644 index 0000000..4b22b6f --- /dev/null +++ b/internal/runtime/trace/graph_phase_test.go @@ -0,0 +1,80 @@ +package trace + +import ( + "testing" + "time" +) + +// TestFlowStatusForPhase 锁住 phase → 节点状态的映射（design/04 §1.1 +// 细粒度 phase、design/19 §1 统一枚举）：三个已知 phase 原样通过， +// 未知 phase 返回空串让调用方回落到 span 终态推导。 +func TestFlowStatusForPhase(t *testing.T) { + t.Parallel() + cases := map[string]string{ + FlowStatusRunning: FlowStatusRunning, + FlowStatusDone: FlowStatusDone, + FlowStatusError: FlowStatusError, + "started": "", + "": "", + } + for phase, want := range cases { + if got := FlowStatusForPhase(phase); got != want { + t.Fatalf("FlowStatusForPhase(%q) = %q, want %q", phase, got, want) + } + } +} + +// TestAddSpanToolPhaseDrivesStatus 锁住裁决 03.4.4：tool 节点的终态由 +// span 携带的细粒度 phase 驱动（与 WS 的 tool_execution.end 帧同一 +// 取值），缺 phase 的 tool span 回落到 span 终态推导，两路绝不矛盾。 +func TestAddSpanToolPhaseDrivesStatus(t *testing.T) { + t.Parallel() + now := time.Now() + tool := func(id, phase string, status SpanStatus) SpanRecord { + rec := SpanRecord{ID: id, Name: "tool.echo", Kind: SpanKindTool, + Status: status, StartedAt: now, EndedAt: now} + if phase != "" { + rec.Props = map[string]any{"phase": phase} + } + return rec + } + cases := []struct { + name string + rec SpanRecord + want string + }{ + {"phase done", tool("t1", FlowStatusDone, SpanStatusOK), FlowStatusDone}, + {"phase error overrides ok status", tool("t2", FlowStatusError, SpanStatusOK), FlowStatusError}, + {"no phase falls back to span status", tool("t3", "", SpanStatusOK), FlowStatusDone}, + {"no phase falls back to span error", tool("t4", "", SpanStatusError), FlowStatusError}, + {"unknown phase falls back to span status", tool("t5", "started", SpanStatusError), FlowStatusError}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + g := NewRunGraph("web") + g.Start(1, now) + g.AddSpan(tc.rec) + if len(g.nodes) != 1 { + t.Fatalf("nodes = %d, want 1", len(g.nodes)) + } + if got := g.nodes[tc.rec.ID].Status; got != tc.want { + t.Fatalf("tool node status = %q, want %q", got, tc.want) + } + }) + } +} + +// TestAddSpanNonToolIgnoresPhase 确保 phase 只驱动 tool 节点：其他 +// kind 的 span 即使带 phase prop（未来不该有），也不改变其 +// 终态推导路径。 +func TestAddSpanNonToolIgnoresPhase(t *testing.T) { + t.Parallel() + now := time.Now() + g := NewRunGraph("web") + g.Start(1, now) + g.AddSpan(SpanRecord{ID: "l1", Name: "llm.call", Kind: SpanKindLLM, + Status: SpanStatusOK, StartedAt: now, EndedAt: now, Props: map[string]any{"phase": FlowStatusError}}) + if got := g.nodes["l1"].Status; got != FlowStatusDone { + t.Fatalf("llm node status = %q, want %q (phase must not drive non-tool nodes)", got, FlowStatusDone) + } +} diff --git a/internal/services/gateway/ws_bridge.go b/internal/services/gateway/ws_bridge.go index 5414e7f..711fa02 100644 --- a/internal/services/gateway/ws_bridge.go +++ b/internal/services/gateway/ws_bridge.go @@ -83,35 +83,71 @@ func subagentProgressEvent(e bus.Event) (wsEvent, bool) { // streamEventData 把流式的 model.StreamKind 映射为其下游 WS 事件 // 类型并构建带类型的 data 载荷。它正是把 loop 的结构化 // 事件变成前端所渲染的 message.*/thinking.*/tool_call.*/tool_execution.* -// 帧的东西。未知类型回退到旧的 stream_chunk 形状，使 +// 与 turn.start/end 帧的东西。模型流帧（message.*/thinking.*/tool_call.*） +// 额外携带 streamId（同一模型流的多帧关联键，design/04 §1.1）； +// tool_execution.* 与 turn.* 不属于某一次模型流，不带 streamId。 +// 未知类型回退到旧的 stream_chunk 形状，使 // 未识别的事件仍能抵达客户端而不是被丢弃。 func streamEventData(ev model.StreamEvent) (string, map[string]any) { switch ev.Kind { case model.StreamKindTextStart: - return "message.start", map[string]any{"index": ev.Index} + return "message.start", withStreamID(map[string]any{"index": ev.Index}, ev) case model.StreamKindTextDelta: - return "message.delta", map[string]any{"delta": ev.Delta, "index": ev.Index} + return "message.delta", withStreamID(map[string]any{"delta": ev.Delta, "index": ev.Index}, ev) case model.StreamKindTextEnd: - return "message.end", map[string]any{"index": ev.Index} + return "message.end", withStreamID(map[string]any{"index": ev.Index}, ev) case model.StreamKindThinkingStart: - return "thinking.start", map[string]any{"index": ev.Index} + return "thinking.start", withStreamID(map[string]any{"index": ev.Index}, ev) case model.StreamKindThinkingDelta: - return "thinking.delta", map[string]any{"delta": ev.Delta, "index": ev.Index} + return "thinking.delta", withStreamID(map[string]any{"delta": ev.Delta, "index": ev.Index}, ev) case model.StreamKindThinkingEnd: - return "thinking.end", map[string]any{"index": ev.Index} + return "thinking.end", withStreamID(map[string]any{"index": ev.Index}, ev) case model.StreamKindToolCallStart: - return "tool_call.start", map[string]any{"id": ev.ID, "index": ev.Index} + return "tool_call.start", withStreamID(map[string]any{"id": ev.ID, "index": ev.Index}, ev) case model.StreamKindToolCallDelta: - return "tool_call.delta", map[string]any{"id": ev.ID, "index": ev.Index, "delta": ev.Delta} + return "tool_call.delta", withStreamID(map[string]any{"id": ev.ID, "index": ev.Index, "delta": ev.Delta}, ev) case model.StreamKindToolCallEnd: - return "tool_call.end", map[string]any{"id": ev.ID, "index": ev.Index} - case model.StreamKindToolExecution: - return "tool_execution.update", map[string]any{"id": ev.ID, "index": ev.Index, "delta": ev.Delta} + return "tool_call.end", withStreamID(map[string]any{"id": ev.ID, "index": ev.Index}, ev) + case model.StreamKindToolExecutionStart: + return "tool_execution.start", toolExecutionData(ev) + case model.StreamKindToolExecutionUpdate: + return "tool_execution.update", toolExecutionData(ev) + case model.StreamKindToolExecutionEnd: + return "tool_execution.end", toolExecutionData(ev) + case model.StreamKindTurnStart: + return "turn.start", nil + case model.StreamKindTurnEnd: + return "turn.end", nil default: return "stream_chunk", map[string]any{"delta": ev.Delta} } } +// withStreamID 把事件携带的 streamId 并入模型流帧的 data（空值省略， +// 使未带流身份的事件保持旧形状）。它原地修改并返回 data， +// 仅供 streamEventData 的字面量使用。 +func withStreamID(data map[string]any, ev model.StreamEvent) map[string]any { + if ev.StreamID != "" { + data["streamId"] = ev.StreamID + } + return data +} + +// toolExecutionData 构建 tool_execution.* 帧的载荷：工具名与细粒度 +// phase 为结构化字段（phase 绝不塞进 delta，design/04 §1.1 裁决 +// 01.4.1）；id 复用 provider 的 tool call id（可关联 tool_call.* 帧）， +// update 帧可再携带 delta 局部输出。键在空值时省略。 +func toolExecutionData(ev model.StreamEvent) map[string]any { + data := map[string]any{"tool": ev.Tool, "phase": ev.Phase} + if ev.ID != "" { + data["id"] = ev.ID + } + if ev.Kind == model.StreamKindToolExecutionUpdate && ev.Delta != "" { + data["delta"] = ev.Delta + } + return data +} + // isBroadcastAgentEvent 报告 subject 是否恰为 4 段广播形态 // cyrene.agent..event，调度器正是这样发布进度的。 // 它刻意不匹配寻址事件（cyrene.agent...event） diff --git a/internal/services/gateway/ws_frame_scope_test.go b/internal/services/gateway/ws_frame_scope_test.go index 7889f70..e168437 100644 --- a/internal/services/gateway/ws_frame_scope_test.go +++ b/internal/services/gateway/ws_frame_scope_test.go @@ -81,7 +81,11 @@ func frameScopeCases() []frameScopeCase { streamed(model.StreamKindToolCallStart, "tool_call.start"), streamed(model.StreamKindToolCallDelta, "tool_call.delta"), streamed(model.StreamKindToolCallEnd, "tool_call.end"), - streamed(model.StreamKindToolExecution, "tool_execution.update"), + streamed(model.StreamKindToolExecutionStart, "tool_execution.start"), + streamed(model.StreamKindToolExecutionUpdate, "tool_execution.update"), + streamed(model.StreamKindToolExecutionEnd, "tool_execution.end"), + streamed(model.StreamKindTurnStart, "turn.start"), + streamed(model.StreamKindTurnEnd, "turn.end"), relayed("stream_chunk", false), relayed("stream_end", false), relayed("thinking", false), diff --git a/internal/services/gateway/ws_stream_events_test.go b/internal/services/gateway/ws_stream_events_test.go new file mode 100644 index 0000000..a8dcca4 --- /dev/null +++ b/internal/services/gateway/ws_stream_events_test.go @@ -0,0 +1,288 @@ +package gateway + +import ( + "context" + "encoding/json" + "sort" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" +) + +// wsEventsAdapter 是一个两步脚本适配器：第一步发出一个文本增量后 +// 返回原生工具调用，第二步再发一个增量并收尾 —— 恰好覆盖 +// v0.50 事件批的全部帧形态（turn 边界、两次模型流的 streamId、 +// tool_execution 三事件中的 start/end）。 +type wsEventsAdapter struct{ calls int } + +func (a *wsEventsAdapter) Name() string { return "ws-events-test" } + +func (a *wsEventsAdapter) Chat(context.Context, model.Request, model.Options) (model.Response, error) { + return model.Response{Content: "done"}, nil +} + +func (a *wsEventsAdapter) ChatStream(_ context.Context, _ model.Request, on func(model.StreamEvent)) (model.Response, error) { + a.calls++ + if a.calls == 1 { + if on != nil { + on(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "Hel", Index: 0}) + } + return model.Response{Model: "m", Content: "", ToolCalls: []model.ToolCall{{ + ID: "call_ws_1", Name: "datetime", Args: map[string]any{}, + }}}, nil + } + if on != nil { + on(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "lo", Index: 0}) + } + return model.Response{Model: "m", Content: "Hello"}, nil +} + +// wsEventsTool 是注册进真实 dispatcher 的最小工具，使 runTurn 走过 +// 完整的工具执行路径（含 tool_execution.start/end 与 phase）。 +type wsEventsTool struct{} + +func (wsEventsTool) Name() string { return "datetime" } +func (wsEventsTool) Description() string { return "test datetime tool" } +func (wsEventsTool) Parameters() (map[string]any, error) { return map[string]any{}, nil } +func (wsEventsTool) Execute(context.Context, any) (any, error) { + return map[string]any{"now": "2026-10-04 12:00:00"}, nil +} + +// newEventsStreamHub 把 hub 接到带工具的真实 loop，OnStream 闭包与 +// cmd/cyrene 的接线一致（整个 StreamEvent 发布到 ws.stream），因此 +// 测试驱动的是生产路径 OnStream → bus → relay → broadcastStream。 +func newEventsStreamHub(t *testing.T) (*hub, *wsConn) { + t.Helper() + b := bus.New() + reg := registry.New() + td, err := tool.New(reg) + if err != nil { + t.Fatalf("tool.New: %v", err) + } + if err := td.Register(wsEventsTool{}); err != nil { + t.Fatalf("register tool: %v", err) + } + ag, err := agent.New("You are a helpful assistant.", td, nil, agent.WithModel(&wsEventsAdapter{}, nil)) + if err != nil { + t.Fatalf("agent.New: %v", err) + } + lp := loop.New(loop.Options{ + Agent: ag, + MaxRounds: 8, + OnStream: func(ev model.StreamEvent) { + _ = b.Publish(context.Background(), wsStreamSubject, ev) + }, + }) + hb := newTestHub(lp, b) + hb.streaming = true + hb.start() + t.Cleanup(hb.close) + c := newOwnerConn(hb, 64) + hb.register(c) + return hb, c +} + +// collectFramesUntilIdle 收集帧直到凑满 want 帧（含 stream_end）后再 +// 静默 200ms 确认没有迟到的 bus 帧。loop 帧经 bus 异步抵达而 +// stream_end 由 runTurn 直接广播，因此不能假定两者相对顺序。 +func collectFramesUntilIdle(t *testing.T, c *wsConn, want int) []wsEvent { + t.Helper() + var frames []wsEvent + decode := func(b []byte) wsEvent { + var ev wsEvent + if err := json.Unmarshal(b, &ev); err != nil { + t.Fatalf("decode frame %s: %v", b, err) + } + return ev + } + deadline := time.After(5 * time.Second) + for { + if len(frames) >= want { + select { + case b := <-c.send: + frames = append(frames, decode(b)) + case <-time.After(200 * time.Millisecond): + return frames + case <-deadline: + return frames + } + continue + } + select { + case b := <-c.send: + frames = append(frames, decode(b)) + case <-deadline: + t.Fatalf("timed out with %d frames, want at least %d: %+v", len(frames), want, frames) + } + } +} + +// runEventsTurn 驱动一个完整的工具回合并返回按 seq 排序的帧， +// 以及其中的 stream_end（loop 帧经 bus 异步抵达而 stream_end 由 +// runTurn 直接广播，因此不能假定两者相对顺序，排序后统一断言）。 +func runEventsTurn(t *testing.T) ([]wsEvent, *wsEvent) { + t.Helper() + hb, c := newEventsStreamHub(t) + if _, err := hb.runTurn(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "time?")}, loop.OwnerEnv()); err != nil { + t.Fatalf("runTurn: %v", err) + } + frames := collectFramesUntilIdle(t, c, 7) + sort.SliceStable(frames, func(i, j int) bool { return frames[i].Seq < frames[j].Seq }) + var streamEnd *wsEvent + for i := range frames { + if frames[i].Type == "stream_end" && streamEnd == nil { + streamEnd = &frames[i] + } + } + return frames, streamEnd +} + +// eventsByType 把 loop 帧（排除 stream_end）按类型归组，同时校验 +// 每帧都带会话与 turn-* 作用域。 +func eventsByType(t *testing.T, frames []wsEvent) map[string][]wsEvent { + t.Helper() + byType := map[string][]wsEvent{} + for _, f := range frames { + if f.Type == "stream_end" { + continue + } + byType[f.Type] = append(byType[f.Type], f) + if f.SessionID != "web" || !strings.HasPrefix(f.TurnID, "turn-") { + t.Fatalf("frame %s scope = session %q turn %q, want web + turn-*", f.Type, f.SessionID, f.TurnID) + } + } + return byType +} + +// TestStreamFramesTurnSequenceAndBoundaries 是 v0.50 回合边界与帧序的 +// 端到端锁定（裁决 01.4.3）：经真实 loop + bus + hub 驱动的一个工具 +// 回合，loop 帧必须按序发出 turn.start → message.delta → +// tool_execution.start → tool_execution.end → message.delta → +// turn.end；turn 边界帧无载荷、整轮 turnId 一致、stream_end 照常收尾。 +func TestStreamFramesTurnSequenceAndBoundaries(t *testing.T) { + t.Parallel() + frames, streamEnd := runEventsTurn(t) + + var seq []string + for _, f := range frames { + if f.Type != "stream_end" { + seq = append(seq, f.Type) + } + } + want := []string{ + "turn.start", "message.delta", + "tool_execution.start", "tool_execution.end", + "message.delta", "turn.end", + } + if len(seq) != len(want) { + t.Fatalf("loop frame types = %v, want %v (frames=%+v)", seq, want, frames) + } + for i := range want { + if seq[i] != want[i] { + t.Fatalf("loop frame[%d] = %q, want %q (all=%v)", i, seq[i], want[i], seq) + } + } + if streamEnd == nil { + t.Fatalf("no stream_end frame arrived (frames=%+v)", frames) + } + + byType := eventsByType(t, frames) + turnID := byType["turn.start"][0].TurnID + if byType["turn.end"][0].TurnID != turnID || byType["message.delta"][0].TurnID != turnID { + t.Fatalf("turnId differs across the turn: %q / %q / %q", turnID, byType["turn.end"][0].TurnID, byType["message.delta"][0].TurnID) + } + for _, boundary := range []wsEvent{byType["turn.start"][0], byType["turn.end"][0]} { + if boundary.Data != nil { + t.Fatalf("%s carries data %+v; the boundary frames must be payload-free", boundary.Type, boundary.Data) + } + } +} + +// TestStreamFramesStreamIdAndToolExecutionPayloads 是 v0.50 streamId +// 与工具执行帧载荷的端到端锁定（裁决 01.4.1 / 01.4.2）：模型流帧带 +// 每流独立的 streamId（同流共享、跨流相异）；tool_execution 帧带 +// 结构化 tool/phase/id，且不带 delta 拼串、不带 streamId。 +func TestStreamFramesStreamIdAndToolExecutionPayloads(t *testing.T) { + t.Parallel() + frames, _ := runEventsTurn(t) + byType := eventsByType(t, frames) + + d1 := dataMap(t, byType["message.delta"][0]) + d2 := dataMap(t, byType["message.delta"][1]) + if d1["delta"] != "Hel" || d2["delta"] != "lo" { + t.Fatalf("deltas = %v / %v, want Hel / lo", d1["delta"], d2["delta"]) + } + s1, _ := d1["streamId"].(string) + s2, _ := d2["streamId"].(string) + if !strings.HasPrefix(s1, "stream-") || !strings.HasPrefix(s2, "stream-") { + t.Fatalf("streamIds = %q / %q, want two stream-* ids (one per model stream)", s1, s2) + } + if s1 == s2 { + t.Fatalf("the tool turn's two model streams share streamId %q; each ChatStream call must mint a fresh one", s1) + } + + startData := dataMap(t, byType["tool_execution.start"][0]) + endData := dataMap(t, byType["tool_execution.end"][0]) + if startData["tool"] != "datetime" || endData["tool"] != "datetime" { + t.Fatalf("tool_execution tool = %v / %v, want datetime", startData["tool"], endData["tool"]) + } + if startData["phase"] != model.ToolPhaseRunning || endData["phase"] != model.ToolPhaseDone { + t.Fatalf("tool_execution phases = %v / %v, want running / done", startData["phase"], endData["phase"]) + } + if startData["id"] != "call_ws_1" || endData["id"] != "call_ws_1" { + t.Fatalf("tool_execution ids = %v / %v, want the provider call id", startData["id"], endData["id"]) + } + for typ, data := range map[string]map[string]any{"start": startData, "end": endData} { + if _, ok := data["delta"]; ok { + t.Fatalf("tool_execution.%s carries a delta key; phase must not ride in delta: %v", typ, data) + } + if _, ok := data["streamId"]; ok { + t.Fatalf("tool_execution.%s carries a streamId; tool execution is not a model stream: %v", typ, data) + } + } +} + +// dataMap 取帧 data 的映射形态，形状不符时直接失败。 +func dataMap(t *testing.T, ev wsEvent) map[string]any { + t.Helper() + data, ok := ev.Data.(map[string]any) + if !ok { + t.Fatalf("frame %s data = %+v, want a map", ev.Type, ev.Data) + } + return data +} + +// TestStreamFramesToolExecutionErrorPhase 是失败路径的帧形状锁定： +// 直接经生产映射路径（broadcastStream，即 relay 消费 ws.stream 的 +// 同一入口）广播一个 error phase 的 end 事件，验证帧携带 +// phase=error（裁决 03.4.4：前端/流程图据此把 tool 节点翻成 +// error 态）。真实的失败产生路径由 loop 包的 +// TestRun_ToolExecutionErrorPhase 端到端覆盖。 +func TestStreamFramesToolExecutionErrorPhase(t *testing.T) { + t.Parallel() + hb, c := newEventsStreamHub(t) + + hb.broadcastStream(model.StreamEvent{ + Kind: model.StreamKindToolExecutionEnd, + Tool: "datetime", + Phase: model.ToolPhaseError, + ID: "call_ws_1", + SessionID: "web", + TurnID: "turn-err", + }) + ev := readFakeFrame(t, c) + if ev.Type != "tool_execution.end" { + t.Fatalf("frame type = %q, want tool_execution.end (frame=%+v)", ev.Type, ev) + } + data := dataMap(t, ev) + if data["phase"] != model.ToolPhaseError || data["tool"] != "datetime" { + t.Fatalf("frame data = %+v, want phase=error tool=datetime", data) + } +}