From 24d5924cee2f311446d2b44df238c078b204cd79 Mon Sep 17 00:00:00 2001 From: Aska Date: Sun, 6 Sep 2026 16:31:28 +0800 Subject: [PATCH] =?UTF-8?q?feat(model):=20Top5#1=20=E7=BB=93=E6=9E=84?= =?UTF-8?q?=E5=8C=96=E6=B5=81=E4=BA=8B=E4=BB=B6=20+=20WS=20=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E4=BD=93=E7=B3=BB=EF=BC=88=E5=AF=B9=E9=BD=90=20OpenCl?= =?UTF-8?q?aw=20AssistantMessageEvent=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 model/stream.go：StreamKind(text/thinking/tool_call 各 start/delta/end + tool_execution) + StreamEvent{Kind,Delta,Index,ID}。 - ChatStream 签名 onChunk func(delta string)→onEvent func(StreamEvent)；loop OnStream 同改； streamStep 转发结构化事件。 - openai：text_delta + text_start/text_end（正文三段式）；anthropic：文本块 text_*/tool_call_* 按块 Index（content_block_stop 分支）真分段；fallback：有事件即交付。 - ws_hub：wsEvent 壳补 turnId/seq；新增 broadcastStream；streamChat 每帧打 turnId+seq； stream.go 把事件 publish 到 bus ws.stream。 - ws_bridge：streamEventData 映射 text_*→message.start/delta/end、thinking_*→thinking.*、 tool_call_*→tool_call.*、tool_execution_update→tool_execution.update，未知回退 stream_chunk。 - 前端 ws.ts：解析 message.start/delta/end、thinking.start/delta/end(pendingThinking)、 tool_call.start/end→addTool、tool_execution.update；保留旧 stream_chunk/thinking/tool_used 兼容； types.ts 补 turnId/seq。Docs/api-client.md 增 §3.1 v0.4。 已知限制（注释/文档标注）：turnId 在 OnStream→bus→relay 路径暂为空（StreamEvent 无 turnId 字段， bus ws.stream 载荷不带 turn）；OpenAI adapter 未解析 tool_call 段；thinking 段两 adapter 暂不发； tool_execution.update 是保留项（随工具循环钩子后置）。go build/vet/test + 前端 tsc 全绿。 --- Docs/api-client.md | 11 ++- cmd/cyrene/stream.go | 35 +++++---- frontend/web/src/lib/types.ts | 11 ++- frontend/web/src/lib/ws.ts | 70 +++++++++++++++++ internal/runtime/agent/agent_test.go | 2 +- internal/runtime/loop/loop.go | 3 +- internal/runtime/loop/loop_test.go | 4 +- internal/runtime/loop/mocks_test.go | 4 +- internal/runtime/loop/options.go | 8 +- internal/runtime/loop/stream.go | 4 +- internal/runtime/memory/llmextractor_test.go | 2 +- internal/runtime/model/adapter.go | 11 ++- internal/runtime/model/anthropic.go | 11 ++- internal/runtime/model/anthropic_stream.go | 60 +++++++++++---- internal/runtime/model/anthropic_test.go | 28 +++++-- internal/runtime/model/fallback.go | 10 +-- internal/runtime/model/openai.go | 23 ++++-- internal/runtime/model/openai_test.go | 14 +++- internal/runtime/model/stream.go | 60 +++++++++++++++ .../runtime/orchestration/scheduler_test.go | 2 +- internal/services/gateway/extra_test.go | 34 +++++++++ internal/services/gateway/ws_bridge.go | 33 ++++++++ internal/services/gateway/ws_hub.go | 75 +++++++++++++++---- 23 files changed, 425 insertions(+), 90 deletions(-) create mode 100644 internal/runtime/model/stream.go diff --git a/Docs/api-client.md b/Docs/api-client.md index f1d607b..ae8c714 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -48,7 +48,15 @@ ## 3. WebSocket `/ws/chat`（实时） - **握手**: `ws://:8080/ws/chat?token=` 或 `Authorization: Bearer`；未认证在握手前回 `401` JSON。 -- **下行（服务端→客户端）**: 经 bus 广播 `{type:"", data:...}`——`stream_chunk` / `stream_end` / `thinking`(思维链) / `mood` / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update`。 +- **下行（服务端→客户端）**: 经 bus 广播 `{type:"", data:..., turnId?, seq?}`（`turnId`/`seq` 为流式关联字段，见 §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` / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update`。 + +### 3.1 结构化流事件（`message.*` / `thinking.*` / `tool_call.*` / `tool_execution.*`） +- **事件壳**：`{type, turnId?, seq?, data}`, `data` 为**类型化结构**（非 `any`）。`seq` 为 hub 级单调递增序号，供客户端将一次突发的流式事件排序关联。 +- **`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 后置实现）。 +- **向后兼容**：未映射/无法确定的事件仍走 `stream_chunk` + `stream_end`（非流式回退路径）。 - **上行（客户端→服务端）**: `mode_switch` / `chat`（跑 loop 并广播流式） / `voice_input` / `subagent_ask` / `task_output_query`。 - ⚠️ 语音/流式完整 schema 随 M9 前端联调补充；会话级定向（当前对所有连接广播）为后续项。 @@ -60,6 +68,7 @@ - **离线**: PWA / 原生端建议缓存 `sessions`/`personas` 列表。 ## 5. 变更记录 +- `v0.4`（结构化流）：WS 下行新增 `message.*` / `thinking.*` / `tool_call.*` / `tool_execution.update` 结构化流事件（event shell 补 `turnId`/`seq`，`data` 类型化），对齐 OpenClaw AssistantMessageEvent；保留 `stream_chunk`/`stream_end` 向后兼容。 - `v0.3`（M10）：管理面板 `/admin` 落地——按 §2 契约封装 admin 只读端点（personas/sessions/memories/plugins/channels/iot）+ owner-only `plugins/policy` 写；其余 M10 端点以契约路径封装、后端未实现时前端显式"待实现"壳。 - `v0.2`（M8）：gateway 落地——登录/health/chat/sessions/personas + admin 只读端点 + `plugins/policy` 写 + WS `/ws/chat`（JWT + bus 广播）。 - `v0.1`（契约先行）：骨架 + 端点登记表。 diff --git a/cmd/cyrene/stream.go b/cmd/cyrene/stream.go index f35d768..0c38713 100644 --- a/cmd/cyrene/stream.go +++ b/cmd/cyrene/stream.go @@ -1,22 +1,31 @@ package main -import "context" +import ( + "context" -// streamChunkSubject is the bus subject the loop's OnStream closure publishes -// each token delta to. The gateway hub subscribes to ws.> and re-broadcasts it -// as a ws.stream_chunk frame to every client. -const streamChunkSubject = "ws.stream_chunk" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) -// streamChunk is the loop's OnStream callback. It publishes one token delta to -// the gateway's streaming subject so the hub fans it out to every client as a -// ws.stream_chunk frame. It is a no-op when no bus is wired and never blocks the -// loop: bus.Publish is asynchronous and a full per-subscription buffer drops the -// event rather than stalling the model stream. -func (a *app) streamChunk(delta string) { +// streamSubject is the bus subject the loop's OnStream closure publishes each +// structured stream event to. The gateway hub subscribes to ws.> and carries a +// dedicated branch for this subject that maps the model.StreamEvent to the typed +// downstream WS event (message.*/thinking.*/tool_call.*) with a seq number. +// +// It MUST stay in sync with wsStreamSubject in internal/services/gateway/ws_hub.go; +// the two packages cannot share a const without pulling the gateway into the +// main binary's import graph, so the literal is duplicated and documented. +const streamSubject = "ws.stream" + +// streamChunk is the loop's OnStream callback. It publishes one structured +// stream event to the gateway's streaming subject so the hub fans it out to +// every client as the matching typed WS event. It is a no-op when no bus is +// wired and never blocks the loop: bus.Publish is asynchronous and a full +// per-subscription buffer drops the event rather than stalling the model stream. +func (a *app) streamChunk(ev model.StreamEvent) { if a.bus == nil { return } - if err := a.bus.Publish(context.Background(), streamChunkSubject, map[string]any{"delta": delta}); err != nil { - a.log.Warn("stream chunk publish", "err", err) + if err := a.bus.Publish(context.Background(), streamSubject, ev); err != nil { + a.log.Warn("stream event publish", "err", err) } } diff --git a/frontend/web/src/lib/types.ts b/frontend/web/src/lib/types.ts index b3d3efa..3f8b152 100644 --- a/frontend/web/src/lib/types.ts +++ b/frontend/web/src/lib/types.ts @@ -143,13 +143,18 @@ export interface ChatSyncResult { /** * Downstream WS frame (server → client). `data` is intentionally `unknown`; - * every consumer narrows it. The backend emits `{type, data}` and types include - * stream_chunk / stream_end / thinking / mood / tool_used / subagent_event / - * graph_update / reminder / device_update / auth / error / task_output. + * every consumer narrows it. The backend emits `{type, data, turnId?, seq?}` + * and types include the structured stream events message.start/delta/end, + * thinking.start/delta/end, tool_call.start/delta/end, tool_execution.update + * (design/04 §1.1), plus the legacy stream_chunk / stream_end / thinking / mood / + * tool_used / subagent_event / graph_update / reminder / device_update / auth / + * error / task_output. */ export interface WsDownEvent { type: string; data: unknown; + turnId?: string; + seq?: number; } /** Upstream WS frame (client → server). */ diff --git a/frontend/web/src/lib/ws.ts b/frontend/web/src/lib/ws.ts index 89158c5..5dd97f4 100644 --- a/frontend/web/src/lib/ws.ts +++ b/frontend/web/src/lib/ws.ts @@ -15,6 +15,11 @@ let shouldRun = false; let reconnectTimer: number | null = null; let attempt = 0; +/** Accumulator for a streamed thinking segment (thinking.* events). Reset on + * thinking.start so a mid-turn reconnect can rebuild the collapsed block + * without relying on prior state. */ +let pendingThinking = ""; + const MAX_BACKOFF = 30_000; const BASE_BACKOFF = 1_000; @@ -45,6 +50,24 @@ function parseTool(v: unknown): ToolUsed | null { }; } +/** Build a tool card from a tool_call.* payload. Unlike the legacy tool_used + * frame these carry `id` (provider block id) rather than a guaranteed `name`, + * so we label the card by id when no name is present. */ +function parseToolCall(v: unknown): ToolUsed | null { + const r = toRecord(v); + if (!r) return null; + const name = typeof r.name === "string" ? r.name : ""; + const id = typeof r.id === "string" ? r.id : ""; + if (!name && !id) return null; + return { + name: name || `tool_${id.slice(0, 4)}`, + summary: typeof r.summary === "string" ? r.summary : undefined, + detail: typeof r.detail === "string" ? r.detail : undefined, + ok: typeof r.ok === "boolean" ? r.ok : undefined, + at: now(), + }; +} + function parseAgent(payload: unknown): AgentInfo | null { const r = toRecord(payload); if (!r) return null; @@ -112,6 +135,53 @@ function dispatch(ev: { type: string; data: unknown }): void { } break; } + // Structured stream events (design/04 §1.1, aligned with OpenClaw). These + // arrive in place of the legacy stream_chunk when the loop streams events. + case "message.start": { + // The assistant bubble is already opened by useChat; nothing to do here. + break; + } + case "message.delta": { + if (r && typeof r.delta === "string") s.appendChunk(r.delta); + break; + } + case "message.end": { + // Body-text segment closed; the turn may continue with tool calls, so we + // do NOT end the stream here (stream_end owns that). + break; + } + case "thinking.start": { + pendingThinking = ""; + break; + } + case "thinking.delta": { + if (r && typeof r.delta === "string") { + pendingThinking += r.delta; + s.setThinking(pendingThinking); + } + break; + } + case "thinking.end": { + if (pendingThinking) s.setThinking(pendingThinking); + pendingThinking = ""; + break; + } + case "tool_call.start": + case "tool_call.end": { + const tool = parseToolCall(ev.data); + if (tool) s.addTool(tool); + break; + } + case "tool_call.delta": { + // Argument fragments stream between start/end; the card is added at the + // boundary so mid-stream reconnects do not drop it. + break; + } + case "tool_execution.update": { + const tool = parseToolCall(ev.data); + if (tool) s.addTool(tool); + break; + } case "thinking": { if (typeof ev.data === "string") { s.setThinking(ev.data); diff --git a/internal/runtime/agent/agent_test.go b/internal/runtime/agent/agent_test.go index d8730dc..ca4895f 100644 --- a/internal/runtime/agent/agent_test.go +++ b/internal/runtime/agent/agent_test.go @@ -32,7 +32,7 @@ func (m *mockAdapter) Chat(_ context.Context, req model.Request, _ model.Options return m.resp, m.err } -func (m *mockAdapter) ChatStream(_ context.Context, _ model.Request, _ func(string)) (model.Response, error) { +func (m *mockAdapter) ChatStream(_ context.Context, _ model.Request, _ func(model.StreamEvent)) (model.Response, error) { return m.resp, m.err } diff --git a/internal/runtime/loop/loop.go b/internal/runtime/loop/loop.go index 3528ba6..8017064 100644 --- a/internal/runtime/loop/loop.go +++ b/internal/runtime/loop/loop.go @@ -20,6 +20,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" @@ -79,7 +80,7 @@ type Loop struct { guards *guardrail.Pipeline personaID string maxRounds int - onStream func(delta string) + onStream func(model.StreamEvent) log *slog.Logger } diff --git a/internal/runtime/loop/loop_test.go b/internal/runtime/loop/loop_test.go index eaf4408..7d4bf22 100644 --- a/internal/runtime/loop/loop_test.go +++ b/internal/runtime/loop/loop_test.go @@ -281,7 +281,7 @@ func TestRun_StreamDeltas(t *testing.T) { var deltas []string l := New(Options{ Agent: a, - OnStream: func(d string) { mu.Lock(); deltas = append(deltas, d); mu.Unlock() }, + OnStream: func(ev model.StreamEvent) { mu.Lock(); deltas = append(deltas, ev.Delta); mu.Unlock() }, MaxRounds: 8, }) @@ -306,7 +306,7 @@ func TestRun_StreamGuardrailBlocks(t *testing.T) { pipe := guardrail.New(guardrail.WithGuard(g)) ad := &mockAdapter{} a := newTestAgent(t, ad, nil) - l := New(Options{Agent: a, Guards: pipe, OnStream: func(string) {}, MaxRounds: 8}) + l := New(Options{Agent: a, Guards: pipe, OnStream: func(model.StreamEvent) {}, MaxRounds: 8}) res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "hi")}) if err != nil { diff --git a/internal/runtime/loop/mocks_test.go b/internal/runtime/loop/mocks_test.go index 15401ce..36bb661 100644 --- a/internal/runtime/loop/mocks_test.go +++ b/internal/runtime/loop/mocks_test.go @@ -59,14 +59,14 @@ func (m *mockAdapter) Chat(_ context.Context, req model.Request, _ model.Options return r.resp, r.err } -func (m *mockAdapter) ChatStream(_ context.Context, req model.Request, on func(string)) (model.Response, error) { +func (m *mockAdapter) ChatStream(_ context.Context, req model.Request, on func(model.StreamEvent)) (model.Response, error) { m.streamReqs = append(m.streamReqs, req) i := m.streamCalls m.streamCalls++ if i < len(m.chunks) { for _, c := range m.chunks[i] { if on != nil { - on(c) + on(model.StreamEvent{Kind: model.StreamKindTextDelta, Delta: c}) } } } diff --git a/internal/runtime/loop/options.go b/internal/runtime/loop/options.go index 4d853a9..dcccf26 100644 --- a/internal/runtime/loop/options.go +++ b/internal/runtime/loop/options.go @@ -6,6 +6,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" @@ -47,9 +48,10 @@ type Options struct { // DefaultMaxRounds (8). MaxRounds int // OnStream, when non-nil, switches the loop to the streaming path: the model - // adapter's ChatStream is driven directly and each token delta is forwarded - // to this callback. - OnStream func(delta string) + // adapter's ChatStream is driven directly and each structured stream event is + // forwarded to this callback (model.StreamEvent), so the frontend can render + // body text, tool calls and (eventually) reasoning segments independently. + OnStream func(model.StreamEvent) // Log is the base logger. Nil is replaced with a discard logger tagged with // source "runtime/loop". Log *slog.Logger diff --git a/internal/runtime/loop/stream.go b/internal/runtime/loop/stream.go index 84fb20e..7efd8cb 100644 --- a/internal/runtime/loop/stream.go +++ b/internal/runtime/loop/stream.go @@ -31,9 +31,9 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO } } - resp, err := adapter.ChatStream(ctx, req, func(delta string) { + resp, err := adapter.ChatStream(ctx, req, func(ev model.StreamEvent) { if l.onStream != nil { - l.onStream(delta) + l.onStream(ev) } }) if err != nil { diff --git a/internal/runtime/memory/llmextractor_test.go b/internal/runtime/memory/llmextractor_test.go index a2ec07d..c7bfab1 100644 --- a/internal/runtime/memory/llmextractor_test.go +++ b/internal/runtime/memory/llmextractor_test.go @@ -25,7 +25,7 @@ func (m *mockAdapter) Chat(ctx context.Context, req model.Request, opts model.Op return model.Response{Content: "[]"}, nil } -func (m *mockAdapter) ChatStream(ctx context.Context, req model.Request, onChunk func(delta string)) (model.Response, error) { +func (m *mockAdapter) ChatStream(ctx context.Context, req model.Request, onEvent func(model.StreamEvent)) (model.Response, error) { return model.Response{}, model.ErrStream } diff --git a/internal/runtime/model/adapter.go b/internal/runtime/model/adapter.go index 85b3173..c2bfe12 100644 --- a/internal/runtime/model/adapter.go +++ b/internal/runtime/model/adapter.go @@ -34,8 +34,11 @@ type Adapter interface { Name() string // Chat performs a non-streaming completion and returns the full response. Chat(ctx context.Context, req Request, opts Options) (Response, error) - // ChatStream streams token-level deltas via onChunk and returns the - // aggregated response (with usage when the provider reports it). onChunk may - // be nil to discard deltas while still aggregating. - ChatStream(ctx context.Context, req Request, onChunk func(delta string)) (Response, error) + // ChatStream streams structured events via onEvent and returns the aggregated + // response (with usage when the provider reports it). onEvent may be nil to + // discard events while still aggregating. Every provider-supported segment is + // bracketed (text_start/delta/end, tool_call_start/delta/end), so the consumer + // can render each independently; segments adapters cannot yet distinguish are + // delivered as text_delta events. + ChatStream(ctx context.Context, req Request, onEvent func(StreamEvent)) (Response, error) } diff --git a/internal/runtime/model/anthropic.go b/internal/runtime/model/anthropic.go index 5105179..9d67c85 100644 --- a/internal/runtime/model/anthropic.go +++ b/internal/runtime/model/anthropic.go @@ -169,9 +169,12 @@ func (a *AnthropicChatAdapter) Chat(ctx context.Context, req Request, opts Optio return out, nil } -// ChatStream streams content-block deltas via onChunk and returns the aggregated -// response with usage when the provider reports it. -func (a *AnthropicChatAdapter) ChatStream(ctx context.Context, req Request, onChunk func(delta string)) (Response, error) { +// ChatStream streams structured events via onEvent and returns the aggregated +// response with usage when the provider reports it. Text blocks are delivered as +// text_start/text_delta/text_end events; tool_use blocks are delivered as +// tool_call_start/tool_call_delta/tool_call_end so the frontend can render each +// segment independently. +func (a *AnthropicChatAdapter) ChatStream(ctx context.Context, req Request, onEvent func(StreamEvent)) (Response, error) { start := time.Now() if a.baseURL == "" { return Response{}, ErrUpstream @@ -197,7 +200,7 @@ func (a *AnthropicChatAdapter) ChatStream(ctx context.Context, req Request, onCh if resp.StatusCode != http.StatusOK { return Response{}, a.upstreamError(resp) } - out, err := scanAnthropicSSE(resp.Body, onChunk) + out, err := scanAnthropicSSE(resp.Body, onEvent) if err != nil { return Response{}, err } diff --git a/internal/runtime/model/anthropic_stream.go b/internal/runtime/model/anthropic_stream.go index 0d6f24c..e113192 100644 --- a/internal/runtime/model/anthropic_stream.go +++ b/internal/runtime/model/anthropic_stream.go @@ -52,11 +52,13 @@ type streamBlock struct { } // scanAnthropicSSE incrementally decodes Anthropic's content-block stream. Text -// deltas are forwarded to onChunk; tool_use input_json_delta fragments are -// accumulated per block and re-assembled into a structured Args map when the block -// stops. The stream must terminate with a message_stop event, otherwise ErrStream -// is returned. -func scanAnthropicSSE(r io.Reader, onChunk func(string)) (Response, error) { +// deltas are forwarded as text_delta events (segment bracketed by +// text_start/text_end); tool_use input_json_delta fragments are forwarded as +// tool_call_delta events (segment bracketed by tool_call_start/tool_call_end) +// and accumulated per block so they re-assemble into a structured Args map when +// the block stops. The stream must terminate with a message_stop event, +// otherwise ErrStream is returned. +func scanAnthropicSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) { scanner := bufio.NewScanner(r) scanner.Buffer(make([]byte, 0, sseBufInit), maxSSELineBytes) agg := Response{} @@ -78,7 +80,7 @@ func scanAnthropicSSE(r io.Reader, onChunk func(string)) (Response, error) { if err := json.Unmarshal([]byte(data), &evt); err != nil { return Response{}, fmt.Errorf("model: %w: decode stream chunk: %v", ErrUpstream, err) } - if err := applyAnthropicEvent(&agg, blocks, &order, &text, evt, onChunk); err != nil { + if err := applyAnthropicEvent(&agg, blocks, &order, &text, evt, onEvent); err != nil { return Response{}, err } if evt.Type == "message_stop" { @@ -102,7 +104,7 @@ func applyAnthropicEvent( order *[]int, text *strings.Builder, evt anthropicStreamEvent, - onChunk func(string), + onEvent func(StreamEvent), ) error { switch evt.Type { case "message_start": @@ -123,11 +125,21 @@ func applyAnthropicEvent( name: evt.Content.Name, initialArgs: evt.Content.Input, } - if evt.Content.Type == "text" && evt.Content.Text != "" { - b.text.WriteString(evt.Content.Text) - text.WriteString(evt.Content.Text) - if onChunk != nil { - onChunk(evt.Content.Text) + switch evt.Content.Type { + case "text": + if onEvent != nil { + onEvent(StreamEvent{Kind: StreamKindTextStart, Index: *evt.Index}) + } + if evt.Content.Text != "" { + b.text.WriteString(evt.Content.Text) + text.WriteString(evt.Content.Text) + if onEvent != nil { + onEvent(StreamEvent{Kind: StreamKindTextDelta, Delta: evt.Content.Text, Index: *evt.Index}) + } + } + case "tool_use": + if onEvent != nil { + onEvent(StreamEvent{Kind: StreamKindToolCallStart, Index: *evt.Index, ID: evt.Content.ID}) } } if _, ok := blocks[*evt.Index]; !ok { @@ -146,11 +158,31 @@ func applyAnthropicEvent( case "text_delta": b.text.WriteString(evt.Delta.Text) text.WriteString(evt.Delta.Text) - if onChunk != nil { - onChunk(evt.Delta.Text) + if onEvent != nil { + onEvent(StreamEvent{Kind: StreamKindTextDelta, Delta: evt.Delta.Text, Index: *evt.Index}) } case "input_json_delta": b.input.WriteString(evt.Delta.PartialJSON) + if onEvent != nil { + onEvent(StreamEvent{Kind: StreamKindToolCallDelta, Delta: evt.Delta.PartialJSON, Index: *evt.Index, ID: b.id}) + } + } + case "content_block_stop": + if evt.Index == nil { + return nil + } + b := blocks[*evt.Index] + if b == nil { + return nil + } + if b.typ == "text" { + if onEvent != nil { + onEvent(StreamEvent{Kind: StreamKindTextEnd, Index: *evt.Index}) + } + } else if b.typ == "tool_use" { + if onEvent != nil { + onEvent(StreamEvent{Kind: StreamKindToolCallEnd, Index: *evt.Index, ID: b.id}) + } } case "message_delta": if evt.Usage != nil { diff --git a/internal/runtime/model/anthropic_test.go b/internal/runtime/model/anthropic_test.go index 4c2bdd3..3f4f7aa 100644 --- a/internal/runtime/model/anthropic_test.go +++ b/internal/runtime/model/anthropic_test.go @@ -323,10 +323,10 @@ func TestAnthropicChat_ChatStream_Text(t *testing.T) { ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"}, WithAnthropicRecorder(func(cr CallRecord) { recs = append(recs, cr) })) - var deltas []string + var events []StreamEvent resp, err := ad.ChatStream(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}, Model: "claude-test", Purpose: "chat"}, - func(d string) { deltas = append(deltas, d) }) + func(ev StreamEvent) { events = append(events, ev) }) if err != nil { t.Fatalf("ChatStream error: %v", err) } @@ -348,8 +348,14 @@ func TestAnthropicChat_ChatStream_Text(t *testing.T) { if resp.TotalTokens != 7 { t.Errorf("resp.TotalTokens = %d, want 7", resp.TotalTokens) } - if !reflect.DeepEqual(deltas, []string{"Hel", "lo"}) { - t.Errorf("deltas = %v, want [Hel lo]", deltas) + wantEvents := []StreamEvent{ + {Kind: StreamKindTextStart, Index: 0}, + {Kind: StreamKindTextDelta, Index: 0, Delta: "Hel"}, + {Kind: StreamKindTextDelta, Index: 0, Delta: "lo"}, + {Kind: StreamKindTextEnd, Index: 0}, + } + if !reflect.DeepEqual(events, wantEvents) { + t.Errorf("events = %+v, want %+v", events, wantEvents) } if len(recs) != 1 || recs[0].Tokens != 7 || recs[0].Purpose != "chat" { t.Errorf("records = %+v, want 1 record tokens=7 purpose=chat", recs) @@ -377,10 +383,10 @@ func TestAnthropicChat_ChatStream_ToolUseReassembly(t *testing.T) { srv := newAnthropicServer(t, http.StatusOK, "text/event-stream", sse, nil) ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"}) - var deltas []string + var events []StreamEvent resp, err := ad.ChatStream(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}, Model: "claude-test"}, - func(d string) { deltas = append(deltas, d) }) + func(ev StreamEvent) { events = append(events, ev) }) if err != nil { t.Fatalf("ChatStream error: %v", err) } @@ -401,8 +407,14 @@ func TestAnthropicChat_ChatStream_ToolUseReassembly(t *testing.T) { if resp.TotalTokens != 18 { t.Errorf("resp.TotalTokens = %d, want 18", resp.TotalTokens) } - if len(deltas) != 0 { - t.Errorf("deltas = %v, want none (only tool_use)", deltas) + wantEvents := []StreamEvent{ + {Kind: StreamKindToolCallStart, Index: 0, ID: "toolu_1"}, + {Kind: StreamKindToolCallDelta, Index: 0, ID: "toolu_1", Delta: `{"location":`}, + {Kind: StreamKindToolCallDelta, Index: 0, ID: "toolu_1", Delta: `"Paris"}`}, + {Kind: StreamKindToolCallEnd, Index: 0, ID: "toolu_1"}, + } + if !reflect.DeepEqual(events, wantEvents) { + t.Errorf("events = %+v, want %+v", events, wantEvents) } } diff --git a/internal/runtime/model/fallback.go b/internal/runtime/model/fallback.go index 73a5979..c9aaf23 100644 --- a/internal/runtime/model/fallback.go +++ b/internal/runtime/model/fallback.go @@ -60,10 +60,10 @@ func (c *Chain) Chat(ctx context.Context, req Request, opts Options) (Response, } // ChatStream tries each step in order. If a step fails before delivering any -// delta, the chain moves to the next step; once a step has delivered output it +// event, the chain moves to the next step; once a step has delivered output it // is committed, so a mid-stream failure surfaces as an error rather than mixing // content across providers. -func (c *Chain) ChatStream(ctx context.Context, req Request, onChunk func(delta string)) (Response, error) { +func (c *Chain) ChatStream(ctx context.Context, req Request, onEvent func(StreamEvent)) (Response, error) { if len(c.steps) == 0 { return Response{}, ErrNoSteps } @@ -74,10 +74,10 @@ func (c *Chain) ChatStream(ctx context.Context, req Request, onChunk func(delta stepReq.Model = s.Model } delivered := false - wrapped := func(delta string) { + wrapped := func(ev StreamEvent) { delivered = true - if onChunk != nil { - onChunk(delta) + if onEvent != nil { + onEvent(ev) } } resp, err := s.Adapter.ChatStream(ctx, stepReq, wrapped) diff --git a/internal/runtime/model/openai.go b/internal/runtime/model/openai.go index 594a74e..b78a908 100644 --- a/internal/runtime/model/openai.go +++ b/internal/runtime/model/openai.go @@ -194,9 +194,9 @@ func (a *OpenAIChatAdapter) Chat(ctx context.Context, req Request, opts Options) return out, nil } -// ChatStream streams token-level deltas via onChunk and returns the aggregated +// ChatStream streams structured events via onEvent and returns the aggregated // response with usage when the provider reports it in the final chunk. -func (a *OpenAIChatAdapter) ChatStream(ctx context.Context, req Request, onChunk func(delta string)) (Response, error) { +func (a *OpenAIChatAdapter) ChatStream(ctx context.Context, req Request, onEvent func(StreamEvent)) (Response, error) { start := time.Now() if a.baseURL == "" { return Response{}, ErrUpstream @@ -222,7 +222,7 @@ func (a *OpenAIChatAdapter) ChatStream(ctx context.Context, req Request, onChunk if resp.StatusCode != http.StatusOK { return Response{}, a.upstreamError(resp) } - out, err := scanSSE(resp.Body, onChunk) + out, err := scanSSE(resp.Body, onEvent) if err != nil { return Response{}, err } @@ -380,12 +380,14 @@ func responseFromWire(wire *openAIChatResponse) Response { // scanSSE incrementally decodes stream chunks, accumulates the full content // with a strings.Builder (never reading the whole body at once) and captures -// usage from the final chunk. -func scanSSE(r io.Reader, onChunk func(string)) (Response, error) { +// usage from the final chunk. Each text delta is delivered as a text_delta +// StreamEvent with the body-text segment bracketed by text_start/text_end. +func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) { scanner := bufio.NewScanner(r) scanner.Buffer(make([]byte, 0, sseBufInit), maxSSELineBytes) agg := Response{} var b strings.Builder + started := false for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) if line == "" || !strings.HasPrefix(line, sseDataPrefix) { @@ -412,13 +414,20 @@ func scanSSE(r io.Reader, onChunk func(string)) (Response, error) { continue } b.WriteString(delta) - if onChunk != nil { - onChunk(delta) + if onEvent != nil { + if !started { + started = true + onEvent(StreamEvent{Kind: StreamKindTextStart}) + } + onEvent(StreamEvent{Kind: StreamKindTextDelta, Delta: delta}) } } if err := scanner.Err(); err != nil { return Response{}, fmt.Errorf("model: %w: read stream: %v", ErrUpstream, err) } + if onEvent != nil && started { + onEvent(StreamEvent{Kind: StreamKindTextEnd}) + } agg.Content = b.String() agg.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: agg.Content}}} return agg, nil diff --git a/internal/runtime/model/openai_test.go b/internal/runtime/model/openai_test.go index dd9bae8..f7b247e 100644 --- a/internal/runtime/model/openai_test.go +++ b/internal/runtime/model/openai_test.go @@ -200,10 +200,10 @@ func TestOpenAIChat_ChatStream_SSE(t *testing.T) { ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, Model: "gpt-test"}, WithRecorder(func(cr CallRecord) { recs = append(recs, cr) })) - var deltas []string + var events []StreamEvent resp, err := ad.ChatStream(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}, Model: "gpt-test", Purpose: "chat"}, - func(d string) { deltas = append(deltas, d) }) + func(ev StreamEvent) { events = append(events, ev) }) if err != nil { t.Fatalf("ChatStream error: %v", err) } @@ -216,8 +216,14 @@ func TestOpenAIChat_ChatStream_SSE(t *testing.T) { if resp.TotalTokens != 7 { t.Errorf("resp.TotalTokens = %d, want 7", resp.TotalTokens) } - if !reflect.DeepEqual(deltas, []string{"Hel", "lo"}) { - t.Errorf("deltas = %v, want [Hel lo]", deltas) + wantEvents := []StreamEvent{ + {Kind: StreamKindTextStart}, + {Kind: StreamKindTextDelta, Delta: "Hel"}, + {Kind: StreamKindTextDelta, Delta: "lo"}, + {Kind: StreamKindTextEnd}, + } + if !reflect.DeepEqual(events, wantEvents) { + t.Errorf("events = %+v, want %+v", events, wantEvents) } if len(recs) != 1 || recs[0].Tokens != 7 || recs[0].Purpose != "chat" { t.Errorf("records = %+v, want 1 record tokens=7 purpose=chat", recs) diff --git a/internal/runtime/model/stream.go b/internal/runtime/model/stream.go new file mode 100644 index 0000000..1704b3b --- /dev/null +++ b/internal/runtime/model/stream.go @@ -0,0 +1,60 @@ +package model + +// StreamKind enumerates the kinds of structured events produced while a +// completion streams. It mirrors OpenClaw's AssistantMessageEvent segments +// (text / thinking / tool_call) plus a tool-execution progress event, so the +// frontend can render each segment into its own surface (body, collapsed +// thinking block, tool card) rather than accumulating a single opaque token +// stream. +type StreamKind string + +const ( + // StreamKindTextStart opens the body-text segment. The single text segment + // of a response is bracketed by TextStart/TextEnd; deltas fill it. + StreamKindTextStart StreamKind = "text_start" + // StreamKindTextDelta is one body-text token fragment. + StreamKindTextDelta StreamKind = "text_delta" + // StreamKindTextEnd closes the body-text segment. + StreamKindTextEnd StreamKind = "text_end" + + // StreamKindThinkingStart opens a chain-of-thought segment (frontend + // collapsed block). Reserved: adapters that report reasoning emit it. + StreamKindThinkingStart StreamKind = "thinking_start" + // StreamKindThinkingDelta is one reasoning token fragment. + StreamKindThinkingDelta StreamKind = "thinking_delta" + // StreamKindThinkingEnd closes a reasoning segment. + StreamKindThinkingEnd StreamKind = "thinking_end" + + // StreamKindToolCallStart opens a tool invocation's streaming form. + StreamKindToolCallStart StreamKind = "tool_call_start" + // StreamKindToolCallDelta is a fragment of the tool arguments (e.g. the + // incremental input_json of an Anthropic tool_use block). + StreamKindToolCallDelta StreamKind = "tool_call_delta" + // 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" +) + +// StreamEvent is a single structured event in a streamed completion. It replaces +// the former plain text delta as the unit delivered by Adapter.ChatStream, so a +// consumer can distinguish body text from tool calls (and, in time, reasoning +// segments) and track each by Index/ID. +type StreamEvent struct { + // Kind classifies the event (see StreamKind). + Kind StreamKind + // Delta carries the incremental payload: body text for TextDelta, reasoning + // text for ThinkingDelta, or a partial_argument fragment for ToolCallDelta. + 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. + ID string +} diff --git a/internal/runtime/orchestration/scheduler_test.go b/internal/runtime/orchestration/scheduler_test.go index a7abd34..68a5e30 100644 --- a/internal/runtime/orchestration/scheduler_test.go +++ b/internal/runtime/orchestration/scheduler_test.go @@ -104,7 +104,7 @@ func (m *mockAdapter) Chat(context.Context, model.Request, model.Options) (model return m.resp, m.err } -func (m *mockAdapter) ChatStream(context.Context, model.Request, func(string)) (model.Response, error) { +func (m *mockAdapter) ChatStream(context.Context, model.Request, func(model.StreamEvent)) (model.Response, error) { return m.resp, m.err } diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index b1d2b34..6d61d3d 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -803,6 +803,40 @@ func TestHubStreamChat(t *testing.T) { } } +func TestHubBroadcastStream(t *testing.T) { + t.Parallel() + hb := newTestHub(nil, nil) + c := newFakeConn(hb, 32) + hb.register(c) + + // 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}) + ev := readFakeFrame(t, c) + if ev.Type != "message.delta" { + t.Fatalf("frame type = %q, want message.delta (frame=%+v)", ev.Type, ev) + } + data, ok := ev.Data.(map[string]any) + if !ok || data["delta"] != "Hel" { + t.Fatalf("frame data = %+v, want delta=Hel", ev.Data) + } + if ev.Seq != 1 { + t.Fatalf("frame seq = %d, want 1", ev.Seq) + } + + // 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}) + ev = readFakeFrame(t, c) + if ev.Type != "tool_call.end" { + t.Fatalf("second frame type = %q, want tool_call.end", ev.Type) + } + hb.broadcastStream("not-a-stream-event") + if err := connHasNoFrame(c); err != nil { + t.Fatalf("unexpected frame from non-StreamEvent payload: %v", err) + } +} + func TestHubStreamChatIdle(t *testing.T) { t.Parallel() hb := newTestHub(nil, nil) // nil runner diff --git a/internal/services/gateway/ws_bridge.go b/internal/services/gateway/ws_bridge.go index 8efa9ab..b4c1bff 100644 --- a/internal/services/gateway/ws_bridge.go +++ b/internal/services/gateway/ws_bridge.go @@ -4,6 +4,7 @@ import ( "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm" ) @@ -79,6 +80,38 @@ func subagentProgressEvent(e bus.Event) (wsEvent, bool) { }}, true } +// streamEventData maps a streamed model.StreamKind to its downstream WS event +// type and builds a typed data payload. It is what turns the loop's structured +// events into the message.*/thinking.*/tool_call.*/tool_execution.* frames the +// frontend renders. Unknown kinds fall back to the legacy stream_chunk shape so +// an unrecognized event still reaches the client instead of being dropped. +func streamEventData(ev model.StreamEvent) (string, map[string]any) { + switch ev.Kind { + case model.StreamKindTextStart: + return "message.start", map[string]any{"index": ev.Index} + case model.StreamKindTextDelta: + return "message.delta", map[string]any{"delta": ev.Delta, "index": ev.Index} + case model.StreamKindTextEnd: + return "message.end", map[string]any{"index": ev.Index} + case model.StreamKindThinkingStart: + return "thinking.start", map[string]any{"index": ev.Index} + case model.StreamKindThinkingDelta: + return "thinking.delta", map[string]any{"delta": ev.Delta, "index": ev.Index} + case model.StreamKindThinkingEnd: + return "thinking.end", map[string]any{"index": ev.Index} + case model.StreamKindToolCallStart: + return "tool_call.start", map[string]any{"id": ev.ID, "index": ev.Index} + case model.StreamKindToolCallDelta: + return "tool_call.delta", map[string]any{"id": ev.ID, "index": ev.Index, "delta": ev.Delta} + 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} + default: + return "stream_chunk", map[string]any{"delta": ev.Delta} + } +} + // isBroadcastAgentEvent reports whether subject has the exact 4-token broadcast // shape cyrene.agent..event, which is how the scheduler publishes progress. // It deliberately does not match addressed events (cyrene.agent...event) diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index bd910d9..4b8b565 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -2,11 +2,14 @@ package gateway import ( "context" + "crypto/rand" + "encoding/hex" "encoding/json" "log/slog" "net/http" "strings" "sync" + "sync/atomic" "time" "github.com/gorilla/websocket" @@ -23,6 +26,7 @@ const ( wsEventsPattern = "ws.>" wsPrefix = "ws." wsChatSubject = "ws.chat" + wsStreamSubject = "ws.stream" wsVoiceInput = "ws.voice_input" wsReadLimit = 4096 wsWriteWait = 10 * time.Second @@ -31,10 +35,14 @@ const ( wsSendBuffer = 32 ) -// wsEvent is the outbound (server→client) frame. +// wsEvent is the outbound (server→client) frame. TurnID/Seq are optional and +// used by the streaming path so a client can correlate a burst of streamed +// events to one assistant turn and keep them in order. type wsEvent struct { - Type string `json:"type"` - Data any `json:"data,omitempty"` + Type string `json:"type"` + TurnID string `json:"turnId,omitempty"` + Seq int `json:"seq,omitempty"` + Data any `json:"data,omitempty"` } // wsInbound is the client→server frame. Data carries the per-type payload. @@ -70,6 +78,7 @@ type hub struct { origins map[string]bool allowAll bool cancel context.CancelFunc + streamSeq atomic.Uint64 } // newHub builds a hub bound to the server's dependencies. @@ -178,6 +187,12 @@ func (h *hub) relay(ctx context.Context, e bus.Event) error { go h.streamChat(ctx, e.Payload) return nil } + if e.Subject == wsStreamSubject { + // OnStream→bus→relay path: the loop already published a structured + // model.StreamEvent; map it to the typed downstream event. + h.broadcastStream(e.Payload) + return nil + } ev, ok := normalizeEvent(e) if !ok { return nil @@ -186,16 +201,20 @@ func (h *hub) relay(ctx context.Context, e bus.Event) error { return nil } -// streamChat runs one agent turn for a chat trigger and broadcasts the result as -// stream_chunk/stream_end. It runs on its own goroutine so a slow model call does -// not stall the bus subscription worker. +// streamChat runs one agent turn for a chat trigger and broadcasts the turn +// terminus (stream_end) with a fresh turnId; on the non-streaming fallback path +// it also emits a single stream_chunk. When the loop is streaming, the per-event +// deltas arrive via the OnStream→bus→relay path as structured message.*/ +// thinking.*/tool_call.* events. It runs on its own goroutine so a slow model +// call does not stall the bus subscription worker. func (h *hub) streamChat(ctx context.Context, payload any) { var req chatStreamRequest if b, err := json.Marshal(payload); err == nil { _ = json.Unmarshal(b, &req) } + turnID := newTurnID() if h.runner == nil || strings.TrimSpace(req.Text) == "" { - h.broadcast(wsEvent{Type: "stream_end", Data: map[string]any{"stopReason": "idle"}}) + h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "idle"}}) return } if req.SessionID == "" { @@ -204,17 +223,33 @@ func (h *hub) streamChat(ctx context.Context, payload any) { res, err := h.runner.Run(ctx, req.SessionID, []model.Message{model.TextMessage(model.RoleUser, req.Text)}) if err != nil { h.log.Error("ws chat run", "err", err, "session", req.SessionID) - h.broadcast(wsEvent{Type: "stream_end", Data: map[string]any{"stopReason": "error", "error": err.Error()}}) + h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": "error", "error": err.Error()}}) return } - // When the loop is streaming (OnStream wired), the deltas already fan out as - // ws.stream_chunk through the OnStream→bus→relay path; a second whole-text - // chunk would duplicate the tail. The non-streaming path (or a stub runner) - // still emits one chunk so the stream contract holds in tests. + // When the loop is streaming (OnStream wired), the structured deltas already + // fan out as message.*/thinking.*/tool_call.* events through the + // OnStream→bus→relay path; a second whole-text chunk would duplicate the + // tail. The non-streaming path (or a stub runner) still emits one chunk so + // the stream contract holds in tests. if !h.streaming { - h.broadcast(wsEvent{Type: "stream_chunk", Data: map[string]any{"delta": res.Text}}) + h.broadcast(wsEvent{Type: "stream_chunk", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"delta": res.Text}}) } - h.broadcast(wsEvent{Type: "stream_end", Data: map[string]any{"stopReason": res.StopReason, "steps": res.Steps}}) + h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), Data: map[string]any{"stopReason": res.StopReason, "steps": res.Steps}}) +} + +// 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). +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}) } // chatStreamRequest is the decoded ws.chat payload. @@ -372,6 +407,18 @@ func eventType(subject string) string { return strings.TrimPrefix(subject, wsPrefix) } +// newTurnID returns a fresh random hex identifier used to correlate a burst of +// streamed WS events to one assistant turn. It burns the crypto/rand source and +// falls back to a nanosecond timestamp if the source is unavailable, so a id is +// always produced (a blank turnId would break client correlation silently). +func newTurnID() string { + var b [8]byte + if _, err := rand.Read(b[:]); err == nil { + return "turn-" + hex.EncodeToString(b[:]) + } + return "turn-" + time.Now().UTC().Format("20060102150405.000000000") +} + // rawData converts a possibly-empty raw payload into a value safe to publish. func rawData(d json.RawMessage) any { if len(d) == 0 {