From 04a8ee07279a2b8575f7ba8a3670369632615e13 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 3 Oct 2026 16:08:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(web):=20=E6=B5=81=E5=BC=8F=E5=B8=A7?= =?UTF-8?q?=E6=8C=89=E4=BC=9A=E8=AF=9D=E8=BF=87=E6=BB=A4=EF=BC=8C=E9=9D=9E?= =?UTF-8?q?=E6=BF=80=E6=B4=BB=E4=BC=9A=E8=AF=9D=E4=B8=8D=E5=86=8D=E4=B8=B2?= =?UTF-8?q?=E5=85=A5=E5=BD=93=E5=89=8D=E6=B0=94=E6=B3=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - WsDownEvent 增 sessionId；流式分支（message/thinking/tool_call/tool_execution/stream_chunk/stream_end 及 legacy thinking/tool_used）统一过 streamFrameApplies - 会话列表未解析(activeSessionId 为空)时保持改前行为，避免误丢主人自己的流 - 抽出 lib/streamScope.ts 承载该判断，ws.ts 回到 500 行红线内 - messageSync.test.ts 增 21 项断言（跨会话丢弃/激活追加/无 session 兼容/未解析回退） --- frontend/web/src/lib/streamScope.ts | 30 ++++++ frontend/web/src/lib/types.ts | 13 ++- frontend/web/src/lib/ws.ts | 19 ++++ frontend/web/src/store/messageSync.test.ts | 116 ++++++++++++++++++++- 4 files changed, 171 insertions(+), 7 deletions(-) create mode 100644 frontend/web/src/lib/streamScope.ts diff --git a/frontend/web/src/lib/streamScope.ts b/frontend/web/src/lib/streamScope.ts new file mode 100644 index 0000000..68a3659 --- /dev/null +++ b/frontend/web/src/lib/streamScope.ts @@ -0,0 +1,30 @@ +import { messageEventApplies } from "../store/useAppStore"; + +/** May a streaming frame write into the conversation the UI is showing? + * + * The hub broadcasts every session's stream to every connection (design/16 §4.2), + * while `appendChunk` / `endStream` / `setThinking` / `addTool` all target the + * *last* assistant message of the one message list — so a background + * conversation's tokens would otherwise be spliced into the bubble on screen. + * The frame's own session (`data.sessionId` when present, else the envelope's) + * decides. + * + * An unresolved active session (the session list has not loaded yet, or its fetch + * failed) keeps the pre-guard behavior rather than dropping the owner's own + * stream: the backend maps an empty request session to the main session, so those + * frames carry "web" while the store still holds "". + * + * Scope note (deliberate): a foreign frame is *dropped*, not buffered — the store + * holds the active session only, and that conversation's text arrives when it is + * switched to and its history is fetched. Buffering live deltas needs the + * per-session cache of design/16 §4.3, a separate step. */ +export function streamFrameApplies( + ev: { sessionId?: string }, + r: Record | null, + activeSessionId: string +): boolean { + // Unresolved session list: keep the pre-guard path (see the doc comment). + if (!activeSessionId) return true; + const sessionId = typeof r?.sessionId === "string" ? r.sessionId : ev.sessionId; + return messageEventApplies(sessionId, activeSessionId); +} diff --git a/frontend/web/src/lib/types.ts b/frontend/web/src/lib/types.ts index 0661aa8..9c77abb 100644 --- a/frontend/web/src/lib/types.ts +++ b/frontend/web/src/lib/types.ts @@ -215,19 +215,28 @@ export interface ChatSyncResult { /** * Downstream WS frame (server → client). `data` is intentionally `unknown`; - * every consumer narrows it. The backend emits `{type, data, turnId?, seq?}` - * and types include the structured stream events message.start/delta/end, + * every consumer narrows it. The backend emits `{type, data, turnId?, seq?, + * sessionId?}` 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, and the capacity/session events from design/14 §5 + 15 §5: * capacity_awaken (状态切换+点亮工具), session_busy / session_pending (她是否在忙+挂起消息数). + * + * `sessionId` (design/16 §4.2) scopes the frame to one conversation. The hub + * broadcasts globally — every connection of the owner sees every session — so the + * client filters by the active session: a frame belonging to another conversation + * (a concurrent stream, an edit, a delete) must never reach the rendered bubble. + * It is absent when the event is not conversation-scoped (panel/device events) or + * when the backend could not resolve the conversation; consumers then keep the old + * permissive behavior rather than dropping the frame. */ export interface WsDownEvent { type: string; data: unknown; turnId?: string; seq?: number; + sessionId?: string; } /** Upstream WS frame (client → server). */ diff --git a/frontend/web/src/lib/ws.ts b/frontend/web/src/lib/ws.ts index ff0149b..1b9e271 100644 --- a/frontend/web/src/lib/ws.ts +++ b/frontend/web/src/lib/ws.ts @@ -14,6 +14,7 @@ import { toRecord, } from "./wsParse"; import { MODES, type AwakenState, type ChatMessage, type Mode, type WsUpEvent } from "./types"; +import { streamFrameApplies } from "./streamScope"; /** * Realtime client for `/ws/chat`. Reconnects with exponential backoff and @@ -83,10 +84,15 @@ export function dispatchFrame(ev: { type: string; sessionId?: string; data: unkn switch (ev.type) { case "stream_chunk": { + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; if (r && typeof r.delta === "string") s.appendChunk(r.delta); break; } case "stream_end": { + // Another session's turn ending must not close the streaming bubble on + // screen or clear this conversation's busy badge (both are single-slot + // state: the last message's flag, and one shell-wide `sessionBusy`). + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; s.endStream(); // The assistant reply stopped streaming, so the immediate turn is over: // close the busy/pending window so the badge does not stick. If the backend @@ -104,6 +110,7 @@ export function dispatchFrame(ev: { type: string; sessionId?: string; data: unkn break; } case "message.delta": { + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; if (r && typeof r.delta === "string") s.appendChunk(r.delta); break; } @@ -167,10 +174,14 @@ export function dispatchFrame(ev: { type: string; sessionId?: string; data: unkn break; } case "thinking.start": { + // Guarded like the deltas: only the active conversation may reset the + // accumulator (which also bounds a switched-away session's stale text). + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; pendingThinking = ""; break; } case "thinking.delta": { + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; if (r && typeof r.delta === "string") { pendingThinking += r.delta; s.setThinking(pendingThinking); @@ -178,12 +189,14 @@ export function dispatchFrame(ev: { type: string; sessionId?: string; data: unkn break; } case "thinking.end": { + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; if (pendingThinking) s.setThinking(pendingThinking); pendingThinking = ""; break; } case "tool_call.start": case "tool_call.end": { + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; const tool = parseToolCall(ev.data); if (tool) s.addTool(tool); break; @@ -194,11 +207,15 @@ export function dispatchFrame(ev: { type: string; sessionId?: string; data: unkn break; } case "tool_execution.update": { + // Same card as tool_call.* — same guard. + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; const tool = parseToolCall(ev.data); if (tool) s.addTool(tool); break; } case "thinking": { + // Legacy single-shot CoT frame — renders into the same bubble. + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; if (typeof ev.data === "string") { s.setThinking(ev.data); } else if (r && typeof r.text === "string") { @@ -212,6 +229,8 @@ export function dispatchFrame(ev: { type: string; sessionId?: string; data: unkn break; } case "tool_used": { + // Legacy tool echo — same card as tool_call.*. + if (!streamFrameApplies(ev, r, s.activeSessionId)) break; const tool = parseTool(ev.data); if (tool) s.addTool(tool); break; diff --git a/frontend/web/src/store/messageSync.test.ts b/frontend/web/src/store/messageSync.test.ts index 7953f2a..512410e 100644 --- a/frontend/web/src/store/messageSync.test.ts +++ b/frontend/web/src/store/messageSync.test.ts @@ -1,7 +1,10 @@ -// Logic test for the D3 realtime-sync frames (`message_edited` / `message_deleted`). -// The project has no frontend test runner and no jsdom, so this is a dependency-free -// plain-node check: it drives the real frame decoder (`dispatchFrame` in lib/ws.ts) -// and the real store, and throws on the first failure. +// Logic test for the realtime-sync frames: the D3 edit/delete pair +// (`message_edited` / `message_deleted`) and the streaming session guard +// (design/16 §4.2 — a stream frame for another conversation must not land in the +// bubble on screen). The project has no frontend test runner and no jsdom, so this +// is a dependency-free plain-node check: it drives the real frame decoder +// (`dispatchFrame` in lib/ws.ts) and the real store, and throws on the first +// failure. // // Run it with the esbuild that already ships with vite (no new dependency, no // package.json change); the banner supplies the `window.localStorage` the store's @@ -110,4 +113,107 @@ useAppStore.getState().updateMessageById("m-ghost", { content: "x" }); useAppStore.getState().removeMessageById("m-ghost"); same(messages().length, 2, "store id-actions no-op on an unknown id"); -console.log(`message_edited/message_deleted logic test: ${checks} checks passed`); +// ⑤ Streaming session guard (design/16 §4.2). The hub broadcasts every session's +// stream to every connection, and `appendChunk` writes into the last assistant +// message of the single active list — so a delta carrying another sessionId +// would otherwise splice that conversation's tokens into this bubble. +function streaming(): ChatMessage[] { + return [ + { id: "s1", sessionId: MAIN, role: "user", content: "hi", createdAt: 3000 }, + { + id: "s2", + sessionId: MAIN, + role: "assistant", + content: "partial", + streaming: true, + tools: [], + createdAt: 4000, + }, + ]; +} + +// Another session's delta never touches this bubble … +seed(streaming(), MAIN); +dispatchFrame({ type: "message.delta", sessionId: OTHER, data: { delta: "LEAK", index: 0 } }); +same(messages().map((m) => m.content), ["hi", "partial"], "delta for another session is dropped"); + +// … the active session's delta is appended as before … +seed(streaming(), MAIN); +dispatchFrame({ type: "message.delta", sessionId: MAIN, data: { delta: " world", index: 0 } }); +same(messages()[1].content, "partial world", "delta for the active session is appended"); + +// … a payload-carried sessionId is honored too … +seed(streaming(), MAIN); +dispatchFrame({ type: "message.delta", sessionId: OTHER, data: { sessionId: MAIN, delta: "!" } }); +same(messages()[1].content, "partial!", "a payload sessionId overrides the envelope's"); + +// … and a frame with no sessionId at all keeps the pre-guard behavior. +seed(streaming(), MAIN); +dispatchFrame({ type: "message.delta", data: { delta: " old", index: 0 } }); +same(messages()[1].content, "partial old", "an unscoped delta keeps the legacy behavior"); + +// … as does the owner's own stream before the session list has resolved the active +// session: the store still holds "" while the frames carry the main session, so +// dropping them would hide a reply the user is watching. +seed(streaming(), ""); +dispatchFrame({ type: "message.delta", sessionId: MAIN, data: { delta: " early", index: 0 } }); +same( + messages()[1].content, + "partial early", + "a delta before the active session resolves keeps the legacy behavior" +); + +// The legacy text channel (stream_chunk) is filtered the same way. +seed(streaming(), MAIN); +dispatchFrame({ type: "stream_chunk", sessionId: OTHER, data: { delta: "LEAK" } }); +same(messages()[1].content, "partial", "legacy stream_chunk for another session is dropped"); +seed(streaming(), MAIN); +dispatchFrame({ type: "stream_chunk", data: { delta: " tail" } }); +same(messages()[1].content, "partial tail", "an unscoped stream_chunk keeps the legacy behavior"); + +// Thinking segments: the accumulator and the rendered block belong to the active +// session only. +seed(streaming(), MAIN); +dispatchFrame({ type: "thinking.start", sessionId: MAIN, data: { index: 0 } }); +dispatchFrame({ type: "thinking.delta", sessionId: OTHER, data: { delta: "other-thought", index: 0 } }); +ok(messages()[1].thinking === undefined, "thinking.delta for another session is dropped"); +dispatchFrame({ type: "thinking.delta", sessionId: MAIN, data: { delta: "mine", index: 0 } }); +same(messages()[1].thinking, "mine", "thinking.delta for the active session accumulates"); +dispatchFrame({ type: "thinking.end", sessionId: OTHER, data: { index: 0 } }); +same(messages()[1].thinking, "mine", "thinking.end for another session leaves the block alone"); +dispatchFrame({ type: "thinking.end", sessionId: MAIN, data: { index: 0 } }); +same(messages()[1].thinking, "mine", "thinking.end for the active session keeps the text"); + +// Tool cards land on the last assistant message too. +seed(streaming(), MAIN); +dispatchFrame({ type: "tool_call.start", sessionId: OTHER, data: { id: "toolu_1", name: "web_search" } }); +same(messages()[1].tools, [], "tool_call.start for another session adds no card"); +dispatchFrame({ type: "tool_call.start", sessionId: MAIN, data: { id: "toolu_1", name: "web_search" } }); +same( + (messages()[1].tools ?? []).map((t) => t.name), + ["web_search"], + "tool_call.start for the active session adds the card" +); +seed(streaming(), MAIN); +dispatchFrame({ type: "tool_used", sessionId: OTHER, data: { name: "shell" } }); +same(messages()[1].tools, [], "legacy tool_used for another session adds no card"); + +// stream_end carries single-slot state (the streaming flag + the busy badge), so +// another session finishing its turn must not close this one. +seed(streaming(), MAIN); +useAppStore.setState({ sessionBusy: true, pendingCount: 2 }); +dispatchFrame({ type: "stream_end", sessionId: OTHER, data: { stopReason: "end_turn" } }); +ok(messages()[1].streaming === true, "a foreign stream_end does not close the active bubble"); +ok(useAppStore.getState().sessionBusy === true, "a foreign stream_end does not clear the busy badge"); +same(useAppStore.getState().pendingCount, 2, "a foreign stream_end does not clear the pending count"); + +dispatchFrame({ type: "stream_end", sessionId: MAIN, data: { stopReason: "end_turn" } }); +ok(messages()[1].streaming === false, "the active session's stream_end closes the bubble"); +ok(useAppStore.getState().sessionBusy === false, "the active session's stream_end clears busy"); +same(useAppStore.getState().pendingCount, 0, "the active session's stream_end clears the pending count"); + +seed(streaming(), MAIN); +dispatchFrame({ type: "stream_end", data: { stopReason: "end_turn" } }); +ok(messages()[1].streaming === false, "an unscoped stream_end keeps the legacy behavior"); + +console.log(`message sync logic test: ${checks} checks passed`);