diff --git a/Docs/api-client.md b/Docs/api-client.md index e75fcdc..6a5b251 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -144,12 +144,12 @@ ## 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）新增**，能确定会话的事件带上，供客户端按激活会话过滤）——`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）。**会话编排状态（design/15，均为广播，`data.sessionId` 区分会话）**：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（见 §3.2）。语音回复走 `voice_output`（定向回请求连接，非广播）。 +- **下行（服务端→客户端）**: 经 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）。**会话编排状态（design/15，均为广播，`data.sessionId` 区分会话）**：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（见 §3.2）。语音回复走 `voice_output`（定向回请求连接，非广播）。 > **`history` 回放事件（design/16，新增）**：WS 握手升级成功后、该连接下一帧即收到 `{type:"history", sessionId:"web", data:{messages:[{id,sessionId,role,content,mode?,mood?,image?,platform?,deviceId?,createdAt}]}}`——回放主会话最新 50 条（`thinking` 已剥离，不泄给非 owner；默认主会话，次级会话由客户端按激活会话走 `GET /sessions/{id}/messages` 拉取）。客户端据此在重连/新开连接后补全上下文。消息体含 `senderName?`（第三方昵称，design/16 §4.6，主人私聊为空）。 ### 3.1 结构化流事件（`message.*` / `thinking.*` / `tool_call.*` / `tool_execution.*`） -- **事件壳**：`{type, turnId?, seq?, data}`, `data` 为**类型化结构**（非 `any`）。`seq` 为 hub 级单调递增序号，供客户端将一次突发的流式事件排序关联。 +- **事件壳**：`{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`）。 @@ -253,6 +253,7 @@ - 语音：上行 `voice_input`（base64 音频或已转写 text），下行 `voice_output`（合成音频 base64 + text）。 ## 5. 变更记录 +- `v0.36`（**A1 流式帧附加会话 / 回合标识**，对客户端**新增两个可选字段、无端点变更**）：结构化流帧 `message.start/delta/end`、`thinking.start/delta/end`、`tool_call.start/delta/end`、`tool_execution.update` 现在带 `sessionId` + `turnId`（**向后兼容**：只新增帧字段，`data` 结构与既有事件形状不变）。`sessionId` 由 loop 在**回合边界**从运行上下文注入（模型适配器在 loop 之下、天然不感知会话，保持零值；loop 外无会话上下文时该字段为空）；`turnId` 由网关**每回合**新生成（`newTurnID`）并经 `loop.WithTurnID` 挂进运行上下文，`runTurn` 是编排层与 legacy 直连路径**唯一**的回合收口点，两路都覆盖——同一会话同时只有一个 in-flight 回合，故同轮内所有帧的 `turnId` 一致。此前这几帧属于 design/16 §4.2 标注的「若知道就带上」缺口（v0.10 起记录）：**多会话并发流式时前端会把别的会话的 delta 接进当前会话气泡**，现在客户端按 `sessionId === activeSessionId` 过滤即可路由到激活会话。`stream_chunk` / `stream_end` 的 `sessionId`/`turnId` 行为不变。详见 §3.1。 - `v0.35`（**可观测性收尾：后端日志 tail 端点**，对客户端**新增一个 admin 只读端点**）：`GET /api/v1/admin/observability/logs?level=&limit=`（owner/admin）→ `200 {data:[{time,level,source,msg}]}`，**最新在前**。数据源为 logger 新增的**有界环形缓冲**（`internal/logger`，容量 2000，满则丢最旧、并发安全、不阻塞、不改动既有控制台/文件输出），组合根在构建 logger 时挂上、在网关就绪后经 `SetLogBuffer` 注入；缓冲未注入回 `501 not_implemented`（不伪造空 tail）。`time` 为 **RFC3339 UTC（秒精度）**，与 `trace.startedAt`/`audit.when` 一致；`level ∈ debug|info|warn|error`（大小写不敏感）为**最低等级过滤**，空 = 全部，未知值 `400 invalid_level`；`limit` 缺省 `200`、上限 `1000`。`source` 取 `logger.WithSource` 的 `source=`（`Scoped` 记录回落 `module/sub`），无来源为空串。详见 §2「可观测性契约」。 - `v0.34`（**D3 收尾：消息编辑 / 删除的 WS 实时广播**，对客户端**新增两个下行事件、无 REST 契约变更**）：`PATCH /api/v1/messages/{id}` 成功后广播 **`message_edited`**（`{type, sessionId, data:<更新后的完整消息DTO>}`，`thinking` 已剥离）、`DELETE /api/v1/messages/{id}` 成功后广播 **`message_deleted`**（`{type, sessionId, data:{id}}`）；两者复用 `message_persisted` 的同一 hub 广播路径（**全连接广播 + 客户端按 `sessionId` 过滤**，hub 无按会话订阅），客户端**按 `id` 原位改 `content` / 按 `id` 移除**，不再需要「改/删后重拉」的绕行。`message_deleted` 的 `sessionId` 由后端在删除前读取消息所属会话填入（`HistoryStore` 新增 `GetMessage`；读不到时客户端按全局唯一的 `id` 兜底）。`404`/`400`/`501` 等失败路径**不广播**。Web 端已消费（`lib/ws.ts` + `store/useAppStore` 新增 `updateMessageById`/`removeMessageById`）。详见 §3.8。 - `v0.33`（**A2 记忆异步提取 / A4 可观测性 Tracing 接线 / A6 通知页后端**）：① **A6 新增 `GET /api/v1/admin/notifications`**（owner/admin 读，**全新端点**）：把**现有源**（审计 `audit_logs` + backthink 主动推送记录）聚合为**统一只读通知 feed**，`type`/`before`/`limit` 过滤与游标见 §2「通知聚合契约」；**不新建表**，未注入的源该类型为空、不伪造；未读游标后置。② **A4 trace 端点由恒空变真实数据**（**无契约变更**，`GET /admin/observability/trace` 与 `metrics` 的 `llmCalls`/`subagentEvents` 从此反映 loop/工具/编排的真实 span）：此前 `TraceSink` 从未接线、环形缓冲无人喂；现在组合根把 loop 产生的 span 扇出到网关观测缓冲。响应字段/形状不变。③ **A2 LLM 异步记忆提取接线**（**对客户端无变更**）：loop 回合结束后改由 LLM 提取器**后台异步**抽取记忆（不阻塞回复、失败仅记日志），**群聊/陌生人仍只读不写**；无可用模型凭据时回落确定性 `ManualExtractor`（`记住 X` 路径不变）。 @@ -279,7 +280,7 @@ - `v0.13`（每条消息带 deviceID/client 标识，design/16 §4.5）：**`chat_messages` 增 `device_id` 列**（迁移 0005，默认 `''`）；`POST /api/v1/chat` 请求体新增**可选 `deviceId`**（客户端生成的端标识，如 `web:chrome:a1b2c3`）；消息 DTO（REST 消息、WS `history` 回放、WS `message_persisted`）均带 **`deviceId`**。`platform`（平台）与 `deviceId`（具体端）互补：同一平台多个端 → 相同 `platform`、不同 `deviceId`。**Web 端**由 `lib/client.ts` 用 `localStorage` 持久化每个浏览器一个的 `web::<随机短id>`，发送时自动带上，聊天气泡默认小字标注来源（仅当 `deviceId` 存在才显示）；**IM 端**按 `:<本地配置唯一id>:<用户平台唯一uid>` 构造（当前 `UnifiedMessage` 尚无 deviceId 概念，channelbridge 以 `::` 占位，`<本地配置唯一id>` 待适配器提供后替换）；Web 端聊天不回流 IM。 - `v0.12`（IM 渠道并入主会话 + 平台标记，design/16 §4.4）：**`chat_messages` 增 `platform` 列**（迁移 0004，默认 `'web'`）；`GET /api/v1/sessions/{id}/messages` 支持可选 **`?platform=`** 过滤（Web 端不传=全量共享主会话；IM 适配器传本平台只拉本渠道）；消息 DTO（REST 与 WS `message_persisted`）均带 `platform`。**IM 渠道主人私聊**（`!IsGroup && Owner`）经 channelbridge 落**主会话 `web`**（`platform=<渠道>`）并广播 `message_persisted` → Web 端免刷新镜像；**群聊/陌生人**保持独立隔离上下文、不落主会话。Web 端聊天（`platform=web`）不回流 IM。 - `v0.11`（跨端实时同步，design/16 §4.2）：WS 下行新增 **`message_persisted`** 事件——每条 user/assistant 消息落库后广播 `{type, sessionId, data:{message:{id,sessionId,role,content,mode?,mood?,image?,createdAt}}}`（`id` 为后端生成，`thinking` 已剥离）。客户端按 `sessionId === activeSessionId` 过滤 + 按 `id` upsert（含本端乐观气泡 `uid`→后端 `id` 归一）去重，实现「A 发消息、A 与 B 都无需刷新且各出现一次」。后端 `history.Store` 新增 `AppendFull` 供编排层取含 `id` 的落库消息。 -- `v0.10`（会话消息持久化 — WS 契约，design/16，后端-2）：WS 下行帧新增可选 `sessionId` 字段（能确定会话的事件带上）；新增 **`history` 回放事件**——WS 握手成功即回放主会话最新 50 条（`{type:"history", sessionId:"web", data:{messages:[...]}}`，`thinking` 已剥离）。`stream_end`/`stream_chunk`/会话编排状态帧（`session_busy` 等）在能确定 session 时带 `sessionId`；`message.delta` 等结构化流帧因 `model.StreamEvent` 无 session 上下文不带（即"若知道就带上"边界）。 +- `v0.10`（会话消息持久化 — WS 契约，design/16，后端-2）：WS 下行帧新增可选 `sessionId` 字段（能确定会话的事件带上）；新增 **`history` 回放事件**——WS 握手成功即回放主会话最新 50 条（`{type:"history", sessionId:"web", data:{messages:[...]}}`，`thinking` 已剥离）。`stream_end`/`stream_chunk`/会话编排状态帧（`session_busy` 等）在能确定 session 时带 `sessionId`；`message.delta` 等结构化流帧当时因 `model.StreamEvent` 无 session 上下文**不带**（即"若知道就带上"边界）——**该缺口已在 v0.36 补齐**（结构化流帧现在带 `sessionId` + `turnId`，见 §3.1）。 - `v0.9`（会话消息持久化 + 主/次级会话，design/16，后端-1）：新增 `POST /api/v1/sessions`（新建次级会话）、`GET /api/v1/sessions/{id}/messages?limit=&before=`（游标热分页拉历史，时间倒序）、`DELETE /api/v1/sessions/{id}`（删次级会话及其消息）；`GET /api/v1/sessions` 在注入 `HistoryStore` 后改返回持久化会话元数据数组（主 + 次级，`type` 区分），未注入时保持原渠道会话列表向后兼容。读消息/写会话为 owner/admin（用户侧私有历史，thinking 不达 viewer）；会话元数据列表保持任意认证角色可读。 - `v0.8`（管理面板写操作 CRUD）：补齐管理面板**写端点**（见 §2「管理面板写操作契约」，均 owner-only + 审计）——`POST/PUT /admin/personas`（真实注册 + 持久化 `configs/personas/.yaml`）、`POST /admin/users/apikey` + `DELETE /admin/users/apikey/{id}`（真实签发/撤销）、`POST /admin/iot/{id}/control`（真实设备控制）、`POST /admin/automations` + `/run` + `DELETE`（真实注册/触发/移除规则）、`POST /admin/llm`（持久化 `models.json` + 热更新，v0.15 起）+ `/admin/llm/test`（真实模型调用）、`POST /admin/channels/{open,close,test}`（真实连接/断开 + 结构性身份识别）、`POST /admin/plugins/uninstall`（真实 registry 移除）、`POST /admin/memories/search`（真实检索）、`GET /admin/sessions/{id}/export`。无存储的写（新增用户 / IoT 场景·规则 / 插件安装）如实返回 `501 not_implemented`。 - `v0.7`（会话消息编排 design/15）：`POST /api/v1/chat` 非流式且会话「正忙」时改回 `202 {data:{queued:true,queueDepth}}`（原同步 `{data:{reply}}` 仍为空闲时返回）；pending 满回 `409 chat_busy`。WS 上行新增 `interrupt`（`{sessionId}`，中断当前任务+清 pending）；WS 下行新增会话编排状态事件 `session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（见 §3.2）。既有 `stream_chunk`/`stream_end` 与结构化流事件不变，契约向后兼容。 diff --git a/internal/runtime/loop/session_ctx.go b/internal/runtime/loop/session_ctx.go index eecbf0c..5d11191 100644 --- a/internal/runtime/loop/session_ctx.go +++ b/internal/runtime/loop/session_ctx.go @@ -19,3 +19,22 @@ func loopSession(ctx context.Context) string { s, _ := ctx.Value(loopSessionIDKey{}).(string) return s } + +// loopTurnIDKey is the context key for the assistant turn the current Run is +// producing. The gateway attaches it (WithTurnID) before starting a turn so the +// streaming events the loop emits can be correlated to that turn by the client. +type loopTurnIDKey struct{} + +// WithTurnID attaches turnID to ctx so the streaming events the loop emits carry +// the turn they belong to. It is exported because the gateway calls it once per +// turn, before Run; a caller that attaches none leaves the events' TurnID empty, +// which keeps the correlation fields backward compatible. +func WithTurnID(ctx context.Context, turnID string) context.Context { + return context.WithValue(ctx, loopTurnIDKey{}, turnID) +} + +// loopTurn returns the turn id on ctx, or "" when none was attached. +func loopTurn(ctx context.Context) string { + s, _ := ctx.Value(loopTurnIDKey{}).(string) + return s +} diff --git a/internal/runtime/loop/stream.go b/internal/runtime/loop/stream.go index fe6a2a6..5210952 100644 --- a/internal/runtime/loop/stream.go +++ b/internal/runtime/loop/stream.go @@ -32,6 +32,12 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO } resp, err := adapter.ChatStream(ctx, req, func(ev model.StreamEvent) { + // Stamp the session/turn identity at the loop boundary: the adapter runs + // below the loop and cannot know which conversation the stream belongs to, + // so the loop fills it from the context it was driven with (design/16 §4.2) + // and the frontend can route the delta to the active session. + ev.SessionID = loopSession(ctx) + ev.TurnID = loopTurn(ctx) // Suppress UI broadcast for internal autonomous passes (backthink), // whose intermediate reasoning must not leak into the chat stream. if l.onStream != nil && !isMutedBroadcast(ctx) { diff --git a/internal/runtime/loop/stream_scope_test.go b/internal/runtime/loop/stream_scope_test.go new file mode 100644 index 0000000..59a8661 --- /dev/null +++ b/internal/runtime/loop/stream_scope_test.go @@ -0,0 +1,119 @@ +package loop + +import ( + "context" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// collectStream returns an OnStream callback plus a reader for the events it saw, +// so a test can assert the scope the loop stamped on each emitted event. +func collectStream() (func(model.StreamEvent), func() []model.StreamEvent) { + var mu sync.Mutex + var events []model.StreamEvent + on := func(ev model.StreamEvent) { + mu.Lock() + events = append(events, ev) + mu.Unlock() + } + read := func() []model.StreamEvent { + mu.Lock() + defer mu.Unlock() + return append([]model.StreamEvent(nil), events...) + } + return on, read +} + +// TestRun_StreamCarriesSessionAndTurn locks the A1 contract: the loop stamps the +// session it was driven for and the turn id attached with WithTurnID on every +// streamed event, so the WS hub can route message.delta/thinking.delta/tool_call.* +// to the right conversation and turn (design/16 §4.2). +func TestRun_StreamCarriesSessionAndTurn(t *testing.T) { + t.Parallel() + ad := &mockAdapter{ + stream: []mgResp{{resp: model.Response{Content: "Hello world"}}}, + chunks: [][]string{{"Hel", "lo"}}, + } + a := newTestAgent(t, ad, nil) + onStream, streams := collectStream() + l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8}) + + ctx := WithTurnID(context.Background(), "turn-xyz") + if _, err := l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}); err != nil { + t.Fatalf("Run: %v", err) + } + events := streams() + if len(events) != 2 { + t.Fatalf("stream events = %d, want 2", 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) + } +} + +// TestRun_StreamWithoutTurnIDLeavesItEmpty keeps the backward-compatible default: +// a caller that attaches no turn id still gets the session on each event, but an +// empty TurnID — the field's zero value is the old behavior. +func TestRun_StreamWithoutTurnIDLeavesItEmpty(t *testing.T) { + t.Parallel() + ad := &mockAdapter{ + stream: []mgResp{{resp: model.Response{Content: "ok"}}}, + chunks: [][]string{{"o", "k"}}, + } + 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.Fatalf("Run: %v", err) + } + events := streams() + if len(events) == 0 { + t.Fatal("no stream events were emitted") + } + for i, ev := range events { + if ev.SessionID != "web" || ev.TurnID != "" { + t.Fatalf("event[%d] scope = session %q turn %q, want session web turn \"\"", i, ev.SessionID, ev.TurnID) + } + } +} + +// TestRun_ToolExecutionCarriesScope covers the loop's other stream emitter: the +// tool_execution_update progress frame must be scoped too, otherwise it would be +// the one frame a multi-session client could not route. +func TestRun_ToolExecutionCarriesScope(t *testing.T) { + t.Parallel() + ad := &mockAdapter{stream: []mgResp{ + {resp: nativeDatetimeCall()}, + {resp: model.Response{Model: "m", Content: "done"}}, + }} + td := dispatcherWith(t, &mockTool{name: "datetime", result: map[string]any{"now": "2026-09-08 22:34:00"}}) + a := newTestAgent(t, ad, td) + onStream, streams := collectStream() + l := New(Options{Agent: a, OnStream: onStream, MaxRounds: 8}) + + ctx := WithTurnID(context.Background(), "turn-tool") + if _, err := l.Run(ctx, "web", []model.Message{model.TextMessage(model.RoleUser, "time?")}); err != nil { + t.Fatalf("Run: %v", err) + } + found := 0 + for _, ev := range streams() { + if ev.Kind != model.StreamKindToolExecution { + continue + } + found++ + if ev.SessionID != "web" || ev.TurnID != "turn-tool" { + t.Fatalf("tool_execution scope = session %q turn %q, want session web turn turn-tool", ev.SessionID, ev.TurnID) + } + } + if found == 0 { + t.Fatal("no tool_execution_update event was emitted") + } +} diff --git a/internal/runtime/loop/tools.go b/internal/runtime/loop/tools.go index a7be672..ac3b623 100644 --- a/internal/runtime/loop/tools.go +++ b/internal/runtime/loop/tools.go @@ -312,7 +312,10 @@ func (l *Loop) emitToolExecution(ctx context.Context, name, phase string) { if l.onStream == nil || isMutedBroadcast(ctx) { return } - l.onStream(model.StreamEvent{Kind: model.StreamKindToolExecution, Delta: phase + ":" + name}) + ev := model.StreamEvent{Kind: model.StreamKindToolExecution, Delta: phase + ":" + name} + ev.SessionID = loopSession(ctx) + ev.TurnID = loopTurn(ctx) + l.onStream(ev) } // invokeParallel runs calls at idxs concurrently, capped at the loop's diff --git a/internal/runtime/model/stream.go b/internal/runtime/model/stream.go index 1704b3b..b95f96f 100644 --- a/internal/runtime/model/stream.go +++ b/internal/runtime/model/stream.go @@ -57,4 +57,15 @@ type StreamEvent struct { // ID is the provider block/index identifier for a tool call (e.g. Anthropic's // tool_use block id). Empty for text segments. ID 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 + // notion of a session, leaves it zero. A consumer that routes the event (the + // WS hub) reads it to target the right conversation. + SessionID string + // TurnID is the assistant turn the event belongs to. Like SessionID it is + // stamped by the loop, which reads the turn identity the caller attached with + // loop.WithTurnID; it is zero when no turn id was attached, so a consumer must + // treat both fields as best-effort correlation rather than a hard guarantee. + TurnID string } diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index 6846252..54eef56 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -948,8 +948,10 @@ func TestHubBroadcastStream(t *testing.T) { // A text_delta StreamEvent on the OnStream→bus→relay path becomes a typed // message.delta frame carrying the delta text and a monotonically increasing - // seq, leaving the legacy stream_chunk shape out of the way. - hb.broadcastStream(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "Hel", Index: 0}) + // seq, leaving the legacy stream_chunk shape out of the way. The loop stamps + // the session/turn identity on the event and the hub must carry it onto the + // frame, so a client can route the delta to the active session (design/16 §4.2). + hb.broadcastStream(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "Hel", Index: 0, SessionID: "web", TurnID: "turn-abc"}) ev := readFakeFrame(t, c) if ev.Type != "message.delta" { t.Fatalf("frame type = %q, want message.delta (frame=%+v)", ev.Type, ev) @@ -961,6 +963,9 @@ func TestHubBroadcastStream(t *testing.T) { if ev.Seq != 1 { t.Fatalf("frame seq = %d, want 1", ev.Seq) } + if ev.SessionID != "web" || ev.TurnID != "turn-abc" { + t.Fatalf("frame scope = session %q turn %q, want session web turn turn-abc", ev.SessionID, ev.TurnID) + } // A tool_call event maps to tool_call.*; a non-StreamEvent payload is dropped. hb.broadcastStream(model.StreamEvent{Kind: model.StreamKindToolCallEnd, ID: "toolu_1", Index: 0}) diff --git a/internal/services/gateway/ws_hub_stream_scope_test.go b/internal/services/gateway/ws_hub_stream_scope_test.go new file mode 100644 index 0000000..1c22928 --- /dev/null +++ b/internal/services/gateway/ws_hub_stream_scope_test.go @@ -0,0 +1,151 @@ +package gateway + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "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" +) + +// scopeStreamAdapter is a model.Adapter that emits two text deltas so a test can +// drive the real loop and observe the framed events it produces. +type scopeStreamAdapter struct{} + +func (scopeStreamAdapter) Name() string { return "scope-test" } + +func (scopeStreamAdapter) Chat(context.Context, model.Request, model.Options) (model.Response, error) { + return model.Response{Content: "Hel lo"}, nil +} + +func (scopeStreamAdapter) ChatStream(_ context.Context, _ model.Request, on func(model.StreamEvent)) (model.Response, error) { + if on != nil { + on(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "Hel"}) + on(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: "lo"}) + } + return model.Response{Content: "Hel lo"}, nil +} + +// newScopedStreamHub wires a hub to a real loop whose OnStream closure mirrors +// cmd/cyrene's streamChunk (publish the whole StreamEvent value to ws.stream), so +// a test exercises the production path OnStream → bus → relay → frame. It returns +// the hub and a registered fake connection. +func newScopedStreamHub(t *testing.T) (*hub, *wsConn) { + t.Helper() + b := bus.New() + ag, err := agent.New("You are a helpful assistant.", nil, nil, agent.WithModel(scopeStreamAdapter{}, 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 := newFakeConn(hb, 64) + hb.register(c) + return hb, c +} + +// readDeltas scans the connection's frames until want message.delta frames have +// arrived. The deltas reach the conn asynchronously through the bus while +// runTurn's own stream_end is sent straight to the conn, so a test cannot assume +// an order. +func readDeltas(t *testing.T, c *wsConn, want int) []wsEvent { + t.Helper() + var deltas []wsEvent + deadline := time.After(3 * time.Second) + for len(deltas) < want { + select { + case raw := <-c.send: + var ev wsEvent + if err := json.Unmarshal(raw, &ev); err != nil { + t.Fatalf("decode frame %s: %v", raw, err) + } + if ev.Type == "message.delta" { + deltas = append(deltas, ev) + } + case <-deadline: + t.Fatalf("timed out with %d message.delta frames, want %d", len(deltas), want) + } + } + return deltas +} + +// TestStreamFramesCarrySessionScope is the end-to-end lock for A1: a turn driven +// through the hub's runTurn must surface message.delta frames that carry the +// session and the turn id, so a client can route them to the active session. It +// runs the real loop instead of calling broadcastStream directly, which also +// proves the bus carries the new fields untouched — no publisher change needed. +func TestStreamFramesCarrySessionScope(t *testing.T) { + t.Parallel() + hb, c := newScopedStreamHub(t) + + if _, err := hb.runTurn(context.Background(), "web", []model.Message{model.TextMessage(model.RoleUser, "hi")}, loop.OwnerEnv()); err != nil { + t.Fatalf("runTurn: %v", err) + } + deltas := readDeltas(t, c, 2) + + for i, ev := range deltas { + if ev.SessionID != "web" { + t.Fatalf("delta[%d] sessionId = %q, want web (frame=%+v)", i, ev.SessionID, ev) + } + if !strings.HasPrefix(ev.TurnID, "turn-") { + t.Fatalf("delta[%d] turnId = %q, want a turn-* id (frame=%+v)", i, ev.TurnID, ev) + } + } + if deltas[0].TurnID != deltas[1].TurnID { + t.Fatalf("turnId differs within one turn: %q vs %q", deltas[0].TurnID, deltas[1].TurnID) + } + if deltas[0].Seq == 0 || deltas[1].Seq <= deltas[0].Seq { + t.Fatalf("seq not increasing: %d then %d", deltas[0].Seq, deltas[1].Seq) + } + if data, ok := deltas[0].Data.(map[string]any); !ok || data["delta"] != "Hel" { + t.Fatalf("first delta data = %+v, want delta=Hel", deltas[0].Data) + } +} + +// TestStreamFramesTurnScopeChangesPerTurn covers the orchestrated resume shape: a +// pending turn drained after the first one completes is a second runTurn call, so +// it must take a fresh turnId. Without that, deltas from the resumed turn would +// be indistinguishable from the previous turn's. +func TestStreamFramesTurnScopeChangesPerTurn(t *testing.T) { + t.Parallel() + hb, c := newScopedStreamHub(t) + + msgs := []model.Message{model.TextMessage(model.RoleUser, "hi")} + if _, err := hb.runTurn(context.Background(), "web", msgs, loop.OwnerEnv()); err != nil { + t.Fatalf("first runTurn: %v", err) + } + first := readDeltas(t, c, 2) + if _, err := hb.runTurn(context.Background(), "web", msgs, loop.OwnerEnv()); err != nil { + t.Fatalf("second runTurn: %v", err) + } + second := readDeltas(t, c, 2) + + for i, ev := range second { + if ev.SessionID != "web" { + t.Fatalf("second turn delta[%d] sessionId = %q, want web", i, ev.SessionID) + } + if ev.TurnID != second[0].TurnID { + t.Fatalf("second turn delta[%d] turnId = %q, want %q for the whole turn", i, ev.TurnID, second[0].TurnID) + } + } + if first[0].TurnID == second[0].TurnID { + t.Fatalf("turnId reused across turns: %q", first[0].TurnID) + } + if second[0].Seq <= first[1].Seq { + t.Fatalf("seq not monotonic across turns: %d then %d", first[1].Seq, second[0].Seq) + } +} diff --git a/internal/services/gateway/ws_hub_turn.go b/internal/services/gateway/ws_hub_turn.go index e3a13eb..3c1392c 100644 --- a/internal/services/gateway/ws_hub_turn.go +++ b/internal/services/gateway/ws_hub_turn.go @@ -80,6 +80,12 @@ func (h *hub) streamChat(ctx context.Context, payload any) { // emits a single whole-text chunk to hold the stream contract. func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) { turnID := newTurnID() + // Thread the turn identity into the run context so the loop stamps it on every + // streamed event it emits (message.*/thinking.*/tool_call.*); the hub then puts + // it on the outbound frame. Every turn takes a fresh id — a session has at most + // one in-flight turn — so a client can correlate one burst of deltas to one + // assistant turn (design/16 §4.2). + ctx = loop.WithTurnID(ctx, turnID) if h.runner == nil || len(msgs) == 0 { h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": "idle"}}) return &loop.Result{StopReason: "idle"}, nil @@ -117,16 +123,17 @@ func (h *hub) handleInterrupt(payload json.RawMessage) { // broadcastStream maps a streamed model.StreamEvent to its downstream WS event // type and fans it out to every client. It runs on the bus subscription worker; // seq is assigned atomically so a burst stays ordered even if the loop emits -// events from more than one goroutine. TurnId is left empty here because the -// OnStream bus payload does not carry the turn identity (future work would -// thread a turn id through loop → streamChunk → payload). +// events from more than one goroutine. SessionID/TurnID ride on the event itself +// (the loop stamps them at the turn boundary from the run context), so the frame +// carries the scope a client needs to route it to the active session and to one +// assistant turn. func (h *hub) broadcastStream(payload any) { ev, ok := payload.(model.StreamEvent) if !ok { return } typ, data := streamEventData(ev) - h.broadcast(wsEvent{Type: typ, Seq: int(h.streamSeq.Add(1)), Data: data}) + h.broadcast(wsEvent{Type: typ, SessionID: ev.SessionID, TurnID: ev.TurnID, Seq: int(h.streamSeq.Add(1)), Data: data}) } // chatStreamRequest is the decoded ws.chat payload.