feat(web): 前端 P0 收尾(多路忙态收口 / 握手 history 落 store / 清理幽灵键)
This commit is contained in:
1 parent
e852e60af9
commit
012e9f3ad4
10 files changed
+365
-31
No files matched your search
@@ -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;
|
||||
}
|
||||
@@ -0,0 +1,176 @@
|
||||
// 握手 `history` 回放帧的逻辑测试(契约 §3.2 + §4.4)。
|
||||
//
|
||||
// 为什么需要它:服务端在 owner/admin 连接注册后**定向**发一帧
|
||||
// `{"type":"history","sessionId":"web","data":{"messages":[<historyMessage>…]}}`(主会话
|
||||
// 最近 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<T>(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<string, unknown>[]): 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`);
|
||||
@@ -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
|
||||
|
||||
@@ -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(不重复渲染) ─────────────────────
|
||||
|
||||
@@ -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 } });
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -28,6 +28,72 @@ export const BACKTHINK_SESSION_ID = "sess-backthink";
|
||||
/** 一个会话的消息列表;键是会话 id。 */
|
||||
export type SessionMessages = Record<string, ChatMessage[]>;
|
||||
|
||||
/** 握手 `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 去重)。
|
||||
* 后端消息是内容的真源,本地气泡保留流式/工具/思考等本端状态。
|
||||
|
||||
@@ -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<AppStore>()(
|
||||
)
|
||||
),
|
||||
|
||||
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))
|
||||
|
||||
Reference in new issue
Block a user