From 012e9f3ad4ec3ac884faf86c92b3a022001a9df0 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 5 Oct 2026 11:29:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(web):=20=E5=89=8D=E7=AB=AF=20P0=20?= =?UTF-8?q?=E6=94=B6=E5=B0=BE=EF=BC=88=E5=A4=9A=E8=B7=AF=E5=BF=99=E6=80=81?= =?UTF-8?q?=E6=94=B6=E5=8F=A3=20/=20=E6=8F=A1=E6=89=8B=20history=20?= =?UTF-8?q?=E8=90=BD=20store=20/=20=E6=B8=85=E7=90=86=E5=B9=BD=E7=81=B5?= =?UTF-8?q?=E9=94=AE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- frontend/web/src/lib/types/ws.ts | 26 ++- .../web/src/lib/ws/handlers/history.test.ts | 176 ++++++++++++++++++ frontend/web/src/lib/ws/handlers/message.ts | 21 ++- .../web/src/lib/ws/handlers/parallel.test.ts | 26 +++ .../web/src/lib/ws/handlers/session.test.ts | 25 ++- frontend/web/src/lib/ws/handlers/session.ts | 17 +- frontend/web/src/lib/ws/handlers/stream.ts | 9 +- frontend/web/src/lib/wsParse.ts | 14 ++ frontend/web/src/store/sessionCache.ts | 66 +++++++ frontend/web/src/store/useAppStore.ts | 16 ++ 10 files changed, 365 insertions(+), 31 deletions(-) create mode 100644 frontend/web/src/lib/ws/handlers/history.test.ts diff --git a/frontend/web/src/lib/types/ws.ts b/frontend/web/src/lib/types/ws.ts index 889428f..b24c69c 100644 --- a/frontend/web/src/lib/types/ws.ts +++ b/frontend/web/src/lib/types/ws.ts @@ -17,15 +17,22 @@ export interface AwakenPayload { tool?: string | { name: string }; } -/** session_busy / session_pending 下行事件 payload（design/15 §5）：她是否在忙 + 挂起消息数。 +/** 会话编排状态帧（`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / + * `session_interrupted` / `pending_full`）的 data（design/15 §5，契约 v0.65 §3.4.1）。 * - * 忙时并行（design/15 §2.5，契约 v0.65 §3.4.1）后这些帧还带 `activeTurns`（在途路数）、 - * `mode`（忙碌通道）、`queueDepth` / `reason` / `dropped` / `cancelledTurns` 等计数键。 - * 全部可选：老后端不发则逐字沿用旧行为。`reason` 目前唯一取值 `"parallel_limit"` - * （声明并行却落回排队）。 */ + * **这是这些帧线上出现过的键的并集**（每帧只带自己那几个）： + * - `session_busy` / `session_idle`：`{sessionId, mode, activeTurns}` + * （`internal/services/gateway/session_parallel.go` 的 `busyPayload`）； + * - `pending_ack`：`{sessionId, mode, queueDepth, reason?}`； + * - `pending_drain`：`{sessionId, mode, queueDepth, activeTurns}`； + * - `session_interrupted`：`{sessionId, mode, dropped, cancelledTurns, activeTurns}`； + * - `pending_full`：`{sessionId, mode, max, clientMsgId?, messageId?}`。 + * + * 全部可选：缺席时逐字沿用旧行为。忙 = `activeTurns > 0`。`reason` 目前唯一取值 + * `"parallel_limit"`（声明并行却落回排队）。 + * 注：曾有的 `busy` / `pendingCount` 两个键**后端从未发出过**（全仓 `git log -S'"busy":'` + * 零命中，`busyPayload` 只发三个键），已删除；`session_pending` 帧同样不存在。 */ export interface SessionStatePayload { - busy?: boolean; - pendingCount?: number; /** 该帧发出时的在途路数（含正在收尾的路）。忙 = activeTurns > 0。 */ activeTurns?: number; /** 忙碌通道：thinking / executing / subagent；`session_idle` 为 "idle"。 */ @@ -132,9 +139,10 @@ export interface WsDownEvent { sessionId?: string; } -/** Upstream WS frame (client → server). */ +/** Upstream WS frame (client → server). 帧壳 = `{type, data?}`；契约 §3.3 里的 `reqId?` + * 后端只声明不消费（`ws_hub.go` 的 `ReqID` 字段无任何读取点），本端也从未设置， + * 因此不再出现在类型里。 */ export interface WsUpEvent { type: "mode_switch" | "chat" | "voice_input" | "subagent_ask" | "task_output_query"; - reqId?: string; data?: unknown; } diff --git a/frontend/web/src/lib/ws/handlers/history.test.ts b/frontend/web/src/lib/ws/handlers/history.test.ts new file mode 100644 index 0000000..2f35df6 --- /dev/null +++ b/frontend/web/src/lib/ws/handlers/history.test.ts @@ -0,0 +1,176 @@ +// 握手 `history` 回放帧的逻辑测试（契约 §3.2 + §4.4）。 +// +// 为什么需要它：服务端在 owner/admin 连接注册后**定向**发一帧 +// `{"type":"history","sessionId":"web","data":{"messages":[…]}}`（主会话 +// 最近 50 条），可此前前端**没有 history case** ⇒ 落到未知帧兜底被丢掉：WS 断线窗口内 +// 落库的消息永久缺失，只能靠切会话/刷新触发 REST 重取才补上。本测试锁四件事： +// ① 回放真的落进 store（并按**帧壳 sessionId** 进对应会话桶，不是激活会话）； +// ② 按 id 去重、幂等（重复并入不产生第二份）； +// ③ 认领本端乐观气泡只用**精确键**（用户行 = `clientMsgId`、助手行 = `turnId`）； +// ④ 精确键对不上时**绝不用内容/时间启发式** —— 否则一条旧助手回放会把本端正在流式的 +// 气泡顶掉（正文被旧内容覆盖、流式态丢失）。 +// +// 项目没有前端测试框架/jsdom，所以这是零依赖的纯 node 检查：它驱动真实的帧解码器 +// （lib/ws.ts 的 dispatchFrame）与真实的 store，并在第一个失败处抛出。 +// +// 跑法（用 vite 自带的 esbuild，不新增依赖、不改 package.json；banner 补上 store 的 +// persist 中间件在 node 下探测的 window.localStorage）： +// +// cd frontend/web \ +// && ./node_modules/.bin/esbuild src/lib/ws/handlers/history.test.ts \ +// --bundle --platform=node --format=esm \ +// --banner:js='globalThis.window={localStorage:{getItem:()=>null,setItem:()=>{},removeItem:()=>{}}};' \ +// --outfile=/tmp/cyrene-history.test.mjs \ +// && node /tmp/cyrene-history.test.mjs +// +// `tsc --noEmit`（npm run build 的一部分）会对本文件做类型检查，因此它只从项目内导入、 +// 不 import node:*。 + +import { dispatchFrame } from "../../ws"; +import { useAppStore } from "../../../store/useAppStore"; +import type { ChatMessage } from "../../types"; + +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 SUB = "web:sub"; +const T1 = "turn-aaaa"; + +/** 每个场景前把 store 复位成「新连接、无在途路」。 */ +function reset(activeSessionId = MAIN, messages: ChatMessage[] = []): void { + useAppStore.setState({ + activeSessionId, + messages, + sessionMessages: {}, + turnBindings: {}, + sessionBusy: false, + pendingCount: 0, + activeTurns: 0, + queueNotice: null, + deviceUpdates: [], + }); +} + +function msgs(): ChatMessage[] { + return useAppStore.getState().messages; +} + +/** 一个握手回放帧（形状与真实 socket 层一致：帧壳带 sessionId，data 里是 messages）。 */ +function replay(sessionId: string, list: Record[]): void { + dispatchFrame({ + type: "history", + sessionId, + data: { messages: list.map((m) => ({ sessionId, createdAt: 1, ...m })) }, + }); +} + +// ── ① 回放落进 store：断线窗口内落库的消息被补回来（按 id 去重） ──────────────────── +{ + reset(MAIN, [ + { id: "m1", sessionId: MAIN, role: "assistant", content: "旧回复", createdAt: 1 }, + ]); + replay(MAIN, [ + { id: "m1", role: "assistant", content: "旧回复" }, + { id: "m2", role: "user", content: "断线期间发的", createdAt: 2 }, + ]); + same(msgs().length, 2, "回放补回缺失的行，同 id 不重复"); + same(msgs().map((m) => m.id), ["m1", "m2"], "顺序 = 回放顺序（既有行不动，新行追加）"); + same(useAppStore.getState().deviceUpdates.length, 0, "history 被消费，不落未知帧兜底"); + + // 幂等：同一帧再来一次不产生第二份。 + replay(MAIN, [{ id: "m2", role: "user", content: "断线期间发的", createdAt: 2 }]); + same(msgs().length, 2, "重复回放不新增行"); +} + +// ── ② 回放按帧壳 sessionId 进桶，不落到激活会话 ────────────────────────────────── +{ + reset(SUB, []); + replay(MAIN, [{ id: "m1", role: "user", content: "主会话的消息" }]); + same(msgs().length, 0, "激活的是子会话 ⇒ 主会话回放不写进当前列表"); + same( + (useAppStore.getState().sessionMessages[MAIN] ?? []).map((m) => m.id), + ["m1"], + "回放进了主会话自己的桶（切过去即可见）" + ); +} + +// ── ③ 精确键认领：用户行按 clientMsgId ─────────────────────────────────────────── +{ + reset(MAIN, [ + { + id: "local-u", + role: "user", + content: "重发的话", + clientMsgId: "c-1", + delivery: "sending", + createdAt: 1, + }, + ]); + replay(MAIN, [{ id: "srv-u", role: "user", content: "重发的话", clientMsgId: "c-1" }]); + same(msgs().length, 1, "clientMsgId 命中 ⇒ 认领本端气泡，不新增"); + same(msgs()[0].id, "srv-u", "气泡对齐到服务端 id"); + same(msgs()[0].delivery, "sent", "落库 ⇒ 投递态转「已送达」"); +} + +// ── ③ 精确键认领：助手行按 turnId（并把在飞气泡停止流式） ───────────────────────── +{ + reset(MAIN, [ + { + id: "local-a", + role: "assistant", + content: "半截回", + streaming: true, + turnId: T1, + createdAt: 1, + }, + ]); + replay(MAIN, [{ id: "srv-a", role: "assistant", content: "完整回复", turnId: T1, createdAt: 5 }]); + same(msgs().length, 1, "turnId 命中 ⇒ 认领本路助手气泡，不新增"); + same(msgs()[0].id, "srv-a", "助手气泡对齐到服务端 id"); + same(msgs()[0].content, "完整回复", "正文以落库行为准"); + same(msgs()[0].streaming, false, "该路已收尾落库 ⇒ 停止流式（不再永久转圈）"); +} + +// ── ④ 精确键对不上绝不猜：不回放里用内容/时间启发式 ─────────────────────────────── +{ + reset(MAIN, [ + { + id: "local-a", + role: "assistant", + content: "正在流式的这一路", + streaming: true, + createdAt: 100, + }, + ]); + // 一条旧助手回放：没有 turnId、内容也不同 —— 启发式会按 createdAt 最近去配它。 + replay(MAIN, [{ id: "srv-old", role: "assistant", content: "很久以前的一条回复", createdAt: 2 }]); + same(msgs().length, 2, "精确键对不上 ⇒ 追加，绝不错配在飞气泡"); + same(msgs()[0].id, "local-a", "在飞气泡保持原样"); + same(msgs()[0].streaming, true, "在飞气泡的流式态不能被旧回放顶掉"); + same(msgs()[1].id, "srv-old", "旧行作为新行追加"); +} + +// ── ⑤ 畸形帧：没有 messages 数组（或坏行）时不动状态，也不进兜底 ────────────────── +{ + reset(MAIN, []); + dispatchFrame({ type: "history", sessionId: MAIN, data: { messages: "坏形状" } }); + same(msgs().length, 0, "非数组 messages 被忽略"); + same(useAppStore.getState().deviceUpdates.length, 0, "仍是已消费帧，不落兜底"); + + replay(MAIN, [{ role: "user", content: "没有 id 的坏行" }, { id: "ok-1", role: "user", content: "好行" }]); + same(msgs().map((m) => m.id), ["ok-1"], "坏行（无 id）被丢弃，好行照常并入"); +} + +console.log(`history replay logic test: ${checks} checks passed`); diff --git a/frontend/web/src/lib/ws/handlers/message.ts b/frontend/web/src/lib/ws/handlers/message.ts index b412fb0..585fdd1 100644 --- a/frontend/web/src/lib/ws/handlers/message.ts +++ b/frontend/web/src/lib/ws/handlers/message.ts @@ -1,6 +1,11 @@ import { MODES, type ChatMessage, type Mode } from "../../types"; -import { parseAdminChannelMessage, parsePersistedMessage, toRecord } from "../../wsParse"; -import type { FrameEvent, FramePayload, StoreState } from "./shared"; +import { + parseAdminChannelMessage, + parseHistoryMessages, + parsePersistedMessage, + toRecord, +} from "../../wsParse"; +import { frameSessionId, type FrameEvent, type FramePayload, type 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 @@ -40,6 +45,18 @@ function parseMessageEdited(data: unknown): MessageEdited | null { * 未命中返回 false 交给下一组。 */ export function handleMessageEvent(ev: FrameEvent, r: FramePayload, s: StoreState): boolean { switch (ev.type) { + case "history": { + // 握手回放（契约 §3.2，仅 owner/admin，只回放主会话最近 50 条）：此前这帧落到 + // 未知帧兜底被丢掉 ⇒ WS 断线窗口内落库的消息永久缺失（要等切会话/刷新才补）。 + // 并入时按 id 去重、只按精确键（clientMsgId / turnId）认领本端气泡（见 + // sessionCache.mergeHistory）—— 整批回放绝不用内容/时间启发式，以免把正在 + // 流式的气泡错配到某条旧行上。帧壳 sessionId（"web"）就是回放所属会话。 + const msgs = parseHistoryMessages(r?.messages); + if (msgs.length) { + s.mergeHistoryPage(frameSessionId(ev, r) ?? msgs[0].sessionId ?? "", msgs); + } + return true; + } case "message_persisted": { // design/16 §4.2 — a user or assistant message was persisted and broadcast // with its backend id. 裁决 16.4.5: the frame is written into the bucket its diff --git a/frontend/web/src/lib/ws/handlers/parallel.test.ts b/frontend/web/src/lib/ws/handlers/parallel.test.ts index b2e925a..5860e37 100644 --- a/frontend/web/src/lib/ws/handlers/parallel.test.ts +++ b/frontend/web/src/lib/ws/handlers/parallel.test.ts @@ -247,6 +247,32 @@ function turnBegin(turnId: string, kind: string, messageIds: string[], leg: numb down("session_busy", undefined, { sessionId: MAIN, mode: "thinking" }); down("stream_end", undefined, { stopReason: "end_turn" }); same(useAppStore.getState().sessionBusy, false, "无 turnId 的 stream_end 照旧收口忙态"); + + // activeTurns 兜底：绑定表证明不了「没有别的路」——重连后内存绑定已清空，但服务端 + // 报的在途路数 > 1 ⇒ 先收尾的那一路同样不得关掉忙徽标。 + reset(); + down("session_busy", undefined, { sessionId: MAIN, mode: "thinking", activeTurns: 2 }); + down("stream_end", T1, { stopReason: "end_turn" }); + same( + useAppStore.getState().sessionBusy, + true, + "绑定表为空但在途路数 2 ⇒ stream_end 不误置闲" + ); + + // session_interrupted：判定以 activeTurns 为准（mode 缺席也不误判为空闲）。 + down("session_interrupted", undefined, { + sessionId: MAIN, + dropped: 0, + cancelledTurns: 1, + activeTurns: 1, + }); + same(useAppStore.getState().sessionBusy, true, "activeTurns>0 的 session_interrupted 不当空闲"); + same(useAppStore.getState().activeTurns, 1, "被打断后如实刷新在途路数"); + + // 缺席 activeTurns 的会话忙碌帧按忙处理（不得因缺字段读成空闲）。 + reset(); + down("session_busy", undefined, { sessionId: MAIN, mode: "thinking" }); + same(useAppStore.getState().sessionBusy, true, "缺 activeTurns 的 session_busy 仍置忙"); } // ── ⑥ chat_duplicate：本地乐观气泡对齐到既有 id（不重复渲染） ───────────────────── diff --git a/frontend/web/src/lib/ws/handlers/session.test.ts b/frontend/web/src/lib/ws/handlers/session.test.ts index 2c8fa1a..e5c7053 100644 --- a/frontend/web/src/lib/ws/handlers/session.test.ts +++ b/frontend/web/src/lib/ws/handlers/session.test.ts @@ -57,24 +57,29 @@ function pending(): number { // ① 活动会话：置忙 → 空闲帧收口。这正是 16.4.1 的预约切换所依赖的「她静默下来」。 seed(MAIN); -dispatchFrame({ type: "session_busy", sessionId: MAIN, data: { sessionId: MAIN, mode: "thinking", busy: true, pendingCount: 0 } }); +dispatchFrame({ type: "session_busy", sessionId: MAIN, data: { sessionId: MAIN, mode: "thinking", activeTurns: 1 } }); 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} 时也不误判为闲）。 +// 忙闲口径 = `activeTurns`（契约 §3.4.1：这些帧的 data 只有 {sessionId, mode, activeTurns}）。 +// 曾读的 `busy` / `pendingCount` 键**后端从未发过**（`busyPayload` 只发三键），已删除。 +seed(MAIN); +dispatchFrame({ type: "session_busy", sessionId: MAIN, data: { sessionId: MAIN, mode: "executing", activeTurns: 2 } }); +ok(busy(), "session_busy{activeTurns:2} 置忙"); +same(useAppStore.getState().activeTurns, 2, "在途路数被读出来"); +dispatchFrame({ type: "session_idle", sessionId: MAIN, data: { sessionId: MAIN, mode: "idle", activeTurns: 0 } }); +ok(!busy(), "session_idle{activeTurns:0} 置闲"); + +// 字段缺席（老帧）时按忙处理，不得因缺字段读成空闲。 seed(MAIN); dispatchFrame({ type: "session_busy", sessionId: MAIN, data: { sessionId: MAIN, mode: "executing" } }); -ok(busy(), "session_busy 缺 busy 字段时默认置忙"); +ok(busy(), "session_busy 缺 activeTurns 时默认置忙"); -// 显式 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 字段，语义就是「不再忙」）。 +// ② interrupt：当前任务取消 + pending 清空（Docs/api-client.md §3.4.1。 +// 帧带 {sessionId, mode, dropped, cancelledTurns, activeTurns}，scope:"all" 时 mode=idle +// 且 activeTurns=0 ⇒ 语义就是「不再忙」）。 seed(MAIN); useAppStore.setState({ sessionBusy: true, pendingCount: 3 }); dispatchFrame({ type: "session_interrupted", sessionId: MAIN, data: { sessionId: MAIN, mode: "idle", dropped: 2 } }); diff --git a/frontend/web/src/lib/ws/handlers/session.ts b/frontend/web/src/lib/ws/handlers/session.ts index dad1647..e0fb8f6 100644 --- a/frontend/web/src/lib/ws/handlers/session.ts +++ b/frontend/web/src/lib/ws/handlers/session.ts @@ -24,14 +24,12 @@ export function handleSessionEvent(ev: FrameEvent, r: FramePayload, s: StoreStat case "session_pending": { // design/15 §5 — 她忙/有挂起消息时的轻状态（不阻塞输入框）。 // 忙态是**单槽全局**状态，只认屏幕上这个会话（见下方 session_idle 的说明）。 - // v0.65 §3.4.1 起帧里还带 `activeTurns`（在途路数）：忙 = activeTurns > 0， - // 比 `busy` 更能表达「并行 N 路」；老后端不发时回落 `busy`（缺省按忙处理）。 + // v0.65 §3.4.1 起这些帧的 data = `{sessionId, mode, activeTurns}`：忙 = + // `activeTurns > 0`，比任何布尔更能表达「并行 N 路」。这两个键**后端从不发** + // （`busyPayload` 只发三键；`"busy":` 全仓零命中），所以不再读，缺失时按忙处理。 if (!frameIsActive(frameSessionId(ev, r), s.activeSessionId)) return true; const turns = frameNumber(r?.activeTurns); - const busy = turns !== undefined ? turns > 0 : typeof r?.busy === "boolean" ? r.busy : true; - const pendingCount = - typeof r?.pendingCount === "number" ? r.pendingCount : undefined; - s.setSessionBusy(busy, pendingCount, turns); + s.setSessionBusy(turns !== undefined ? turns > 0 : true, undefined, turns); return true; } case "session_idle": { @@ -54,12 +52,15 @@ export function handleSessionEvent(ev: FrameEvent, r: FramePayload, s: StoreStat // scope:"current" ⇒ mode 是当时忙碌的通道（**非 idle**）、dropped:0、 // activeTurns 可能 > 0（并行的其它路仍在跑）——置闲就会 // 显示「她闲了」而实际还在答。 + // 判定口径与 `session_busy` 一致：**activeTurns 优先**（>0 = 仍在忙）， + // 它缺席时才看 mode（非 idle = 忙）；两者都缺（老帧）沿用旧行为当作不再忙。 const sid = frameSessionId(ev, r); if (!frameIsActive(sid, s.activeSessionId)) return true; const mode = frameText(r?.mode); const turns = frameNumber(r?.activeTurns); - if (mode !== undefined && mode !== "idle") { - s.setSessionBusy(turns === undefined ? true : turns > 0, undefined, turns); + const busy = turns !== undefined ? turns > 0 : mode !== undefined ? mode !== "idle" : false; + if (busy) { + s.setSessionBusy(true, undefined, turns); return true; } s.setSessionBusy(false, 0, 0); diff --git a/frontend/web/src/lib/ws/handlers/stream.ts b/frontend/web/src/lib/ws/handlers/stream.ts index 6066367..dcbd0b5 100644 --- a/frontend/web/src/lib/ws/handlers/stream.ts +++ b/frontend/web/src/lib/ws/handlers/stream.ts @@ -63,9 +63,14 @@ export function handleStreamEvent(ev: FrameEvent, r: FramePayload, s: StoreState /** 除了刚收尾的这一路，该会话还有别的路在跑吗？有就**不能**把忙态置闲 —— 并行时 * 先答完的那一路会把「她正在忙」徽标误关掉（v0.65 §4.6 的 `session_idle` 时点收紧 - * 就是为这个）。没有 `turnId` 的旧帧（单路老后端）逐字沿用旧行为：照样清忙。 */ + * 就是为这个）。两个判据取并集： + * ① 绑定表里还有别的 `turnId`（最强证据，`turn_begin` 建立）； + * ② 服务端报出的在途路数 `activeTurns > 1`（绑定表证明不了「没有别的路」： + * 重连后内存绑定清空、或那一路的 `turn_begin` 丢帧时，仍要靠它兜住）。 + * 没有 `turnId` 的旧帧（单路老后端）逐字沿用旧行为：照样清忙。 */ function otherLegsRunning(s: StoreState, sessionId: string | undefined, turnId: string | undefined): boolean { if (!turnId) return false; const sid = sessionId ?? s.activeSessionId; - return turnsInSession(s.turnBindings, sid).some((b) => b.turnId !== turnId); + if (turnsInSession(s.turnBindings, sid).some((b) => b.turnId !== turnId)) return true; + return s.activeTurns > 1; } diff --git a/frontend/web/src/lib/wsParse.ts b/frontend/web/src/lib/wsParse.ts index aeed53a..bc953e5 100644 --- a/frontend/web/src/lib/wsParse.ts +++ b/frontend/web/src/lib/wsParse.ts @@ -229,6 +229,20 @@ export function parsePersistedMessage(v: unknown): ChatMessage | null { }; } +/** 解析握手 `history` 回放帧的 `data.messages` 数组（契约 §3.2）：服务端在 owner/admin + * 连接注册后**定向**发一帧，内含主会话（`web`）最近 50 条（`wsReplayLimit`）。每条就是 + * 同一个 `historyMessage` DTO（§4.4），因此复用 `parsePersistedMessage` 收窄；坏行走 + * `null` 被丢弃，绝不把一条畸形行塞进列表。 */ +export function parseHistoryMessages(v: unknown): ChatMessage[] { + if (!Array.isArray(v)) return []; + const out: ChatMessage[] = []; + for (const item of v) { + const msg = parsePersistedMessage(item); + if (msg) out.push(msg); + } + return out; +} + /** Build an AdminChannelMessage from a `channel_persisted` payload (design/16 * §4.6 §4): the backend broadcast a fully-persisted isolated (group/stranger) * channel message carrying the backend id + adapter id, for the IM adapter log diff --git a/frontend/web/src/store/sessionCache.ts b/frontend/web/src/store/sessionCache.ts index 5b7e28c..b03615f 100644 --- a/frontend/web/src/store/sessionCache.ts +++ b/frontend/web/src/store/sessionCache.ts @@ -28,6 +28,72 @@ export const BACKTHINK_SESSION_ID = "sess-backthink"; /** 一个会话的消息列表；键是会话 id。 */ export type SessionMessages = Record; +/** 握手 `history` 回放（契约 §3.2）的**整批**并入：一次最多 50 条**已落库**的权威行。 + * + * 为什么不直接复用 `mergeMessage`：单条 `message_persisted` 允许用「同 role + 内容全等 / + * `createdAt` 最近」的启发式去认领本端乐观气泡；整批回放套这条启发式，会把本端 + * **正在流式**的气泡错配到某条旧助手行上（正文被旧内容顶掉、流式态丢失）—— + * 回放越大、越容易配错。因此这里认领本端气泡只用**精确键**： + * - 用户行：`clientMsgId`（本端生成、随 `chat.clientMsgId` 上行，回放 DTO 也带）； + * - 助手行：`turnId`（`turn_begin` 已把这个 id 钉在本端气泡上）。 + * 精确键对不上一律**追加**（猜错比多一条更糟），同 id 则 upsert。 */ +export function mergeHistory(list: ChatMessage[], msgs: ChatMessage[]): ChatMessage[] { + let next = list; + for (const msg of msgs) { + const copy = next === list ? list.slice() : next; + const idx = copy.findIndex((m) => m.id === msg.id); + if (idx !== -1) { + copy[idx] = { ...copy[idx], ...msg }; + next = copy; + continue; + } + const adopt = historyAdoptIndex(copy, msg); + if (adopt !== -1) { + const m = copy[adopt]; + copy[adopt] = { + ...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, + documents: msg.documents ?? m.documents, + turnId: msg.turnId ?? m.turnId, + clientMsgId: msg.clientMsgId ?? m.clientMsgId, + personaId: msg.personaId ?? m.personaId, + createdAt: msg.createdAt ?? m.createdAt, + // 回放里的**助手行** ⇒ 这一路已收尾落库：本端在飞气泡必须停止流式， + // 否则那一帧 `stream_end` 丢在断线窗口里会让它永远转圈。 + ...(msg.role === "assistant" ? { streaming: false } : {}), + // 用户行落库 ⇒ 不再是「发送中」，转「已送达」；已被某路接手（answering） + // 或已排队（queued）的不降级 —— 那些态来自比「落库」更晚的帧。 + ...(msg.role === "user" && m.delivery === "sending" + ? { delivery: "sent" as const } + : {}), + }; + next = copy; + continue; + } + copy.push(msg); + next = copy; + } + return next; +} + +/** 回放行只按**精确键**认领本端乐观气泡：用户行认 `clientMsgId`、助手行认 `turnId`。 + * 两者都没有、或本端没有对应气泡时返回 -1（调用方改为追加）。 */ +function historyAdoptIndex(list: ChatMessage[], msg: ChatMessage): number { + if (!msg.clientMsgId && !msg.turnId) return -1; + for (let i = list.length - 1; i >= 0; i -= 1) { + const m = list[i]; + if (m.sessionId || m.role !== msg.role) continue; + if (msg.role === "user" && msg.clientMsgId && m.clientMsgId === msg.clientMsgId) return i; + if (msg.role === "assistant" && msg.turnId && m.turnId === msg.turnId) return i; + } + return -1; +} + /** 把一条后端消息并入消息列表：按 id upsert；没有同 id 时采纳本端同 role 的本地 * 乐观气泡（本地 `uid` → 后端 `id`），使同一句话不会渲染两遍（design/16 §4.3 去重）。 * 后端消息是内容的真源，本地气泡保留流式/工具/思考等本端状态。 diff --git a/frontend/web/src/store/useAppStore.ts b/frontend/web/src/store/useAppStore.ts index dbaa549..d4e5e37 100644 --- a/frontend/web/src/store/useAppStore.ts +++ b/frontend/web/src/store/useAppStore.ts @@ -18,6 +18,7 @@ import { now, uid } from "../lib/util"; import { currentGraph, dropMessage, + mergeHistory, mergeMessage, patchAssistantTail, patchMessage, @@ -144,6 +145,13 @@ export interface AppStore extends TurnSlice { * 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; + /** Merge one handshake `history` replay page (contract §3.2) into the session the + * frame names — up to 50 already-persisted rows, once per (re)connect. Dedup by + * id and adopt a local optimistic bubble only on an exact key (`clientMsgId` for + * users / `turnId` for assistants), never by the content/time heuristic: this is + * the reconnect catch-up path, so a wrong adopt would rewrite a streaming bubble + * with an old reply. */ + mergeHistoryPage: (sessionId: string, msgs: ChatMessage[]) => void; /** Patch one already-rendered message in place by id — the `message_edited` WS * 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 @@ -280,6 +288,14 @@ export const useAppStore = create()( ) ), + mergeHistoryPage: (sessionId, list) => + set((s) => { + if (!list.length) return s; + return writeSession(s, sessionId || s.activeSessionId, (cur) => + mergeHistory(cur, list) + ); + }), + updateMessageById: (id, patch, sessionId) => set((s) => writeSessionById(s, sessionId, (list) => patchMessage(list, id, patch))