【取证(症状成立,且比登记描述更糟)】两路真并行时两路最终落盘**同一个 turn_id**,
其中一路的图彻底消失,且 A 路的工具被画在 B 路的图上。逐层定位:发布器 graphs 只按
sessionID 建键(后起的路整张顶掉前者)· 帧载荷基线的键集被既有测试精确断言为
{id,nodes,edges,sessionId,startedAt,endedAt}(无任何区分路的键)· 前端每会话单桶 ·
落盘 (session_id,turn_id) 唯一键两路相撞 · 冷启动 LastGraph 只有一张。
该 bug 早已公开登记(design/15 §2.5.10-1「本轮未解决」、契约 §6-19「仍**未修**」)。
【修法】
- 归属桶 {会话}[#{路}](loop.FlowOwnerKey,与 checkpoint 键同字符口径);路只来自
ParallelLegFrom(ctx) ⇒ 单路逐字不变(桶 = 会话 id),既有调用点语义零改动
- 图 id 的第二分量改用**建图序号**(每次 +1、永不复用):同一轮里可以有并行路 +
队首续跑的串行路,用轮次号会让两条串行路撞同一个 web-1
- 轮次号只在上一轮的路全部收尾后推进,并丢掉更早轮次的桶(否则内存随路数无界增长)
- 新增只读出口 LatestLanes ⇒ GET /admin/flow/graph 的 lanes[](最近一轮每一路),
graph 仍是「最新一路」= 向后兼容
- 同源洞一并修:子代理归属原取 AgentSpec.ParentSession(会话、不含路)⇒ 并行路里派的
子代理 span 被静默丢弃;新增 AgentSpec.FlowOwner 承载「会话#路」,ParentSession 继续
只承载会话(总线载荷 subagent_result.sessionId 不受影响)
- 前端:store 增 flowLanes(按 turnId 分格、按轮淘汰、硬上限 4)+ LaneTracks 泳道卡;
单路/未给 lanes 时渲染路径与改前逐字相同
【无迁移】turn_id 是自由文本、每路 id 唯一 ⇒ 0022 的唯一索引天然分两行(0023 仍空着)
【我的验收】① 隔离 worktree(30 文件)build/vet/linux/darwin 全过、69 包全绿 0 FAIL;
② 我亲手打红变异:让 FlowOwnerKey 忽略路 ⇒ TestFlowPublisher_ParallelLegsDoNotOverwriteEachOther
FAIL;③ 前端独立复跑 npx tsc --noEmit + npm run build 通过(4.12s)
⚠️ 契约 v0.77 的登记在紧随其后的 docs(contract) 提交里(同一个工作时段内)。
335 lines
15 KiB
TypeScript
335 lines
15 KiB
TypeScript
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` 取图,而不是所有页面共用一张全局图——后台思考的完整链路图与
|
||
* 「用户侧前端发不发」无关,也绝不顶掉管理员看的「当前会话」图。真并行时同一
|
||
* 会话同一轮有**多条路**(切片 FV2),于是会话桶之外还有**路桶**:同一个会话的
|
||
* 每条路各占一格(`putLane`),两路交错的帧因此不会互相覆盖。
|
||
*
|
||
* 从 `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<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 去重)。
|
||
* 后端消息是内容的真源,本地气泡保留流式/工具/思考等本端状态。
|
||
*
|
||
* 采纳顺序(忙时并行后很重要,规范草案 §1.7):
|
||
* ① 同 `turnId` 且尚未落库的气泡 —— `turn_begin` 已把本端气泡绑到这一路,turnId
|
||
* 才是权威配对(并行时两条回复内容可能相似,靠内容判等会**配错路**);
|
||
* ② 内容完全相同的那条(旧启发式,保持老后端行为);
|
||
* ③ createdAt 最近的那条。 */
|
||
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;
|
||
}
|
||
if (msg.turnId) {
|
||
for (let i = list.length - 1; i >= 0; i -= 1) {
|
||
const m = list[i];
|
||
if (!m.sessionId && m.role === msg.role && m.turnId === msg.turnId) {
|
||
return adopt(list, i, msg);
|
||
}
|
||
}
|
||
}
|
||
// 没有同 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];
|
||
return adopt(list, best, msg);
|
||
}
|
||
|
||
/** 把本端乐观气泡归一为后端消息(id 换成服务端的,其余保留本端状态)。 */
|
||
function adopt(list: ChatMessage[], at: number, msg: ChatMessage): ChatMessage[] {
|
||
const m = list[at];
|
||
const next = list.slice();
|
||
next[at] = {
|
||
...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,
|
||
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>
|
||
): 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<string, FlowGraph>,
|
||
sessionId: string,
|
||
graph: FlowGraph
|
||
): Record<string, FlowGraph> {
|
||
const next: Record<string, FlowGraph> = {};
|
||
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<string, FlowGraph> = {};
|
||
for (const k of keys) if (!drop.has(k)) trimmed[k] = next[k];
|
||
return trimmed;
|
||
}
|
||
|
||
/** 一条**路**在 store 里的键(切片 FV2):图自己的 `turnId`;没有权威标识时用空串
|
||
* —— 那是「会话那一张」(单路 / 旧后端),因此旧形状的帧仍然只占一个桶。 */
|
||
export function laneKeyOf(graph: FlowGraph): string {
|
||
return graph.turnId ?? "";
|
||
}
|
||
|
||
/** 一个会话最多同时保留多少条路的图。真并行的硬上限是 3 路(后端
|
||
* `maxParallelTurnsHardCap`),留一格余量,绝不无界增长。导出它是因为前端逻辑测试
|
||
* 必须能钉住这条上限,而不是在测试里抄一个字面量。 */
|
||
export const MAX_LANES = 4;
|
||
|
||
/** 把一帧图并进该会话**本轮**的路列表:按路键就地更新(同一条路的高频帧只占一格),
|
||
* 并丢掉**更早轮次**的路 —— 轮次号更大的帧到达即表示新一轮开始,上一轮的路该退场
|
||
* (缺席轮次号的旧帧不参与淘汰,按「保留」处理,向后兼容)。
|
||
*
|
||
* 这是「两路互不覆盖」的前端半边:后端给每条路各发一帧(各带自己的 `turnId`),
|
||
* 这里按路分格存放,于是两路交错到达时 store 里是两条泳道而不是一条。 */
|
||
export function putLane(
|
||
lanes: readonly FlowGraph[] | undefined,
|
||
graph: FlowGraph
|
||
): FlowGraph[] {
|
||
const key = laneKeyOf(graph);
|
||
const list = [...(lanes ?? [])];
|
||
const at = list.findIndex((g) => laneKeyOf(g) === key);
|
||
if (at >= 0) list[at] = graph;
|
||
else list.push(graph);
|
||
const round = graph.round;
|
||
const kept =
|
||
round === undefined ? list : list.filter((g) => g.round === undefined || g.round >= round);
|
||
return kept.slice(-MAX_LANES);
|
||
}
|
||
|
||
/** 一次图写入的完整结果:会话桶(向后兼容的「该会话最新一张」)+ 路桶(本轮每一路)。 */
|
||
export function writeFlowGraph(
|
||
graphs: Record<string, FlowGraph>,
|
||
lanes: Record<string, FlowGraph[]>,
|
||
sessionId: string,
|
||
graph: FlowGraph
|
||
): { flowGraphs: Record<string, FlowGraph>; flowLanes: Record<string, FlowGraph[]> } {
|
||
return {
|
||
flowGraphs: putGraph(graphs, sessionId, graph),
|
||
flowLanes: { ...lanes, [sessionId]: putLane(lanes[sessionId], graph) },
|
||
};
|
||
}
|
||
|
||
/** 兼容字段 `flowGraph` 的口径(管理端 Flow / Agents 页读它):**激活会话**的图;
|
||
* 激活会话还没有图时回落主会话的最近一张,使页面不至于空着。
|
||
* 后台思考会话的图不会顶掉它 —— 这正是裁决 16.4.2 要修的:每条思考看自己的图,
|
||
* 「当前会话」的图只随当前会话变化。 */
|
||
export function currentGraph(
|
||
graphs: Record<string, FlowGraph>,
|
||
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> | 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> | 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;
|
||
}
|