feat(web): 流式帧按会话过滤,非激活会话不再串入当前气泡
- 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 兼容/未解析回退)
This commit is contained in:
1 parent
61cc32c44b
commit
04a8ee0727
4 files changed
+171
-7
No files matched your search
@@ -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<string, unknown> | 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);
|
||||
}
|
||||
@@ -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). */
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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`);
|
||||
Reference in new issue
Block a user