diff --git a/frontend/web/src/admin/modules/Flow.tsx b/frontend/web/src/admin/modules/Flow.tsx index 3fdf5f7..7e785fb 100644 --- a/frontend/web/src/admin/modules/Flow.tsx +++ b/frontend/web/src/admin/modules/Flow.tsx @@ -7,6 +7,7 @@ import { Icon } from "../../components/Icon"; import { FlowGraphView } from "./FlowGraph"; import { NodeDetail } from "./flow/NodeDetail"; import { LaneSummary } from "./flow/LaneSummary"; +import { LaneTracks } from "./flow/LaneTracks"; import { FlowSnapshotHint } from "./flow/FlowSnapshotHint"; import { useFlowSnapshot, type FlowGraphSource } from "./flow/useFlowSnapshot"; import { useFlowTurns } from "./flow/useFlowTurns"; @@ -38,10 +39,16 @@ import type { AdminFlowTurn } from "../../lib/api"; export function FlowModule(): React.ReactNode { const sessions = useAppStore((s) => s.sessions); const flowGraphs = useAppStore((s) => s.flowGraphs); + const flowLanes = useAppStore((s) => s.flowLanes); // 选择器选中值：默认主会话 `web`（后端固定 MainSessionID），不随激活会话漂移。 const [pickedSessionId, setPickedSessionId] = useState(MAIN_SESSION_ID); - // 取数接线（纯函数，flowSession.ts）：用哪个 sessionId 取数 / 画哪张图 / 有哪些选项。 - const { sessionId, graph, options } = flowPageData(sessions, flowGraphs, pickedSessionId); + // 取数接线（纯函数，flowSession.ts）：用哪个 sessionId 取数 / 画哪张图 / 本轮有哪些路 / 选项。 + const { sessionId, graph, lanes, options } = flowPageData( + sessions, + flowGraphs, + pickedSessionId, + flowLanes, + ); // 最新一轮：F1 的实时图 + 只读快照补拉（补拉会写进图桶 ⇒ 配合图桶持久化「刷新即刻有图」）。 const graphSource = useFlowSnapshot(sessionId); // 历史轮（契约 v0.73 多轮历史回放）：已落盘轮次列表 + 选中轮的只读图。它**不写 store**， @@ -50,6 +57,7 @@ export function FlowModule(): React.ReactNode { return ( 1 ? laneList : []; return (
@@ -172,24 +193,32 @@ export function FlowBody({ /> - - - setSelectedId(n.id)} + {parallelLanes.length > 0 ? ( + - - {showingHistory ? null : ( - + + setSelectedId(n.id)} /> - )} - + + {showingHistory ? null : ( + + )} + + )} - {selected ? ( + {selected && parallelLanes.length === 0 ? ( + {lanes.map((lane, i) => ( + + ))} +
+ ); +} + +/** 一条路的泳道卡：标题（第几路 + 权威 turnId）、图、子代理泳道摘要、节点详情。 */ +function LaneTrack({ + lane, + index, + sessionId, + source, +}: { + lane: FlowGraph; + /** 展示用的路序号（1 基，最旧的为 1）。 */ + index: number; + sessionId: string; + source: FlowGraphSource; +}): React.ReactNode { + const [selectedId, setSelectedId] = useState(null); + const grouping = groupFlowLanes(lane); + const selected = lane.nodes.find((n) => n.id === selectedId) ?? null; + return ( + +
+ + 路 {index} + + + {lane.turnId ? `turnId ${lane.turnId}` : "无路标识（旧后端 / 单路形状）"} + + + {lane.nodes.length} 节点 · {lane.edges.length} 连线 + +
+ setSelectedId(n.id)} + /> + + + {selected ? ( + setSelectedId(null)} /> + ) : null} + + ); +} diff --git a/frontend/web/src/admin/modules/flow/flowSession.ts b/frontend/web/src/admin/modules/flow/flowSession.ts index 905b645..5e7a5a1 100644 --- a/frontend/web/src/admin/modules/flow/flowSession.ts +++ b/frontend/web/src/admin/modules/flow/flowSession.ts @@ -28,12 +28,16 @@ export interface FlowSessionOption { label: string; } -/** Flow 页的取数接线：`sessionId` 给取数层（hook / REST）用，`graph` 是屏幕上要画的图。 */ +/** Flow 页的取数接线：`sessionId` 给取数层（hook / REST）用，`graph` 是屏幕上要画的图， + * `lanes` 是该会话**本轮每一路**的图（真并行时不止一张，见切片 FV2）。 */ export interface FlowPageData { /** 取数用的会话 id（`useFlowSnapshot` / `useFlowTurns` / REST 的 `sessionId`）。 */ sessionId: string; /** 该会话的图（取自**它自己的**图桶；没有就是没有，绝不编图）。 */ graph: FlowGraph | null; + /** 该会话本轮每一路的图（最旧的在前）。路桶还没有内容时（单路 / 旧后端 / 快照补拉 + * 前）回落到 `graph` 那一条 —— 单路下「每一路」就是那一路。 */ + lanes: FlowGraph[]; /** 会话选择器选项（只含白名单内会话）。 */ options: FlowSessionOption[]; } @@ -96,17 +100,23 @@ export function resolveFlowSessionId( return options.some((o) => o.id === picked) ? picked : MAIN_SESSION_ID; } -/** Flow 页的取数接线（纯函数）：会话列表 + 图桶 + 选择器选中值 ⇒ - * 「用哪个 sessionId 取数 / 画哪张图 / 有哪些选项」。 +/** Flow 页的取数接线（纯函数）：会话列表 + 图桶 + 路桶 + 选择器选中值 ⇒ + * 「用哪个 sessionId 取数 / 画哪张图 / 本轮有哪些路 / 有哪些选项」。 * * `graph` **只看选中会话自己的图桶**，绝不回落到兼容字段 `flowGraph`（它按 - * `activeSessionId` 取 ⇒ 切了会话图也不变，本切片要修的就是这个洞）。 */ + * `activeSessionId` 取 ⇒ 切了会话图也不变，本切片要修的就是这个洞）。 + * + * `lanes` 是该会话**本轮每一路**的图（真并行时两条以上）；路桶还没有内容时回落到 + * 单个 `graph` —— 单路/旧后端下「每一路」就是那一路，绝不让页面因为路桶为空而空掉。 */ export function flowPageData( sessions: Record, flowGraphs: Record, picked: string, + flowLanes?: Record, ): FlowPageData { const options = flowSessionOptions(sessions); const sessionId = resolveFlowSessionId(picked, options); - return { sessionId, graph: flowGraphs[sessionId] ?? null, options }; + const graph = flowGraphs[sessionId] ?? null; + const lanes = flowLanes?.[sessionId]?.length ? flowLanes[sessionId] : graph ? [graph] : []; + return { sessionId, graph, lanes, options }; } diff --git a/frontend/web/src/admin/modules/flow/laneTracks.render.test.ts b/frontend/web/src/admin/modules/flow/laneTracks.render.test.ts new file mode 100644 index 0000000..6e07042 --- /dev/null +++ b/frontend/web/src/admin/modules/flow/laneTracks.render.test.ts @@ -0,0 +1,112 @@ +// 「真并行两路」的**真实渲染**测试（切片 FV2；design/19 §2、design/15 §2.5.10-1）。 +// +// 为什么需要它：逻辑测试（store/flowLanes.test.ts）钉住「两路进了两个桶」，而主人报的 +// 症状是**看得见**的 —— 面板上只出现一路的图（后起的那一路把前一路顶掉）。这里钉住 +// 「桶里有两路 ⇒ 页面上真的并列画出两条泳道」，并且单路时渲染路径与引入本键之前一致 +// （既有渲染测试因此不受影响）。 +// +// 渲染的是 **props 驱动**的 `FlowBody`：本项目没有 jsdom，store 驱动的组件在测试里只能 +// 看到初始 state ⇒ 只能喂 props。 +// +// 跑法（渲染类测试必须 `--format=cjs`，见 messageSync.test.ts 顶部注释）： +// +// cd frontend/web \ +// && ./node_modules/.bin/esbuild src/admin/modules/flow/laneTracks.render.test.ts \ +// --bundle --platform=node --format=cjs \ +// --outfile=/tmp/cyrene-flow-lanes-render.test.cjs \ +// && node /tmp/cyrene-flow-lanes-render.test.cjs + +import { createElement } from "react"; +import { renderToStaticMarkup } from "react-dom/server"; +import type { FlowGraph } from "../../../lib/types"; +import { MAIN_SESSION_ID } from "../../../store/sessionCache"; +import { FlowBody } from "../Flow"; + +let checks = 0; + +function ok(cond: boolean, what: string): void { + checks += 1; + if (!cond) throw new Error(`FAILED: ${what}`); +} + +const has = (haystack: string, needle: string): boolean => haystack.includes(needle); + +/** 一路的图：输入 + 一个以该路命名的工具节点（label 是断言用的哨兵）。 */ +function laneGraph(id: string, turnId: string, sentinel: string): FlowGraph { + return { + id, + sessionId: MAIN_SESSION_ID, + turnId, + round: 1, + nodes: [ + { id: "input", label: `${sentinel}输入`, kind: "input", status: "done" }, + { id: `span-${turnId}`, label: sentinel, kind: "tool", status: "done" }, + { id: "output", label: `${sentinel}回复`, kind: "output", status: "done" }, + ], + edges: [ + { from: "input", to: `span-${turnId}` }, + { from: `span-${turnId}`, to: "output" }, + ], + }; +} + +const laneA = laneGraph("web-1#turn-a", "turn-a", "哨兵A"); +const laneB = laneGraph("web-2#turn-b", "turn-b", "哨兵B"); + +// ── ① 两路 ⇒ 两条泳道（两路的哨兵都在，且各有各的 turnId 标题） ───────────── +const twoLanes = renderToStaticMarkup( + createElement(FlowBody, { + graph: laneB, + lanes: [laneA, laneB], + sessionId: MAIN_SESSION_ID, + source: "live", + }) +); +ok(has(twoLanes, "路 1") && has(twoLanes, "路 2"), "两路时画出两条泳道卡"); +ok(has(twoLanes, "turn-a") && has(twoLanes, "turn-b"), "每条泳道标出自己那一路的 turnId"); +ok( + has(twoLanes, "哨兵A") && has(twoLanes, "哨兵B"), + "两路的图都在页面上（后起的那一路绝不顶掉前一路）", +); +ok( + twoLanes.split("3 节点").length === 3, + "两条泳道各自带自己的节点统计（两张图各画一次，不是同一张画两遍）", +); + +// ── ② 单路 / 未给 lanes ⇒ 与引入本键之前的单图呈现一致（没有泳道标题） ────── +const single = renderToStaticMarkup( + createElement(FlowBody, { + graph: laneB, + lanes: [laneB], + sessionId: MAIN_SESSION_ID, + source: "live", + }) +); +ok(has(single, "哨兵B"), "单路仍画出那一张图"); +ok(!has(single, "路 1"), "单路不出现「路 N」泳道标题（保持既有呈现）"); + +const withoutLanes = renderToStaticMarkup( + createElement(FlowBody, { + graph: laneB, + sessionId: MAIN_SESSION_ID, + source: "live", + }) +); +ok(!has(withoutLanes, "路 1"), "省略 lanes 时按单路渲染（向后兼容）"); + +// ── ③ 历史轮：翻历史轮时走那一轮的落盘图，不按「本轮多路」铺开 ────────────── +const history = renderToStaticMarkup( + createElement(FlowBody, { + graph: laneB, + lanes: [laneA, laneB], + sessionId: MAIN_SESSION_ID, + source: "live", + selectedTurnId: "web-1#turn-a", + historyGraph: laneA, + onSelectTurn: () => {}, + }) +); +ok(has(history, "哨兵A"), "翻历史轮画的是选中的那一轮"); +ok(!has(history, "路 1"), "历史轮不走「本轮多路」的泳道铺开"); + +console.log(`flow lanes render test: ${checks} checks passed`); diff --git a/frontend/web/src/admin/modules/flow/useFlowSnapshot.ts b/frontend/web/src/admin/modules/flow/useFlowSnapshot.ts index ba0be78..9686a1f 100644 --- a/frontend/web/src/admin/modules/flow/useFlowSnapshot.ts +++ b/frontend/web/src/admin/modules/flow/useFlowSnapshot.ts @@ -48,6 +48,11 @@ export function useFlowSnapshot(sessionId: string): FlowGraphSource { adminFlowGraph(sessionId) .then((resp) => { if (cancelled || !resp.graph) return; + // 真并行时同一轮有多条路（契约 v0.77 的 `lanes[]`）：每一路都写进图桶， + // 于是冷启动的页面同样看到「两路各自的图」，而不是只剩最新那一路。 + for (const lane of resp.lanes ?? []) { + if (lane.graph) setFlowGraph(lane.graph, sessionId); + } setFlowGraph(resp.graph, sessionId); setPulled(resp.graph); }) diff --git a/frontend/web/src/lib/api/adminSystem.ts b/frontend/web/src/lib/api/adminSystem.ts index 9021fca..c28b312 100644 --- a/frontend/web/src/lib/api/adminSystem.ts +++ b/frontend/web/src/lib/api/adminSystem.ts @@ -68,9 +68,20 @@ export interface AdminFlowGraphResponse { turnId: string; source: "live" | "persisted" | "none"; graph: FlowGraph | null; + /** 最近一轮**每一路**的图（契约 v0.77，纯增量；真并行 `chat.parallel` 时不止一条）。 + * `graph` 只是其中最新的一条（向后兼容），`lanes` 才是全貌 —— 面板据此把两路并列 + * 画出来，而不是让后起的那一路把前一路顶掉。没有图时是 `[]`（后端保证非 null）。 */ + lanes: AdminFlowLane[]; turns: AdminFlowTurn[]; } +/** `lanes[]` 里的一条路：`turnId` 是该路的权威回合标识（design/15 §2.5.9）， + * `graph` 与响应的 `graph` / 帧的 data 逐字同形。 */ +export interface AdminFlowLane { + turnId: string; + graph: FlowGraph | null; +} + /** GET /api/v1/admin/flow/graph?sessionId=&turnId= — **链路图快照 + 多轮历史回放**（契约 * v0.71 起，v0.73 增量；只读、不审计，owner/admin 档）。 * diff --git a/frontend/web/src/lib/types/flow.ts b/frontend/web/src/lib/types/flow.ts index baea80c..860f5a3 100644 --- a/frontend/web/src/lib/types/flow.ts +++ b/frontend/web/src/lib/types/flow.ts @@ -23,6 +23,17 @@ export interface FlowGraph { edges: FlowEdge[]; /** The graph's owning session id (backend `sessionId`). */ sessionId?: string; + /** The **lane** this graph belongs to: the authoritative turn id of the leg that + * produced it (backend `turnId`, design/15 §2.5.9). Absent when that leg has no + * authoritative id (single leg / older backend) — then the client keeps the + * per-session bucket only (compat), never inventing an id. Real parallel turns + * (`chat.parallel`) publish one graph per lane, and this key is what keeps them + * apart (design/15 §2.5.10-1). */ + turnId?: string; + /** The session round number this graph belongs to (backend `round`, monotonic per + * session; every lane of one round shares it). The store uses it to drop the + * lanes of earlier rounds. Absent (older backend) keeps every lane it received. */ + round?: number; /** Round start/end times (RFC3339), when the backend provides them. */ startedAt?: string; endedAt?: string; diff --git a/frontend/web/src/lib/wsParse.ts b/frontend/web/src/lib/wsParse.ts index 6c3233f..f396d87 100644 --- a/frontend/web/src/lib/wsParse.ts +++ b/frontend/web/src/lib/wsParse.ts @@ -172,6 +172,10 @@ export function parseGraph(payload: unknown): FlowGraph | null { sessionId: (typeof r.sessionId === "string" ? r.sessionId : undefined) ?? (typeof graphRec.sessionId === "string" ? graphRec.sessionId : undefined), + // 路标识与轮次号（契约 v0.77，纯增量）：`turnId` 缺席表示这一路没有权威标识 + // （单路 / 旧后端）⇒ 客户端只按会话归一张图；`round` 缺席按「未知轮次」处理。 + turnId: typeof graphRec.turnId === "string" ? graphRec.turnId : undefined, + round: typeof graphRec.round === "number" ? graphRec.round : undefined, startedAt: typeof graphRec.startedAt === "string" ? graphRec.startedAt : undefined, endedAt: typeof graphRec.endedAt === "string" ? graphRec.endedAt : undefined, }; diff --git a/frontend/web/src/store/flowLanes.test.ts b/frontend/web/src/store/flowLanes.test.ts new file mode 100644 index 0000000..433935e --- /dev/null +++ b/frontend/web/src/store/flowLanes.test.ts @@ -0,0 +1,166 @@ +// Logic test for the **per-lane** flow buckets (切片 FV2): a real parallel turn +// (`chat.parallel`) publishes one graph per 路 (leg), each carrying its own +// `turnId`; the store must hold them as separate swimlanes instead of letting the +// later frame overwrite the earlier one (the reported symptom: 「一路的节点长在另 +// 一路的图上」, design/15 §2.5.10-1). +// +// It drives the real frame decoder (`dispatchFrame` in lib/ws.ts) and the real +// store, and throws on the first failure. The project has no frontend test runner +// and no jsdom, so this is a dependency-free plain-node check; the banner supplies +// the `window.localStorage` the store's persist middleware probes (absent under node): +// +// cd frontend/web \ +// && ./node_modules/.bin/esbuild src/store/flowLanes.test.ts \ +// --bundle --platform=node --format=esm \ +// --banner:js='globalThis.window={localStorage:{getItem:()=>null,setItem:()=>{},removeItem:()=>{}}};' \ +// --outfile=/tmp/cyrene-flow-lanes.test.mjs \ +// && node /tmp/cyrene-flow-lanes.test.mjs +// +// `tsc --noEmit` (part of `npm run build`) type-checks this file, which is why it +// imports nothing from node:*. + +import { dispatchFrame } from "../lib/ws"; +import { useAppStore } from "./useAppStore"; +import { MAIN_SESSION_ID, MAX_LANES } from "./sessionCache"; +import { flowPageData } from "../admin/modules/flow/flowSession"; +import type { FlowGraph, SessionMeta } from "../lib/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 { + const g = JSON.stringify(got); + const w = JSON.stringify(want); + ok(g === w, `${what} — got ${g}, want ${w}`); +} + +/** 一帧 `graph_update`（真后端发的形状：图自带 sessionId / turnId / round）。 */ +function laneFrame(id: string, turnId: string | undefined, round: number | undefined, label: string): void { + dispatchFrame({ + type: "graph_update", + data: { + id, + sessionId: MAIN_SESSION_ID, + ...(turnId === undefined ? {} : { turnId }), + ...(round === undefined ? {} : { round }), + nodes: [{ id: "input", label, kind: "input", status: "done" }], + edges: [], + }, + }); +} + +/** 当前会话的路桶（最旧的在前）。 */ +function lanesOf(sessionId: string): FlowGraph[] { + return useAppStore.getState().flowLanes[sessionId] ?? []; +} + +// ① 两路帧**交错**到达 ⇒ 两条泳道，而不是一条（这是本切片的核心断言）。 +{ + laneFrame("web-1#turn-a", "turn-a", 1, "标记A"); + laneFrame("web-2#turn-b", "turn-b", 1, "标记B"); + // 两路各自继续推进（同一个 turnId 再发一帧）：只能就地更新，绝不能新开一格。 + laneFrame("web-1#turn-a", "turn-a", 1, "标记A（更新）"); + laneFrame("web-2#turn-b", "turn-b", 1, "标记B（更新）"); + + same( + lanesOf(MAIN_SESSION_ID).map((g) => g.turnId), + ["turn-a", "turn-b"], + "两条路各占一格（交错到达也绝不互相覆盖）", + ); + same( + lanesOf(MAIN_SESSION_ID).map((g) => g.nodes[0]?.label), + ["标记A（更新）", "标记B（更新）"], + "同一条路的后续帧就地更新自己那一格", + ); + same( + useAppStore.getState().flowGraphs[MAIN_SESSION_ID]?.id, + "web-2#turn-b", + "兼容字段（会话桶）仍是最近那一张 —— 老读法不受影响", + ); +} + +// ② 新一轮（round 更大）到达 ⇒ 上一轮的路退场；单路（无 turnId）也只占一格。 +{ + laneFrame("web-5", undefined, 2, "单路的下一轮"); + same( + lanesOf(MAIN_SESSION_ID).map((g) => g.turnId ?? ""), + [""], + "新一轮开始后只留本轮的路（上一轮的两条路已退场）", + ); + laneFrame("web-9", undefined, undefined, "旧后端的帧（无 round）"); + same( + lanesOf(MAIN_SESSION_ID).map((g) => g.id), + ["web-9"], + "无轮次号 / 无路标识的旧帧按空路键就地更新，绝不多开一格", + ); +} + +// ③ 路数有硬上限：绝不随帧无界增长。 +{ + for (let i = 0; i < MAX_LANES + 3; i += 1) { + laneFrame(`web-x${i}`, `turn-x${i}`, 3, `路${i}`); + } + same( + lanesOf(MAIN_SESSION_ID).length, + MAX_LANES, + "路桶有硬上限（不随并行路数无界增长）", + ); + same( + lanesOf(MAIN_SESSION_ID).map((g) => g.turnId), + ["turn-x3", "turn-x4", "turn-x5", "turn-x6"], + "超限时丢掉最旧的路", + ); +} + +// ④ Flow 页取数：≥2 路时 `flowPageData` 交出每一路；只有单张图时回落到那一条。 +{ + const sessions: Record = {}; + const page = flowPageData( + sessions, + useAppStore.getState().flowGraphs, + MAIN_SESSION_ID, + useAppStore.getState().flowLanes, + ); + same(page.sessionId, MAIN_SESSION_ID, "取数会话 = 选中会话"); + same(page.lanes.length, MAX_LANES, "页面拿到本轮每一路（供并列泳道渲染）"); + + const single: Record = { + [MAIN_SESSION_ID]: { id: "web-1", nodes: [], edges: [], sessionId: MAIN_SESSION_ID }, + }; + const fallback = flowPageData(sessions, single, MAIN_SESSION_ID, {}); + same( + fallback.lanes.map((g) => g.id), + ["web-1"], + "路桶为空时回落为那一张图（单路 / 旧后端 / 快照补拉前）", + ); +} + +// ⑤ 会话隔离：另一会话的路桶不会被本会话的帧写脏（裁决 16.4.2 的口径不变）。 +{ + dispatchFrame({ + type: "graph_update", + data: { + id: "sess-backthink-1", + sessionId: "sess-backthink", + turnId: "turn-bt", + round: 1, + nodes: [], + edges: [], + }, + }); + same( + lanesOf("sess-backthink").map((g) => g.turnId), + ["turn-bt"], + "别的会话的路进它自己的桶", + ); + ok( + lanesOf(MAIN_SESSION_ID).every((g) => g.turnId !== "turn-bt"), + "别的会话的路绝不混进本会话", + ); +} + +console.log(`flow lanes logic test: ${checks} checks passed`); diff --git a/frontend/web/src/store/sessionCache.ts b/frontend/web/src/store/sessionCache.ts index b03615f..3ed4623 100644 --- a/frontend/web/src/store/sessionCache.ts +++ b/frontend/web/src/store/sessionCache.ts @@ -11,7 +11,9 @@ import type { AppStore } from "./useAppStore"; * 桶、把切到的会话取出，于是「切会话不再清空」。 * ② **链路图桶**（裁决 16.4.2）：一个会话最近一张 `graph_update` 快照。管理后台按 * `sessionId` 取图，而不是所有页面共用一张全局图——后台思考的完整链路图与 - * 「用户侧前端发不发」无关，也绝不顶掉管理员看的「当前会话」图。 + * 「用户侧前端发不发」无关，也绝不顶掉管理员看的「当前会话」图。真并行时同一 + * 会话同一轮有**多条路**（切片 FV2），于是会话桶之外还有**路桶**：同一个会话的 + * 每条路各占一格（`putLane`），两路交错的帧因此不会互相覆盖。 * * 从 `useAppStore.ts` 拆出来（AGENTS §10 文件 ≤500 行 / 三次复制必须抽取）：下面每条 * 规则都被「激活会话」与「非激活会话」两条写入路径共用，放在一处才不会走样。 @@ -235,6 +237,51 @@ export function putGraph( 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, + lanes: Record, + sessionId: string, + graph: FlowGraph +): { flowGraphs: Record; flowLanes: Record } { + return { + flowGraphs: putGraph(graphs, sessionId, graph), + flowLanes: { ...lanes, [sessionId]: putLane(lanes[sessionId], graph) }, + }; +} + /** 兼容字段 `flowGraph` 的口径（管理端 Flow / Agents 页读它）：**激活会话**的图； * 激活会话还没有图时回落主会话的最近一张，使页面不至于空着。 * 后台思考会话的图不会顶掉它 —— 这正是裁决 16.4.2 要修的：每条思考看自己的图， diff --git a/frontend/web/src/store/useAppStore.ts b/frontend/web/src/store/useAppStore.ts index d4e5e37..61830c1 100644 --- a/frontend/web/src/store/useAppStore.ts +++ b/frontend/web/src/store/useAppStore.ts @@ -22,7 +22,7 @@ import { mergeMessage, patchAssistantTail, patchMessage, - putGraph, + writeFlowGraph, replacePage, withBucket, writeSession, @@ -75,6 +75,12 @@ export interface AppStore extends TurnSlice { * conversation the admin panel is looking at. Bounded by `MAX_GRAPHS` * (store/sessionCache.ts), most recently updated first. */ flowGraphs: Record; + /** Per-session **lane** graphs (切片 FV2): every 路 (leg) of the session's latest + * round, oldest first, keyed by the graph's own `turnId`. Real parallel turns + * publish one graph per leg; they used to overwrite each other in the single + * session bucket. `flowGraphs` keeps its «session → newest graph» meaning (compat), + * the Flow page reads this list to draw each leg of the round as its own lane. */ + flowLanes: 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`. */ @@ -212,6 +218,7 @@ export const useAppStore = create()( ], backthink: [], flowGraphs: {}, + flowLanes: {}, flowGraph: null, deviceUpdates: [], awakenState: "quiet", @@ -426,10 +433,14 @@ export const useAppStore = create()( setFlowGraph: (graph, sessionId) => set((s) => { // 归属会话：图自己带的 sessionId（后端 RunGraph 建图时写入）优先，其次帧壳的， - // 最后落到当前会话。每张图只进它自己的桶，互不覆盖（裁决 16.4.2）。 + // 最后落到当前会话。每张图只进它自己的桶，互不覆盖（裁决 16.4.2）；真并行时 + // 同一条会话的每一路再各占一格（切片 FV2，见 sessionCache.writeFlowGraph）。 const sid = graph.sessionId ?? sessionId ?? s.activeSessionId; - const flowGraphs = putGraph(s.flowGraphs, sid, graph); - return { flowGraphs, flowGraph: currentGraph(flowGraphs, s.activeSessionId) }; + const next = writeFlowGraph(s.flowGraphs, s.flowLanes, sid, graph); + return { + ...next, + flowGraph: currentGraph(next.flowGraphs, s.activeSessionId), + }; }), addDeviceUpdate: (text) => set((s) => ({ deviceUpdates: [text, ...s.deviceUpdates].slice(0, 50) })), @@ -468,6 +479,7 @@ export const useAppStore = create()( sessionMessages: {}, moods: {}, flowGraphs: {}, + flowLanes: {}, flowGraph: null, backthink: s.backthink, awakenState: "quiet", diff --git a/internal/runtime/flowgraph/flowgraph.go b/internal/runtime/flowgraph/flowgraph.go index e19c202..09ccd2b 100644 --- a/internal/runtime/flowgraph/flowgraph.go +++ b/internal/runtime/flowgraph/flowgraph.go @@ -12,7 +12,9 @@ // （`cmd/cyrene` 的 `flowSessionAllowed`，design/19 §1 裁决 01.4.5）决定 —— // 本包只被**已经过白名单的** `EndRound` 调用，因此不再存一份白名单。 // - **写入时机 = 每轮一条**（`EndRound`），**不是每 span 一条**：span 级写放大是纯浪费， -// 图在收尾时已完整（`output` 节点已并入）。 +// 图在收尾时已完整（`output` 节点已并入）。真并行时一轮有多条路（design/15 §2.5.5）， +// 因此实际是**每一路一条** —— 两路各有各的 `turn_id`，唯一键 `(session_id, turn_id)` +// 因此不会把其中一路顶上（FV2；design/15 §2.5.10-1）。 // - **不是图的真源**：实时图仍由发布器持有（`graph_update` 帧的唯一来源）；本包只是 // 一份**收尾存档**，供「打开页面后翻历史轮」。读写失败都**不许**影响会话收尾。 // diff --git a/internal/runtime/flowgraph/flowgraph_lane_test.go b/internal/runtime/flowgraph/flowgraph_lane_test.go new file mode 100644 index 0000000..7faa570 --- /dev/null +++ b/internal/runtime/flowgraph/flowgraph_lane_test.go @@ -0,0 +1,134 @@ +package flowgraph + +// 本文件钉住**真并行两路的落盘不撞 key**（design/15 §2.5.10-1 + design/19 §4 P3； +// 切片 FV2）。症状：两路在同一会话各收一轮，若两路的 `turn_id` 相同， +// `(session_id, turn_id)` 唯一键会让两路 upsert 成**一行** —— 后收尾的那一路把前一路 +// 整张图顶掉，翻历史轮时那一路的图彻底消失。 +// +// 断言的是**库里的事实**（几行、每行 id 是什么、读回来的图属于哪一路），不按日志断言， +// 与 flowgraph_test.go 同一口径。夹具用 `trace.RunGraph` 走真建图路径（两路各自 +// SetTurnID + SetSeq），因此库里的 turn_id 与线上图 id 逐字同源。 + +import ( + "context" + "database/sql" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// laneGraph 用真建图器造一条路的收尾图：seq 是会话级建图序号（两路必不同）， +// lane 是这一路的权威 turnId。 +func laneGraph(session string, seq int, lane, input, output string) *trace.FlowGraph { + g := trace.NewRunGraph(session) + g.SetTurnID(lane) + g.SetSeq(seq) + g.Start(seq, testNow().Add(-time.Minute)) + g.SeedInput(input) + g.Upsert(trace.NodeSpec{ID: "span-" + lane, Label: lane, Kind: "tool", Status: trace.FlowStatusDone, ParentID: "input"}) + g.SeedOutput(output) + return g.Snapshot() +} + +// TestStore_ParallelLanesPersistSeparateRows_正常 是本文件的核心断言：同一会话同一轮 +// 的两路各占**一行**，按各自的路标识查得回来，`Turns` 列出两条，任一路都没有被 +// 另一路顶掉。 +func TestStore_ParallelLanesPersistSeparateRows_正常(t *testing.T) { + t.Parallel() + st, conn := newTestStore(t, nil) + ctx := context.Background() + + laneA := laneGraph("web", 1, "turn-a", "标记A", "答复A") + laneB := laneGraph("web", 2, "turn-b", "标记B", "答复B") + if laneA.ID == laneB.ID { + t.Fatalf("夹具自身就撞了 id：%q", laneA.ID) + } + for _, g := range []*trace.FlowGraph{laneA, laneB} { + if err := st.RecordRound(g); err != nil { + t.Fatalf("落盘 %s: %v", g.ID, err) + } + } + + if got := countRows(t, conn, "web"); got != 2 { + t.Fatalf("flow_graphs 里该会话的行数 = %d，want 2（两路各一行）", got) + } + for _, want := range []struct { + graph *trace.FlowGraph + input string + output string + }{{laneA, "标记A", "答复A"}, {laneB, "标记B", "答复B"}} { + got, err := st.Graph(ctx, "web", want.graph.ID) + if err != nil { + t.Fatalf("查 %s: %v", want.graph.ID, err) + } + if got == nil { + t.Fatalf("查不到 %s —— 这一路被另一路顶掉了", want.graph.ID) + } + if got.TurnID != want.graph.TurnID || got.Round != want.graph.Round { + t.Fatalf("读回 %s 的路/轮次 = %q/%d，want %q/%d", + want.graph.ID, got.TurnID, got.Round, want.graph.TurnID, want.graph.Round) + } + if !hasNodeLabel(got, want.input) || !hasNodeLabel(got, want.output) { + t.Fatalf("读回 %s 的图不是这一路的（输入/回复节点对不上）：%+v", want.graph.ID, got.Nodes) + } + } + + turns, err := st.Turns(ctx, "web", 0) + if err != nil { + t.Fatalf("Turns: %v", err) + } + if len(turns) != 2 { + t.Fatalf("Turns 条数 = %d，want 2（两路都能翻到）", len(turns)) + } + if latest, err := st.Latest(ctx, "web"); err != nil || latest == nil { + t.Fatalf("Latest = %+v / %v，want 一条真图", latest, err) + } +} + +// TestStore_SameLaneTwiceIsOneRow_边界 钉住唯一键的另一半语义：**同一路**二次收尾 +// 是同轮覆盖（一行），这正是分桶改造**不该改动**的既有行为 —— 两路分行靠的是两路 +// 各有各的 id，而不是靠放弃 upsert。 +func TestStore_SameLaneTwiceIsOneRow_边界(t *testing.T) { + t.Parallel() + st, conn := newTestStore(t, nil) + first := laneGraph("web", 7, "turn-x", "标记A", "答复A") + if err := st.RecordRound(first); err != nil { + t.Fatalf("首次落盘: %v", err) + } + again := laneGraph("web", 7, "turn-x", "标记A", "答复A（重放）") + if err := st.RecordRound(again); err != nil { + t.Fatalf("二次落盘: %v", err) + } + if got := countRows(t, conn, "web"); got != 1 { + t.Fatalf("同一路二次收尾后行数 = %d，want 1（同轮覆盖）", got) + } + back, err := st.Graph(context.Background(), "web", first.ID) + if err != nil || back == nil { + t.Fatalf("Graph = %+v / %v", back, err) + } + if !hasNodeLabel(back, "答复A（重放）") { + t.Fatalf("覆盖后的图不是最新那份：%+v", back.Nodes) + } +} + +// countRows 数该会话在 flow_graphs 里的行数（结构级断言，不经过 store 的读方法）。 +func countRows(t *testing.T, conn *sql.DB, session string) int { + t.Helper() + var n int + if err := conn.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM flow_graphs WHERE session_id = ?`, session).Scan(&n); err != nil { + t.Fatalf("COUNT 行数: %v", err) + } + return n +} + +// hasNodeLabel 报告图里有没有标签等于 label 的节点。 +func hasNodeLabel(g *trace.FlowGraph, label string) bool { + for _, n := range g.Nodes { + if n.Label == label { + return true + } + } + return false +} diff --git a/internal/runtime/loop/flow.go b/internal/runtime/loop/flow.go index 4ca660e..dc9adc5 100644 --- a/internal/runtime/loop/flow.go +++ b/internal/runtime/loop/flow.go @@ -2,6 +2,7 @@ package loop import ( "log/slog" + "sort" "sync" "time" @@ -21,6 +22,9 @@ const flowLogSource = "runtime/loop/flow" // allowedSessions 是用于控制哪些会话可发布 // flow 图的谓词签名（主人的 web 会话），使 IM/陌生人/子代理轮次绝不 // 泄露进管理面板。谓词为 nil 时允许所有会话。 +// +// 谓词收的是**会话 id**（不是归属桶）：调用方先经 splitFlowOwner 拆掉路标识， +// 因此白名单永远只按会话判 —— 路的存在绝不放宽或收紧任何一档。 type allowedSessions func(sessionID string) bool // RoundRecorder 把一条**已收尾轮次**的图落盘（design/19 §4 的 P3「多轮历史回放」； @@ -30,6 +34,10 @@ type allowedSessions func(sessionID string) bool // 上调用），失败**上抛**给发布器（发布器记 WARN 后继续）—— 图是可视化资产，落盘失败 // 绝不允许影响会话收尾，也绝不改变实时帧。 // +// 「每轮一条」在真并行下的含义是**每一路一条**：两路各有自己的图 id（见 +// trace.RunGraph 的 id 口径），因此 `flow_graphs` 的 `(session_id, turn_id)` 唯一键 +// 不会被两路撞上（design/15 §2.5.10-1）。 +// // 白名单不在实现方：EndRound 对发布会话范围外的会话**直接返回**，因此落盘出口天然 // 只被白名单内的会话触发（复用 flowSessionAllowed，不存在第二份白名单）。 type RoundRecorder interface { @@ -50,22 +58,39 @@ func WithFlowLogger(log *slog.Logger) FlowPublisherOption { return func(p *FlowPublisher) { p.log = log } } -// FlowPublisher 从循环的 trace span 累积每会话 flow 图， +// FlowPublisher 从循环的 trace span 累积**按路**的 flow 图， // 并在每个 span 开始/结束时发布 ws.graph_update，给前端 // 一张实时的 input→think→tool→output 图（design/19 §1）。它容忍 nil -// （nil 发布者为空操作）且并发安全；会话图 +// （nil 发布者为空操作）且并发安全；同一会话的图 // 在同一把互斥锁下串行化，因为并行的工具 span 会并发完成。 // -// 归属：一个会话拥有一个实时轮次。声明自己是另一会话分支的 -// Run（loop.WithFlowBranch）绝不调用 StartRound/EndRound， -// 而是把其 span 发布到所属会话下，因此子代理渲染为 +// 归属：**一个 Run 拥有一个路 bucket**。单路时 bucket 就是会话 id（行为与引入 +// 「路」之前逐字相同）；真并行时每一路各有自己的 bucket（`{会话}#{路}`，见 +// FlowOwnerKey），因此两路的节点绝不互相长在对方的图上、也绝不互相重置 +// （design/15 §2.5.10-1 登记的那条「graph_update 互相顶掉」）。声明自己是另一 +// 会话分支的 Run（loop.WithFlowBranch）绝不调用 StartRound/EndRound， +// 而是把其 span 发布到所属路的 bucket 下，因此子代理渲染为 // 父图的一条泳道，而不是第二张图（design/19 §2）。 type FlowPublisher struct { mu sync.Mutex publish func(subject string, payload any) allow allowedSessions - graphs map[string]*trace.RunGraph - rounds map[string]int + // graphs 是「归属桶 → 该桶当前这张图」。键见 FlowOwnerKey。 + graphs map[string]*trace.RunGraph + // seq 是**每会话**的建图序号：每次 StartRound 加一，并写进图 id + // （`{会话}-{序号}[#{路}]`）。用序号而不是轮次号做 id 的第二个分量，是为了让 + // 「同一轮里的第二条路」也拿到一个全新的 id —— 否则两路会撞同一个 + // `flow_graphs.turn_id`（唯一键），后收尾的那一路把前一路整行顶掉。 + seq map[string]int + // rounds 是每会话当前的**轮次号**（= 该轮第一条路的建图序号）。同一轮的各条路 + // 共享它，客户端据此只保留最新一轮的泳道。 + rounds map[string]int + // roundOf 是「归属桶 → 它属于哪一轮」。 + roundOf map[string]int + // inflight 是每会话**在飞的桶**集合。新一轮的判据是「上一轮的路全部收尾」 + // （集合为空），因此它是轮次号推进的唯一依据，也是内存不随路数无界增长的 + // 依据（新一轮起时丢掉更早轮次的桶）。 + inflight map[string]map[string]bool // record 是每轮收尾的落盘出口（design/19 §4 P3；切片 F2）。nil = 不落盘 // （保持 F1 之前的纯内存行为，且测试可只测发布）。 record RoundRecorder @@ -81,10 +106,13 @@ type FlowPublisher struct { // 可变项经 Functional Options 传入（不改变既有两参调用）：落盘出口（F2）与日志。 func NewFlowPublisher(publish func(subject string, payload any), allow func(sessionID string) bool, opts ...FlowPublisherOption) *FlowPublisher { p := &FlowPublisher{ - publish: publish, - allow: allow, - graphs: make(map[string]*trace.RunGraph), - rounds: make(map[string]int), + publish: publish, + allow: allow, + graphs: make(map[string]*trace.RunGraph), + seq: make(map[string]int), + rounds: make(map[string]int), + roundOf: make(map[string]int), + inflight: make(map[string]map[string]bool), } for _, opt := range opts { if opt != nil { @@ -94,40 +122,83 @@ func NewFlowPublisher(publish func(subject string, payload any), allow func(sess return p } -// StartRound 为 sessionID 开始一张新图，用用户文本 -// 播种一个 input 节点并发布初始帧。它重置该会话的上一 -// 轮次，因此每个用户 turn 都是自己的一张图。 -func (p *FlowPublisher) StartRound(sessionID, inputText string) { - if sessionID == "" || !p.allowed(sessionID) { +// StartRound 为 owner 开始一张新图，用用户文本 +// 播种一个 input 节点并发布初始帧。 +// +// owner 是**归属桶**（`{会话}[#{路}]`，见 FlowOwnerKey）：单路调用方传的就是会话 +// id（引入本切片之前的所有调用点原样可用），真并行路由经 flowOwner 传 +// `{会话}#{路}`。两条路各起各的图 ⇒ 后起的那一路绝不重置前一路的图。 +// +// 轮次号只在**上一轮的路全部收尾**之后推进（inflight 为空），因此「同一条消息 +// 引出的两路」共享同一个轮次号；新一轮开始时丢掉更早轮次的桶，内存因此不随路数 +// 无界增长（图的历史在落盘里，见 flowgraph 包）。 +func (p *FlowPublisher) StartRound(owner, inputText string) { + session, lane := splitFlowOwner(owner) + if session == "" || owner == "" || !p.allowed(session) { return } p.mu.Lock() defer p.mu.Unlock() - round := p.rounds[sessionID] + 1 - g := trace.NewRunGraph(sessionID) + round := p.beginRoundLocked(session, owner) + g := trace.NewRunGraph(session) + g.SetTurnID(lane) + g.SetSeq(p.seq[session]) g.Start(round, time.Now()) g.SeedInput(inputText) - p.graphs[sessionID] = g - p.rounds[sessionID] = round + p.graphs[owner] = g + p.roundOf[owner] = round + if p.inflight[session] == nil { + p.inflight[session] = make(map[string]bool, 2) + } + p.inflight[session][owner] = true p.publishLocked(g) } +// beginRoundLocked 分配（或沿用）本轮次号并清掉更早轮次的桶。调用方必须持有 p.mu。 +func (p *FlowPublisher) beginRoundLocked(session, owner string) int { + p.seq[session]++ + if len(p.inflight[session]) == 0 { + p.rounds[session] = p.seq[session] + p.dropEarlierRoundsLocked(session) + } + return p.rounds[session] +} + +// dropEarlierRoundsLocked 丢掉该会话**更早轮次**留下的桶与图。图的历史由落盘承载 +// （`flow_graphs`），实时桶只需覆盖当前这一轮 —— 否则每一条并行路（turnId 每回合 +// 全新、bucket 键因此永不复用）都会在内存里永久留一张图。调用方必须持有 p.mu。 +func (p *FlowPublisher) dropEarlierRoundsLocked(session string) { + current := p.rounds[session] + for owner, round := range p.roundOf { + if round >= current { + continue + } + if s, _ := splitFlowOwner(owner); s != session { + continue + } + delete(p.roundOf, owner) + delete(p.graphs, owner) + } +} + // EndRound 用回复播种一个 output 节点并发布最终帧， -// 以此收尾该轮次。它保留该图，使错过最终帧的客户端能重放最后一份快照 —— -// **重放的出口是 `LastGraph`**（网关只读查询，契约 v0.71），而不是又发一帧： -// 实时帧向来只推给当时在线的连接，刷新/晚开的页面只能靠只读拉取补齐 -// （`graph_update` 本身不重放）。 +// 以此收尾该路的图。它保留该图，使错过最终帧的客户端能重放最后一份快照 —— +// **重放的出口是 `LastGraph` / `LatestLanes`**（网关只读查询，契约 v0.71/v0.77）， +// 而不是又发一帧：实时帧向来只推给当时在线的连接，刷新/晚开的页面只能靠只读拉取 +// 补齐（`graph_update` 本身不重放）。 // -// 收尾也是**落盘**的唯一时机（切片 F2；design/19 §4 的 P3「多轮历史回放」）：每轮落 -// **一条**（绝不是每 span 一条），且只在该会话处于发布会话白名单内时发生（本方法对 -// 白名单外的会话直接返回 ⇒ 落盘出口天然只被白名单会话触发）。深拷贝与写库在**放锁 -// 之后**做：一次库写绝不把其他会话的 span 发布挡住；失败只记 WARN，绝不影响本轮收尾。 -func (p *FlowPublisher) EndRound(sessionID, outputText string) { - if sessionID == "" || !p.allowed(sessionID) { +// 收尾也是**落盘**的唯一时机（切片 F2；design/19 §4 的 P3「多轮历史回放」）：每轮 +// 落**一条**（绝不是每 span 一条），真并行下**每一路各一条**；且只在该会话处于发布 +// 会话白名单内时发生（本方法对白名单外的会话直接返回 ⇒ 落盘出口天然只被白名单会话 +// 触发）。深拷贝与写库在**放锁之后**做：一次库写绝不把其他会话的 span 发布挡住； +// 失败只记 WARN，绝不影响本轮收尾。 +func (p *FlowPublisher) EndRound(owner, outputText string) { + session, _ := splitFlowOwner(owner) + if session == "" || owner == "" || !p.allowed(session) { return } p.mu.Lock() - g := p.graphs[sessionID] + g := p.graphs[owner] if g == nil { p.mu.Unlock() return @@ -138,29 +209,33 @@ func (p *FlowPublisher) EndRound(sessionID, outputText string) { if p.record != nil { snap = g.Snapshot().Clone() } + // 本路退出在飞集合（**只删自己的桶**）：集合空了才说明整轮收尾，下一轮才会 + // 拿到新的轮次号 —— 因此轮次号在同轮并行的两路之间保持同一个值。 + delete(p.inflight[session], owner) p.mu.Unlock() if snap != nil { if err := p.record.RecordRound(snap); err != nil { - p.logPersistError(sessionID, snap.ID, err) + p.logPersistError(session, snap.ID, err) } } } -// MarkInputSteered 给本轮的 input 节点打上 `steered:true` 标记并发布一帧 -// （design/15 §5 / 裁决 13.4.5「注入时给 input 节点打 steered:true 标记」）。 -// 它让链路图如实表达「这一轮的输入在回合中途被追加过」——图仍然只有一条主干 +// MarkInputSteered 给本路的 input 节点打上 `steered:true` 标记并发布一帧 +// （design/15 §5 / 裁决 13.4.5「注入时给 input 节点贴 steered:true 标记」）。 +// 它让链路图如实表达「这一路的输入在回合中途被追加过」——图仍然只有一条主干 // （I1：同会话恒只有一个在飞回合），而不是被画成第二轮。 // -// nil 接收者、空会话、发布白名单外的会话、或该会话还没有图时都是空操作；标记 +// nil 接收者、空归属、发布白名单外的会话、或该路还没有图时都是空操作；标记 // 只并入既有 input 节点（Upsert 的合并语义），绝不新建节点、绝不改动它的标签。 -func (p *FlowPublisher) MarkInputSteered(sessionID string) { - if p == nil || sessionID == "" || !p.allowed(sessionID) { +func (p *FlowPublisher) MarkInputSteered(owner string) { + session, _ := splitFlowOwner(owner) + if p == nil || owner == "" || session == "" || !p.allowed(session) { return } p.mu.Lock() defer p.mu.Unlock() - g := p.graphs[sessionID] + g := p.graphs[owner] if g == nil { return } @@ -174,13 +249,13 @@ func (p *FlowPublisher) MarkInputSteered(sessionID string) { // SpanStarted 为一个刚开始的 span 记录一个运行中的占位 // 节点。state_transition span 会被图构建器省略。 -func (p *FlowPublisher) SpanStarted(sessionID string, sp *trace.Span) { - if sp == nil || sessionID == "" || !p.allowed(sessionID) { +func (p *FlowPublisher) SpanStarted(owner string, sp *trace.Span) { + if sp == nil { return } p.mu.Lock() defer p.mu.Unlock() - g := p.graphs[sessionID] + g := p.bucketLocked(owner) if g == nil { return } @@ -190,14 +265,14 @@ func (p *FlowPublisher) SpanStarted(sessionID string, sp *trace.Span) { } // SpanFinished 把一个刚结束的 span 作为最终节点（状态 done/error） -// 并入会话图并发布。必须在 span 的 Finish 之后调用。 -func (p *FlowPublisher) SpanFinished(sessionID string, sp *trace.Span) { - if sp == nil || sessionID == "" || !p.allowed(sessionID) { +// 并入该路的图并发布。必须在 span 的 Finish 之后调用。 +func (p *FlowPublisher) SpanFinished(owner string, sp *trace.Span) { + if sp == nil { return } p.mu.Lock() defer p.mu.Unlock() - g := p.graphs[sessionID] + g := p.bucketLocked(owner) if g == nil { return } @@ -205,16 +280,30 @@ func (p *FlowPublisher) SpanFinished(sessionID string, sp *trace.Span) { p.publishLocked(g) } -// LastGraph 返回 sessionID **最近一轮**链路图的一份只读深拷贝快照，供 +// bucketLocked 取归属桶当前那张图；归属非法或该路还没有图时返回 nil。调用方必须 +// 持有 p.mu。白名单判据与建图处同一处（bucket 拆出的会话），因此**不可能**让 +// 白名单外的会话经 span 折叠进图。 +func (p *FlowPublisher) bucketLocked(owner string) *trace.RunGraph { + session, _ := splitFlowOwner(owner) + if owner == "" || session == "" || !p.allowed(session) { + return nil + } + return p.graphs[owner] +} + +// LastGraph 返回 sessionID **最近一路**链路图的一份只读深拷贝快照，供 // 管理面板在「轮次已经结束、实时帧已经错过」时补拉（契约 v0.71 的 // `GET /api/v1/admin/flow/graph`；design/19 §4 的 P3「多轮历史回放」的 // 最小一步）。 // +// 「最近」= 该会话所有在内存的路里**开始最晚**的那一路（同刻用图 id 定序）。 +// 真并行时它只是最近的一路，**全貌**是 `LatestLanes`（契约 v0.77 的 `lanes[]`）。 +// // 它是**只读**的：不发布、不改图、不动轮次计数 —— 因此补拉绝不会在 // 管理面板里凭空造出一帧 `graph_update`，也绝不改任何会话的实时图。 // // 该会话还没有图时返回 nil：从未跑过带 span 的轮次、或该会话本就**不在 -// 发布会话范围内**（`StartRound` 对白名单外的会话直接返回 ⇒ 从不建图， +// 发布会话范围**（`StartRound` 对白名单外的会话直接返回 ⇒ 从不建图， // 隔离渠道会话因此天然查不到）。调用方**必须**把 nil 读作「确实没有图」， // 绝不允许据此补画任何节点或假图。 // @@ -226,11 +315,81 @@ func (p *FlowPublisher) LastGraph(sessionID string) *trace.FlowGraph { } p.mu.Lock() defer p.mu.Unlock() - g := p.graphs[sessionID] - if g == nil { + latest := p.latestLocked(sessionID) + if latest == nil { return nil } - return g.Snapshot().Clone() + return latest.Clone() +} + +// LatestLanes 返回 sessionID **最近一轮每一路**的图快照（每条路一份深拷贝）， +// 按路的开始时间升序（最早的在前）。契约 v0.77 的 `lanes[]` 就是它：真并行时 +// 同一会话同一轮有多条路，`LastGraph` 只给出其中最新的一条，管理面板要靠本方法 +// 才看得全（design/19 §2「并行分支独立显示」）。 +// +// 没有图时返回空切片（不是 nil）：调用方据此渲染诚实空态，绝不补画。 +// 与 LastGraph 一样是**只读**的，且返回深拷贝（读者在锁外编码）。 +func (p *FlowPublisher) LatestLanes(sessionID string) []*trace.FlowGraph { + if p == nil || sessionID == "" { + return nil + } + p.mu.Lock() + defer p.mu.Unlock() + round := p.rounds[sessionID] + owners := make([]string, 0, 2) + for owner, r := range p.roundOf { + if r != round { + continue + } + if s, _ := splitFlowOwner(owner); s == sessionID { + owners = append(owners, owner) + } + } + out := make([]*trace.FlowGraph, 0, len(owners)) + for _, owner := range owners { + g := p.graphs[owner] + if g == nil { + continue + } + out = append(out, g.Snapshot().Clone()) + } + sortLanes(out) + return out +} + +// latestLocked 取该会话开始最晚的那一路的活图（没有时 nil）。调用方必须持有 p.mu。 +func (p *FlowPublisher) latestLocked(sessionID string) *trace.FlowGraph { + var best *trace.FlowGraph + for owner, g := range p.graphs { + if s, _ := splitFlowOwner(owner); s != sessionID { + continue + } + snap := g.Snapshot() + if best == nil || laneNewer(snap, best) { + best = snap + } + } + return best +} + +// laneNewer 报告 a 是否比 b 更「新」：先比开始时刻，同刻用图 id 定序（图 id 里含 +// 会话级单调序号，因此定序不依赖时间分辨率）。 +func laneNewer(a, b *trace.FlowGraph) bool { + if a.StartedAt.Equal(b.StartedAt) { + return a.ID > b.ID + } + return a.StartedAt.After(b.StartedAt) +} + +// sortLanes 按「开始时刻升序、同刻按图 id」把一路路图排序，使一个轮次里的多条路 +// 有一个确定的展示次序（绝不让展示次序随 map 遍历序抖动）。 +func sortLanes(lanes []*trace.FlowGraph) { + sort.SliceStable(lanes, func(i, j int) bool { + if lanes[i].StartedAt.Equal(lanes[j].StartedAt) { + return lanes[i].ID < lanes[j].ID + } + return lanes[i].StartedAt.Before(lanes[j].StartedAt) + }) } // publishLocked 发布 g 的一份快照。调用方必须持有 p.mu。 diff --git a/internal/runtime/loop/flow_branch.go b/internal/runtime/loop/flow_branch.go index 44b416d..5df4f93 100644 --- a/internal/runtime/loop/flow_branch.go +++ b/internal/runtime/loop/flow_branch.go @@ -1,6 +1,40 @@ package loop -import "context" +import ( + "context" + "strings" +) + +// flowOwnerSep 是 flow 图**归属桶**（`{会话}#{路}`）的分隔符。它与 +// design/15 §2.5.9 的 checkpoint 归属键（`{sessionID}#{turnID}`）以及 +// trace 的图 id 拼接口径（`{会话}-{轮次}#{路}`）取同一个字符：**一条路** +// 在流程图、checkpoint 与落盘行三处因此是同一个标识。 +// +// 之所以能在同一个字符串里编码「会话 + 路」：会话 id 的取值族是 +// `web` / `web:` / `sess-backthink`（`cmd/cyrene` 的 flowSessionAllowed）， +// 权威 turnId 是 `turn-`（`gateway.newTurnID`），两者都不含 `#`。 +const flowOwnerSep = "#" + +// FlowOwnerKey 拼出「会话[#路]」的 flow 归属桶；lane 为空时就是会话本身 +// （单路 / 未声明并行的运行，逐字等于引入本特性之前的会话 id）。 +// +// 导出它是因为**归属桶的串形状只有一个权威**：发布器的建图出口、子代理 span 的 +// 归属、以及跨包的验收测试都按它拼串，任何一处自己拼 `#` 都会在换分隔符时静默走偏。 +func FlowOwnerKey(sessionID, lane string) string { + if lane == "" { + return sessionID + } + return sessionID + flowOwnerSep + lane +} + +// splitFlowOwner 把归属桶拆回（会话, 路）。没有 `#` 时整串就是会话、路为空 —— +// 这正是单路运行与**旧调用方**传给发布器的形态，因此拆分是向后兼容的。 +func splitFlowOwner(owner string) (string, string) { + if i := strings.Index(owner, flowOwnerSep); i >= 0 { + return owner[:i], owner[i+1:] + } + return owner, "" +} // flowBranchKey 是标记某个 Run 为另一会话 flow 图分支的 context key。 // 它由 orchestration 层（经 WithFlowBranch）在驱动 @@ -8,24 +42,28 @@ import "context" type flowBranchKey struct{} // WithFlowBranch 返回 ctx，把当前 Run 标记为分支：它的 span 并入 -// parentSession 的实时 flow 图，且该 Run 自身既不开始也不结束 +// parent 的实时 flow 图，且该 Run 自身既不开始也不结束 // 轮次 —— 由所属会话的轮次负责。orchestration 层正是靠它把 // 子代理的工作归属到派生它的会话，使并行的 // 子代理渲染为同一张图的泳道，而不是用各自的图 // 替换前端唯一的 flowGraph（design/19 §2）。 // -// 空的 parentSession 仍把该 Run 标记为分支：它不拥有轮次、 +// parent 是**归属桶**（`{会话}#{路}`，见 FlowOwnerKey）：并行路里派发的子代理必须 +// 并进**那一路**的图，而不是会话主干那张 —— 否则两路的子代理节点会混进同一张图， +// 正是本切片要治的「两路互相覆盖」（design/15 §2.5.10-1）。单路时它逐字等于会话 id。 +// +// 空的 parent 仍把该 Run 标记为分支：它不拥有轮次、 // 不发布任何东西。这正是无人归属到会话的后台子代理的 // 状态，它不得悄悄变成属于自己的一张图。nil ctx // 原样返回。 -func WithFlowBranch(ctx context.Context, parentSession string) context.Context { +func WithFlowBranch(ctx context.Context, parent string) context.Context { if ctx == nil { return ctx } - return context.WithValue(ctx, flowBranchKey{}, parentSession) + return context.WithValue(ctx, flowBranchKey{}, parent) } -// flowBranch 报告 WithFlowBranch 记录的父会话以及 +// flowBranch 报告 WithFlowBranch 记录的父归属桶以及 // 该 Run 是否本就是分支。非分支的 Run 拥有自己的轮次并在 // 自己的 session id 下发布。 func flowBranch(ctx context.Context) (string, bool) { @@ -40,27 +78,55 @@ func flowBranch(ctx context.Context) (string, bool) { return s, true } -// flowOwner 返回应当接收该 Run span 的 flow 图所属会话： +// flowOwner 返回应当接收该 Run span 的 flow 归属桶： // 当该 Run 是子代理分支时取分支父级（分支未归属时为空， -// 此时抑制发布），否则取该 Run 自身的 sessionID。 +// 此时抑制发布），否则取该 Run 自身的 session 加上它所属的**路** +// （真并行路的标识由网关经 WithParallelLeg 盖上；单路时为会话本身）。 func flowOwner(ctx context.Context, sessionID string) string { if parent, isBranch := flowBranch(ctx); isBranch { return parent } - return sessionID + return FlowOwnerKey(sessionID, ParallelLegFrom(ctx)) } -// FlowOwnerFromContext 返回拥有从 ctx 启动的工作的 flow 图所属会话： -// 当 ctx 运行于子代理分支内时取分支父级，否则 -// 取外层 Run 被驱动的 session id（两者都未设置时为空）。 +// FlowOwnerFromContext 返回拥有从 ctx 启动的工作的 flow 图**归属桶** +// （`{会话}#{路}`，见 FlowOwnerKey）：当 ctx 运行于子代理分支内时取分支父级， +// 否则取外层 Run 被驱动的 session id 加上它所属的路（两者都未设置时为空）。 // -// 派发后台工作的工具 —— 子代理派生工具 —— 把它 -// 作为父会话传入，使在另一个子代理内部派生的子代理 -// 保持在同一个顶层图的归属之下，而不是归属到一个不拥有 -// 任何图的 id。 +// 派发后台工作的工具 —— 子代理派生工具 —— 把它作为父归属桶传入，使在另一个 +// 子代理内部派生的子代理保持在同一个顶层图的归属之下，而不是归属到一个不拥有 +// 任何图的 id；真并行路里派发的子代理也因此落进**那一路**的图。 +// +// ⚠️ 它返回的是**归属桶**而不是会话 id。要把子代理的归属会话交给总线载荷 +// （`subagent_result` 的 sessionId 必须是真会话 id）时必须用 +// FlowSessionFromContext 取会话那一半。 func FlowOwnerFromContext(ctx context.Context) string { - if parent, isBranch := flowBranch(ctx); isBranch { - return parent + if owner, isBranch := flowBranch(ctx); isBranch { + return owner } - return SessionFromContext(ctx) + session := SessionFromContext(ctx) + if session == "" { + // 没有会话就没有归属：绝不回一个 `#路` 形态的串（那既查不到图，也会让 + // 调用方误以为有归属）。分支与子代理因此保持「不发布」的既有语义。 + return "" + } + return FlowOwnerKey(session, FlowLaneFromContext(ctx)) +} + +// FlowSessionFromContext 返回从 ctx 启动的工作的**归属会话 id**（不含路）： +// 真并行时同一会话的每一路都是同一个会话 id，总线载荷与日志要的就是它。 +func FlowSessionFromContext(ctx context.Context) string { + session, _ := splitFlowOwner(FlowOwnerFromContext(ctx)) + return session +} + +// FlowLaneFromContext 返回从 ctx 启动的工作所属的**路**标识（design/15 §2.5.9 +// 的 turnId）；单路 / 未声明并行时为空。分支运行取分支归属桶里的那一半（子代理 +// 因此继承派生它的那一路），非分支运行取本 Run 的并发路。 +func FlowLaneFromContext(ctx context.Context) string { + if owner, isBranch := flowBranch(ctx); isBranch { + _, lane := splitFlowOwner(owner) + return lane + } + return ParallelLegFrom(ctx) } diff --git a/internal/runtime/loop/flow_lane_test.go b/internal/runtime/loop/flow_lane_test.go new file mode 100644 index 0000000..c2ab93d --- /dev/null +++ b/internal/runtime/loop/flow_lane_test.go @@ -0,0 +1,232 @@ +package loop + +import ( + "context" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// 本文件钉住**按路分桶**这条改造的其余口径（核心行为已在 flow_parallel_test.go 用真 +// 循环跑过）：桶 / 图 id / 帧上 `turnId` 键三者同源，最近一轮的每一路都拿得回来， +// 且新一轮开始后不再把上一轮的路留在内存里。这些用例走发布器的公开面（真调用点 +// 传给它的就是这些字符串），不依赖任何数据库。 + +// laneIDs 返回每一路图的 id 与 turnId 键，供断言用。 +func laneIDs(lanes []*trace.FlowGraph) []string { + out := make([]string, 0, len(lanes)) + for _, g := range lanes { + out = append(out, g.ID+"@"+g.TurnID) + } + return out +} + +// TestFlowPublisher_ParallelLanesGetOwnGraphs 钉住真并行两路各有各的桶： +// 图 id 不同（落盘不撞 key）、都归属同一个会话、都带上自己的路标识（帧上 `turnId`）， +// 且最近一轮的两路都能经 LatestLanes 拿回来。 +func TestFlowPublisher_ParallelLanesGetOwnGraphs(t *testing.T) { + p := NewFlowPublisher(nil, func(string) bool { return true }) + p.StartRound(FlowOwnerKey("web", "turn-a"), "标记A") + p.SpanStarted(FlowOwnerKey("web", "turn-a"), spanNamed("span-a", "tool.probe_a")) + p.SpanFinished(FlowOwnerKey("web", "turn-a"), finishedSpan("span-a", "tool.probe_a")) + + p.StartRound(FlowOwnerKey("web", "turn-b"), "标记B") + p.SpanStarted(FlowOwnerKey("web", "turn-b"), spanNamed("span-b", "tool.probe_b")) + p.SpanFinished(FlowOwnerKey("web", "turn-b"), finishedSpan("span-b", "tool.probe_b")) + + ga := p.LastGraph("web") + if ga == nil { + t.Fatal("LastGraph = nil，但该会话刚起了两路") + } + // 最近一路 = 后起的那一路；它的图里只有自己的工具。 + if ga.TurnID != "turn-b" || !hasNode(ga, "probe_b") || hasNode(ga, "probe_a") { + t.Fatalf("LastGraph 不是 B 那一路的图：id=%s turnId=%q nodes=%s", ga.ID, ga.TurnID, describeNodes(ga)) + } + + lanes := p.LatestLanes("web") + if len(lanes) != 2 { + t.Fatalf("最近一轮的路数 = %d，want 2（两路各自一张图）：%v", len(lanes), laneIDs(lanes)) + } + if got := laneIDs(lanes); got[0] != "web-1#turn-a@turn-a" || got[1] != "web-2#turn-b@turn-b" { + t.Fatalf("两路的 id@turnId = %v，want [web-1#turn-a@turn-a web-2#turn-b@turn-b]", got) + } + if lanes[0].ID == lanes[1].ID { + t.Fatalf("两路撞同一个图 id %q（落盘会互相顶掉）", lanes[0].ID) + } + for _, g := range lanes { + if g.SessionID != "web" { + t.Fatalf("图 %s 的 sessionId = %q，want web（路标识绝不改变会话归属）", g.ID, g.SessionID) + } + if g.Round != lanes[0].Round { + t.Fatalf("同一轮的两路轮次号不同：%d vs %d", g.Round, lanes[0].Round) + } + } + // 收尾后仍是最近一轮的每一路（页面在这之后才打开也看得到两条）。 + p.EndRound(FlowOwnerKey("web", "turn-a"), "答复A") + p.EndRound(FlowOwnerKey("web", "turn-b"), "答复B") + if got := p.LatestLanes("web"); len(got) != 2 { + t.Fatalf("收轮后最近一轮的路数 = %d，want 2", len(got)) + } +} + +// TestFlowPublisher_NewRoundDropsEarlierLanes 钉住轮次推进与内存边界：新一轮只在 +// 上一轮的路全部收尾之后开始（该轮的路因此共享轮次号），并丢掉更早轮次的桶 —— +// 否则每一条真并行的路（turnId 每回合全新、桶键永不复用）都会在内存里永久留一张图。 +func TestFlowPublisher_NewRoundDropsEarlierLanes(t *testing.T) { + p := NewFlowPublisher(nil, func(string) bool { return true }) + p.StartRound(FlowOwnerKey("web", "turn-a"), "标记A") + p.StartRound(FlowOwnerKey("web", "turn-b"), "标记B") + round := p.LatestLanes("web")[0].Round + p.EndRound(FlowOwnerKey("web", "turn-a"), "答复A") + p.EndRound(FlowOwnerKey("web", "turn-b"), "答复B") + + // 上一轮尚未被新轮次顶掉时，两条路都还在。 + if got := p.LatestLanes("web"); len(got) != 2 { + t.Fatalf("上一轮的路数 = %d，want 2（收轮后仍可补拉）", len(got)) + } + + p.StartRound("web", "单路的下一轮") + next := p.LatestLanes("web") + if len(next) != 1 || next[0].TurnID != "" { + t.Fatalf("新一轮的路 = %v，want 恰好一条无路标识的图（上一轮的桶必须被丢掉）", laneIDs(next)) + } + if next[0].Round <= round { + t.Fatalf("新一轮轮次号 = %d，want > %d", next[0].Round, round) + } + if g := p.LastGraph("web"); g == nil || g.TurnID != "" { + t.Fatalf("新一轮的 LastGraph = %+v，want 单路那张（上一轮的路已不在内存）", g) + } +} + +// TestFlowPublisher_RoundsShareRoundNumberButNeverID 钉住「轮次号」与「图 id 的序号」 +// 是两个量：同一轮里可以有**不止一条路**（真并行的那一路 + 队首续跑的串行路），它们 +// 共享轮次号（客户端据此把同一轮的路并排画），但**各有各的图 id** —— 否则落盘的 +// `(session_id, turn_id)` 唯一键会把同一会话的两轮 upsert 成一行，前一轮整张图消失。 +func TestFlowPublisher_RoundsShareRoundNumberButNeverID(t *testing.T) { + rec := &fakeRecorder{} + p := NewFlowPublisher(nil, func(string) bool { return true }, WithRoundRecorder(rec)) + // 一条并行路先起（占用本轮），随后两条串行路在它仍飞着的时候先后跑完。 + p.StartRound(FlowOwnerKey("web", "turn-a"), "并行的一路") + p.StartRound("web", "串行第一轮") + p.EndRound("web", "答复1") + p.StartRound("web", "串行第二轮") + p.EndRound("web", "答复2") + + lines := recordedRounds(rec) + if len(lines) != 2 { + t.Fatalf("落盘 %d 条，want 2（两条串行路各一条）", len(lines)) + } + if lines[0].ID == lines[1].ID { + t.Fatalf("同一轮的两条串行路撞同一个 id %q ⇒ 落盘会把前一轮整行顶掉", lines[0].ID) + } + if lines[0].Round != lines[1].Round { + t.Fatalf("同一轮的两条路轮次号不同：%d vs %d", lines[0].Round, lines[1].Round) + } +} + +// TestFlowPublisher_SingleLegUnchanged 钉住**单路逐字不变**：归属桶就是会话 id， +// 图 id 仍是 `{会话}-{轮次}`（没有路后缀），帧上 `turnId` 键缺席。既有回归 +// （flow_snapshot_test.go 的 web-1/web-2）与本条是同一口径的两侧。 +func TestFlowPublisher_SingleLegUnchanged(t *testing.T) { + p := NewFlowPublisher(nil, func(string) bool { return true }) + p.StartRound("web", "你好") + p.EndRound("web", "你也好") + g := p.LastGraph("web") + if g == nil || g.ID != "web-1" || g.TurnID != "" { + t.Fatalf("单路图 = %+v，want id=web-1 且无 turnId（逐字不变）", g) + } +} + +// TestFlowOwnershipHelpers 钉住 loop 与 orchestration 之间的归属口径： +// 真并行路由 `{会话}#{路}` 归属、真会话 id 与路标识各自可单独取出（总线载荷要的是 +// 会话、图归属要的是桶），子代理分支继承派生它的那一路。 +func TestFlowOwnershipHelpers(t *testing.T) { + // 与 Run 同形：会话先经 withLoopSession 播种，网关再加路标识（WithParallelLeg）。 + ctx := withLoopSession(WithParallelLeg(WithTurnID(context.Background(), "turn-a"), "turn-a"), "web") + if got := flowOwner(ctx, "web"); got != "web#turn-a" { + t.Fatalf("并行路的归属桶 = %q，want web#turn-a", got) + } + if got := FlowOwnerFromContext(ctx); got != "web#turn-a" { + t.Fatalf("FlowOwnerFromContext = %q，want web#turn-a", got) + } + if got := FlowSessionFromContext(ctx); got != "web" { + t.Fatalf("FlowSessionFromContext = %q，want web（总线载荷要真会话 id）", got) + } + if got := FlowLaneFromContext(ctx); got != "turn-a" { + t.Fatalf("FlowLaneFromContext = %q，want turn-a", got) + } + + // 子代理（分支 Run）：归属继承分支桶，会话与路各自拆得回来。 + branch := WithFlowBranch(context.Background(), FlowOwnerFromContext(ctx)) + if got := flowOwner(branch, "sub-1"); got != "web#turn-a" { + t.Fatalf("子代理 span 的归属桶 = %q，want web#turn-a（必须落进派生它的那一路）", got) + } + if got := FlowSessionFromContext(branch); got != "web" { + t.Fatalf("子代理的 FlowSessionFromContext = %q，want web", got) + } + if got := FlowLaneFromContext(branch); got != "turn-a" { + t.Fatalf("子代理的 FlowLaneFromContext = %q，want turn-a", got) + } + + // 单路：桶、会话、路三者不出意外的形态（路为空）。 + single := withLoopSession(context.Background(), "web") + if got := flowOwner(single, "web"); got != "web" { + t.Fatalf("单路归属桶 = %q，want web", got) + } + if got := FlowOwnerFromContext(single); got != "web" { + t.Fatalf("单路 FlowOwnerFromContext = %q，want web", got) + } + if got := FlowLaneFromContext(single); got != "" { + t.Fatalf("单路 FlowLaneFromContext = %q，want 空", got) + } + // 没有会话就没有归属：绝不回 `#路` 这种既查不到图、又骗调用方的串。 + if got := FlowOwnerFromContext(WithParallelLeg(context.Background(), "turn-a")); got != "" { + t.Fatalf("无会话时 FlowOwnerFromContext = %q，want 空", got) + } + if got := FlowSessionFromContext(WithParallelLeg(context.Background(), "turn-a")); got != "" { + t.Fatalf("无会话时 FlowSessionFromContext = %q，want 空", got) + } + if _, lane := splitFlowOwner("web"); lane != "" { + t.Fatalf("无 # 的归属串应拆出空路，得到 %q", lane) + } +} + +// spanNamed / finishedSpan 造一个刚起 / 刚结束的 span，让图里有真实的节点。 +func spanNamed(id, name string) *trace.Span { + return &trace.Span{ID: id, Name: name, Kind: trace.SpanKindTool} +} + +func finishedSpan(id, name string) *trace.Span { + sp := spanNamed(id, name) + sp.Finish(trace.SpanStatusOK) + return sp +} + +// TestFlowPublisher_LaneFramesCarryLaneKey 钉住**帧载荷**上的 `turnId` 键： +// 客户端据此把两路的帧分成两条泳道（旧客户端忽略未知键仍可用）。 +func TestFlowPublisher_LaneFramesCarryLaneKey(t *testing.T) { + var frames []*trace.FlowGraph + p := NewFlowPublisher(func(_ string, payload any) { + if g, ok := payload.(*trace.FlowGraph); ok { + frames = append(frames, g) + } + }, func(string) bool { return true }) + p.StartRound(FlowOwnerKey("web", "turn-a"), "标记A") + p.StartRound(FlowOwnerKey("web", "turn-b"), "标记B") + + if len(frames) < 2 { + t.Fatalf("发布帧数 = %d，want ≥2", len(frames)) + } + for _, g := range frames { + if g.TurnID == "" { + t.Fatalf("并行路的帧缺 turnId 键（客户端无法分成两条泳道）：id=%s", g.ID) + } + if !strings.Contains(g.ID, g.TurnID) { + t.Fatalf("图 id %q 里找不到路标识 %q（两者必须同源）", g.ID, g.TurnID) + } + if g.Round == 0 { + t.Fatalf("帧缺轮次号：id=%s", g.ID) + } + } +} diff --git a/internal/runtime/loop/flow_parallel_test.go b/internal/runtime/loop/flow_parallel_test.go new file mode 100644 index 0000000..fe3f417 --- /dev/null +++ b/internal/runtime/loop/flow_parallel_test.go @@ -0,0 +1,292 @@ +package loop + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// 本文件是**真并行两路互不覆盖**的回归（design/15 §2.5.10-1 登记的那条真 bug）： +// 同一会话的两条并行路（`chat.parallel`）各自开一轮，`graphs[sessionID]` 只有一张 ⇒ +// 后起的那一路把前一路的图整张顶掉、两路的节点互相长在对方的图上、落盘两路撞同一个 +// `flow_graphs.turn_id`。 +// +// 用例刻意只用**引入本切片之前就存在**的 API 驱动真实路径（真 `Loop.Run` + 网关实际 +// 挂的那两条 ctx 通道 `WithTurnID` / `WithParallelLeg`），因此它在修复前能编译、能跑、 +// 会红 —— 「改前红」的证据就是它（见 FV2-report §4）。判据： +// +// - 两条路各落一条（`RecordRound` 两次）且**图 id 不同**（同一 turn_id 会被 +// `(session_id, turn_id)` 唯一键 upsert 成一行）； +// - 以 A 的输入播种的那张图里只有 A 的工具节点，反之亦然； +// - 任何一帧 `graph_update` 都不同时含两条路的工具节点。 + +// barrierTool 是执行期停在闸门上的工具：它报告「已到执行点」后阻塞，使测试能确定地 +// 构造「A 的工具在飞时 B 起步」的交错时序，而不是靠 sleep 赌调度。 +type barrierTool struct { + name string + arrived chan struct{} + gate chan struct{} + once sync.Once +} + +func newBarrierTool(name string) *barrierTool { + return &barrierTool{name: name, arrived: make(chan struct{}, 1), gate: make(chan struct{})} +} + +func (b *barrierTool) Name() string { return b.name } +func (b *barrierTool) Description() string { return "barrier tool " + b.name } + +func (b *barrierTool) Parameters() (map[string]any, error) { + return map[string]any{"type": "object", "properties": map[string]any{}}, nil +} + +func (b *barrierTool) Execute(ctx context.Context, _ any) (any, error) { + b.once.Do(func() { b.arrived <- struct{}{} }) + select { + case <-b.gate: + return b.name + " 完成", nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// laneAdapter 按「这一路的窗口里是哪个标记文本」分流，先请求该路的工具，工具结果回灌 +// 之后给出该路的最终答复。两条路共用一个实例 —— 与生产同形（同一个 runner 跑两路）。 +type laneAdapter struct { + tool map[string]string // 标记文本 → 该路请求的工具名 + answer map[string]string // 标记文本 → 该路的最终答复 +} + +func (a *laneAdapter) Name() string { return "lane" } + +func (a *laneAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) { + mark := a.markOf(req) + if mark == "" { + return model.Response{Content: "无法归属的请求"}, nil + } + if lastMessageRole(req.Messages) == model.RoleTool { + return model.Response{Content: a.answer[mark]}, nil + } + return model.Response{Content: `{"tool_calls":[{"name":"` + a.tool[mark] + `","args":{}}]}`}, nil +} + +// markOf 找出请求里出现的那一路的标记文本；两路的窗口互不共享（I4），因此最多一命中。 +func (a *laneAdapter) markOf(req model.Request) string { + for _, m := range req.Messages { + if m.Role != model.RoleUser { + continue + } + for mark := range a.tool { + if strings.Contains(messageText(m), mark) { + return mark + } + } + } + return "" +} + +// ChatStream 是流式那一半：本用例的循环走非流式路径（未挂 OnStream），因此它只把 +// 请求委托给 Chat 并原样交出聚合结果 —— 刻意**不**再造第二套脚本，也刻意不发任何 +// 增量事件，避免「哪个入口真的跑过」变成模糊事实。 +func (a *laneAdapter) ChatStream(ctx context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) { + return a.Chat(ctx, req, model.Options{}) +} + +// lastMessageRole 返回末条消息的角色（空窗口时为空串）。 +func lastMessageRole(msgs []model.Message) model.Role { + if len(msgs) == 0 { + return "" + } + return msgs[len(msgs)-1].Role +} + +// parallelLaneFixture 是两路交错跑完之后的全部观测面：落盘的每一路、发布的每一帧、 +// 以及两路各自的最终答复（用来确认哪张图属于哪一路）。 +type parallelLaneFixture struct { + pub *FlowPublisher + recorder *fakeRecorder + mu sync.Mutex + frames []*trace.FlowGraph +} + +// run 让两条平行路交错跑完一轮：A 起轮 → A 的工具在飞 → B 起轮 → B 的工具在飞 → +// 依次放行。返回时两路的 Run 都已返回。 +func (f *parallelLaneFixture) run(t *testing.T) { + t.Helper() + toolA, toolB := newBarrierTool("probe_a"), newBarrierTool("probe_b") + ad := &laneAdapter{ + tool: map[string]string{"标记A": "probe_a", "标记B": "probe_b"}, + answer: map[string]string{"标记A": "答复A", "标记B": "答复B"}, + } + l := New(Options{ + Agent: newTestAgent(t, ad, dispatcherWith(t, toolA, toolB)), + MaxRounds: 8, + Flow: f.pub, + Tracer: trace.New(nil), + }) + + ctxA := WithParallelLeg(WithTurnID(context.Background(), "turn-a"), "turn-a") + ctxB := WithParallelLeg(WithTurnID(context.Background(), "turn-b"), "turn-b") + var wg sync.WaitGroup + var resA, resB *Result + var errA, errB error + wg.Add(2) + go func() { + defer wg.Done() + resA, errA = l.Run(ctxA, "web", []model.Message{model.TextMessage(model.RoleUser, "标记A：请跑一次工具")}) + }() + // A 的工具确实在飞之后才起 B —— 这才叫「同一会话两路并行」，也正是缺陷现场。 + awaitBarrier(t, toolA.arrived, "A 的工具执行点") + go func() { + defer wg.Done() + resB, errB = l.Run(ctxB, "web", []model.Message{model.TextMessage(model.RoleUser, "标记B：请跑一次工具")}) + }() + awaitBarrier(t, toolB.arrived, "B 的工具执行点") + close(toolA.gate) + close(toolB.gate) + wg.Wait() + + if errA != nil || errB != nil { + t.Fatalf("两路 Run 报错：A=%v B=%v", errA, errB) + } + if resA.Text != "答复A" || resB.Text != "答复B" { + t.Fatalf("两路答复 = %q / %q，want 答复A / 答复B（窗口串了）", resA.Text, resB.Text) + } +} + +// awaitBarrier 等一个闸门信号，超时即让测试失败（避免缺陷把它挂死成超时）。 +func awaitBarrier(t *testing.T, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + case <-time.After(20 * time.Second): + t.Fatalf("等待 %s 超时", what) + } +} + +// newParallelLaneFixture 装一台接了假落盘出口的发布器（真 bus 不在本用例范围内）。 +func newParallelLaneFixture(t *testing.T) *parallelLaneFixture { + t.Helper() + f := ¶llelLaneFixture{recorder: &fakeRecorder{}} + f.pub = NewFlowPublisher(func(subject string, payload any) { + if subject != graphUpdateSubject { + return + } + g, ok := payload.(*trace.FlowGraph) + if !ok { + return + } + f.mu.Lock() + f.frames = append(f.frames, g) + f.mu.Unlock() + }, func(string) bool { return true }, WithRoundRecorder(f.recorder)) + return f +} + +// recorded 返回落盘出口收到的每一路图（副本，调用方无需持锁）。 +func (f *parallelLaneFixture) recorded() []*trace.FlowGraph { + return recordedRounds(f.recorder) +} + +// recordedRounds 返回假落盘出口收到的每一路图（副本，调用方无需持锁）。 +func recordedRounds(rec *fakeRecorder) []*trace.FlowGraph { + rec.mu.Lock() + defer rec.mu.Unlock() + return append([]*trace.FlowGraph(nil), rec.graphs...) +} + +// publishedFrames 返回发布过的每一帧（副本）。 +func (f *parallelLaneFixture) publishedFrames() []*trace.FlowGraph { + f.mu.Lock() + defer f.mu.Unlock() + return append([]*trace.FlowGraph(nil), f.frames...) +} + +// TestFlowPublisher_ParallelLegsDoNotOverwriteEachOther 是本切片的核心用例： +// 两路并行 ⇒ 两桶互不覆盖、各自落一条、任何一帧都不含对方节点。 +func TestFlowPublisher_ParallelLegsDoNotOverwriteEachOther(t *testing.T) { + f := newParallelLaneFixture(t) + f.run(t) + + recorded := f.recorded() + if len(recorded) != 2 { + t.Fatalf("落盘 %d 条，want 2（每路一条）：%s", len(recorded), describeGraphs(recorded)) + } + if recorded[0].ID == recorded[1].ID { + t.Fatalf("两路撞同一个 turn_id %q ⇒ flow_graphs 的 (session_id, turn_id) 会把其中一路整行顶掉：%s", + recorded[0].ID, describeGraphs(recorded)) + } + graphA := graphWithInput(recorded, "标记A") + graphB := graphWithInput(recorded, "标记B") + if graphA == nil || graphB == nil { + t.Fatalf("两路没有各自的图（A=%v B=%v）：%s", graphA != nil, graphB != nil, describeGraphs(recorded)) + } + assertLaneIsolated(t, "A", graphA, "probe_a", "probe_b") + assertLaneIsolated(t, "B", graphB, "probe_b", "probe_a") + + for i, frame := range f.publishedFrames() { + if hasNode(frame, "probe_a") && hasNode(frame, "probe_b") { + t.Fatalf("第 %d 帧同时含两条路的工具节点（B 的节点长在 A 的图上）：%s", i, describeNodes(frame)) + } + } +} + +// assertLaneIsolated 断言某一路的图里有自己的工具节点、且**没有**对方的工具节点。 +func assertLaneIsolated(t *testing.T, lane string, g *trace.FlowGraph, own, other string) { + t.Helper() + if !hasNode(g, own) { + t.Fatalf("%s 路的图里没有自己的工具节点 %q：%s", lane, own, describeNodes(g)) + } + if hasNode(g, other) { + t.Fatalf("%s 路的图里出现了另一路的工具节点 %q：%s", lane, other, describeNodes(g)) + } + if g.SessionID != "web" { + t.Fatalf("%s 路图的 sessionId = %q，want web（路不能改会话归属）", lane, g.SessionID) + } +} + +// graphWithInput 找出输入节点标签含 mark 的那张图（两路各自被用户文本播种）。 +func graphWithInput(graphs []*trace.FlowGraph, mark string) *trace.FlowGraph { + for _, g := range graphs { + for _, n := range g.Nodes { + if n.Kind == "input" && strings.Contains(n.Label, mark) { + return g + } + } + } + return nil +} + +// hasNode 报告图里有没有标签等于（或前缀为）name 的工具节点。 +func hasNode(g *trace.FlowGraph, name string) bool { + for _, n := range g.Nodes { + if n.Kind == "tool" && strings.Contains(n.Label, name) { + return true + } + } + return false +} + +// describeGraphs / describeNodes 只打印**结构与标签**，供失败信息定位（测试里没有 +// 敏感内容，但仍只打标识）。 +func describeGraphs(graphs []*trace.FlowGraph) string { + var b strings.Builder + for _, g := range graphs { + b.WriteString("\n id=" + g.ID + " session=" + g.SessionID) + b.WriteString(" nodes=" + describeNodes(g)) + } + return b.String() +} + +func describeNodes(g *trace.FlowGraph) string { + parts := make([]string, 0, len(g.Nodes)) + for _, n := range g.Nodes { + parts = append(parts, n.Kind+":"+n.Label) + } + return "[" + strings.Join(parts, " ") + "]" +} diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index 0a7f19d..394ef20 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -83,9 +83,14 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa // 分支 Run（子代理）把其 span 并入父会话的实时 // 图；只有所属的 Run 开始与结束轮次。没有它，子代理 // 会重置前端唯一的 flowGraph，并让父级自身的节点出现断档。 + // + // 归属取**桶**（`{会话}[#{路}]`，见 FlowOwnerKey）：真并行时本 Run 属于某一条路， + // 两路因此各起各的图、各收各的图 —— 后起的那一路绝不重置前一路（design/15 + // §2.5.10-1）。单路时桶逐字等于 sessionID，行为与引入本切片之前完全一致。 if _, isBranch := flowBranch(ctx); l.flow != nil && !isBranch { - l.flow.StartRound(sessionID, lastUserText(msgs)) - defer func() { l.flow.EndRound(sessionID, res.Text) }() + owner := flowOwner(ctx, sessionID) + l.flow.StartRound(owner, lastUserText(msgs)) + defer func() { l.flow.EndRound(owner, res.Text) }() } return l.driveTurn(ctx, sessionID, msgs, startStep, res) } diff --git a/internal/runtime/loop/steer.go b/internal/runtime/loop/steer.go index 61cc186..acd4775 100644 --- a/internal/runtime/loop/steer.go +++ b/internal/runtime/loop/steer.go @@ -98,7 +98,7 @@ func (l *Loop) steerAtBoundary(ctx context.Context, sessionID string, step int, // 链路图如实标记「本轮输入被中途追加过」（design/15 §5）。子代理 Run 不拥有 // 自己的图（它的 span 并入派生它的会话），因此分支 Run 不打这个标记。 if _, isBranch := flowBranch(ctx); !isBranch { - l.flow.MarkInputSteered(sessionID) + l.flow.MarkInputSteered(flowOwner(ctx, sessionID)) } l.log.Info("steer injected", "session", sessionID, "step", step, "messages", len(msgs)) return append(steered, msgs...) diff --git a/internal/runtime/orchestration/scheduler.go b/internal/runtime/orchestration/scheduler.go index fdb37da..38aa817 100644 --- a/internal/runtime/orchestration/scheduler.go +++ b/internal/runtime/orchestration/scheduler.go @@ -139,6 +139,7 @@ func (s *Scheduler) Spawn(ctx context.Context, spec AgentSpec) (*SubAgent, error Mode: spec.Mode, AttachInfo: spec.AttachInfo, ParentSession: spec.ParentSession, + FlowOwner: spec.FlowOwner, Data: spec.Data, ToolWhitelist: cloneStrings(spec.ToolWhitelist), Context: taskContext(spec), @@ -295,9 +296,10 @@ func (s *Scheduler) run(ctx context.Context, sub *SubAgent) { // 子代理不得把内心独白泄漏进调用方的聊天流 // （design/05 §8: 少说话、多做事; design/04 §2: 不污染前台）：静音抑制 // UI 流/容量广播，而进度仍经总线与泳道到达前端。flow 分支把本次 Run 发射的 - // 每个 span 都归属到派发会话的图，而不是另开第二张图。 + // 每个 span 都归属到派发**那一路**的图（`{会话}[#{路}]`，见 SubAgent.flowOwner）， + // 而不是另开第二张图、也不是落进会话主干那张。 runCtx = loop.WithMutedBroadcast(runCtx) - runCtx = loop.WithFlowBranch(runCtx, sub.ParentSession) + runCtx = loop.WithFlowBranch(runCtx, sub.flowOwner()) // 被派发的目标角色决定本次运行每一步的提示词与记忆作用域（design/22 §8-2）。 // 留空则保持继承来的 env：既有行为，即子代理按派发回合的角色运行。 runCtx = withPersonaEnv(runCtx, sub.PersonaID) @@ -370,18 +372,18 @@ func (s *Scheduler) subCtx(parent context.Context, sub *SubAgent) (context.Conte } // subSpan 在配置了 tracer 时开启一个子代理 span，把它串成 ctx 中已有 span 的子节点。 -// 子代理被归属到某个 flow 会话（AgentSpec.ParentSession）时，该 span 还会作为子代理 -// 泳道根折进那个会话的实时图，使分支渲染在派发它的会话图内，而不是自成一张图 -// （design/19 §2）。它返回携带该 span 的 ctx 与 span 自身（tracer 为 nil 时 span 为 -// nil）。 +// 子代理被归属到某张 flow 图（`SubAgent.flowOwner`，即 `{会话}[#{路}]`）时，该 span +// 还会作为子代理泳道根折进那张图，使分支渲染在派发它的**那一路**图内，而不是自成 +// 一张图、也不会错落进另一路的图（design/19 §2）。它返回携带该 span 的 ctx 与 span +// 自身（tracer 为 nil 时 span 为 nil）。 func (s *Scheduler) subSpan(ctx context.Context, sub *SubAgent) (context.Context, *trace.Span) { if s.tracer == nil { return ctx, nil } sctx, span := s.tracer.StartSpan(ctx, "subagent."+sub.ID, trace.SpanKindSubagent) - if sess := sub.ParentSession; sess != "" && s.flow != nil { - _ = span.SetProp("session", sess) - s.flow.SpanStarted(sess, span) + if owner := sub.flowOwner(); owner != "" && s.flow != nil { + _ = span.SetProp("session", owner) + s.flow.SpanStarted(owner, span) } // 目标角色随分支节点传播，使流程图面板展示子代理以哪个角色作答 // （design/22 §8-2 可观测性），而不只是子代理 id。 @@ -427,13 +429,16 @@ func (s *Scheduler) spanErr(sub *SubAgent, span *trace.Span, err error) { s.foldSpanFinish(sub, span) } -// foldSpanFinish 把一个刚关闭的子代理 span 折进其父会话的流程图，把泳道根翻转为终态。 -// 该 span 已把父会话带在 "session" 属性上，因此发布器知道哪张图拥有它。 +// foldSpanFinish 把一个刚关闭的子代理 span 折进其所属的流程图，把泳道根翻转为终态。 +// 该 span 已把归属桶带在 "session" 属性上，因此发布器知道哪张图拥有它 +// （真并行时就是派发它的那一路的图，见 SubAgent.flowOwner）。 func (s *Scheduler) foldSpanFinish(sub *SubAgent, span *trace.Span) { - if span == nil || s.flow == nil || sub.ParentSession == "" { + if span == nil || s.flow == nil || sub == nil { return } - s.flow.SpanFinished(sub.ParentSession, span) + if owner := sub.flowOwner(); owner != "" { + s.flow.SpanFinished(owner, span) + } } // publish 经总线广播一条进度事件。未配置 comm 探针时是空操作。发布使用后台上下文， diff --git a/internal/runtime/orchestration/scheduler_spans.go b/internal/runtime/orchestration/scheduler_spans.go index e9dfa4d..9c6d913 100644 --- a/internal/runtime/orchestration/scheduler_spans.go +++ b/internal/runtime/orchestration/scheduler_spans.go @@ -22,7 +22,8 @@ const ( // 「派发标记」：与泳道根并列而不是它的父级 —— 若把它当作子代理 span 的父级， // 泳道链会多出一节（图构建器把非 subagent 祖先一路穿透解析泳道），既有泳道 // 契约就被改写了。span 只带非敏感属性（被派发的子代理 id、目标角色 id）， -// 声明了归属会话时同时发布一个「运行中」占位节点。未配置 tracer 时返回 nil。 +// 声明了归属时同时发布一个「运行中」占位节点（真并行时落进派发它的那一路的图）。 +// 未配置 tracer 时返回 nil。 func (s *Scheduler) orchSpan(ctx context.Context, sub *SubAgent) *trace.Span { if s.tracer == nil || sub == nil { return nil @@ -34,15 +35,15 @@ func (s *Scheduler) orchSpan(ctx context.Context, sub *SubAgent) *trace.Span { if sub.PersonaID != "" { _ = span.SetProp("persona", sub.PersonaID) } - if sess := sub.ParentSession; sess != "" && s.flow != nil { - _ = span.SetProp("session", sess) - s.flow.SpanStarted(sess, span) + if owner := sub.flowOwner(); owner != "" && s.flow != nil { + _ = span.SetProp("session", owner) + s.flow.SpanStarted(owner, span) } return span } // finishOrchSpan 收尾一次派发：记录耗时（失败另记原因）与一条脱敏事件， -// 失败时终态为 error，然后把已关闭的 span 折回所属会话的图。 +// 失败时终态为 error，然后把已关闭的 span 折回所属那一路的图。 func (s *Scheduler) finishOrchSpan(sub *SubAgent, span *trace.Span, err error) { if span == nil { return @@ -58,8 +59,10 @@ func (s *Scheduler) finishOrchSpan(sub *SubAgent, span *trace.Span, err error) { status = trace.SpanStatusError } span.Finish(status) - if s.flow == nil || sub == nil || sub.ParentSession == "" { + if s.flow == nil || sub == nil { return } - s.flow.SpanFinished(sub.ParentSession, span) + if owner := sub.flowOwner(); owner != "" { + s.flow.SpanFinished(owner, span) + } } diff --git a/internal/runtime/orchestration/spawnflow_lane_test.go b/internal/runtime/orchestration/spawnflow_lane_test.go new file mode 100644 index 0000000..52105eb --- /dev/null +++ b/internal/runtime/orchestration/spawnflow_lane_test.go @@ -0,0 +1,121 @@ +package orchestration + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// 本文件钉住**真并行路里派发的子代理落进那一路的图**（切片 FV2；design/15 §2.5.10-1）。 +// +// 症状：同一会话真并行时两路各有一张图，而子代理的 span 归属取自 `AgentSpec.ParentSession` +// —— 那是**会话**（总线载荷要用它），不含路 ⇒ 子代理的泳道会落进「会话主干那张」，而那张图 +// 在纯并行轮次里根本不存在（span 被静默丢弃），或者落进另一路。这里断言的是：派发工具从 +// ctx 取出的**归属桶**（`{会话}#{路}`）一路传到发布器，泳道根只长在**那一路**的图上。 + +// laneGraphCapture 按 (会话, 路) 记录发布器发出的每一张图（路的键就是帧上的 `turnId`）。 +type laneGraphCapture struct { + mu sync.Mutex + lanes map[string]*trace.FlowGraph +} + +func laneKey(session, turnID string) string { return session + "|" + turnID } + +func (c *laneGraphCapture) publish(_ string, payload any) { + g, ok := payload.(*trace.FlowGraph) + if !ok || g == nil { + return + } + c.mu.Lock() + defer c.mu.Unlock() + if c.lanes == nil { + c.lanes = make(map[string]*trace.FlowGraph) + } + c.lanes[laneKey(g.SessionID, g.TurnID)] = g +} + +// lane 返回某一路最新那张图；不存在时判测试失败（并列出收到过哪些路）。 +func (c *laneGraphCapture) lane(t *testing.T, session, turnID string) *trace.FlowGraph { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + g := c.lanes[laneKey(session, turnID)] + if g == nil { + t.Fatalf("没有收到会话 %q 路 %q 的图（收到过的路：%v）", session, turnID, c.keysLocked()) + } + return g +} + +func (c *laneGraphCapture) keysLocked() []string { + out := make([]string, 0, len(c.lanes)) + for k := range c.lanes { + out = append(out, k) + } + return out +} + +// branchOf 造一个「运行在某一路上」的 ctx（网关经 WithParallelLeg 盖上会话 + 路； +// 子代理自身则经 WithFlowBranch 带着父归属桶）。两条通道的 `FlowOwnerFromContext` +// 取值同源，因此派发工具的取值路径只有这一条。 +func branchOf(session, lane string) context.Context { + return loop.WithFlowBranch(context.Background(), loop.FlowOwnerKey(session, lane)) +} + +// TestSpawnToolBindsSubAgentToItsLaneGraph 是核心断言：在一路里派发的子代理，其泳道根 +// 只出现在**那一路**的图上，且会话主干那张图**不存在**（＝没有被静默丢弃或错落他路）。 +func TestSpawnToolBindsSubAgentToItsLaneGraph(t *testing.T) { + t.Parallel() + capture := &laneGraphCapture{} + flow := loop.NewFlowPublisher(capture.publish, nil) + tracer := trace.New(nil) + + d, err := tool.New(registry.New()) + if err != nil { + t.Fatalf("tool.New: %v", err) + } + if rerr := d.Register(&echoTool{name: "echo", result: "ok"}); rerr != nil { + t.Fatalf("Register: %v", rerr) + } + s := branchLoop(t, flow, tracer, d, Options{MaxParallel: 2}) + + // 这一路已经开跑（网关在调用 Run 之前就发过 turn_begin，图的归属桶因此已存在）。 + owner := loop.FlowOwnerKey(parentSession, "turn-a") + flow.StartRound(owner, "并行的一路") + sp := NewSpawnTool(s, WithSpawnWait(5*time.Second)) + if _, xerr := sp.Execute(branchOf(parentSession, "turn-a"), map[string]any{"goal": "调研一下"}); xerr != nil { + t.Fatalf("派发: %v", xerr) + } + flow.EndRound(owner, "结果如下") + + g := capture.lane(t, parentSession, "turn-a") + roots := 0 + for _, n := range g.Nodes { + if n.Kind == "subagent" { + roots++ + } + } + if roots != 1 { + t.Fatalf("该路的图里子代理泳道根 = %d 个，want 1（nodes=%v）", roots, nodeIDs(g)) + } + if strings.HasSuffix(g.ID, "#turn-a") == false { + t.Fatalf("该路的图 id = %q，want 带 `#turn-a` 后缀", g.ID) + } + // 会话主干那张（无路标识）绝不该被这一路的子代理污染：它要么不存在，要么不含泳道根。 + capture.mu.Lock() + trunk := capture.lanes[laneKey(parentSession, "")] + capture.mu.Unlock() + if trunk != nil { + for _, n := range trunk.Nodes { + if n.Kind == "subagent" { + t.Fatalf("子代理泳道根长在了会话主干图上（路桶 %v 才是它该在的地方）", capture.keysLocked()) + } + } + } +} diff --git a/internal/runtime/orchestration/spawntool.go b/internal/runtime/orchestration/spawntool.go index 8a5a2a4..d46ad17 100644 --- a/internal/runtime/orchestration/spawntool.go +++ b/internal/runtime/orchestration/spawntool.go @@ -153,10 +153,17 @@ func (t *SpawnTool) Execute(ctx context.Context, args any) (any, error) { if t.sched == nil { return nil, fmt.Errorf("%s: %w", SpawnToolName, ErrNoScheduler) } - specs, err := spawnSpecs(m, loop.FlowOwnerFromContext(ctx), t.timeout) + specs, err := spawnSpecs(m, loop.FlowSessionFromContext(ctx), t.timeout) if err != nil { return nil, fmt.Errorf("%s: %w", SpawnToolName, err) } + // flow 归属桶（`{会话}[#{路}]`）另取一次：ParentSession 必须是**真会话 id** + // （后台批次的 `subagent_result` 帧按它归属到父会话），而图归属要精确到**哪一路**， + // 否则真并行时两路子代理会混进同一张图（design/15 §2.5.10-1）。 + owner := loop.FlowOwnerFromContext(ctx) + for i := range specs { + specs[i].FlowOwner = owner + } // 在任何东西启动之前就拒绝未知的目标角色：批次必须是全有或全无，否则已经派发出的 // 子代理会继续运行，而它们的结果永远不上报给模型（design/22 §8-2: 未知角色明确报错， // 不静默回落到默认角色). diff --git a/internal/runtime/orchestration/spec.go b/internal/runtime/orchestration/spec.go index 8ef5f2a..4dac2a3 100644 --- a/internal/runtime/orchestration/spec.go +++ b/internal/runtime/orchestration/spec.go @@ -58,9 +58,17 @@ type AgentSpec struct { AttachInfo string // ParentSession 是拥有该分支的会话，其实时流程图将承载子代理的 span 及其内部 // 步骤，使它们渲染成派发会话图的一条泳道，而不是自成一张图（design/19 §2）。 - // 派发工具从 loop.FlowOwnerFromContext 取得它。留空则关闭流图归属；它绝不影响 - // 子代理在哪个会话里运行，因此其 checkpoint、记忆与流隔离都不变。 + // 派发工具从 loop.FlowSessionFromContext 取得它（**不含路**：后台批次的 + // `subagent_result` 帧按它归属到父会话，因此它必须是真会话 id）。 + // 留空则关闭流图归属；它绝不影响子代理在哪个会话里运行，因此其 checkpoint、 + // 记忆与流隔离都不变。 ParentSession string + // FlowOwner 是子代理 span 落进**哪一张** flow 图的归属桶 + // （`{会话}[#{路}]`，见 loop.FlowOwnerFromContext）。真并行时同一会话有多条路， + // 派发它的那一路有自己的一张图，子代理的泳道必须落进那一路 —— 否则两路的 + // 子代理节点会互相长在对方的图上（design/15 §2.5.10-1）。派发工具填它； + // 留空时回落到 ParentSession（会话主干那张，即单路与旧调用方的既有行为）。 + FlowOwner string // Data 承载供调用方使用的可选结构化载荷。 Data any // ToolWhitelist 限定该子代理可执行的工具集合（design/05 §7 裁决 02.4.1）： diff --git a/internal/runtime/orchestration/subagent.go b/internal/runtime/orchestration/subagent.go index 3c7640a..88acf4a 100644 --- a/internal/runtime/orchestration/subagent.go +++ b/internal/runtime/orchestration/subagent.go @@ -11,11 +11,11 @@ import ( ) // SubAgent 是单个已派发的子代理实例。身份字段（ID、Goal、RoleID、PersonaID、Mode、 -// AttachInfo、ParentSession、Data、ToolWhitelist、Context、WorkingMemory）在 Spawn 时 -// 一次设定，此后永不改动。生命周期字段（Status、Summary、Result、Err）由子代理的后台 -// goroutine 写入，并经下方同步访问器读取，因此在子代理结束前读取它们是安全的。当且 -// 仅当 Done 报告 true 时子代理处于终结态；Join 在全部子代理终结后返回，Wait 在指名的 -// id 全部终结后返回。 +// AttachInfo、ParentSession、FlowOwner、Data、ToolWhitelist、Context、WorkingMemory） +// 在 Spawn 时一次设定，此后永不改动。生命周期字段（Status、Summary、Result、Err）由 +// 子代理的后台 goroutine 写入，并经下方同步访问器读取，因此在子代理结束前读取它们 +// 是安全的。当且仅当 Done 报告 true 时子代理处于终结态；Join 在全部子代理终结后 +// 返回，Wait 在指名的 id 全部终结后返回。 // // SubAgent 由 (*Scheduler).Spawn 创建；不应直接构造。 type SubAgent struct { @@ -35,6 +35,12 @@ type SubAgent struct { AttachInfo string // ParentSession 是拥有本子代理分支的流程图所属会话（该分支不被可视化时为空）。 ParentSession string + // FlowOwner 是本子代理 span 落进**哪一张** flow 图的归属桶 + // （`{会话}[#{路}]`，见 loop.FlowOwnerFromContext）：真并行时派发它的那一路 + // 有自己的图，本子代理必须落进那一路，而不是会话主干那张 —— 否则两路的 + // 子代理节点会互相长在对方的图上（design/15 §2.5.10-1）。留空时回落到 + // ParentSession（＝会话主干那张，单路/旧调用方的既有行为）。 + FlowOwner string // Data 是派发时提供的可选结构化载荷。 Data any // ToolWhitelist 是该子代理的工具白名单（design/05 §7 裁决 02.4.1）： @@ -117,6 +123,20 @@ func (s *SubAgent) FinishedAt() time.Time { // Done 报告子代理是否已进入终结态。 func (s *SubAgent) Done() bool { return s.Status().Terminal() } +// flowOwner 返回本子代理 span 该落进哪一张 flow 图的归属桶 +// （`{会话}[#{路}]`）：显式声明了 FlowOwner 时用它（真并行路派发的子代理因此落进 +// **那一路**的图），否则回落到 ParentSession（会话主干那张图，即单路与旧调用方 +// 的既有行为）。两者都为空时返回空串 —— 调用方据此抑制发布（不被可视化的分支）。 +func (s *SubAgent) flowOwner() string { + if s == nil { + return "" + } + if s.FlowOwner != "" { + return s.FlowOwner + } + return s.ParentSession +} + // waitDone 返回子代理进入终结态时被关闭的 channel。它是 (*Scheduler).Wait 按 id 阻塞 // 时等待的对象，无需轮询。 func (s *SubAgent) waitDone() <-chan struct{} { return s.finishedCh } diff --git a/internal/runtime/trace/graph.go b/internal/runtime/trace/graph.go index 60bf542..7564c76 100644 --- a/internal/runtime/trace/graph.go +++ b/internal/runtime/trace/graph.go @@ -61,12 +61,22 @@ type FlowEdge struct { // FlowGraph is a live, single-round flow graph published as ws.graph_update. type FlowGraph struct { - ID string `json:"id"` - Nodes []FlowNode `json:"nodes"` - Edges []FlowEdge `json:"edges"` - SessionID string `json:"sessionId,omitempty"` - StartedAt time.Time `json:"startedAt,omitempty"` - EndedAt time.Time `json:"endedAt,omitempty"` + ID string `json:"id"` + Nodes []FlowNode `json:"nodes"` + Edges []FlowEdge `json:"edges"` + // SessionID 是拥有这张图的会话（帧壳与落盘行的归属坐标）。 + SessionID string `json:"sessionId,omitempty"` + // TurnID 是这张图所属**路**的权威回合标识（design/15 §2.5.9）：真并行时 + // 同一会话有多条路，每条路各有自己的图与本键。**缺席**语义：这条路没有 + // 权威标识（单路的历史帧、未接编排器的调用方、旧后端）——客户端据此把图 + // 归到会话那一张（兼容口径），绝不臆造一个 id。 + TurnID string `json:"turnId,omitempty"` + // Round 是这张图所属的**会话轮次号**（每会话单调递增；同一轮的各条路共享它）。 + // 客户端据此只保留最新一轮的泳道，避免旧轮的路留在页面上。 + // **缺席**（0）语义：旧后端/未设置 —— 客户端按「每次帧都算同一轮」的保守口径处理。 + Round int `json:"round,omitempty"` + StartedAt time.Time `json:"startedAt,omitempty"` + EndedAt time.Time `json:"endedAt,omitempty"` } // NodeSpec is the input to RunGraph.Upsert: the full description of a node plus @@ -87,8 +97,16 @@ type NodeSpec struct { // structure: it knows nothing about the bus, sessions or the loop, so it is // unit-testable in isolation. A caller serializes access (see loop.FlowPublisher). type RunGraph struct { - session string - round int + session string + round int + // seq 是本图的**会话级建图序号**（调用方每次建图递增、永不复用）。图 id 的第二 + // 分量取它而不是轮次号：真并行时同一轮里的第二条路必须也拿到全新的 id，否则两路 + // 撞同一个 `flow_graphs.turn_id`（唯一键），后收尾的那一路把前一路整行顶掉。 + // 未经 SetSeq 建图的调用方（图构建器单测）保留 `{会话}-{轮次}` 的老形状。 + seq int + // turnID 是本图所属**路**的权威回合标识（design/15 §2.5.9）。空串表示这条路 + // 没有权威标识（单路 / 未接编排器 / 旧调用方），此时图上该键缺席。 + turnID string nodes map[string]*FlowNode order []string edges map[string]bool @@ -118,7 +136,8 @@ func NewRunGraph(session string) *RunGraph { } // Start marks the round began at now and resets the node/edge sets so a reused -// graph starts fresh for a new round. +// graph starts fresh for a new round. 路口标识不在重置之列：它是**本次建图**的 +// 身份，由 SetTurnID 在 Start 之后立刻盖上（见 loop.FlowPublisher）。 func (g *RunGraph) Start(round int, now time.Time) { g.round = round g.started = now @@ -131,6 +150,15 @@ func (g *RunGraph) Start(round int, now time.Time) { g.laneLast = make(map[string]string) } +// SetTurnID 盖上这张图所属**路**的权威回合标识（design/15 §2.5.9）。空串是 +// 合法入参（单路 / 未接编排器），此时快照里该键缺席（omitempty），客户端回落 +// 到「按会话归一张图」的兼容口径。 +func (g *RunGraph) SetTurnID(turnID string) { g.turnID = turnID } + +// SetSeq 盖上这张图的**会话级建图序号**（见 RunGraph.seq）。未设置（0）时图 id +// 退回 `{会话}-{轮次}` 的老形状，因此只调 Start 的既有调用方/单测逐字不变。 +func (g *RunGraph) SetSeq(seq int) { g.seq = seq } + // SeedInput creates the "input" node for the round. It returns the node id so a // caller can reference it. The label is redacted (design/19 §3), so a message that // happens to carry a credential cannot publish one. @@ -160,22 +188,48 @@ func (g *RunGraph) SeedOutput(text string) string { return "output" } -// Snapshot returns a copy of the graph safe to encode and publish. +// laneIDSep 是图 id 里「会话-轮次」与「路标识」之间的分隔符。它与 `loop` 的 +// flow 归属串（`{会话}#{路}`）以及 design/15 §2.5.9 的 checkpoint 归属键 +// （`{sessionID}#{turnID}`）取同一个字符：**一条路**在流程图、checkpoint 与 +// 落盘行三处因此是同一个拼接口径（loop 侧的同名常量见 loop/flow_branch.go）。 +const laneIDSep = "#" + +// Snapshot returns a copy of the graph safe to encode and publish. 图 id 是 +// `{会话}-{轮次}`；这张图属于一条**路**时追加 `#{路标识}` —— 真并行下同一会话 +// 同一轮的两路必须各有各的 id，否则落盘（`flow_graphs` 的 `(session_id, turn_id)` +// 唯一键）会把其中一路整行顶掉、客户端也分不出两张图（design/15 §2.5.10-1）。 func (g *RunGraph) Snapshot() *FlowGraph { nodes := make([]FlowNode, 0, len(g.order)) for _, id := range g.order { nodes = append(nodes, *g.nodes[id]) } return &FlowGraph{ - ID: fmt.Sprintf("%s-%d", g.session, g.round), + ID: g.graphID(), Nodes: nodes, Edges: append([]FlowEdge(nil), g.edgeOrder...), SessionID: g.session, + TurnID: g.turnID, + Round: g.round, StartedAt: g.started, EndedAt: g.ended, } } +// graphID 拼出这张图的 id：`{会话}-{序号}[#{路标识}]`。序号取会话级建图序号 +// （未设置时退回轮次号，见 RunGraph.seq），因此**同一轮里的每一条路也各有各的 id**； +// 路标识只在「这条路有权威标识」时追加。 +func (g *RunGraph) graphID() string { + n := g.seq + if n == 0 { + n = g.round + } + base := fmt.Sprintf("%s-%d", g.session, n) + if g.turnID == "" { + return base + } + return base + laneIDSep + g.turnID +} + // Clone 深拷贝一份图，供**锁外**的读者使用（契约 v0.71 的 // `GET /api/v1/admin/flow/graph` 就近取材于 `loop.FlowPublisher.LastGraph`）。 // @@ -195,6 +249,8 @@ func (g *FlowGraph) Clone() *FlowGraph { Nodes: make([]FlowNode, 0, len(g.Nodes)), Edges: append([]FlowEdge(nil), g.Edges...), SessionID: g.SessionID, + TurnID: g.TurnID, + Round: g.Round, StartedAt: g.StartedAt, EndedAt: g.EndedAt, } diff --git a/internal/services/gateway/admin_flow.go b/internal/services/gateway/admin_flow.go index 96c78d1..e03376b 100644 --- a/internal/services/gateway/admin_flow.go +++ b/internal/services/gateway/admin_flow.go @@ -42,15 +42,27 @@ type flowTurnDTO struct { CreatedAt string `json:"createdAt"` } +// flowLaneDTO 是「最近一轮里的一条路」（契约 v0.77 的 `lanes[]`）：`turnId` 是**那一路** +// 的权威回合标识（`graph.turnId`，design/15 §2.5.9），`graph` 是那一路的图快照 +// （与 `graph_update` 帧的 data / 本响应的 `graph` 逐字同形）。 +// +// 真并行时同一会话同一轮有多条路，`graph` 只给出最新的一条（向后兼容：老客户端读它 +// 照旧可用），`lanes[]` 才是全貌 —— 并行子代理与并行路各自的图因此不再互相顶掉。 +type flowLaneDTO struct { + TurnID string `json:"turnId"` + Graph any `json:"graph"` +} + // adminFlowGraphResponse 是 GET /api/v1/admin/flow/graph 的响应体 -// （v0.71 起；**v0.73 起新增 `turnId`/`source`/`turns` 三键**，纯增量）。 +// （v0.71 起；**v0.73 起新增 `turnId`/`source`/`turns` 三键**，**v0.77 起新增 `lanes` +// 一键**，均为纯增量）。 // // Graph 为 nil 表示「该轮确实没有图」——这是一个**诚实的空态**，不是错误： // 客户端必须据此渲染空态，绝不允许补画任何节点或假图（§4.11(b)）。 // 该键**恒在**（`null` 而不是缺席），使客户端不必区分「没有这个键」与 // 「这个键是空」。 // -// Turns **恒为非 nil 切片**（没有轮次时是 `[]`），使客户端不必判空。 +// Turns **恒为非 nil 切片**（没有轮次时是 `[]`），使客户端不必判空。Lanes 同此口径。 type adminFlowGraphResponse struct { SessionID string `json:"sessionId"` // TurnID 是**实际返回的那一轮**的标识（`graph.id`）；没有图时为空串。 @@ -61,6 +73,11 @@ type adminFlowGraphResponse struct { // （契约 §3.4.5 / §4.11(a)）。字段以 `any` 承载，因为「没有图」（nil ⇒ // 线上 JSON `null`）与「有图」的区分就是本字段的语义。 Graph any `json:"graph"` + // Lanes 是**最近一轮每一路**的图（契约 v0.77，纯增量；真并行才有意义的全貌）。 + // 它取内存里的实时图（`FlowGraphSource.LatestLanes`），因此与 `?turnId=` 无关 —— + // 翻历史轮时 `lanes` 仍报「当前这一轮的各路」，而 `graph` 才是你指定的那一轮。 + // 图源未接线或该会话还没有图时是空数组（恒非 nil）。 + Lanes []flowLaneDTO `json:"lanes"` // Turns 是该会话**已落盘轮次**的摘要，最新在前（契约 v0.73）。 Turns []flowTurnDTO `json:"turns"` } @@ -120,10 +137,31 @@ func (s *Server) adminFlowGraph(w http.ResponseWriter, r *http.Request) { TurnID: resolvedTurn, Source: source, Graph: graphAny, + Lanes: s.flowLanes(sessionID), Turns: turns, }) } +// flowLanes 取**最近一轮每一路**的图（契约 v0.77 的 `lanes[]`）：真并行时同一会话 +// 同一轮有多条路，只回 `graph` 一棵会把其余路从面板上抹掉（design/15 §2.5.10-1）。 +// +// 它恒返回非 nil 切片（没有路时是 `[]`），使客户端不必判空。图源未接线（`s.flow` 为 +// nil）时回空切片而不是错误：那时端点本就走「只有落盘」的形态（v0.73 口径）， +// `lanes` 缺席语义 = 「没有实时路」。 +func (s *Server) flowLanes(sessionID string) []flowLaneDTO { + out := make([]flowLaneDTO, 0) + if s.flow == nil { + return out + } + for _, g := range s.flow.LatestLanes(sessionID) { + if g == nil { + continue + } + out = append(out, flowLaneDTO{TurnID: g.TurnID, Graph: g}) + } + return out +} + // resolveFlowGraph 解析「这一轮」的图：**先内存、miss 才落盘**（契约 v0.73 §4.11(b)）。 // // 判定顺序（三条都必要）： diff --git a/internal/services/gateway/admin_flow_lanes_test.go b/internal/services/gateway/admin_flow_lanes_test.go new file mode 100644 index 0000000..0e9a7a7 --- /dev/null +++ b/internal/services/gateway/admin_flow_lanes_test.go @@ -0,0 +1,238 @@ +package gateway + +// 本文件钉住契约 v0.77 的**按路**形状（切片 FV2，修 design/15 §2.5.10-1 登记的那条 +// 「graph_update 互相顶掉」）： +// +// - 帧 / 快照的图上多了 `turnId`（这张图属于哪一路的权威回合标识）与 `round` +// （会话轮次号），两者都是**纯增量**（老客户端忽略未知键仍可用）； +// - `GET /api/v1/admin/flow/graph` 多了 `lanes[]`：真并行时同一会话同一轮有多条路， +// 只回 `graph` 一棵会把其余路从面板上抹掉。 + +import ( + "encoding/json" + "net/http" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" +) + +// startLane 让发布器开始**一条路**的一轮（起轮 + 一个在飞的工具 span），并把归属桶与 +// 该 span 交回调用方，供稍后收尾用。两条路必须**交错**起轮（B 在 A 在飞时起）—— +// 那才是真并行的现场：轮次号正是在「上一轮的路全部收尾」之后才推进的。 +func (e *flowTestEnv) startLane(t *testing.T, session, lane, input string) (string, *trace.Span) { + t.Helper() + owner := loop.FlowOwnerKey(session, lane) + e.flow.StartRound(owner, input) + sp := &trace.Span{ID: "span-" + lane, Name: "tool." + lane, Kind: trace.SpanKindTool, StartedAt: time.Now()} + e.flow.SpanStarted(owner, sp) + sp.Finish(trace.SpanStatusOK) + e.flow.SpanFinished(owner, sp) + return owner, sp +} + +// finishLane 收尾一条路（答复播种 output 节点并发布最终帧）。 +func (e *flowTestEnv) finishLane(t *testing.T, owner, output string) { + t.Helper() + e.flow.EndRound(owner, output) +} + +// flowLanesData 解码响应体里的 `lanes[]`（契约 v0.77）。缺键时回 nil（调用方据此判红）。 +func flowLanesData(t *testing.T, body []byte) []map[string]json.RawMessage { + t.Helper() + env := decodeBody(t, body) + var data struct { + Lanes []map[string]json.RawMessage `json:"lanes"` + } + if err := json.Unmarshal(env["data"], &data); err != nil { + t.Fatalf("decode lanes %s: %v", body, err) + } + return data.Lanes +} + +// TestAdminFlowGraph_ReturnsEveryLaneOfLatestRound 是本次快照侧的核心断言： +// 真并行两路各自收轮之后打开页面，`lanes[]` 必须**每一路一条**、彼此不缺， +// 且 `graph`（向后兼容的那一棵）是其中最新的一路。 +func TestAdminFlowGraph_ReturnsEveryLaneOfLatestRound(t *testing.T) { + env := newFlowTestEnv(t, func(string) bool { return true }) + ownerA, _ := env.startLane(t, "web", "turn-a", "标记A") + ownerB, _ := env.startLane(t, "web", "turn-b", "标记B") + env.finishLane(t, ownerA, "答复A") + env.finishLane(t, ownerB, "答复B") + + resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/flow/graph?sessionId=web", nil, env.token(t, auth.RoleOwner)) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + _, graph, present := flowGraphData(t, body) + if !present || graph == nil { + t.Fatalf("`graph` 必须恒在且非空（两路都收轮了）：%s", body) + } + lanes := flowLanesData(t, body) + if len(lanes) != 2 { + t.Fatalf("lanes 条数 = %d，want 2（同一轮的两路各一条）：%s", len(lanes), body) + } + seen := map[string]string{} + for _, lane := range lanes { + var ( + turnID string + gid string + nodes []struct { + Kind string `json:"kind"` + Label string `json:"label"` + } + ) + _ = json.Unmarshal(lane["turnId"], &turnID) + var inner map[string]json.RawMessage + if err := json.Unmarshal(lane["graph"], &inner); err != nil { + t.Fatalf("lanes[].graph 不是图对象：%v (%s)", err, body) + } + _ = json.Unmarshal(inner["id"], &gid) + if err := json.Unmarshal(inner["nodes"], &nodes); err != nil { + t.Fatalf("lanes[].graph.nodes: %v", err) + } + seen[turnID] = gid + for _, n := range nodes { + if n.Kind == "tool" && n.Label != turnID { + t.Fatalf("路 %s 的图里出现了另一路的工具节点 %q（两路又混进同一张图了）", turnID, n.Label) + } + } + } + for _, want := range []string{"turn-a", "turn-b"} { + if seen[want] == "" { + t.Fatalf("lanes[] 里缺 %s 那一路：%v (%s)", want, seen, body) + } + } + // graph（兼容字段）= 最新一路；两条路 id 必须不同（否则落盘会互相顶掉）。 + if seen["turn-a"] == seen["turn-b"] { + t.Fatalf("两路拿到同一个图 id %q：%v", seen["turn-a"], seen) + } + var graphID string + _ = json.Unmarshal(graph["id"], &graphID) + if graphID != seen["turn-b"] { + t.Fatalf("graph.id = %q，want 最新一路 %q（后起的那一路最新）", graphID, seen["turn-b"]) + } +} + +// TestAdminFlowGraph_SingleLaneHasOneEntry 钉住单路口径：没有真并行时 `lanes[]` 是 +// 恰好一条（无路标识），`graph` 与它同源 —— 「每一路」在单路下就是那一路， +// 不因为引入本键而多出一棵假的第二路。 +func TestAdminFlowGraph_SingleLaneHasOneEntry(t *testing.T) { + env := newFlowTestEnv(t, func(string) bool { return true }) + env.runFinishedRound(t, "web") + + resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/flow/graph?sessionId=web", nil, env.token(t, auth.RoleOwner)) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + lanes := flowLanesData(t, body) + if len(lanes) != 1 { + t.Fatalf("单路的 lanes 条数 = %d，want 1：%s", len(lanes), body) + } + var turnID string + _ = json.Unmarshal(lanes[0]["turnId"], &turnID) + if turnID != "" { + t.Fatalf("单路的 lanes[0].turnId = %q，want 空（没有路标识 ⇒ 客户端按会话那一张画）", turnID) + } +} + +// TestAdminFlowGraph_LanesEmptyOnEmptySession 钉住空态：该会话还没有图时 `lanes` +// 是空数组（恒非 null）—— 与 `graph:null` 同一诚实口径，客户端不必判空。 +func TestAdminFlowGraph_LanesEmptyOnEmptySession(t *testing.T) { + env := newFlowTestEnv(t, func(string) bool { return true }) + resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/flow/graph?sessionId=sess-backthink", nil, env.token(t, auth.RoleOwner)) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + lanes := flowLanesData(t, body) + if lanes == nil { + t.Fatalf("`lanes` 键必须恒在且是数组（空时是 []）：%s", body) + } + if len(lanes) != 0 { + t.Fatalf("没有图时会话的 lanes = %s，want []", body) + } +} + +// TestFlowGraphDispatchLaneFields 钉住**帧载荷**上的两个新键与它们的缺席语义： +// 有路时 `turnId` 在场且与图 id 同源，无路时该键**缺席**（老后端/单路的兼容口径）， +// `round` 则每次建图都在（客户端据此只保留最新一轮的泳道）。 +func TestFlowGraphDispatchLaneFields(t *testing.T) { + laneGraph := trace.NewRunGraph("web") + laneGraph.SetTurnID("turn-a") + laneGraph.SetSeq(3) + laneGraph.Start(1, time.Now()) + laneGraph.SeedInput("标记A") + raw, err := json.Marshal(laneGraph.Snapshot()) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var got map[string]json.RawMessage + if err := json.Unmarshal(raw, &got); err != nil { + t.Fatalf("decode: %v", err) + } + var id, turnID string + var round int + _ = json.Unmarshal(got["id"], &id) + if _, ok := got["turnId"]; !ok { + t.Fatalf("有路的帧缺 `turnId` 键（客户端分不出两条泳道）：%s", raw) + } + _ = json.Unmarshal(got["turnId"], &turnID) + _ = json.Unmarshal(got["round"], &round) + if id != "web-3#turn-a" || turnID != "turn-a" { + t.Fatalf("id/turnId = %q/%q，want web-3#turn-a / turn-a（同源）", id, turnID) + } + if round != 1 { + t.Fatalf("round = %d，want 1（轮次号与会话级序号是两个量）", round) + } + + // 无路的图：`turnId` 必须**缺席**（omitempty），客户端据此回落到「会话那一张」。 + plain := trace.NewRunGraph("web") + plain.Start(4, time.Now()) + plainRaw, err := json.Marshal(plain.Snapshot()) + if err != nil { + t.Fatalf("marshal plain: %v", err) + } + var plainGot map[string]json.RawMessage + if err := json.Unmarshal(plainRaw, &plainGot); err != nil { + t.Fatalf("decode plain: %v", err) + } + if _, ok := plainGot["turnId"]; ok { + t.Fatalf("单路的帧不该带 `turnId` 键：%s", plainRaw) + } + var plainRound int + _ = json.Unmarshal(plainGot["round"], &plainRound) + if plainRound != 4 { + t.Fatalf("单路 round = %d，want 4", plainRound) + } +} + +// TestFlowGraphSourceLanesIsReadOnly 钉住 `lanes[]` 的只读性：补拉快照**不会**发布 +// 任何新帧、也不会改任何路的图（与 LastGraph 同一纪律）。 +func TestFlowGraphSourceLanesIsReadOnly(t *testing.T) { + published := 0 + pub := loop.NewFlowPublisher(func(string, any) { published++ }, func(string) bool { return true }) + pub.StartRound(loop.FlowOwnerKey("web", "turn-a"), "标记A") + before := published + v := pub.LatestLanes("web") + if len(v) == 0 { + t.Fatal("LatestLanes 为空，但该会话刚起了一路") + } + if published != before { + t.Fatalf("LatestLanes 发布了 %d 帧新图（只读查询绝不能造帧）", published-before) + } + // 快照是深拷贝：改它不能碰到发布器内部的图。 + v[0].Nodes = append(v[0].Nodes, trace.FlowNode{ID: "ghost", Kind: "input"}) + again := pub.LatestLanes("web") + for _, g := range again { + for _, n := range g.Nodes { + if n.ID == "ghost" { + t.Fatal("读者追加的节点漏进了发布器内部的图 ⇒ 不是深拷贝") + } + } + } +} + +// 保证 flowTestEnv 的图形源仍实现 FlowGraphSource（v0.77 多了 LatestLanes）。 +var _ FlowGraphSource = (*loop.FlowPublisher)(nil) diff --git a/internal/services/gateway/contracts.go b/internal/services/gateway/contracts.go index b1f3cc3..e07d15d 100644 --- a/internal/services/gateway/contracts.go +++ b/internal/services/gateway/contracts.go @@ -293,10 +293,10 @@ type ChannelEventStore interface { // 组合根把**同一个**发布器既交给 loop/orchestration（写入面：每帧 span // 发布 `ws.graph_update`）也交给网关（本查询面）。 // -// 它刻意只声明**一个读方法**：发布、建图、轮次计数与发布会话白名单都留在 -// 发布器自己手里，管理查询面拿不到、也不该拿到它们（最小权限）——因此 -// 只读查询**不可能**改变任何会话的实时图，也不可能把面板看不见的会话 -// 变成可查的（白名单外的会话从未建图，`LastGraph` 自然回 nil）。 +// 它刻意只声明**两个只读方法**（v0.77 起是两个：`LastGraph` + `LatestLanes`）： +// 发布、建图、轮次计数与发布会话白名单都留在发布器自己手里，管理查询面拿不到、 +// 也不该拿到它们（最小权限）——因此只读查询**不可能**改变任何会话的实时图， +// 也不可能把面板看不见的会话变成可查的（白名单外的会话从未建图，两者自然回空）。 // // 为 nil 时端点诚实作答 500 `flow_unavailable`，绝不回 `graph:null` // （那会把「图源没接线」伪装成「这一轮确实没有图」—— @@ -305,6 +305,14 @@ type FlowGraphSource interface { // LastGraph 返回 sessionID 最近一轮图的一份只读快照；该会话还没有图 // （从未跑过带 span 的轮次，或本就不在发布会话范围内）时返回 nil。 LastGraph(sessionID string) *trace.FlowGraph + // LatestLanes 返回 sessionID **最近一轮每一路**的图快照（契约 v0.77 的 + // `lanes[]`），按路的开始时间升序；没有图时返回空切片。 + // + // 真并行时同一会话同一轮有多条路（design/15 §2.5.5），`LastGraph` 只给出 + // 其中最新的一条 —— 管理面板要靠本方法才看得全「两路各自的图」，否则 + // 后起的那一路会把前一路从面板上顶掉（design/15 §2.5.10-1）。它同样是 + // **只读**的：不发布、不改图、不动轮次计数。 + LatestLanes(sessionID string) []*trace.FlowGraph } // FlowGraphStore 是**已落盘轮次图**的只读出口（契约 v0.73 的 diff --git a/internal/services/gateway/flow_update_ws_test.go b/internal/services/gateway/flow_update_ws_test.go index 12b26a8..fb5ea52 100644 --- a/internal/services/gateway/flow_update_ws_test.go +++ b/internal/services/gateway/flow_update_ws_test.go @@ -94,9 +94,14 @@ func TestFlowGraphUpdateWireFields(t *testing.T) { } // assertGraphKeys 钉住 parseGraph/Flow.tsx 所读的 FlowGraph 顶层键。 +// +// v0.77 起多了两个键（纯增量）：`round`（会话轮次号，客户端据此只保留最新一轮的 +// 泳道）与 `turnId`（这张图所属**路**的权威回合标识，真并行时两路各一个值）。 +// 前者在每次建图时必有；后者只在「这条路有权威标识」时出现（omitempty）， +// 因此本断言只钉它在场时可读，缺席语义由 TestFlowGraphDispatchLaneFields 钉住。 func assertGraphKeys(t *testing.T, graph map[string]json.RawMessage) { t.Helper() - want := []string{"id", "nodes", "edges", "sessionId", "startedAt", "endedAt"} + want := []string{"id", "nodes", "edges", "sessionId", "round", "startedAt", "endedAt"} if got := sortedKeys(graph); !equalKeys(got, want) { t.Fatalf("graph keys = %v, want %v", got, want) }