AskaEth/Cyrene_v2
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(web,ws): 会话级消息缓存 + 图桶 + 忙态按会话过滤（裁决 16.4.5/16.4.2 + 修 16.4.1 的帧缺口）

Browse Source 
- 16.4.5：非激活会话的帧**不再丢弃**，按自己的 sessionId 落进该会话桶（新增 store/sessionCache.ts 纯函数 +
  useAppStore.sessionMessages/flowGraphs）；切会话改为「存回/取出」而非清空；流式增量 O(1) 定位桶、
  只延长已存在的助手气泡（不凭一帧 delta 凭空造消息）
- 16.4.2：图按会话入桶（上限 12 个），Backthink 页取 sess-backthink 的图而不再被用户会话图顶替；
  空态区分「实时通道未连接」与「尚未产生链路图」；wire_runtime 的图源白名单注释按裁决口径重写（零行为改动）
- 修 16.4.1 的缺口：handlers/session.ts 补 `session_idle` / `session_interrupted`（此前落到未知帧兜底 ⇒
  sessionBusy 可能卡 true ⇒ 预约永不落地）；忙态**置忙与置闲两侧**统一按 frameIsActive 过滤（只过滤置闲
  会让后台会话的忙帧点亮徽标而它自己的 idle 被丢 ⇒ 永久卡忙）
- 清理：删除 lib/streamScope.ts（口径已被 16.4.5 推翻）与只剩测试在用的 messageEventApplies
- 复核：tsc 0 错 + vite build 成功；messageSync 69 / session 19 项逻辑测试；4 条变异反证
This commit is contained in:
AskaEth committed 2026-10-04 13:41:02 +08:00
1 parent f454273c5a
commit b480b04dd0
15 files changed
+733 -300
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

frontend/web/src/store/messageSync.test.ts
+116 -35
View File
@@ -1,9 +1,11 @@
// Logic test for the realtime-sync frames: the D3 edit/delete pair
// (`message_edited` / `message_deleted`), the streaming session guard
// (design/16 §4.2 — a stream frame for another conversation must not land in the
// bubble on screen), and the session read cursor (Android 协同 B3/D4 — the
// `session_read` frame clears exactly the conversation it names, and opening a
// session pushes its forward-only cursor). The project has no frontend test runner
// (`message_edited` / `message_deleted`), the per-session message cache
// (裁决 16.4.5 — a frame is written into the bucket its own `sessionId` names, and
// switching sessions keeps the list instead of clearing it), the per-session flow
// graphs (裁决 16.4.2 — a backthink graph never overwrites the active session's),
// and the session read cursor (Android 协同 B3/D4 — the `session_read` frame clears
// exactly the conversation it names, and opening a session pushes its forward-only
// cursor). 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), the real read path
// (`markSessionOpenedRead` in lib/sessionRead.ts, via lib/api) and the real store,
@@ -25,7 +27,9 @@
import { dispatchFrame } from "../lib/ws";
import { markSessionOpenedRead } from "../lib/sessionRead";
import { messageEventApplies, useAppStore } from "./useAppStore";
import { frameIsActive } from "../lib/ws/handlers/shared";
import { useAppStore } from "./useAppStore";
import { BACKTHINK_SESSION_ID } from "./sessionCache";
import type { ChatMessage, SessionMeta } from "../lib/types";
let checks = 0;
@@ -42,15 +46,21 @@ function same<T>(got: T, want: T, what: string): void {
const MAIN = "web";
const OTHER = "web:sub-1";
/** Reset the store to a focused session holding a fixed message list. */
/** Reset the store to a focused session holding a fixed message list (and no
* buckets, so each case starts from a known cache state). */
function seed(messages: ChatMessage[], activeSessionId: string): void {
useAppStore.setState({ activeSessionId, messages });
useAppStore.setState({ activeSessionId, messages, sessionMessages: {} });
}
function messages(): ChatMessage[] {
return useAppStore.getState().messages;
}
/** The cached message bucket of a session that is not on screen. */
function bucket(sessionId: string): ChatMessage[] {
return useAppStore.getState().sessionMessages[sessionId] ?? [];
}
function base(): ChatMessage[] {
return [
{ id: "m1", sessionId: MAIN, role: "user", content: "old", mode: "chat", createdAt: 1000 },
@@ -59,9 +69,13 @@ function base(): ChatMessage[] {
}
// ① Session filter: a frame scoped to another conversation never touches this one.
ok(messageEventApplies(MAIN, MAIN), "same session applies");
ok(!messageEventApplies(OTHER, MAIN), "another session does not apply");
ok(messageEventApplies(undefined, MAIN), "an unscoped (id-only) frame applies");
// 这是单槽全局状态（忙态 / 上下文 / 代审小字 / 回合结束的收口）共用的判据，
// 实现在 lib/ws/handlers/shared.ts。
ok(frameIsActive(MAIN, MAIN), "same session applies");
ok(!frameIsActive(OTHER, MAIN), "another session does not apply");
ok(frameIsActive(undefined, MAIN), "an unscoped (id-only) frame applies");
// 激活会话尚未解析（会话列表没回来）时沿用旧行为，不把主人正看着的流丢掉。
ok(frameIsActive(MAIN, ""), "an unresolved active session keeps the legacy behavior");
// ② message_edited patches the addressed bubble in place, by id, in the active
// session — and leaves its siblings alone.
@@ -117,10 +131,10 @@ 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");
// ⑤ 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.
// ⑤ 会话级消息缓存（裁决 16.4.5 / design/16 §4.3）。hub 把每个会话的帧都广播给每条
// 连接，因此一帧要落到**它自己那个会话**：属于屏幕上这个会话就写当前列表，属于别的
// 会话就写它的消息桶（切过去即见），而不是被丢掉。只有单槽全局状态（忙态 / 上下文）
// 仍只认「屏幕上这个会话」。
function streaming(): ChatMessage[] {
return [
{ id: "s1", sessionId: MAIN, role: "user", content: "hi", createdAt: 3000 },
@@ -136,10 +150,36 @@ function streaming(): ChatMessage[] {
];
}
// Another session's delta never touches this bubble …
// A foreign delta never touches this bubble, and it does not invent a bubble in the
// other session's bucket either (a delta can only extend an assistant bubble that
// already exists there — the persisted row covers the rest).
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");
same(messages().map((m) => m.content), ["hi", "partial"], "delta for another session does not touch the active bubble");
same(bucket(OTHER), [], "a delta without a bubble in that session invents nothing");
// … but when the bucket already holds that turn's bubble — the common case, because
// switching away stashes the optimistic bubbles into it — the delta lands there and
// is on screen the moment the session is switched to.
seed(streaming(), MAIN);
useAppStore.setState({ sessionMessages: { [OTHER]: streaming() } });
dispatchFrame({ type: "message.delta", sessionId: OTHER, data: { delta: "+bg", index: 0 } });
same(bucket(OTHER)[1].content, "partial+bg", "a background turn keeps streaming into its own bucket");
same(messages()[1].content, "partial", "…and the active bubble stays untouched");
useAppStore.getState().setActiveSession(OTHER);
same(messages()[1].content, "partial+bg", "switching to the background session shows its live text");
useAppStore.getState().setActiveSession(MAIN);
same(messages().map((m) => m.content), ["hi", "partial"], "switching back restores the first session (nothing cleared)");
// A persisted message for another session is cached as well (never dropped).
seed(base(), MAIN);
dispatchFrame({
type: "message_persisted",
sessionId: OTHER,
data: { message: { id: "m9", sessionId: OTHER, role: "assistant", content: "bg", createdAt: 5000 } },
});
same(bucket(OTHER).map((m) => m.content), ["bg"], "message_persisted for another session is cached in its bucket");
same(messages().length, 2, "…and leaves the active list alone");
// … the active session's delta is appended as before …
seed(streaming(), MAIN);
@@ -156,31 +196,30 @@ 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.
// Before the session list has resolved the active session the store still holds "",
// while the frames carry the main session: the text is filed under that session (not
// spliced into the unresolved list) and appears as soon as the list resolves.
seed(streaming(), "");
useAppStore.setState({ sessionMessages: { [MAIN]: 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"
);
same(bucket(MAIN)[1].content, "partial early", "a delta before the active session resolves lands in the named session's bucket");
useAppStore.getState().setActiveSession(MAIN);
same(messages()[1].content, "partial early", "…and is on screen once the active session resolves");
// The legacy text channel (stream_chunk) is filtered the same way.
// The legacy text channel (stream_chunk) is routed 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");
same(messages()[1].content, "partial", "legacy stream_chunk for another session does not touch the active bubble");
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.
// Thinking segments: the accumulator and the rendered block are per session, so a
// background turn's reasoning never lands in the block on screen.
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");
ok(messages()[1].thinking === undefined, "thinking.delta for another session does not touch this bubble");
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 } });
@@ -188,10 +227,11 @@ same(messages()[1].thinking, "mine", "thinking.end for another session leaves th
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.
// Tool cards land on the last assistant message of their own session 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");
same(messages()[1].tools, [], "tool_call.start for another session adds no card here");
same(bucket(OTHER), [], "…and invents no bucket for it");
dispatchFrame({ type: "tool_call.start", sessionId: MAIN, data: { id: "toolu_1", name: "web_search" } });
same(
(messages()[1].tools ?? []).map((t) => t.name),
@@ -200,17 +240,21 @@ same(
);
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");
same(messages()[1].tools, [], "legacy tool_used for another session adds no card here");
// stream_end carries single-slot state (the streaming flag + the busy badge), so
// another session finishing its turn must not close this one.
// another session finishing its turn must not close this one — it only closes its
// own bubble, in its own bucket.
seed(streaming(), MAIN);
useAppStore.setState({ sessionBusy: true, pendingCount: 2 });
useAppStore.setState({ sessionBusy: true, pendingCount: 2, sessionMessages: { [OTHER]: streaming() } });
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(bucket(OTHER)[1].streaming === false, "…it closes that turn's bubble in its own bucket");
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");
seed(streaming(), MAIN);
useAppStore.setState({ sessionBusy: true, pendingCount: 2 });
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");
@@ -328,4 +372,41 @@ same(calls.length, 1, "a missing session metadata entry sends no read call");
globalThis.fetch = originalFetch;
globalThis.localStorage = originalLocalStorage;
// ⑦ 会话级链路图（裁决 16.4.2）。`graph_update` 快照按它自己的 sessionId 入桶：
// 后台思考的图不会顶掉管理端 Flow / Agents 页正在看的「当前会话」图，管理后台
// 因此能同时拿到每一路（用户会话 / 后台思考）的完整链路。
seed([], MAIN);
dispatchFrame({
type: "graph_update",
data: { id: "web-1", nodes: [], edges: [], sessionId: MAIN },
});
dispatchFrame({
type: "graph_update",
data: { id: "sess-backthink-1", nodes: [], edges: [], sessionId: BACKTHINK_SESSION_ID },
});
same(
useAppStore.getState().flowGraphs[MAIN]?.id,
"web-1",
"a graph is stored under its own sessionId"
);
same(
useAppStore.getState().flowGraphs[BACKTHINK_SESSION_ID]?.id,
"sess-backthink-1",
"the backthink graph lives in its own bucket"
);
same(
useAppStore.getState().flowGraph?.id,
"web-1",
"the compatibility alias stays on the active session (a backthink graph never takes it over)"
);
// A graph without a sessionId (legacy/unscoped frame) falls back to the active session.
seed([], MAIN);
dispatchFrame({ type: "graph_update", data: { id: "bare-1", nodes: [], edges: [] } });
same(
useAppStore.getState().flowGraphs[MAIN]?.id,
"bare-1",
"an unscoped graph falls back to the active session"
);
console.log(`message sync logic test: ${checks} checks passed`);