From b480b04dd0bed53a6538ff4df5cb791e9367a2fd Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 13:41:02 +0800 Subject: [PATCH] =?UTF-8?q?feat(web,ws):=20=E4=BC=9A=E8=AF=9D=E7=BA=A7?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E7=BC=93=E5=AD=98=20+=20=E5=9B=BE=E6=A1=B6?= =?UTF-8?q?=20+=20=E5=BF=99=E6=80=81=E6=8C=89=E4=BC=9A=E8=AF=9D=E8=BF=87?= =?UTF-8?q?=E6=BB=A4=EF=BC=88=E8=A3=81=E5=86=B3=2016.4.5/16.4.2=20+=20?= =?UTF-8?q?=E4=BF=AE=2016.4.1=20=E7=9A=84=E5=B8=A7=E7=BC=BA=E5=8F=A3?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 条变异反证 --- cmd/cyrene/wire_runtime.go | 18 +- frontend/web/src/admin/modules/Backthink.tsx | 17 +- frontend/web/src/lib/streamScope.ts | 30 -- frontend/web/src/lib/ws/handlers/agent.ts | 5 +- frontend/web/src/lib/ws/handlers/approval.ts | 8 +- frontend/web/src/lib/ws/handlers/message.ts | 43 ++- .../web/src/lib/ws/handlers/session.test.ts | 134 ++++++++ frontend/web/src/lib/ws/handlers/session.ts | 22 +- frontend/web/src/lib/ws/handlers/shared.ts | 22 ++ frontend/web/src/lib/ws/handlers/stream.ts | 34 +-- frontend/web/src/lib/ws/handlers/thinking.ts | 44 +-- frontend/web/src/lib/ws/handlers/tool.ts | 19 +- frontend/web/src/store/messageSync.test.ts | 151 ++++++--- frontend/web/src/store/sessionCache.ts | 199 ++++++++++++ frontend/web/src/store/useAppStore.ts | 287 +++++++++--------- 15 files changed, 733 insertions(+), 300 deletions(-) delete mode 100644 frontend/web/src/lib/streamScope.ts create mode 100644 frontend/web/src/lib/ws/handlers/session.test.ts create mode 100644 frontend/web/src/store/sessionCache.ts diff --git a/cmd/cyrene/wire_runtime.go b/cmd/cyrene/wire_runtime.go index 0f5f221..6a5d479 100644 --- a/cmd/cyrene/wire_runtime.go +++ b/cmd/cyrene/wire_runtime.go @@ -173,15 +173,21 @@ func (a *app) wireRuntime() error { tracer := trace.New(a.traceWriter) // Realtime flow visualization (design/19 §1): the publisher accumulates the // loop's trace spans into a per-session flow graph and publishes - // ws.graph_update on each span transition. It is gated to the owner's web - // sessions ("web" and web sub-sessions), so IM / stranger / sub-agent turns - // never leak into the admin panel. A nil flow keeps the loop's prior behavior. + // ws.graph_update on each span transition. + // + // 图源白名单（裁决 16.4.2）：它决定**哪些会话产出实时链路图**，也是后端能提供 + // 完整链路导图的那一侧 —— 主人 web 会话族（"web" 主会话 + "web:" 次级会话， + // IM 主人私聊镜像也落主会话）与自主后台思考会话（design/20 §2）。子代理的 span 由 + // loop.WithFlowBranch 并进**派生它的那个会话**，渲染成该会话图的一条泳道（design/19 + // §2），因此也在这份白名单里。 + // + // 注意口径（裁决 16.4.2）：design/19 §1 的「仅发布/不发」说的是**用户侧前端**观察到 + // 的现象，不是「后端可以少画图」——管理后台的图源由此保持完整，与用户端发布策略 + // 是两件事。隔离的渠道会话（群聊/陌生人，sess-xxx）不在此白名单内：那是 design/19 §1 + // 发布会话范围（裁决 01.4.5）的既有裁决，本批未改。nil flow 保持 loop 原行为。 flowPub := loop.NewFlowPublisher( func(subject string, payload any) { _ = a.bus.Publish(context.Background(), subject, payload) }, func(sessionID string) bool { - // Publish the web/main flow graph (web chat + IM owner-private mirror) AND the - // autonomous-backthink session, so the backthink page's flow link graph renders - // too (design/20 §2). Other isolated channel sessions (sess-xxx) stay filtered. return sessionID == history.MainSessionID || strings.HasPrefix(sessionID, history.MainSessionID+":") || sessionID == backthink.DefaultSessionID diff --git a/frontend/web/src/admin/modules/Backthink.tsx b/frontend/web/src/admin/modules/Backthink.tsx index 17c70af..85c81bc 100644 --- a/frontend/web/src/admin/modules/Backthink.tsx +++ b/frontend/web/src/admin/modules/Backthink.tsx @@ -8,10 +8,15 @@ import { Icon } from "../../components/Icon"; import { FlowGraphView } from "./FlowGraph"; import { adminBackthink } from "../../lib/api"; import { useAsyncData } from "../../lib/useAsyncData"; +import { BACKTHINK_SESSION_ID } from "../../store/sessionCache"; function BackthinkRow({ item }: { item: BackthinkItem }): React.ReactNode { const markPushed = useAppStore((s) => s.markBackthinkPushed); - const flowGraph = useAppStore((s) => s.flowGraph); + // 裁决 16.4.2：这条思考的链路图取自**后台思考会话**（`sess-backthink`，design/19 §1 + // 的发布会话范围）的快照，而不是管理端 Flow 页那张「当前会话」图 —— 后者属于用户端 + // 会话，与后台思考的完整链路是两件事，且不会被思考图顶掉。 + const graph = useAppStore((s) => s.flowGraphs[BACKTHINK_SESSION_ID] ?? null); + const wsStatus = useAppStore((s) => s.wsStatus); const [open, setOpen] = useState(false); return ( @@ -39,11 +44,13 @@ function BackthinkRow({ item }: { item: BackthinkItem }): React.ReactNode { {open ? "收起链路" : "查看完整链路导图"} {open ? ( - flowGraph ? ( - + graph ? ( + ) : (
- 暂无链路导图数据 —— 等待 graph_update 事件。 + {wsStatus === "connected" + ? "后台思考会话尚未产生链路图 —— 该会话还没有 graph_update 快照（本轮思考未跑 span，或后端未发布）。" + : `实时通道未连接（${wsStatus}），链路图无法更新 —— 这不等于「本轮思考没有链路」。`}
) ) : null} @@ -110,7 +117,7 @@ export function BackthinkModule(): React.ReactNode { )} - 数据源：/ws/chat 下行 backthink / thinking 事件（实时，主驱动）+ 打开时一次拉取 GET /api/v1/admin/backthink 历史，两者按 id 去重、按时间倒序，新记录增量插入（不整页重渲）。干预为后端 M10+ 提供，当前为只读展示。 + 数据源：/ws/chat 下行 backthink / thinking 事件（实时，主驱动）+ 打开时一次拉取 GET /api/v1/admin/backthink 历史，两者按 id 去重、按时间倒序，新记录增量插入（不整页重渲）。链路导图取自后台思考会话（sess-backthink）的 graph_update 快照 —— 后端完整链路图始终发布，与管理端 Flow 页的「当前会话」图相互独立（裁决 16.4.2）。干预为后端 M10+ 提供，当前为只读展示。 ); } diff --git a/frontend/web/src/lib/streamScope.ts b/frontend/web/src/lib/streamScope.ts deleted file mode 100644 index 68a3659..0000000 --- a/frontend/web/src/lib/streamScope.ts +++ /dev/null @@ -1,30 +0,0 @@ -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/ws/handlers/agent.ts b/frontend/web/src/lib/ws/handlers/agent.ts index 9a36378..cb2bf1c 100644 --- a/frontend/web/src/lib/ws/handlers/agent.ts +++ b/frontend/web/src/lib/ws/handlers/agent.ts @@ -11,8 +11,11 @@ export function handleAgentEvent(ev: FrameEvent, r: FramePayload, s: StoreState) return true; } case "graph_update": { + // 裁决 16.4.2：图按它自己的 sessionId 存进会话图桶，管理后台按会话取图 —— + // 后台思考 / 子会话的图不再顶掉管理员正在看的「当前会话」图（store 的两条 + // 取图口径见 store/sessionCache.ts#currentGraph）。 const g = parseGraph(ev.data); - if (g) s.setFlowGraph(g); + if (g) s.setFlowGraph(g, ev.sessionId); return true; } case "backthink": { diff --git a/frontend/web/src/lib/ws/handlers/approval.ts b/frontend/web/src/lib/ws/handlers/approval.ts index e6b3618..f9e0875 100644 --- a/frontend/web/src/lib/ws/handlers/approval.ts +++ b/frontend/web/src/lib/ws/handlers/approval.ts @@ -1,5 +1,4 @@ -import { streamFrameApplies } from "../../streamScope"; -import type { FrameEvent, FramePayload, StoreState } from "./shared"; +import { frameIsActive, frameSessionId, type FrameEvent, type FramePayload, type StoreState } from "./shared"; /** 代审帧处理组：approval_notice —— 主代理代审小字（design/11 §4，裁决 13.4.3）。 * 敏感操作（如 terminal）经代审后后端广播该帧，`explain` 是后端产出的固定文案 @@ -7,8 +6,9 @@ import type { FrameEvent, FramePayload, StoreState } from "./shared"; * 命中返回 true，未命中返回 false 交给下一组。 */ export function handleApprovalEvent(ev: FrameEvent, r: FramePayload, s: StoreState): boolean { if (ev.type !== "approval_notice") return false; - // 与流式帧同一作用域规则：别的会话发生的代审不写进当前会话的小字。 - if (!streamFrameApplies(ev, r, s.activeSessionId)) return true; + // 小字是单槽状态，只认屏幕上这个会话：别的会话发生的代审不写进当前会话 + // （与 stream_end / 忙态同一判据：shared 的 frameIsActive）。 + if (!frameIsActive(frameSessionId(ev, r), s.activeSessionId)) return true; // 没有文案就没有可渲染的小字 —— 丢掉这帧，而不是编造一句。 const explain = typeof r?.explain === "string" ? r.explain : ""; if (!explain) return true; diff --git a/frontend/web/src/lib/ws/handlers/message.ts b/frontend/web/src/lib/ws/handlers/message.ts index f20111d..fecc9ff 100644 --- a/frontend/web/src/lib/ws/handlers/message.ts +++ b/frontend/web/src/lib/ws/handlers/message.ts @@ -1,12 +1,12 @@ -import { messageEventApplies } from "../../../store/useAppStore"; import { MODES, type ChatMessage, type Mode } from "../../types"; import { parseAdminChannelMessage, parsePersistedMessage, toRecord } from "../../wsParse"; import type { FrameEvent, FramePayload, StoreState } from "./shared"; /** The fields of a `message_edited` payload the chat needs: the edited message's * id and its new content. `sessionId` is carried inside the broadcast DTO (and on - * the frame envelope) so the frame can be filtered; `mode`/`mood` are optional — - * an edit never changes them, but the DTO carries them for completeness. */ + * the frame envelope) so the store can confine the patch to that session's bucket; + * `mode`/`mood` are optional — an edit never changes them, but the DTO carries them + * for completeness. */ interface MessageEdited { id: string; content: string; @@ -42,39 +42,36 @@ export function handleMessageEvent(ev: FrameEvent, r: FramePayload, s: StoreStat switch (ev.type) { case "message_persisted": { // design/16 §4.2 — a user or assistant message was persisted and broadcast - // with its backend id. Only surface messages for the active session; the - // store dedups by id and adopts the local optimistic bubble (uid → backend - // id), so both the originating and remote browsers converge on one message. + // with its backend id. 裁决 16.4.5: the frame is written into the bucket its + // own `sessionId` names (the store routes it), so a background conversation + // keeps accumulating instead of being dropped. Within that bucket the store + // dedups by id and adopts the local optimistic bubble (uid → backend id), so + // both the originating and remote browsers converge on one message. const msg = parsePersistedMessage(r?.message); - if (msg && msg.sessionId === s.activeSessionId) { - s.appendOrUpdateMessage(msg); - } + if (msg) s.appendOrUpdateMessage(msg); return true; } case "message_edited": { // design/16 §4.2 (D3 edit path) — the stored message's content was changed // and broadcast with the updated DTO. The backend row stays the single source - // of truth; this only re-renders the bubble, and only in the session it - // belongs to (same filter as message_persisted). Patch by id — an id we do not - // hold is another session's (or an already-gone) message, never a new row. + // of truth; this only re-renders the bubble of the session the frame names + // (the store confines the patch to that session's bucket). Patch by id — an + // id we do not hold is another session's (or an already-gone) message, never + // a new row. const edit = parseMessageEdited(ev.data); if (!edit) return true; - if (messageEventApplies(edit.sessionId ?? ev.sessionId, s.activeSessionId)) { - const patch: Partial = { content: edit.content }; - if (edit.mode) patch.mode = edit.mode; - if (edit.mood) patch.mood = edit.mood; - s.updateMessageById(edit.id, patch); - } + const patch: Partial = { content: edit.content }; + if (edit.mode) patch.mode = edit.mode; + if (edit.mood) patch.mood = edit.mood; + s.updateMessageById(edit.id, patch, edit.sessionId ?? ev.sessionId); return true; } case "message_deleted": { // design/16 §4.2 (D3 delete path) — the message was removed server-side, so - // drop it by id instead of waiting for a manual re-read. The id is globally - // unique; sessionId is advisory and, when present, must be the active one. + // drop it by id instead of waiting for a manual re-read. The frame's session + // scopes the removal; without one the store searches by id (globally unique). const id = r && typeof r.id === "string" ? r.id : ""; - if (id && messageEventApplies(ev.sessionId, s.activeSessionId)) { - s.removeMessageById(id); - } + if (id) s.removeMessageById(id, ev.sessionId); return true; } case "channel_persisted": { diff --git a/frontend/web/src/lib/ws/handlers/session.test.ts b/frontend/web/src/lib/ws/handlers/session.test.ts new file mode 100644 index 0000000..2c8fa1a --- /dev/null +++ b/frontend/web/src/lib/ws/handlers/session.test.ts @@ -0,0 +1,134 @@ +// 会话编排忙闲帧的逻辑测试（design/15 §3/§4/§5；裁决 16.4.1）。 +// +// 锁两件事： +// ① `session_idle` / `session_interrupted` 必须把忙态置回空闲 —— 此前二者没有 case， +// 落到未知帧兜底，忙态只能靠 `stream_end` 回落；回合卡住/被中断时 +// `useAppStore.sessionBusy` 一直为 true，`components/modeSwitch.ts` 的 +// 「预约切换」判据（`useHerActive`：sessionBusy || streaming || sending）永不落地； +// ② 置忙与置闲必须用**同一把尺子**（屏幕上的那个会话），否则后台会话的 bus 帧点亮徽标、 +// 它自己的 idle 帧又被过滤掉 ⇒ 永久卡忙（design/16 §4.2 已登记的单槽债）。 +// +// 项目没有前端测试框架/jsdom（`useHerActive` 是钩子，需要 DOM，故不在此覆盖），所以这是一个 +// 零依赖的纯 node 检查：它驱动真实的帧解码器（lib/ws.ts 的 dispatchFrame）与真实的 store， +// 断言 `SessionStatusHint` / `useHerActive` 所读的那几个字段，并在第一个失败处抛出。 +// +// 跑法（用 vite 自带的 esbuild，不新增依赖、不改 package.json；banner 补上 store 的 +// persist 中间件在 node 下探测的 window.localStorage）： +// +// cd frontend/web \ +// && ./node_modules/.bin/esbuild src/lib/ws/handlers/session.test.ts \ +// --bundle --platform=node --format=esm \ +// --banner:js='globalThis.window={localStorage:{getItem:()=>null,setItem:()=>{},removeItem:()=>{}}};' \ +// --outfile=/tmp/cyrene-session-busy.test.mjs \ +// && node /tmp/cyrene-session-busy.test.mjs +// +// `tsc --noEmit`（npm run build 的一部分）会对本文件做类型检查，因此它 +// 只从项目内导入、不 import node:*。 + +import { dispatchFrame } from "../../ws"; +import { useAppStore } from "../../../store/useAppStore"; + +let checks = 0; + +function ok(cond: boolean, what: string): void { + checks += 1; + if (!cond) throw new Error(`FAILED: ${what}`); +} + +function same(got: T, want: T, what: string): void { + ok(JSON.stringify(got) === JSON.stringify(want), `${what} — got ${JSON.stringify(got)}, want ${JSON.stringify(want)}`); +} + +const MAIN = "web"; +const OTHER = "web:sub-1"; + +/** 每个场景前把忙态复位成「她静默、无挂起」，并把焦点放在 `active` 上。 */ +function seed(active: string): void { + useAppStore.setState({ activeSessionId: active, sessionBusy: false, pendingCount: 0 }); +} + +function busy(): boolean { + return useAppStore.getState().sessionBusy; +} + +function pending(): number { + return useAppStore.getState().pendingCount; +} + +// ① 活动会话：置忙 → 空闲帧收口。这正是 16.4.1 的预约切换所依赖的「她静默下来」。 +seed(MAIN); +dispatchFrame({ type: "session_busy", sessionId: MAIN, data: { sessionId: MAIN, mode: "thinking", busy: true, pendingCount: 0 } }); +ok(busy(), "session_busy 置忙"); +dispatchFrame({ type: "session_idle", sessionId: MAIN, data: { sessionId: MAIN, mode: "idle" } }); +ok(!busy(), "session_idle 必须把忙态置回空闲（此前的缺口：忙态卡到 stream_end）"); +same(pending(), 0, "session_idle 顺带清空挂起数（队列已清空）"); + +// 缺 `busy` 字段的 session_busy 按忙处理（后端帧只有 {sessionId, mode} 时也不误判为闲）。 +seed(MAIN); +dispatchFrame({ type: "session_busy", sessionId: MAIN, data: { sessionId: MAIN, mode: "executing" } }); +ok(busy(), "session_busy 缺 busy 字段时默认置忙"); + +// 显式 busy:false 仍是空闲。 +seed(MAIN); +dispatchFrame({ type: "session_busy", sessionId: MAIN, data: { sessionId: MAIN, busy: false } }); +ok(!busy(), "session_busy{busy:false} 读作空闲"); + +// ② interrupt：当前任务取消 + pending 清空（Docs/api-client.md §3.2。 +// 帧带 {sessionId, mode, dropped}，没有 busy 字段，语义就是「不再忙」）。 +seed(MAIN); +useAppStore.setState({ sessionBusy: true, pendingCount: 3 }); +dispatchFrame({ type: "session_interrupted", sessionId: MAIN, data: { sessionId: MAIN, mode: "idle", dropped: 2 } }); +ok(!busy(), "session_interrupted 必须把忙态置回空闲"); +same(pending(), 0, "session_interrupted 清空挂起数（pending 被丢弃）"); + +// ③ 会话级过滤：单槽忙态跨会话会串，所以置忙与置闲都只认屏幕上这个会话。 +// 别的会话忙：不动这里的徽标。 +seed(MAIN); +dispatchFrame({ type: "session_busy", sessionId: OTHER, data: { sessionId: OTHER, mode: "thinking" } }); +ok(!busy(), "别的会话置忙不得点亮当前会话的忙态"); + +// 别的会话回空闲：不清掉当前会话的忙（否则当前会话看上去凭空静默了）。 +seed(MAIN); +useAppStore.setState({ sessionBusy: true, pendingCount: 2 }); +dispatchFrame({ type: "session_idle", sessionId: OTHER, data: { sessionId: OTHER, mode: "idle" } }); +ok(busy(), "别的会话回空闲不得清掉当前会话的忙态"); +same(pending(), 2, "别的会话回空闲不得清掉当前会话的挂起数"); +dispatchFrame({ type: "session_interrupted", sessionId: OTHER, data: { sessionId: OTHER, mode: "idle", dropped: 9 } }); +ok(busy(), "别的会话被中断不得清掉当前会话的忙态"); + +// 置忙 + 置闲配对：后台会话从忙到闲全程不影响前台（这是「同一把尺子」的关键—— +// 若只过滤置闲，后台会话的忙态会永久卡在徽标上）。 +seed(MAIN); +dispatchFrame({ type: "session_busy", sessionId: OTHER, data: { sessionId: OTHER, mode: "thinking" } }); +dispatchFrame({ type: "session_idle", sessionId: OTHER, data: { sessionId: OTHER, mode: "idle" } }); +ok(!busy(), "后台会话的忙 → 闲不得在当前会话留下卡住的徽标"); + +// ④ 无会话归属的帧（旧后端 / 会话列表未解析）沿用旧行为：不因为对不上就丢掉。 +seed(MAIN); +dispatchFrame({ type: "session_busy", data: { mode: "thinking" } }); +ok(busy(), "无 sessionId 的 session_busy 沿用旧行为（置忙）"); +dispatchFrame({ type: "session_idle", data: { mode: "idle" } }); +ok(!busy(), "无 sessionId 的 session_idle 沿用旧行为（置闲）"); + +// 激活会话尚未解析（列表没回来，仍是 ""）时同样沿用旧行为。 +seed(""); +dispatchFrame({ type: "session_busy", sessionId: MAIN, data: { sessionId: MAIN, mode: "thinking" } }); +ok(busy(), "激活会话未解析时 session_busy 照旧置忙"); +dispatchFrame({ type: "session_idle", sessionId: MAIN, data: { sessionId: MAIN, mode: "idle" } }); +ok(!busy(), "激活会话未解析时 session_idle 照旧置闲"); + +// ⑤ 这两帧必须被消费（不得落进未知帧兜底，往 deviceUpdates 里堆垃圾）。 +seed(MAIN); +const before = useAppStore.getState().deviceUpdates.length; +dispatchFrame({ type: "session_idle", sessionId: MAIN, data: { sessionId: MAIN, mode: "idle" } }); +dispatchFrame({ type: "session_interrupted", sessionId: MAIN, data: { sessionId: MAIN, mode: "idle" } }); +same(useAppStore.getState().deviceUpdates.length, before, "idle/interrupted 不进未知帧兜底"); + +// ⑥ 与 stream_end 的关系：回合正常结束仍按老路子收口（两条路互不打架）。 +seed(MAIN); +useAppStore.setState({ sessionBusy: true, pendingCount: 1 }); +dispatchFrame({ type: "stream_end", sessionId: MAIN, data: { stopReason: "end_turn" } }); +ok(!busy(), "stream_end 仍然收口忙态"); +same(pending(), 0, "stream_end 仍然清空挂起数"); + +console.log(`session busy logic test: ${checks} checks passed`); diff --git a/frontend/web/src/lib/ws/handlers/session.ts b/frontend/web/src/lib/ws/handlers/session.ts index b61e993..564103d 100644 --- a/frontend/web/src/lib/ws/handlers/session.ts +++ b/frontend/web/src/lib/ws/handlers/session.ts @@ -2,10 +2,11 @@ import { useModeStore } from "../../../store/useModeStore"; import { sessionReadTarget } from "../../sessionRead"; import { MODES, type AwakenState, type Mode } from "../../types"; import { parseAwakenTool } from "../../wsParse"; -import type { FrameEvent, FramePayload, StoreState } from "./shared"; +import { frameIsActive, frameSessionId, type FrameEvent, type FramePayload, type StoreState } from "./shared"; -/** 会话/角色状态帧处理组：capacity_awaken / session_busy / session_pending / - * session_read / mode_switch / mood。命中返回 true，未命中返回 false 交给下一组。 */ +/** 会话/角色状态帧处理组：capacity_awaken / 会话编排忙闲（session_busy / + * session_pending / session_idle / session_interrupted）/ session_read / + * mode_switch / mood。命中返回 true，未命中返回 false 交给下一组。 */ export function handleSessionEvent(ev: FrameEvent, r: FramePayload, s: StoreState): boolean { switch (ev.type) { case "capacity_awaken": { @@ -23,12 +24,27 @@ export function handleSessionEvent(ev: FrameEvent, r: FramePayload, s: StoreStat case "session_busy": case "session_pending": { // design/15 §5 — 她忙/有挂起消息时的轻状态（不阻塞输入框）。两者同构：都是 {busy, pendingCount}。 + // 忙态是**单槽全局**状态，只认屏幕上这个会话（见下方 session_idle 的说明）。 + if (!frameIsActive(frameSessionId(ev, r), s.activeSessionId)) return true; const busy = typeof r?.busy === "boolean" ? r.busy : true; const pendingCount = typeof r?.pendingCount === "number" ? r.pendingCount : s.pendingCount; s.setSessionBusy(busy, pendingCount); return true; } + case "session_idle": + case "session_interrupted": { + // design/15 §3/§4/§5 — 会话回到空闲：`session_idle`（队列清空）与 + // `session_interrupted`（用户 interrupt：当前任务取消、pending 丢弃）都自带 + // `{sessionId, mode}`、**没有 busy 字段**，语义就是「这个会话不再忙」。 + // 缺这两个 case 时忙态只能靠 `stream_end` 回落：回合卡住/被中断时 + // `useAppStore.sessionBusy` 一直为 true，裁决 16.4.1 的「预约切换」永不落地。 + const sid = frameSessionId(ev, r); + // 单槽忙态跨会话会串（design/16 §4.2 已登记的债）：置忙与置闲必须用同一把尺子， + // 否则后台会话的 bus 帧把徽标点亮、它自己的 idle 帧又被过滤掉 ⇒ 永久卡忙。 + if (frameIsActive(sid, s.activeSessionId)) s.setSessionBusy(false, 0); + return true; + } case "session_read": { // 协同 B3/D4 — one end marked the conversation read, so the unread badge goes // out on every end. Only the addressed session is touched (a frame naming diff --git a/frontend/web/src/lib/ws/handlers/shared.ts b/frontend/web/src/lib/ws/handlers/shared.ts index 0cedb84..c5ab66f 100644 --- a/frontend/web/src/lib/ws/handlers/shared.ts +++ b/frontend/web/src/lib/ws/handlers/shared.ts @@ -12,3 +12,25 @@ export type FramePayload = Record | null; /** dispatchFrame 入口处取到的 store 快照类型，按参数传给各帧处理组。 */ export type StoreState = ReturnType; + +/** 一帧所属的会话 id：payload 里带的优先（结构化流帧在回合边界由 loop 注入）， + * 其次帧壳的；两者都没有时返回 undefined，调用方据此落回「当前激活会话」。 + * + * 裁决 16.4.5 后这里返回的 id 决定**写进哪个会话的消息桶**（而不是决定丢弃）。 */ +export function frameSessionId(ev: FrameEvent, r: FramePayload): string | undefined { + if (r && typeof r.sessionId === "string" && r.sessionId) return r.sessionId; + return ev.sessionId || undefined; +} + +/** 这帧是不是屏幕上这个会话的？只用于**单槽全局状态**的判定（忙态 / 上下文 / + * 思考累积器）—— 消息内容本身按各自的会话桶写入，不再受此限制。 + * + * 激活会话尚未解析（列表没加载出来，仍是 ""）时返回 true，沿用旧行为：不因为 + * 会话 id 还没对上就把主人正看着的流丢掉。 */ +export function frameIsActive( + sessionId: string | undefined, + activeSessionId: string +): boolean { + if (!sessionId || !activeSessionId) return true; + return sessionId === activeSessionId; +} diff --git a/frontend/web/src/lib/ws/handlers/stream.ts b/frontend/web/src/lib/ws/handlers/stream.ts index 31b55e2..c827296 100644 --- a/frontend/web/src/lib/ws/handlers/stream.ts +++ b/frontend/web/src/lib/ws/handlers/stream.ts @@ -1,27 +1,26 @@ -import { streamFrameApplies } from "../../streamScope"; -import type { FrameEvent, FramePayload, StoreState } from "./shared"; +import { frameIsActive, frameSessionId, type FrameEvent, type FramePayload, type StoreState } from "./shared"; /** 流式正文帧处理组：stream_chunk / stream_end / message.start / message.delta / - * message.end。命中（含组内主动丢弃）返回 true，未命中返回 false 交给下一组。 */ + * message.end。命中（含组内主动丢弃）返回 true，未命中返回 false 交给下一组。 + * + * 裁决 16.4.5：帧按自己的 `sessionId` 落进对应会话的消息桶，**不再**因为不属于 + * 激活会话而被丢掉 —— 后台会话的正文继续累积，切过去就能看到完整的这一轮。 */ export function handleStreamEvent(ev: FrameEvent, r: FramePayload, s: StoreState): boolean { switch (ev.type) { case "stream_chunk": { - if (!streamFrameApplies(ev, r, s.activeSessionId)) return true; - if (r && typeof r.delta === "string") s.appendChunk(r.delta); + if (r && typeof r.delta === "string") s.appendChunk(r.delta, frameSessionId(ev, r)); return true; } 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)) return true; - 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 - // keeps a task/subagent running it re-emits session_busy (design/15 §5). - s.setSessionBusy(false, 0); - if (r && typeof r.stopReason === "string") { - s.setContext({ lastAgentMode: s.context.lastAgentMode }); + const sid = frameSessionId(ev, r); + s.endStream(sid); + // 忙态 / 上下文是**单槽全局**状态（design/15 §5），所以只有屏幕上这个会话的 + // 回合结束才收口；别的会话结束只能收它自己气泡的流，不能清掉这里的徽标。 + if (frameIsActive(sid, s.activeSessionId)) { + s.setSessionBusy(false, 0); + if (r && typeof r.stopReason === "string") { + s.setContext({ lastAgentMode: s.context.lastAgentMode }); + } } return true; } @@ -32,8 +31,7 @@ export function handleStreamEvent(ev: FrameEvent, r: FramePayload, s: StoreState return true; } case "message.delta": { - if (!streamFrameApplies(ev, r, s.activeSessionId)) return true; - if (r && typeof r.delta === "string") s.appendChunk(r.delta); + if (r && typeof r.delta === "string") s.appendChunk(r.delta, frameSessionId(ev, r)); return true; } case "message.end": { diff --git a/frontend/web/src/lib/ws/handlers/thinking.ts b/frontend/web/src/lib/ws/handlers/thinking.ts index 5ea2551..a477b7a 100644 --- a/frontend/web/src/lib/ws/handlers/thinking.ts +++ b/frontend/web/src/lib/ws/handlers/thinking.ts @@ -1,43 +1,45 @@ -import { streamFrameApplies } from "../../streamScope"; -import type { FrameEvent, FramePayload, StoreState } from "./shared"; +import { frameSessionId, type FrameEvent, type FramePayload, type StoreState } from "./shared"; -/** 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 = ""; +/** Accumulator for a streamed thinking segment (thinking.* events), **per session**: + * a background turn's reasoning must not be spliced into the segment being built + * for the conversation on screen (裁决 16.4.5). Reset on thinking.start so a + * mid-turn reconnect can rebuild the collapsed block without relying on prior state. */ +const pendingThinking: Record = {}; + +/** 累积器与写入用的会话键：帧带的 sessionId 优先，缺席时归当前激活会话。 */ +function thinkKey(ev: FrameEvent, r: FramePayload, s: StoreState): string { + return frameSessionId(ev, r) ?? s.activeSessionId; +} /** 思考流帧处理组：thinking.start / thinking.delta / thinking.end + 遗留单帧 thinking。 * 命中（含组内主动丢弃）返回 true，未命中返回 false 交给下一组。 */ export function handleThinkingEvent(ev: FrameEvent, r: FramePayload, s: StoreState): boolean { switch (ev.type) { 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)) return true; - pendingThinking = ""; + pendingThinking[thinkKey(ev, r, s)] = ""; return true; } case "thinking.delta": { - if (!streamFrameApplies(ev, r, s.activeSessionId)) return true; - if (r && typeof r.delta === "string") { - pendingThinking += r.delta; - s.setThinking(pendingThinking); - } + if (!r || typeof r.delta !== "string") return true; + const key = thinkKey(ev, r, s); + const text = (pendingThinking[key] ?? "") + r.delta; + pendingThinking[key] = text; + s.setThinking(text, frameSessionId(ev, r)); return true; } case "thinking.end": { - if (!streamFrameApplies(ev, r, s.activeSessionId)) return true; - if (pendingThinking) s.setThinking(pendingThinking); - pendingThinking = ""; + const key = thinkKey(ev, r, s); + const text = pendingThinking[key] ?? ""; + if (text) s.setThinking(text, frameSessionId(ev, r)); + delete pendingThinking[key]; return true; } case "thinking": { // Legacy single-shot CoT frame — renders into the same bubble. - if (!streamFrameApplies(ev, r, s.activeSessionId)) return true; if (typeof ev.data === "string") { - s.setThinking(ev.data); + s.setThinking(ev.data, frameSessionId(ev, r)); } else if (r && typeof r.text === "string") { - s.setThinking(r.text); + s.setThinking(r.text, frameSessionId(ev, r)); } return true; } diff --git a/frontend/web/src/lib/ws/handlers/tool.ts b/frontend/web/src/lib/ws/handlers/tool.ts index b3432ef..fc3a2f2 100644 --- a/frontend/web/src/lib/ws/handlers/tool.ts +++ b/frontend/web/src/lib/ws/handlers/tool.ts @@ -1,17 +1,18 @@ -import { streamFrameApplies } from "../../streamScope"; +import { frameSessionId, type FrameEvent, type FramePayload, type StoreState } from "./shared"; import { parseTool, parseToolCall } from "../../wsParse"; -import type { FrameEvent, FramePayload, StoreState } from "./shared"; /** 工具帧处理组：tool_call.start / tool_call.end / tool_call.delta / * tool_execution.update + 遗留 tool_used。命中（含组内主动丢弃）返回 true， - * 未命中返回 false 交给下一组。 */ + * 未命中返回 false 交给下一组。 + * + * 裁决 16.4.5：工具卡按帧的 `sessionId` 落在该会话的助手气泡上（其它会话的调用 + * 卡不会跑到屏幕上的这一轮里）。 */ export function handleToolEvent(ev: FrameEvent, r: FramePayload, s: StoreState): boolean { switch (ev.type) { case "tool_call.start": case "tool_call.end": { - if (!streamFrameApplies(ev, r, s.activeSessionId)) return true; const tool = parseToolCall(ev.data); - if (tool) s.addTool(tool); + if (tool) s.addTool(tool, frameSessionId(ev, r)); return true; } case "tool_call.delta": { @@ -20,17 +21,15 @@ export function handleToolEvent(ev: FrameEvent, r: FramePayload, s: StoreState): return true; } case "tool_execution.update": { - // Same card as tool_call.* — same guard. - if (!streamFrameApplies(ev, r, s.activeSessionId)) return true; + // Same card as tool_call.* — same routing. const tool = parseToolCall(ev.data); - if (tool) s.addTool(tool); + if (tool) s.addTool(tool, frameSessionId(ev, r)); return true; } case "tool_used": { // Legacy tool echo — same card as tool_call.*. - if (!streamFrameApplies(ev, r, s.activeSessionId)) return true; const tool = parseTool(ev.data); - if (tool) s.addTool(tool); + if (tool) s.addTool(tool, frameSessionId(ev, r)); return true; } default: diff --git a/frontend/web/src/store/messageSync.test.ts b/frontend/web/src/store/messageSync.test.ts index 50a3128..1048f15 100644 --- a/frontend/web/src/store/messageSync.test.ts +++ b/frontend/web/src/store/messageSync.test.ts @@ -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(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`); diff --git a/frontend/web/src/store/sessionCache.ts b/frontend/web/src/store/sessionCache.ts new file mode 100644 index 0000000..8c12a44 --- /dev/null +++ b/frontend/web/src/store/sessionCache.ts @@ -0,0 +1,199 @@ +import type { ChatMessage, FlowGraph } from "../lib/types"; +import type { AppStore } from "./useAppStore"; + +/** + * 会话级缓存助手（裁决 16.4.5 / 16.4.2）——纯函数，不碰 store，便于逐条做逻辑测试。 + * + * 两件事： + * ① **消息桶**（design/16 §4.3）：一个会话一份消息列表。store 把**激活会话**的列表 + * 放在 `messages`，把**非激活会话**的列表放在 `sessionMessages`；这样任何一帧都能 + * 按自己的 `sessionId` 落进对应的桶（而不是被丢弃），切会话时把离开的会话存回它的 + * 桶、把切到的会话取出，于是「切会话不再清空」。 + * ② **链路图桶**（裁决 16.4.2）：一个会话最近一张 `graph_update` 快照。管理后台按 + * `sessionId` 取图，而不是所有页面共用一张全局图——后台思考的完整链路图与 + * 「用户侧前端发不发」无关，也绝不顶掉管理员看的「当前会话」图。 + * + * 从 `useAppStore.ts` 拆出来（AGENTS §10 文件 ≤500 行 / 三次复制必须抽取）：下面每条 + * 规则都被「激活会话」与「非激活会话」两条写入路径共用，放在一处才不会走样。 + */ + +/** 后端固定主会话 id（`internal/runtime/history` 的 `MainSessionID`，design/16 §2.1）。 */ +export const MAIN_SESSION_ID = "web"; + +/** 后台思考会话 id（`internal/services/backthink` 的 `DefaultSessionID`；design/19 §1 的 + * 发布会话范围、design/20 §2）。后台思考的完整链路图发在这个会话上，管理后台的 + * Backthink 页据此取图（裁决 16.4.2）。 */ +export const BACKTHINK_SESSION_ID = "sess-backthink"; + +/** 一个会话的消息列表；键是会话 id。 */ +export type SessionMessages = Record; + +/** 把一条后端消息并入消息列表：按 id upsert；没有同 id 时采纳本端同 role 的本地 + * 乐观气泡（本地 `uid` → 后端 `id`），使同一句话不会渲染两遍（design/16 §4.3 去重）。 + * 后端消息是内容的真源，本地气泡保留流式/工具/思考等本端状态。 */ +export function mergeMessage(list: ChatMessage[], msg: ChatMessage): ChatMessage[] { + const idx = list.findIndex((m) => m.id === msg.id); + if (idx !== -1) { + const next = list.slice(); + next[idx] = { ...next[idx], ...msg }; + return next; + } + // 没有同 id：找一个本端尚未落库的同 role 气泡（无 sessionId）来归一化。 + // 优先内容完全相同的那条，否则取 createdAt 最近的；从新到旧扫描。 + let best = -1; + let bestDist = Number.MAX_SAFE_INTEGER; + for (let i = list.length - 1; i >= 0; i--) { + const m = list[i]; + if (m.sessionId || m.role !== msg.role || m.id === msg.id) continue; + const dist = m.content === msg.content ? 0 : Math.abs(m.createdAt - msg.createdAt); + if (dist < bestDist) { + bestDist = dist; + best = i; + } + } + if (best === -1) return [...list, msg]; + const m = list[best]; + const next = list.slice(); + next[best] = { + ...m, + id: msg.id, + sessionId: msg.sessionId ?? m.sessionId, + content: msg.content || m.content, + mode: msg.mode ?? m.mode, + mood: msg.mood ?? m.mood, + image: msg.image ?? m.image, + platform: msg.platform ?? m.platform, + deviceId: msg.deviceId ?? m.deviceId, + personaId: msg.personaId ?? m.personaId, + createdAt: msg.createdAt ?? m.createdAt, + }; + return next; +} + +/** 按 id 就地打补丁。目标不在列表里返回 `null` —— 调用方据此不动状态，另一个会话 + * 的编辑帧绝不会在这里凭空插一行。 */ +export function patchMessage( + list: ChatMessage[], + id: string, + patch: Partial +): ChatMessage[] | null { + const idx = list.findIndex((m) => m.id === id); + if (idx === -1) return null; + const next = list.slice(); + next[idx] = { ...next[idx], ...patch }; + return next; +} + +/** 按 id 删除。不在列表里返回 `null`（重复帧、或那条消息本就属于别的会话）。 */ +export function dropMessage(list: ChatMessage[], id: string): ChatMessage[] | null { + const next = list.filter((m) => m.id !== id); + return next.length === list.length ? null : next; +} + +/** 对**最后一条助手气泡**做一次变换：流式增量、思考文本、工具卡、流结束都落在它上面。 + * 列表为空或最后一条不是助手气泡时返回 `null` —— 帧不能改到用户气泡或更早的历史上。 */ +export function patchAssistantTail( + list: ChatMessage[], + fn: (m: ChatMessage) => ChatMessage +): ChatMessage[] | null { + const last = list[list.length - 1]; + if (!last || last.role !== "assistant") return null; + const next = list.slice(); + next[next.length - 1] = fn(last); + return next; +} + +/** 把一个会话的消息列表写回桶：空列表就删键（省内存，也代表「该会话没有缓存」）。 */ +export function withBucket( + buckets: SessionMessages, + sessionId: string, + list: ChatMessage[] +): SessionMessages { + if (list.length) return { ...buckets, [sessionId]: list }; + if (!(sessionId in buckets)) return buckets; + const next = { ...buckets }; + delete next[sessionId]; + return next; +} + +/** 用后端的一页**替换**当前列表，但保留本端尚未落库的流式助手气泡：切走再切回时它 + * 已随会话桶回来，后端要等这一轮结束才落库，直接丢掉会让用户正看着的回复凭空消失 + * （design/16 §4.3）。该气泡若已出现在后端这一页里（同 id）则不重复追加。 */ +export function replacePage(list: ChatMessage[], page: ChatMessage[]): ChatMessage[] { + const tail = list[list.length - 1]; + if (!tail?.streaming || tail.role !== "assistant") return page; + if (page.some((m) => m.id === tail.id)) return page; + return [...page, tail]; +} + +/** 最多保留多少个会话的链路图。web 子会话可能很多，桶不能无限长（图随每次 span + * 迁移更新，是高频写入）；超限时丢弃最久没更新的那个会话。 */ +const MAX_GRAPHS = 12; + +/** 存一张会话图：键按最近更新重排（先删后插），超出上限时丢掉最久没更新的会话。 */ +export function putGraph( + graphs: Record, + sessionId: string, + graph: FlowGraph +): Record { + const next: Record = {}; + for (const [k, v] of Object.entries(graphs)) if (k !== sessionId) next[k] = v; + next[sessionId] = graph; + const keys = Object.keys(next); + if (keys.length <= MAX_GRAPHS) return next; + const drop = new Set(keys.slice(0, keys.length - MAX_GRAPHS)); + const trimmed: Record = {}; + for (const k of keys) if (!drop.has(k)) trimmed[k] = next[k]; + return trimmed; +} + +/** 兼容字段 `flowGraph` 的口径（管理端 Flow / Agents 页读它）：**激活会话**的图； + * 激活会话还没有图时回落主会话的最近一张，使页面不至于空着。 + * 后台思考会话的图不会顶掉它 —— 这正是裁决 16.4.2 要修的：每条思考看自己的图， + * 「当前会话」的图只随当前会话变化。 */ +export function currentGraph( + graphs: Record, + activeSessionId: string +): FlowGraph | null { + return graphs[activeSessionId] ?? graphs[MAIN_SESSION_ID] ?? null; +} + +/** 一次会话级写入：`fn` 返回新列表就写回该会话（激活会话写 `messages`，其它写它的 + * 桶），返回 `null` 表示这次不改动（帧与我们持有的消息对不上）。所有会话级写入 + * 都走这里，两条路径的规则因此不会走样（裁决 16.4.5）。 + * + * 放在本模块而非 store 里：写入的落点就是缓存策略，且 `AppStore` 只作类型引入 + * （编译后被擦除，不构成运行时循环依赖）；`useAppStore.ts` 也要守住 ≤500 行。 */ +export function writeSession( + s: AppStore, + sessionId: string, + fn: (list: ChatMessage[]) => ChatMessage[] | null +): Partial | AppStore { + if (sessionId === s.activeSessionId) { + const next = fn(s.messages); + return next === null ? s : { messages: next }; + } + const next = fn(s.sessionMessages[sessionId] ?? []); + if (next === null) return s; + return { sessionMessages: withBucket(s.sessionMessages, sessionId, next) }; +} + +/** 按 id 的会话级写入（编辑/删除帧）。帧带 `sessionId` 时才在它能命中的会话里找； + * 没带则说明后端解析不出会话（id 全局唯一）—— 先看屏幕上这个会话，再逐桶找， + * 命中的那一桶只改它自己（裁决 16.4.5）。 */ +export function writeSessionById( + s: AppStore, + sessionId: string | undefined, + fn: (list: ChatMessage[]) => ChatMessage[] | null +): Partial | AppStore { + if (sessionId) return writeSession(s, sessionId, fn); + const active = fn(s.messages); + if (active !== null) return { messages: active }; + for (const [sid, list] of Object.entries(s.sessionMessages)) { + const next = fn(list); + if (next !== null) { + return { sessionMessages: withBucket(s.sessionMessages, sid, next) }; + } + } + return s; +} diff --git a/frontend/web/src/store/useAppStore.ts b/frontend/web/src/store/useAppStore.ts index 2989ae9..b46e6d6 100644 --- a/frontend/web/src/store/useAppStore.ts +++ b/frontend/web/src/store/useAppStore.ts @@ -15,6 +15,19 @@ import type { WsStatus, } from "../lib/types"; import { now, uid } from "../lib/util"; +import { + currentGraph, + dropMessage, + mergeMessage, + patchAssistantTail, + patchMessage, + putGraph, + replacePage, + withBucket, + writeSession, + writeSessionById, + type SessionMessages, +} from "./sessionCache"; interface ChatContext { topic?: string; @@ -30,9 +43,19 @@ export interface AppStore { * session list is loaded. Messages in `messages` always belong to this id. */ activeSessionId: string; wsStatus: WsStatus; - /** Messages of the *active* session, in display order (oldest first). Reset on - * session switch and refilled from the backend (design/16 §4/§5). */ + /** Messages of the *active* session, in display order (oldest first). Switching + * sessions no longer clears it: the list on screen is stashed into that session's + * bucket in `sessionMessages` and the target session's bucket is taken out + * (design/16 §4.3; 裁决 16.4.5). */ messages: ChatMessage[]; + /** Message buckets of the sessions that are **not** on screen — the active + * session's list lives in `messages`. A frame (persisted message, stream delta, + * edit, delete) is written into the bucket its `sessionId` names instead of being + * dropped, so a background conversation keeps accumulating and is rendered whole + * the moment it is switched to (裁决 16.4.5). Bounded by what the backend pushed + * during this page's lifetime; the backend stays the message truth (design/16 §1) + * and a switch-in is refetched page by page. Never persisted. */ + sessionMessages: SessionMessages; /** Latest mood per role id (design/22 §4.6). The `mood` WS event is * persona-scoped and carries no `sessionId`, so each role gets its own bucket * and the UI shows the bucket of the *active session's* role — role X's mood @@ -43,6 +66,14 @@ export interface AppStore { context: ChatContext; agents: AgentInfo[]; backthink: BackthinkItem[]; + /** Per-session flow graphs, keyed by the graph's own `sessionId` (裁决 16.4.2): + * a backthink / sub-session graph no longer overwrites the graph of the + * conversation the admin panel is looking at. Bounded by `MAX_GRAPHS` + * (store/sessionCache.ts), most recently updated first. */ + flowGraphs: Record; + /** Compatibility alias for the admin Flow / Agents pages: the graph of the + * *active* session, falling back to the main session's latest one. Kept in sync + * by `setFlowGraph` / `setActiveSession` / `resetChat`. */ flowGraph: FlowGraph | null; deviceUpdates: string[]; /** 唤醒式能力三态（design/14 §5）——当前唤醒状态、上次切换的来源态、最近点亮的工具。 */ @@ -91,37 +122,50 @@ export interface AppStore { * frame another end broadcasts — a no-op for an already-read or unknown * session, so neither path can invent a row or re-render the list needlessly. */ clearUnread: (sessionId: string) => void; - /** Focus a session and clear its message cache (refilled by the caller). */ + /** Focus a session: stash the list on screen into its own bucket and take the + * target session's bucket out (裁决 16.4.5). Never drops a visited session's + * loaded page, so switching back shows it immediately while the backend page + * is refetched. */ setActiveSession: (id: string) => void; - /** Replace the active session's message list (initial page on switch). */ + /** Replace the active session's message list with the backend page. A streaming + * assistant bubble that has not been persisted yet is carried over (see + * `replacePage`). An empty list clears the panel (the fetch-failed path). */ setMessages: (list: ChatMessage[]) => void; /** Prepend an older page (chronological) onto the active session's messages. */ prependMessages: (list: ChatMessage[]) => void; - /** Append a message, or merge by id if it already exists. On message_persisted - * it also adopts a same-role local optimistic bubble (rewriting its uid to the - * backend id) so a message never renders twice (design/16 §4.3 dedup). */ + /** Append a message to the bucket its `sessionId` names, or merge by id if it + * already exists there. On `message_persisted` it also adopts a same-role local + * optimistic bubble (rewriting its uid to the backend id) so a message never + * renders twice (design/16 §4.3 dedup). A message without a session id (a local + * optimistic bubble) belongs to the active session. */ appendOrUpdateMessage: (msg: ChatMessage) => void; /** Patch one already-rendered message in place by id — the `message_edited` WS - * path. A no-op when the id is not in the active session, so an edit broadcast - * for another conversation can never inject a row here. */ - updateMessageById: (id: string, patch: Partial) => void; - /** Drop one message by id — the `message_deleted` WS path. A no-op when the id - * is absent (another session's message, or a duplicate frame). */ - removeMessageById: (id: string) => void; + * path. With `sessionId` the patch is confined to that session's bucket; without + * one the id is searched across the active list and every bucket (ids are + * globally unique). A miss is a no-op, so an edit can never inject a row. */ + updateMessageById: (id: string, patch: Partial, sessionId?: string) => void; + /** Drop one message by id — the `message_deleted` WS path. Same scoping rules as + * `updateMessageById`; a miss is a no-op (another session's message, or a + * duplicate frame). */ + removeMessageById: (id: string, sessionId?: string) => void; addUserMessage: (text: string, mode?: Mode, image?: string, deviceId?: string) => void; beginAssistant: (mode?: Mode, deviceId?: string) => string; - appendChunk: (delta: string) => void; - endStream: () => void; - setThinking: (text: string) => void; - addTool: (tool: ToolUsed) => void; + /** Append a streamed delta to the assistant bubble of `sessionId` (default: the + * active session) — a background turn keeps streaming into its own bucket. */ + appendChunk: (delta: string, sessionId?: string) => void; + endStream: (sessionId?: string) => void; + setThinking: (text: string, sessionId?: string) => void; + addTool: (tool: ToolUsed, sessionId?: string) => void; setAgents: (agents: AgentInfo[]) => void; upsertAgent: (agent: AgentInfo) => void; addBackthink: (item: BackthinkItem) => void; revealBackthink: (id: string, conclusion: string) => void; markBackthinkPushed: (id: string) => void; - setFlowGraph: (graph: FlowGraph) => void; + /** Store one `graph_update` snapshot under the session it belongs to and refresh + * the `flowGraph` alias when that session is the one on screen (裁决 16.4.2). */ + setFlowGraph: (graph: FlowGraph, sessionId?: string) => void; addDeviceUpdate: (text: string) => void; setAwaken: ( from: AwakenState | undefined, @@ -146,12 +190,14 @@ export const useAppStore = create()( activeSessionId: "", wsStatus: "idle", messages: [], + sessionMessages: {}, moods: {}, context: {}, agents: [ { id: "main", name: "昔涟", status: "idle", role: "owner" }, ], backthink: [], + flowGraphs: {}, flowGraph: null, deviceUpdates: [], awakenState: "quiet", @@ -182,7 +228,24 @@ export const useAppStore = create()( set({ sessions: Object.fromEntries(list.map((s) => [s.id, s])), }), - setActiveSession: (id) => set({ activeSessionId: id, messages: [] }), + setActiveSession: (id) => + set((s) => { + if (id === s.activeSessionId) return s; + // 离开的会话：把屏幕上的列表存回它自己的桶（未落库的流式内容随之保留）。 + // 还没有归属（""）时不建桶，与切换前一致。 + const buckets = s.activeSessionId + ? withBucket(s.sessionMessages, s.activeSessionId, s.messages) + : s.sessionMessages; + const next = { ...buckets }; + const cached = next[id] ?? []; + delete next[id]; + return { + activeSessionId: id, + messages: cached, + sessionMessages: next, + flowGraph: currentGraph(s.flowGraphs, id), + }; + }), clearUnread: (sessionId) => set((s) => { const meta = s.sessions[sessionId]; @@ -191,76 +254,28 @@ export const useAppStore = create()( sessions: { ...s.sessions, [sessionId]: { ...meta, unreadCount: 0 } }, }; }), - setMessages: (list) => set({ messages: list }), + setMessages: (list) => + set((s) => ({ messages: replacePage(s.messages, list) })), prependMessages: (list) => set((s) => { const seen = new Set(s.messages.map((m) => m.id)); const older = list.filter((m) => !seen.has(m.id)); - return { messages: [...older, ...s.messages] }; + return older.length ? { messages: [...older, ...s.messages] } : s; }), appendOrUpdateMessage: (msg) => - set((s) => { - // Exact id match → merge. Keeps local streaming/tools/thinking, takes - // the backend's content/mode/mood as truth. - const idx = s.messages.findIndex((m) => m.id === msg.id); - if (idx !== -1) { - const messages = s.messages.slice(); - messages[idx] = { ...messages[idx], ...msg }; - return { messages }; - } - // No exact id: adopt a local optimistic bubble (no sessionId) of the - // same role, so a message the local browser already rendered - // optimistically is never duplicated when message_persisted arrives - // (design/16 §4.3 dedup by id). Prefer a content match, else the - // nearest by createdAt; scan newest-first. - let best = -1; - let bestDist = Number.MAX_SAFE_INTEGER; - for (let i = s.messages.length - 1; i >= 0; i--) { - const m = s.messages[i]; - if (m.sessionId || m.role !== msg.role || m.id === msg.id) continue; - const dist = m.content === msg.content ? 0 : Math.abs(m.createdAt - msg.createdAt); - if (dist < bestDist) { - bestDist = dist; - best = i; - } - } - if (best !== -1) { - const m = s.messages[best]; - const messages = s.messages.slice(); - messages[best] = { - ...m, - id: msg.id, - sessionId: msg.sessionId ?? m.sessionId, - content: msg.content || m.content, - mode: msg.mode ?? m.mode, - mood: msg.mood ?? m.mood, - image: msg.image ?? m.image, - platform: msg.platform ?? m.platform, - deviceId: msg.deviceId ?? m.deviceId, - personaId: msg.personaId ?? m.personaId, - createdAt: msg.createdAt ?? m.createdAt, - }; - return { messages }; - } - // No local bubble (e.g. a remote browser that never optimistically - // rendered this message): append. - return { messages: [...s.messages, msg] }; - }), + set((s) => + writeSession(s, msg.sessionId ?? s.activeSessionId, (list) => + mergeMessage(list, msg) + ) + ), - updateMessageById: (id, patch) => - set((s) => { - const idx = s.messages.findIndex((m) => m.id === id); - if (idx === -1) return s; - const messages = s.messages.slice(); - messages[idx] = { ...messages[idx], ...patch }; - return { messages }; - }), + updateMessageById: (id, patch, sessionId) => + set((s) => + writeSessionById(s, sessionId, (list) => patchMessage(list, id, patch)) + ), - removeMessageById: (id) => - set((s) => { - const messages = s.messages.filter((m) => m.id !== id); - return messages.length === s.messages.length ? s : { messages }; - }), + removeMessageById: (id, sessionId) => + set((s) => writeSessionById(s, sessionId, (list) => dropMessage(list, id))), addUserMessage: (text, mode, image, deviceId) => set((s) => ({ @@ -298,47 +313,44 @@ export const useAppStore = create()( return id; }, - appendChunk: (delta) => - set((s) => { - const messages = s.messages.slice(); - const last = messages[messages.length - 1]; - if (!last || last.role !== "assistant") return s; - messages[messages.length - 1] = { ...last, content: last.content + delta }; - return { messages }; - }), + appendChunk: (delta, sessionId) => + set((s) => + writeSession(s, sessionId ?? s.activeSessionId, (list) => + patchAssistantTail(list, (last) => ({ + ...last, + content: last.content + delta, + })) + ) + ), - endStream: () => - set((s) => { - const messages = s.messages.slice(); - const last = messages[messages.length - 1]; - if (!last || last.role !== "assistant") return s; - messages[messages.length - 1] = { ...last, streaming: false }; - return { messages }; - }), + endStream: (sessionId) => + set((s) => + writeSession(s, sessionId ?? s.activeSessionId, (list) => + patchAssistantTail(list, (last) => ({ ...last, streaming: false })) + ) + ), - setThinking: (text) => - set((s) => { - const messages = s.messages.slice(); - const last = messages[messages.length - 1]; - if (!last || last.role !== "assistant") return s; - messages[messages.length - 1] = { ...last, thinking: text }; - return { messages }; - }), + setThinking: (text, sessionId) => + set((s) => + writeSession(s, sessionId ?? s.activeSessionId, (list) => + patchAssistantTail(list, (last) => ({ ...last, thinking: text })) + ) + ), - addTool: (tool) => - set((s) => { - const messages = s.messages.slice(); - const last = messages[messages.length - 1]; - if (!last || last.role !== "assistant") return s; - const existing = last.tools ?? []; - const idx = existing.findIndex((t) => t.name === tool.name); - const tools = - idx === -1 - ? [...existing, tool] - : existing.map((t, i) => (i === idx ? { ...t, ...tool } : t)); - messages[messages.length - 1] = { ...last, tools }; - return { messages }; - }), + addTool: (tool, sessionId) => + set((s) => + writeSession(s, sessionId ?? s.activeSessionId, (list) => + patchAssistantTail(list, (last) => { + const existing = last.tools ?? []; + const idx = existing.findIndex((t) => t.name === tool.name); + const tools = + idx === -1 + ? [...existing, tool] + : existing.map((t, i) => (i === idx ? { ...t, ...tool } : t)); + return { ...last, tools }; + }) + ) + ), setAgents: (agents) => set({ agents }), upsertAgent: (agent) => @@ -365,7 +377,14 @@ export const useAppStore = create()( ), })), - setFlowGraph: (graph) => set({ flowGraph: graph }), + setFlowGraph: (graph, sessionId) => + set((s) => { + // 归属会话：图自己带的 sessionId（后端 RunGraph 建图时写入）优先，其次帧壳的， + // 最后落到当前会话。每张图只进它自己的桶，互不覆盖（裁决 16.4.2）。 + const sid = graph.sessionId ?? sessionId ?? s.activeSessionId; + const flowGraphs = putGraph(s.flowGraphs, sid, graph); + return { flowGraphs, flowGraph: currentGraph(flowGraphs, s.activeSessionId) }; + }), addDeviceUpdate: (text) => set((s) => ({ deviceUpdates: [text, ...s.deviceUpdates].slice(0, 50) })), setAwaken: (from, to, tool) => @@ -405,7 +424,9 @@ export const useAppStore = create()( resetChat: () => set((s) => ({ messages: [], + sessionMessages: {}, moods: {}, + flowGraphs: {}, flowGraph: null, backthink: s.backthink, awakenState: "quiet", @@ -440,7 +461,8 @@ export const useAppStore = create()( }, partialize: (s) => ({ // messages deliberately excluded: the backend is the message truth - // (design/16 §1), so we never persist the whole chat into localStorage. + // (design/16 §1), so we never persist the whole chat into localStorage — + // and neither is `sessionMessages` (the per-session buckets of 裁决 16.4.5). // We keep the (small) session metadata + the focused session id so a // refresh lands on the same conversation, plus the role catalog so the // avatar/name/wallpaper resolve before the first fetch returns. @@ -454,26 +476,3 @@ export const useAppStore = create()( } ) ); - -/** Convenience selector for the last streaming assistant message id. */ -export function lastStreamingMessage(s: AppStore): ChatMessage | undefined { - const last = s.messages[s.messages.length - 1]; - return last?.role === "assistant" ? last : undefined; -} - -/** Does an edit/delete frame scoped to `sessionId` belong to the conversation the - * UI is showing? `message_edited` / `message_deleted` carry the owning session so - * the rule matches `message_persisted`: only the active session's messages are - * rendered, so another conversation's frame must not touch this list. - * - * A frame without a session id is id-scoped (the backend could not resolve the - * conversation — see the gateway's optional message lookup). It still applies: - * message ids are globally unique, so the by-id update/removal can only ever match - * a message of its own session, and an id we do not hold is a no-op either way. */ -export function messageEventApplies( - sessionId: string | undefined, - activeSessionId: string -): boolean { - if (!sessionId) return true; - return sessionId === activeSessionId; -}