diff --git a/packages/webui/server/lib/mcode-acp.js b/packages/webui/server/lib/mcode-acp.js index 6df89a2c..02ea784e 100644 --- a/packages/webui/server/lib/mcode-acp.js +++ b/packages/webui/server/lib/mcode-acp.js @@ -5,7 +5,15 @@ import { McodeAcpClient } from "../../acp.mjs"; import { DEFAULT_WORKSPACE, DEFAULT_MODEL, PROMPT_IDLE_TIMEOUT_MS } from "./config.js"; import { createIdleWatchdog } from "./idle-watchdog.js"; import { streamUpdateLine } from "./chat-line.js"; -import { bindDraftToMcodeSid, computeContextPercent } from "./sessions.js"; +import { + createRunChat, + runChatLinesFor, +} from "./state-bus.js"; +import { + bindDraftToMcodeSid, + bindRecordToMcodeSid, + computeContextPercent, +} from "./sessions.js"; import { setActiveChild, clearActiveChild, @@ -166,6 +174,20 @@ export async function runMcodeAcp(content, opts = {}) { const existingSid = opts.sessionId || null; const cs = opts.cs; const cid = opts.cid; + // session-isolation/02 (run-mirror): the webui record this turn + // belongs to, captured BEFORE any await. Mid-run the user can switch + // sessions, which re-points the live `cs` (sessionId / mcodeSessionId + // / chat) at ANOTHER record — every cs-mutation downstream (draft + // promotion, finalize's sid binding and title write-back) must be + // gated on "the user is still looking at the session that ran", and + // the owning record is addressed by this id instead when they did + // not. `handleSend` passes the id it captured right after creating + // the turn's draft record; the cs fallback covers direct callers. + const owningWebuiSessionId = + (typeof opts.owningWebuiSessionId === "string" && + opts.owningWebuiSessionId) || + (cs && cs.sessionId) || + null; // Uploaded files, already validated to be inside UPLOAD_DIR by the route. const attachments = Array.isArray(opts.attachments) ? opts.attachments : []; const workspace = @@ -236,9 +258,23 @@ export async function runMcodeAcp(content, opts = {}) { // 等到 finalize。之前长任务全程草稿是 uuid 孤儿 —— sidebar 同时显示 // uuid 草稿和 mvs_ 引擎条目两条;此时点 mvs_ 条目会走 new_from_mcode // 建壳,把同一对话永久分裂成两条记录(审计日志实锤)。幂等。 + // + // session-isolation/02 (run-mirror): bindDraftToMcodeSid mutates `cs` + // AND renames/merges records through cs.sessionId — both are only + // correct while the user is still viewing the session that ran. A + // mid-run switch re-points cs at the OTHER session; promoting + // through it would rename that session's record or merge its chat. + // Still viewing → cs path as before; switched away → the same + // promotion targeted at the OWNING record by id (cs untouched). + const stillViewingAtBind = + !owningWebuiSessionId || cs.sessionId === owningWebuiSessionId; if (sid) { try { - bindDraftToMcodeSid(cs, sid); + if (stillViewingAtBind) { + bindDraftToMcodeSid(cs, sid); + } else { + bindRecordToMcodeSid(owningWebuiSessionId, sid); + } } catch (e) { console.warn(`[webui] bindDraftToMcodeSid: ${e.message}`); } @@ -253,7 +289,16 @@ export async function runMcodeAcp(content, opts = {}) { // failed session/load fell back to a fresh engine session above). updateRunSid(cid, sid); } - return await streamAcpPrompt(client, sid, content, label, cs, cid, attachments); + return await streamAcpPrompt( + client, + sid, + content, + label, + cs, + cid, + attachments, + owningWebuiSessionId, + ); } catch (e) { // v2.0 (lease B02): §AP5 — surface subprocess start / session // failures on the anomaly channel instead of swallowing them @@ -337,12 +382,16 @@ export function applyConfigOptionUpdate(cs, update) { export function applyToolUpdate(r, cs, update) { const u = update || {}; if (!r.toolIndexById) r.toolIndexById = new Map(); + // session-isolation/02: route tool-update writes into the runChat + // buffer (not cs.chat), so the viewing-session sees no cross- + // contamination when the user switches mid-run. + const chat = r && typeof r.chatArray === "function" ? r.chatArray() : cs.chat; let insertAfter = r.toolIndexById.get(u.toolCallId); if (insertAfter == null) { const name = u.title || u.name || u.toolName || "tool"; - cs.chat = [...cs.chat, `→ ${name}`]; - insertAfter = cs.chat.length - 1; + chat.push(`→ ${name}`); + insertAfter = chat.length - 1; r.toolIndexById.set(u.toolCallId, insertAfter); } @@ -376,11 +425,19 @@ export function applyToolUpdate(r, cs, update) { ` ! ${typeof u.error === "string" ? u.error : u.error.message || JSON.stringify(u.error)}`, ); - cs.chat = [ - ...cs.chat.slice(0, insertAfter + 1), - ...newLines, - ...cs.chat.slice(insertAfter + 1), - ]; + // Insert newLines into the chat array (runChat buffer when in a + // turn, cs.chat otherwise). Match the splice-with-index-shift that + // would happen on a regular array mutation. + const before = chat.slice(0, insertAfter + 1); + const after = chat.slice(insertAfter + 1); + for (let i = 0; i < before.length; i += 1) chat[i] = before[i]; + for (let j = 0; j < newLines.length; j += 1) { + chat[before.length + j] = newLines[j]; + } + for (let k = 0; k < after.length; k += 1) { + chat[before.length + newLines.length + k] = after[k]; + } + chat.length = before.length + newLines.length + after.length; if (r.toolIndexById) { for (const [k, v] of r.toolIndexById) { if (v > insertAfter) r.toolIndexById.set(k, v + newLines.length); @@ -391,7 +448,16 @@ export function applyToolUpdate(r, cs, update) { // streamAcpPrompt — like collectExecResult, but the event source is // the acp client's prompt callback rather than a child-process stdout // stream. -function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) { +function streamAcpPrompt( + client, + sid, + content, + label, + cs, + cid, + attachments = [], + owningWebuiSessionId = null, +) { return new Promise((resolve) => { const r = { answer: null, @@ -413,6 +479,32 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) // starts with lastChunkKind === null and the first chunk of // any kind triggers a clean accumulator. lastChunkKind: null, + // session-isolation/02 (run-mirror): the owning session id for + // this turn, captured at entry — the ENGINE sid the turn runs on + // (identical to `cs.mcodeSessionId` once the draft is bound). + // Every stream write (▲ / ● / tool_call / tool_update / plan / + // empty-turn note) lands in the runChat buffer keyed by this id, + // NEVER directly in cs.chat: mid-run the user can switch sessions + // and cs.chat then belongs to whichever session they opened. The + // wire snapshots re-attach the buffer for the owning view + // (state-bus.snapshotViewFields) and the route's finalize drain + // flushes the buffer to the owning session's view-or-record. + owningSessionId: sid, + // session-isolation/02 (run-mirror): the webui record id this + // turn belongs to, captured by runMcodeAcp before its first + // await. Finalize consults this to detect "the user switched + // away mid-run" before any cs mutation. + owningWebuiSessionId, + // `chatArray()` — the write target for every stream line. ALWAYS + // the runChat buffer while the turn's buffer exists (created just + // below, before the first engine event can arrive); cs.chat is + // only a fallback for calls outside a buffered turn (unit tests, + // non-turn helpers). View delivery is the snapshot's job, not the + // write target's. + chatArray() { + const m = runChatLinesFor(cid, sid); + return m !== null ? m : cs.chat; + }, }; const t0 = Date.now(); cs.running = { @@ -428,6 +520,14 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) cs.context.thinkingStatus = "Running"; setActiveChild(cid, client); pushStateFor(cid); + // session-isolation/02 (run-mirror): create the per-(cid, + // owning-session) buffer that captures every stream write + // during this turn. Lines DO NOT go to cs.chat directly — the + // viewing session might be a different one (the user may have + // switched mid-run). The buffer is drained at finalize back + // into either cs.chat (same session still viewing) or the + // owning session's persisted record (user switched away). + createRunChat(cid, sid, []); // Idle watchdog — every stream event (thought / message / tool_call / // tool_update / usage / other) refreshes cs.running.lastDeltaAt, // so a long but healthy turn never trips this; only a silent @@ -473,13 +573,12 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) // upstream (no per-turn history replay), and we mirror that scope here. if ( typeof r.durationMs === "number" && - r.durationMs > 0 && - Array.isArray(cs.chat) + r.durationMs > 0 ) { - cs.chat = [ - ...cs.chat, - `§§ processed_duration=${Math.round(r.durationMs)}ms`, - ]; + const chatTarget = r && typeof r.chatArray === "function" ? r.chatArray() : cs.chat; + if (Array.isArray(chatTarget)) { + chatTarget.push(`§§ processed_duration=${Math.round(r.durationMs)}ms`); + } } clearActiveChild(cid); cs.running = { @@ -497,12 +596,14 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) // Strip the streaming cursor ▍ from every line — streamUpdateLine // adds it on every push, finalize must clear it or the thinking // / answer lines stay marked as streaming forever. - if (Array.isArray(cs.chat)) { - cs.chat = cs.chat.map((line) => - typeof line === "string" && line.endsWith(" ▍") - ? line.slice(0, -2) - : line, - ); + const cursorTarget = r && typeof r.chatArray === "function" ? r.chatArray() : cs.chat; + if (Array.isArray(cursorTarget)) { + for (let i = 0; i < cursorTarget.length; i += 1) { + const line = cursorTarget[i]; + if (typeof line === "string" && line.endsWith(" ▍")) { + cursorTarget[i] = line.slice(0, -2); + } + } } if (r.usage) { cs.context.tokens = @@ -554,7 +655,21 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) `[finalize.usage] cid=${cid} r.usage=${JSON.stringify(r.usage)} r.answerLen=${(r.answer || "").length} r.thinkingLen=${(r.thinking || "").length}`, ); } - if (r.sessionId) cs.mcodeSessionId = r.sessionId; + // session-isolation/02 (run-mirror): only re-point the VIEWED + // session's engine binding when the user is still looking at the + // session that ran. A mid-run switch left cs bound to the OTHER + // session — writing r.sessionId over it would redirect that + // session's next turn onto this turn's engine conversation. + // The still-viewing test covers both id forms: the pre-promotion + // draft id (owningWebuiSessionId) and the post-promotion engine + // id the record was renamed to at bind time. + const stillViewingAtFinalize = + !owningWebuiSessionId || + cs.sessionId === owningWebuiSessionId || + (r.sessionId != null && cs.sessionId === r.sessionId); + if (r.sessionId && stillViewingAtFinalize) { + cs.mcodeSessionId = r.sessionId; + } // Fire-and-forget — the mavis hook writes a // local_runtime_token_usage row after acp completes. Wait ~400ms // for it to land, then query the db for the real numbers. @@ -601,16 +716,30 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) // v0.5.bx: prompt 完成后用 mcodeSessionId 反查 mcode 真实 title if (r.sessionId) { const finalSid = r.sessionId; + // session-isolation/02 (run-mirror): binding and title belong to + // the record that RAN, addressed by id — cs.sessionId is only + // the right target while the user still views the owning + // session (both id forms count; see stillViewingAtFinalize). + // After a mid-run switch, writing through cs would stamp this + // turn's engine sid (and title) onto the session the user + // switched TO. + const bindTargetId = stillViewingAtFinalize + ? cs.sessionId + : owningWebuiSessionId; getMcodeSessionTitle(finalSid) .then((title) => { // qa (两条记录): mcodeSessionId 绑定与 title 查询解耦 —— 之前 // `if (!title) return` 提前退出会连绑定一起跳过,titleCustom // 守卫也曾把绑定一并挡住(该守卫只应保护标题本身)。绑定 // 无条件写入。 - if (cs.sessionId) { + if (bindTargetId) { try { const all = loadSessions(); - const item = all.find((s) => s.id === cs.sessionId); + // The owning record may have been promoted mid-run — its + // id is then the engine sid, not the captured webui id. + const item = + all.find((s) => s && s.id === bindTargetId) || + all.find((s) => s && s.mcodeSessionId === finalSid); if (item && item.mcodeSessionId !== finalSid) { item.mcodeSessionId = finalSid; item.updatedAt = Date.now(); @@ -628,24 +757,35 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) cs.sessionTitle === "Untitled"; if (isDefault && cs.mcodeSessionId === finalSid) { cs.sessionTitle = title; - // 同步到 webui session db(写 title,让 sidebar 能 1:1 找回来) - if (cs.sessionId) { - try { - const all = loadSessions(); - const item = all.find((s) => s.id === cs.sessionId); - // qa (session-workspace-crud): titleCustom 是用户显式改名 - // (POST /api/sessions/rename) 的留痕 — 自动标题永不覆盖 - // 用户标题。isDefault 的 cs 侧判定之外再守一道 item 侧, - // 封住"改名发生在 title RPC 在途时"的竞态窗口。 - if (item && !item.titleCustom) { - item.title = title; - item.updatedAt = Date.now(); - saveSessions(all); - } - } catch (e) { - console.warn(`[bx] save title failed: ${e.message}`); + } + // The record's title follows the conversation (the session + // that ran), regardless of which session is on screen; the + // cs-side mirror above only applies while still viewing. + // qa (session-workspace-crud): titleCustom 是用户显式改名 + // (POST /api/sessions/rename) 的留痕 — 自动标题永不覆盖 + // 用户标题。isDefault 的 cs 侧判定之外再守一道 item 侧, + // 封住"改名发生在 title RPC 在途时"的竞态窗口。 + try { + const all = loadSessions(); + const item = + all.find((s) => s && s.id === bindTargetId) || + all.find((s) => s && s.mcodeSessionId === finalSid); + if (item && !item.titleCustom && item.title !== title) { + const recordIsDefault = + !item.title || + item.title === "New session" || + item.title === "Untitled" || + item.title === "Mcode session"; + if (recordIsDefault) { + item.title = title; + item.updatedAt = Date.now(); + saveSessions(all); } } + } catch (e) { + console.warn(`[bx] save title failed: ${e.message}`); + } + if (stillViewingAtFinalize) { pushStateFor(cid); } }) @@ -720,23 +860,27 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) r.thinking += c.text; r.lastChunkKind = "thought"; const oneLine = r.thinking.replace(/\n+/g, " ").trim(); - streamUpdateLine(cs.chat, "▲", oneLine); + streamUpdateLine(r.chatArray(), "▲", oneLine); } else if (c.kind === "message" && typeof c.text === "string") { if (r.lastChunkKind !== "message") r.answer = ""; r.answer += c.text; r.lastChunkKind = "message"; const oneLine = r.answer.replace(/\n+/g, " ").trim(); - streamUpdateLine(cs.chat, "●", oneLine); + streamUpdateLine(r.chatArray(), "●", oneLine); } else if (c.kind === "tool_call" && c.update) { // v0.5.bs: 工具调用开始 — 写 `→ toolName` 行到 chat const u = c.update; const name = u.title || u.name || u.toolName || "tool"; const input = u.rawInput ? JSON.stringify(u.rawInput) : ""; const line = input ? `→ ${name} ${input}` : `→ ${name}`; - cs.chat = [...cs.chat, line]; + // session-isolation/02: route into the runChat buffer (not + // cs.chat) when in a turn. The viewing-session sees no + // cross-contamination when the user switches mid-run. + const tcChat = r && typeof r.chatArray === "function" ? r.chatArray() : cs.chat; + tcChat.push(line); // 记下这行在 chat 里的位置(之后 tool_update 用来在它后面插输出) if (!r.toolIndexById) r.toolIndexById = new Map(); - r.toolIndexById.set(u.toolCallId, cs.chat.length - 1); + r.toolIndexById.set(u.toolCallId, tcChat.length - 1); // session-isolation/06: tool_call (and tool_update, // plan_update, error, anything else) breaks the same-prefix // chain. Without this update, the next message chunk would @@ -833,7 +977,10 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = []) // v2.3: 思考链超长回合(思维耗尽输出预算)以无正文结束 — 界面上 // 表现为"思考戛然而止"。落一条 system 提示行说明结局与续法。 const note = buildEmptyTurnNote(r.stopReason, r.answer); - if (note) cs.chat = [...(cs.chat || []), note]; + if (note) { + const noteChat = r && typeof r.chatArray === "function" ? r.chatArray() : cs.chat; + noteChat.push(note); + } finalize(); }) .catch((e) => { diff --git a/packages/webui/server/lib/sessions.js b/packages/webui/server/lib/sessions.js index c46ad0e9..c4e03a33 100644 --- a/packages/webui/server/lib/sessions.js +++ b/packages/webui/server/lib/sessions.js @@ -145,6 +145,14 @@ export function promoteDraftToMcodeSid(cs) { * finalize may call it again. Leaving the draft unbound for the whole turn * makes the sidebar show two records for one conversation (uuid draft + the * mvs_ engine entry), and clicking the latter forks it into two. + * + * session-isolation/02 (run-mirror): callers capture the turn's owning + * webui session id at send time. When the user has already switched to + * another session by the time the engine session id is known, `cs` no + * longer points at the owning record — promoting through `cs` would + * rename/merge whichever record the user switched TO. Use + * `bindRecordToMcodeSid` for that case; this one stays the cs-driven + * path for the still-viewing case. */ export function bindDraftToMcodeSid(cs, sid) { if (!cs || !sid) return false; @@ -152,6 +160,53 @@ export function bindDraftToMcodeSid(cs, sid) { return promoteDraftToMcodeSid(cs); } +/** + * session-isolation/02 (run-mirror): `promoteDraftToMcodeSid` for a + * record addressed BY ID, without touching `cs`. + * + * Mid-run the live `cs` can belong to a different session (the user + * switched away while the engine session id was still unknown). The + * turn's draft record must still be promoted — engine identity bound, + * id rewritten to the mvs id (or merged into an existing overlay) — + * exactly what `promoteDraftToMcodeSid` does, but targeted at the + * owning record so the switched-to session's record is never renamed + * or merged by someone else's turn. + * + * Idempotent: a record already carrying `sid` (or an already-promoted + * record that no longer matches `webuiId`) is left alone. + * + * @returns {string|null} the owning record's id after promotion + * (null when there was nothing to bind — record gone, or already + * bound to another engine session). + */ +export function bindRecordToMcodeSid(webuiId, sid) { + if (!webuiId || !sid) return null; + const all = loadSessions(); + // Already promoted for this turn? (webuiId may BE the mvs id after a + // previous promotion, or the record may carry the binding already.) + const bound = findOverlayForMcodeSid(all, sid); + if (bound) return bound.id; + const draft = all.find((s) => s && s.id === webuiId && !s.mcodeSessionId); + if (!draft) return null; + const existing = findOverlayForMcodeSid(all, sid); + if (existing) { + const draftChat = Array.isArray(draft.chat) ? draft.chat : []; + if (draftChat.length > 0) { + existing.chat = [...(existing.chat || []), ...draftChat]; + } + existing.updatedAt = Date.now(); + const idx = all.indexOf(draft); + if (idx >= 0) all.splice(idx, 1); + saveSessions(all); + return existing.id; + } + draft.id = sid; + draft.mcodeSessionId = sid; + draft.updatedAt = Date.now(); + saveSessions(all); + return draft.id; +} + // Memoize parsed content by (mtimeMs, size). pushStateFor calls // loadSessions on EVERY snapshot (per SSE push, up to 60Hz), and the // switch/persist paths read too — re-reading + JSON.parsing a @@ -305,6 +360,33 @@ export function persistCurrentChat(cs) { saveSessions(all); } +/** + * session-isolation/02 (run-mirror): append finished-turn lines to a + * session's PERSISTED record, addressed by id or engine session id. + * + * Used by the finalize drain when the user switched away mid-run: the + * live `cs` belongs to whichever session the user is looking at now, + * so `persistCurrentChat(cs)` would persist the wrong view. The turn's + * lines belong to the session that RAN, so they are written to that + * session's record directly — found by webui id, or by mcodeSessionId + * when the record was promoted mid-run (its id is then the mvs id). + * + * No-op when there is nothing to append or no record matches (the + * record was deleted mid-run — nothing sensible to resurrect). + */ +export function appendChatToSession(sessionId, lines) { + if (!sessionId || !Array.isArray(lines) || lines.length === 0) return false; + const all = loadSessions(); + const item = + all.find((s) => s && s.id === sessionId) || + findOverlayForMcodeSid(all, sessionId); + if (!item) return false; + item.chat = [...(item.chat || []), ...lines]; + item.updatedAt = Date.now(); + saveSessions(all); + return true; +} + // Boot-time cleanup of empty / default-titled session entries (the // residue of "+ New session" presses that never sent a message). // Keep entries that have chat OR a real (non-default) title; for diff --git a/packages/webui/server/lib/state-bus.js b/packages/webui/server/lib/state-bus.js index 96907c00..7e7ff270 100644 --- a/packages/webui/server/lib/state-bus.js +++ b/packages/webui/server/lib/state-bus.js @@ -230,6 +230,10 @@ function ensureMcodeSessionsFetchedAndPush(workspace) { // 直接 loadSessions(),把每个 session 的完整 chat 数组推进 SSE, // 是 v2.3 修掉的主负载;两处(本处 + pushOnlineCount)漏改。 sessions: sessionsListForSnapshot(), + // session-isolation/02 (run-mirror): re-attach the live run's + // buffered lines when the viewed session owns the run, and scope + // the run indicator to the viewed session otherwise. + ...snapshotViewFields(c, ccs), mcodeSessions: cached, mcodeSessionsPending: false, availableCommands: getCachedMcodeCommands(), @@ -309,6 +313,11 @@ export function pushStateFor(cid, opts = {}) { const snapshot = { ...cs, sessions: sessionsListForSnapshot(), + // session-isolation/02 (run-mirror): re-attach the live run's + // buffered lines when the viewed session owns the run (live view), + // and scope the run indicator to the viewed session otherwise (a + // foreign turn never claims "thinking" in this view). + ...snapshotViewFields(cid, cs), ...fields, availableCommands: cachedCmds, onlineCount: sseByCid.size, @@ -548,6 +557,10 @@ export function pushOnlineCount(lanBroadcast) { // qa (session-workspace-crud): 同上 — 复用瘦身投影,别把 chat 数组 // 随 onlineCount 广播出去。 sessions: sessionsListForSnapshot(), + // session-isolation/02 (run-mirror): same view contract as every + // other snapshot builder — a broadcast landing mid-run must not + // flash a foreign turn's lines (or its run claim) into this view. + ...snapshotViewFields(c, cs), ...mcodeSessionsSnapshotFields((cs.workspace && cs.workspace.dir) || ""), availableCommands: cachedCmds, onlineCount: sseByCid.size, @@ -639,6 +652,208 @@ export function beginRun(cid, sid) { return { ok: true }; } +// session-isolation/02 (run-mirror port): per-(cid, owning-session) +// line buffer that captures every stream write during a single +// turn. The buffer is independent of `cs.chat` (the VIEWING session's +// chat) — a switch to another session while T1 is streaming keeps T1's +// lines landing here rather than spilling into T2's view. At finalize, +// the buffer is drained back into either `cs.chat` (if the user is +// still viewing the owning session) or the owning session's persisted +// record (if the user switched away mid-run). The buffer is +// intentionally simple — keyed by session id, not by run id — +// because session-id is the natural boundary for "where does this +// turn's content live". +const runChatByCid = new Map(); // cid -> Map + +/** + * Create the run-time chat buffer for (cid, sessionId). Lines written + * by the engine during this turn land here instead of in `cs.chat`. + * + * Replaces any existing buffer for this cid: `beginRun` allows at most + * ONE live turn per cid, so the previous entry is either the same turn + * re-created before its first write (handleSend seeds the buffer at + * claim time, streamAcpPrompt re-creates it at stream start) or a stale + * empty entry from a turn whose `session/load` failed and fell back to + * a fresh engine session (the buffer re-keys with the new sid — the old + * key never received a line and must not linger). Replacing before the + * first engine write loses nothing. + */ +export function createRunChat(cid, sessionId, baseLines = []) { + if (!cid || !sessionId) return; + runChatByCid.set(cid, new Map([ + [sessionId, { + chat: Array.isArray(baseLines) ? [...baseLines] : [], + }], + ])); +} + +/** Append one line to the run-time buffer. No-op if the buffer does + * not exist (e.g. caller is not in a turn). */ +export function appendRunChatLine(cid, sessionId, line) { + if (!cid || !sessionId) return; + const m = runChatByCid.get(cid); + if (!m) return; + const entry = m.get(sessionId); + if (!entry) return; + entry.chat.push(line); +} + +/** Read-only access to the buffer's chat array. Returns null if no + * buffer exists for this (cid, sessionId). The reference is shared + * with the buffer — callers must not mutate. */ +export function runChatLinesFor(cid, sessionId) { + const m = runChatByCid.get(cid); + if (!m) return null; + const entry = m.get(sessionId); + return entry ? entry.chat : null; +} + +/** Remove the buffer and return its lines (or null). Caller is now + * responsible for appending those lines to whatever destination + * (cs.chat, the owning session's persisted record, etc.). */ +export function drainRunChat(cid, sessionId) { + const m = runChatByCid.get(cid); + if (!m) return null; + const entry = m.get(sessionId); + if (!entry) return null; + const lines = entry.chat.slice(); + m.delete(sessionId); + if (m.size === 0) runChatByCid.delete(cid); + return lines; +} + +/** Convenience for finalize + drain + safe-cleanup in one call. */ +export function hasRunChat(cid, sessionId) { + const m = runChatByCid.get(cid); + return m !== undefined && m.has(sessionId); +} + +/** Remove the buffer without reading it. Safe to call when nothing + * is held (no-op). */ +export function removeRunChat(cid, sessionId) { + const m = runChatByCid.get(cid); + if (!m) return; + m.delete(sessionId); + if (m.size === 0) runChatByCid.delete(cid); +} + +// ============================================================ +// View routing (run-mirror) — what a snapshot shows for the session +// the user is LOOKING at. +// +// While a turn runs, the engine's lines accumulate in the runChat +// buffer (keyed by the OWNING session's engine id), never in +// `cs.chat` — `cs.chat` mirrors whichever session the user has open, +// and that can change at any moment mid-run. The wire snapshots must +// therefore re-attach the buffer at push time: +// +// viewed session OWNS the live run → chat = cs.chat + buffer +// (the live view; lines stream in exactly as they did before +// run-mirror, they just travel via the buffer); +// +// viewed session is a DIFFERENT session → chat stays cs.chat (its +// own record), and the run indicator fields are scoped to idle so +// the foreign turn never claims "thinking" in this view. +// +// Everything goes through `snapshotViewFields` so every snapshot +// builder (SSE push, authoritative mcode-sessions push, online-count +// broadcast, /api/state, the SSE first frame, the switch response) +// renders the same view contract. +// ============================================================ + +// Idle `running` shape — byte-mirrored from makeClientState() and the +// runners' finalize(). Frozen: it is shared across snapshots and must +// never be mutated into. +const IDLE_RUNNING_VIEW = Object.freeze({ + active: false, + prompt: null, + pid: null, + startedAt: null, + model: null, + sessionId: null, + lastDeltaAt: null, + tps: 0, +}); + +/** True when the session `cs` currently displays OWNS this cid's live + * run (the run registry carries the turn's engine sid, and the viewed + * session is bound to that same engine sid). */ +export function viewOwnsLiveRun(cid, cs) { + const run = runsByCid.get(cid || "default"); + if (!run || !run.sid) return false; + return !!cs && cs.mcodeSessionId === run.sid; +} + +/** The chat array a snapshot should carry for `cs`: the viewed + * session's own chat, plus the live run's buffered lines when the + * viewed session owns the run. Returns the base array untouched (same + * reference) when there is nothing to merge. */ +export function runChatViewChat(cid, cs) { + const base = Array.isArray(cs && cs.chat) ? cs.chat : []; + if (!viewOwnsLiveRun(cid, cs)) return base; + const buf = runChatLinesFor(cid || "default", cs.mcodeSessionId); + if (!buf || buf.length === 0) return base; + return [...base, ...buf]; +} + +/** + * View-scoped snapshot overrides for `cid`/`cs`, or {} when the plain + * client state already renders correctly. Spread AFTER `...cs`: + * + * - `chat` — cs.chat with the live buffer re-attached (owning view); + * - `running` / `context` — idle-shaped when a turn is live on this + * cid but the viewed session is a different one (T2 stays visually + * idle while T1 streams elsewhere; ticket 02 acceptance #3). + */ +export function snapshotViewFields(cid, cs) { + const key = cid || "default"; + const chat = runChatViewChat(cid, cs); + const chatChanged = chat !== (cs && cs.chat); + if (viewOwnsLiveRun(cid, cs)) { + const fields = chatChanged ? { chat } : {}; + // The user is looking at the session that runs — including the case + // "switched away and back mid-run": a switch's resetContext healed + // cs.running to idle, but this view's turn is demonstrably live + // (the run registry holds it). Project the live run back onto the + // snapshot so the owning view keeps its running indicator (ticket 02 + // acceptance #3) without mutating cs behind the switch route's back. + const run = runsByCid.get(key); + if (run && run.sid && cs && cs.running && !cs.running.active) { + fields.running = { + active: true, + prompt: "prompt", + pid: null, + startedAt: run.startedAt, + model: (cs.model && cs.model.name) || null, + sessionId: run.sid, + // The stream callback keeps refreshing cs.running.lastDeltaAt / + // tps even after a switch's resetContext — project the live + // values so the diff gate still suppresses identical frames. + lastDeltaAt: cs.running.lastDeltaAt || run.startedAt, + tps: cs.running.tps || 0, + }; + fields.context = { + ...(cs.context || {}), + thinkingStatus: "Running", + }; + } + return fields; + } + const run = runsByCid.get(key); + if (!run || !run.sid) { + // No live run anywhere on this cid — nothing to scope. + return chatChanged ? { chat } : {}; + } + // A turn is live on this cid and the viewed session is NOT the + // owning one: keep this view's own chat and force the run claim off. + const context = cs && cs.context + ? { ...cs.context, thinkingStatus: "Idle", tps: 0 } + : undefined; + const fields = { chat, running: IDLE_RUNNING_VIEW }; + if (context) fields.context = context; + return fields; +} + /** Release a turn claimed by `beginRun`. Safe to call when nothing is held. */ export function endRun(cid) { const key = cid || "default"; diff --git a/packages/webui/server/routes/chat.js b/packages/webui/server/routes/chat.js index 2e976094..d095e30a 100644 --- a/packages/webui/server/routes/chat.js +++ b/packages/webui/server/routes/chat.js @@ -9,8 +9,17 @@ import { saveSessions, persistCurrentChat, promoteDraftToMcodeSid, + appendChatToSession, } from "../lib/sessions.js"; -import { pushStateFor, pushAlert, getActiveChild, beginRun, endRun } from "../lib/state-bus.js"; +import { + pushStateFor, + pushAlert, + getActiveChild, + beginRun, + endRun, + createRunChat, + drainRunChat, +} from "../lib/state-bus.js"; // 2026-09-20 rigor fix (G1 bypass finding): import the lib/slash.js shell, // NOT interaction/commands.js directly. The shell carries the B03 // authorize("slash.clear") gate + write-ahead audit (slash.clear.intent / @@ -171,6 +180,14 @@ export async function handleSend(req, res, ctx) { cs.sessionId = id; } + // session-isolation/02 (run-mirror): the webui record this turn + // belongs to, captured before any await (draft creation above just + // made sure it exists). Mid-run switches re-point cs (sessionId / + // mcodeSessionId / chat) at another record — the finalize drain and + // the engine-side bind/title writes must know where the turn CAME + // FROM, not where the user is looking now. + const owningWebuiSessionId = (cs && cs.sessionId) || null; + // Detect slash commands that we can satisfy without spawning mcode const slashResult = await handleLocalSlash(content, cs, cid); if (slashResult.handled) { @@ -188,6 +205,14 @@ export async function handleSend(req, res, ctx) { console.log( `[send] cid=${cid} content=${JSON.stringify(content.slice(0, 80))} model=${modelToUse} sessionId=${cs.mcodeSessionId} workspace=${(cs && cs.workspace && cs.workspace.dir) || "null"}`, ); + // session-isolation/02 (run-mirror): seed the per-(cid, owning + // session) line buffer right before the engine runs (a no-op while + // mcodeSessionId is still null — the buffer is keyed by the engine + // sid, which `streamAcpPrompt` creates the moment it is known; a + // locally-handled slash command above never leaves a stale buffer + // behind). The engine's stream writes land in this buffer, never + // directly in cs.chat; the finalize drain below flushes it. + createRunChat(cid, cs && cs.mcodeSessionId, []); const t0 = Date.now(); const r = process.env.MCODE_USE_ACP === "0" @@ -208,6 +233,7 @@ export async function handleSend(req, res, ctx) { cs, cid, attachments, + owningWebuiSessionId, }); console.log( `[send] result ${Date.now() - t0}ms:`, @@ -218,24 +244,80 @@ export async function handleSend(req, res, ctx) { sessionId: r.sessionId, }).slice(0, 500), ); + // session-isolation/02 (run-mirror): finalize drain. The turn's + // stream lines accumulated in the runChat buffer keyed by the + // OWNING engine session; the user's `›` line was persisted up front. + // Where the buffer flushes depends on where the user is looking: + // still viewing the owning session → append into cs.chat (the + // live view) — the success-branch ● rewrite below then lands on + // the drained line and persistCurrentChat persists the record; + // switched away mid-run → cs.chat belongs to ANOTHER session — + // never touched. The drained lines (with the final ● text + // patched in) go to the owning session's persisted record via + // appendChatToSession; the final persistCurrentChat(cs) below + // only re-writes the viewed session's own (unchanged) chat. + // stillViewing keys on the engine sid the turn actually ran on + // (r.sessionId — it can differ from the beginRun claim when a stale + // session/load fell back to a fresh engine session), with the + // owning webui record id as the fallback view test for a turn whose + // draft never got bound. + const owningSid = (r && r.sessionId) || cs.mcodeSessionId || null; + const drainedLines = owningSid ? drainRunChat(cid, owningSid) : null; + const stillViewing = + !owningSid || + cs.mcodeSessionId === owningSid || + (owningWebuiSessionId != null && cs.sessionId === owningWebuiSessionId); + // The flushed line list, normalized once: a successful turn's last + // ● line is rewritten to the authoritative answer text — the same + // normalization the still-viewing path has always applied to + // cs.chat — no matter which destination the lines end up in. + const flushDrainedLines = (oneLine) => { + if (!drainedLines || drainedLines.length === 0) return null; + const lines = drainedLines.slice(); + if (oneLine != null) { + let patched = false; + for (let i = lines.length - 1; i >= 0; i--) { + if (typeof lines[i] === "string" && lines[i].startsWith("● ")) { + lines[i] = `● ${oneLine}`; + patched = true; + break; + } + } + if (!patched) lines.push(`● ${oneLine}`); + } + if (stillViewing) { + cs.chat = [...cs.chat, ...lines]; + } else { + try { + appendChatToSession(owningSid, lines); + } catch (e) { + console.warn(`[chat] appendChatToSession failed: ${e.message}`); + } + } + return lines; + }; if (r.status === "succeeded" && r.answer) { // v0.5.bx-4: 流式输出已经在 streamAcpPrompt/streamUpdateLine 里把 ▲ 和 ● 行写进 chat 了 const oneLine = r.answer.replace(/\n+/g, " ").trim(); - let lastAnsIdx = -1; - for (let i = cs.chat.length - 1; i >= 0; i--) { - if (typeof cs.chat[i] === "string" && cs.chat[i].startsWith("● ")) { - lastAnsIdx = i; - break; + const flushed = flushDrainedLines(oneLine); + if (stillViewing) { + let lastAnsIdx = -1; + for (let i = cs.chat.length - 1; i >= 0; i--) { + if (typeof cs.chat[i] === "string" && cs.chat[i].startsWith("● ")) { + lastAnsIdx = i; + break; + } + } + if (lastAnsIdx >= 0) { + cs.chat[lastAnsIdx] = `● ${oneLine}`; + } else { + cs.chat = [...cs.chat, `● ${oneLine}`]; } - } - if (lastAnsIdx >= 0) { - cs.chat[lastAnsIdx] = `● ${oneLine}`; - } else { - cs.chat = [...cs.chat, `● ${oneLine}`]; } cs.context.assistantLast = oneLine; cs.context.assistantAt = Date.now(); - } else if (r.status === "failed" || r.error) { + } else { + if (r.status === "failed" || r.error) { const rawMsg = (r.error?.message || r.status).replace(/\n+/g, " "); let oneLine = rawMsg; let hint = ""; @@ -271,10 +353,20 @@ export async function handleSend(req, res, ctx) { // has already run inside runMcodeAcp/collectExecResult and put // cs into exactly this idle shape. resetThinkingClaim(cs); + } + // Non-success turn (failed, timeout, or succeeded with no answer + // text — e.g. the empty-turn note line): the buffered lines are + // still the owning session's content — flush them exactly like + // the success path, minus the ● normalization. + flushDrainedLines(null); } // v2.4 单一基础会话:回合绑定了 mcode 会话(cs.mcodeSessionId 由 acp // finalize 写入)后,把草稿记录晋升为引擎身份(id → mvs_…),或并入 // 该 mcode 会话既有的叠加记录——保证一次对话在存储里只有一条记录。 + // (session-isolation/02: when the user switched away mid-run, cs + // belongs to the OTHER session; this is a no-op for it — the + // owning record was already promoted at bind time via + // bindRecordToMcodeSid inside runMcodeAcp.) if (cs.mcodeSessionId) { try { promoteDraftToMcodeSid(cs); diff --git a/packages/webui/server/routes/sessions.js b/packages/webui/server/routes/sessions.js index 1130615c..77d6333f 100644 --- a/packages/webui/server/routes/sessions.js +++ b/packages/webui/server/routes/sessions.js @@ -28,7 +28,11 @@ import { import { loadTranscriptChatLines } from "../lib/transcript.js"; import { applyMavisUsageToCs } from "../lib/mavis-usage.js"; import { getMcodeModelLimit } from "../lib/models.js"; -import { pushStateFor, clients } from "../lib/state-bus.js"; +import { + pushStateFor, + clients, + runChatViewChat, +} from "../lib/state-bus.js"; import { MCODE_RUNTIME_DB } from "../lib/config.js"; import { getSessionTree, invalidateSessionTree } from "../lib/session-tree.js"; import { authorize } from "../lib/authorize.js"; @@ -473,7 +477,12 @@ export async function handleSwitchSession(req, res, ctx) { id: target.id, mcodeSessionId: cs.mcodeSessionId, title: cs.sessionTitle, - chat: cs.chat, + // session-isolation/02 (run-mirror): switching back to the + // session that is mid-run must show what it produced so far. + // cs.chat holds the record's lines; the live turn's output is + // still in the runChat buffer — re-attach it for the owning + // view (same contract as every state snapshot). + chat: runChatViewChat(cid, cs), }, }), ); diff --git a/packages/webui/server/routes/state.js b/packages/webui/server/routes/state.js index ea491230..d0151d92 100644 --- a/packages/webui/server/routes/state.js +++ b/packages/webui/server/routes/state.js @@ -7,6 +7,7 @@ import { pushStateFor, pushOnlineCount, mcodeSessionsSnapshotFields, + snapshotViewFields, getSseClient, setSseClient, endSseClient, @@ -44,6 +45,11 @@ export async function handleEvents(req, res, ctx) { const snapshot = { ...cs, sessions: sessionsListForSnapshot(), + // session-isolation/02 (run-mirror): the first frame follows the same + // view contract as the push path — a client connecting mid-run sees + // the owning session's buffered lines (or a clean idle view of the + // session it opened instead). + ...snapshotViewFields(cid, cs), ...mcodeSessionsSnapshotFields((cs.workspace && cs.workspace.dir) || ""), lanBroadcast: getLanBroadcast(), readOnly: getReadOnly(), @@ -113,6 +119,9 @@ export async function handleState(req, res, ctx) { JSON.stringify({ ...cs, sessions: sessionsListForSnapshot(), + // session-isolation/02 (run-mirror): same view contract as the SSE + // push path (see handleEvents). + ...snapshotViewFields(ctx.cid, cs), mcodeSessions, availableCommands: getCachedMcodeCommands(), lanBroadcast: getLanBroadcast(), diff --git a/packages/webui/test/helpers/_setup.js b/packages/webui/test/helpers/_setup.js index 8647ac77..6b1dc8ec 100644 --- a/packages/webui/test/helpers/_setup.js +++ b/packages/webui/test/helpers/_setup.js @@ -285,6 +285,77 @@ export async function setupMocks(t, overrides = {}) { } }, persistCurrentChat: () => {}, + // session-isolation/02 (run-mirror): the buffer-drain finalize path + // (routes/chat.js) writes the turn back to the owning session's + // persisted record; mirror the real lookup (by webui id, then by + // engine sid) against the in-memory store. + appendChatToSession: (sessionId, lines) => { + if (!sessionId || !Array.isArray(lines) || lines.length === 0) { + return false; + } + const item = + _sessionsStore.find((r) => r && r.id === sessionId) || + _sessionsStore.find((r) => r && r.mcodeSessionId === sessionId); + if (!item) return false; + item.chat = [...(item.chat || []), ...lines]; + item.updatedAt = Date.now(); + return true; + }, + // session-isolation/02 (run-mirror): cs-driven draft bind — mirrors + // bindDraftToMcodeSid via the mock's promoteDraftToMcodeSid. + bindDraftToMcodeSid: (cs, sid) => { + if (!cs || !sid) return false; + cs.mcodeSessionId = sid; + // replicate the mock's promoteDraftToMcodeSid body + if (!cs.mcodeSessionId || !cs.sessionId) return false; + if (cs.sessionId === cs.mcodeSessionId) return false; + const draft = _sessionsStore.find( + (r) => r && r.id === cs.sessionId && !r.mcodeSessionId, + ); + const existing = _sessionsStore.find( + (r) => r && r.mcodeSessionId === cs.mcodeSessionId, + ); + if (existing) { + if (draft && Array.isArray(draft.chat) && draft.chat.length) { + existing.chat = [...(existing.chat || []), ...draft.chat]; + } + existing.updatedAt = Date.now(); + if (draft) _sessionsStore.splice(_sessionsStore.indexOf(draft), 1); + cs.sessionId = existing.id; + return true; + } + if (!draft) return false; + draft.id = cs.mcodeSessionId; + draft.mcodeSessionId = cs.mcodeSessionId; + draft.updatedAt = Date.now(); + cs.sessionId = draft.id; + return true; + }, + // session-isolation/02 (run-mirror): record-targeted bind (no cs). + bindRecordToMcodeSid: (webuiId, sid) => { + if (!webuiId || !sid) return null; + const bound = _sessionsStore.find((r) => r && r.mcodeSessionId === sid); + if (bound) return bound.id; + const draft = _sessionsStore.find( + (r) => r && r.id === webuiId && !r.mcodeSessionId, + ); + if (!draft) return null; + const existing = _sessionsStore.find( + (r) => r && r.mcodeSessionId === sid, + ); + if (existing) { + if (Array.isArray(draft.chat) && draft.chat.length) { + existing.chat = [...(existing.chat || []), ...draft.chat]; + } + existing.updatedAt = Date.now(); + _sessionsStore.splice(_sessionsStore.indexOf(draft), 1); + return existing.id; + } + draft.id = sid; + draft.mcodeSessionId = sid; + draft.updatedAt = Date.now(); + return draft.id; + }, // v2.4: single-identity helpers — the mock keeps the in-memory store // shape so promotion/overlay logic is testable through handlers too. promoteDraftToMcodeSid: (cs) => { diff --git a/packages/webui/test/routes/chat-run-mirror.check.mjs b/packages/webui/test/routes/chat-run-mirror.check.mjs new file mode 100644 index 00000000..8e75e17f --- /dev/null +++ b/packages/webui/test/routes/chat-run-mirror.check.mjs @@ -0,0 +1,556 @@ +// webui/test/routes/chat-run-mirror.check.mjs +// Route-level contract test for the run-mirror session isolation +// (session-isolation/02). +// +// Bug shape (acceptance evidence): mid-run the user switches from +// session T1 to session T2 in the same browser (same cid). The +// engine's streamed lines used to be written into `cs.chat` — the +// VIEWED session's chat — so T2's live view filled with T1's tool and +// answer lines, and the turn-end persistence wrote T1's content (and +// engine binding) into T2's record. +// +// The fix: every stream write lands in a per-(cid, owning engine +// session) runChat buffer; the wire snapshots re-attach the buffer for +// the OWNING view only (state-bus.snapshotViewFields); the route's +// finalize drain flushes the buffer into cs.chat when the user still +// views the owning session, or into the owning session's persisted +// record (appendChatToSession) when they switched away. +// +// Like chat-first-turn-session-guard.check.mjs, this file mocks ONLY +// the ACP transport (acp.mjs) and the heavy peripherals, and runs the +// REAL chat.js → runMcodeAcp → sessions.js → state-bus chain plus the +// REAL switch route — the switch is performed by handleSwitchSession +// itself, exactly as the browser triggers it. +// +// Contract pinned here: +// 1. Mid-run switch away → the switched-to view receives no lines of +// the running session (pushed snapshot chat + running indicator), +// while the buffer keeps accumulating the running session's lines. +// 2. Switch back mid-run → the view shows the record lines plus the +// buffered lines so far (switch response and SSE snapshot). +// 3. Finalize while still viewing → full turn lands in cs.chat and +// in the OWNING record; buffer drained; no duplicate ● line. +// 4. Finalize while switched away → the full turn lands in the +// OWNING record only; the viewed session's cs.chat and record stay +// clean; cs.mcodeSessionId is NOT re-pointed at the run's engine +// sid (the finalize clobber that made T2 inherit T1's engine +// session). +// 5. First-turn draft promotion under a pre-bind switch: the DRAFT +// record (captured owning webui id) is promoted and receives the +// turn — never the record the user switched to. + +import { test, describe, before, beforeEach, after } from "node:test"; +import assert from "node:assert/strict"; +import { Readable } from "node:stream"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { pathToFileURL } from "node:url"; + +// Isolation FIRST — lib/config.js resolves SESSIONS_DB / UPLOAD_DIR from +// MCODE_WEBUI_DATA_DIR at import time. Neither this check nor the +// operator's real ~/.mcode-webui may see the other. +const _tmpDataDir = mkdtempSync(join(tmpdir(), "webui-run-mirror-")); +process.env.MCODE_WEBUI_DATA_DIR = _tmpDataDir; +process.env.MCODE_WEBUI_EVENTS_PATH = join(_tmpDataDir, "events.ndjson"); + +const SERVER_DIR = resolve(import.meta.dirname, "..", "..", "server"); +const absPath = (rel) => pathToFileURL(resolve(SERVER_DIR, rel)).href; + +// ------------------------------------------------------------------ +// Fake ACP transport — same surface as the guard test's, plus: +// - static emit(chunk): drives the live prompt's onChunk callback +// under test control (streamed lines); +// - a gateable newSession: the pre-bind switch test parks the turn +// BEFORE the engine session id exists. +// ------------------------------------------------------------------ +class FakeMcodeAcpClient { + static instances = []; + static sessionCounter = 0; + static pending = []; + static lastOnChunk = null; + static newSessionGate = null; // fn set → newSession awaits gate() + + static reset() { + this.instances = []; + this.sessionCounter = 0; + this.pending = []; + this.lastOnChunk = null; + this.newSessionGate = null; + } + + /** Feed one stream chunk into the live prompt callback. */ + static emit(chunk) { + const cb = this.lastOnChunk; + if (!cb) throw new Error("no live prompt callback"); + cb(chunk); + } + + /** Release the oldest parked prompt (FIFO — matches claim order). */ + static release(extra = {}) { + const resolveFn = this.pending.shift(); + if (resolveFn) { + resolveFn({ + answer: "ok", + thinking: null, + stopReason: "end_turn", + usage: null, + ...extra, + }); + } + } + + constructor() { + FakeMcodeAcpClient.instances.push(this); + } + async start() {} + async loadSession(sessionId) { + return { sessionId, configOptions: [] }; + } + async newSession() { + if (FakeMcodeAcpClient.newSessionGate) { + await FakeMcodeAcpClient.newSessionGate(); + } + FakeMcodeAcpClient.sessionCounter += 1; + return { + sessionId: `mvs_fake_${FakeMcodeAcpClient.sessionCounter}`, + configOptions: [], + }; + } + async request() { + return {}; + } + prompt(sessionId, _blocks, onChunk) { + FakeMcodeAcpClient.lastOnChunk = onChunk; + return new Promise((resolveFn) => { + FakeMcodeAcpClient.pending.push((extra) => { + resolveFn({ + answer: "ok", + thinking: null, + stopReason: "end_turn", + usage: null, + ...extra, + }); + }); + }); + } + stop() {} +} + +// Register the mock modules. Must run before the SUTs are imported. +async function setupMocks(t) { + t.mock.module(absPath("../acp.mjs"), { + namedExports: { McodeAcpClient: FakeMcodeAcpClient }, + }); + t.mock.module(absPath("lib/acp-client.js"), { + namedExports: { + getCachedMcodeCommands: () => [], + getMcodeSessionsForWorkspace: async () => [], + getMcodeSessionsCacheSync: () => null, + getMcodeSessionsStaleSync: () => null, + getMcodeSessionTitle: async () => "Engine title", + deleteMcodeSessionFromDb: () => ({ ok: true }), + getMcodeAcpClient: async () => null, + listAllMcodeSessions: async () => [], + getMcodeServerInfo: () => null, + invalidateMcodeSessionsCache: () => {}, + shutdownMcodeAcpSingleton: () => {}, + dropMcodeSessionFromCache: () => {}, + ensureMcodeCommands: async () => ({ + mcode: [], webui: [], fetchedAt: 0, source: "test-default", + }), + }, + }); + t.mock.module(absPath("lib/mavis-usage.js"), { + namedExports: { + getMavisTokenUsage: async () => null, + getMavisTokenUsageModel: async () => null, + applyMavisUsageToCs: async () => false, + }, + }); + t.mock.module(absPath("lib/slash.js"), { + namedExports: { + handleLocalSlash: async () => ({ handled: false, continueMcode: false }), + handleCmdCommand: async () => ({ ok: true }), + matchSlash: (content) => { + const m = content.match(/^\/([a-zA-Z][\w-]*)\b\s*(.*)/); + if (!m) return null; + return { cmd: m[1], rest: m[2] || "" }; + }, + }, + }); +} + +let handleSend; +let handleSwitchSession; +let sb; // real state-bus +let sessions; // real sessions lib (redirected store) +let alerts; + +function fakeReq(body) { + return Readable.from([Buffer.from(JSON.stringify(body), "utf8")]); +} + +function fakeRes() { + return { + _status: 200, + _headers: {}, + _body: null, + writeHead(s, h) { + this._status = s; + if (h) this._headers = h; + }, + end(b) { + this._body = b; + }, + }; +} + +// Fake SSE client — lets the test observe the exact snapshot bytes the +// server pushes for this cid (peekLastPushed returns the last payload). +function fakeSse() { + return { write() {}, end() {} }; +} + +function lastSnapshot(cid) { + const raw = sb.peekLastPushed(cid); + return raw ? JSON.parse(raw) : null; +} + +async function waitFor(fn, what, timeoutMs = 2000) { + const deadline = Date.now() + timeoutMs; + for (;;) { + const v = fn(); + if (v) return v; + if (Date.now() > deadline) { + throw new Error(`waitFor timed out: ${what}`); + } + await new Promise((r) => setTimeout(r, 5)); + } +} + +const WS = join(_tmpDataDir, "ws"); + +function makeClient(cid, { sessionId = null, mcodeSessionId = null } = {}) { + const cs = sb.makeClientState(); + cs.workspace = { dir: WS, branch: null, tree: null }; + cs.sessionId = sessionId; + cs.mcodeSessionId = mcodeSessionId; + cs.chat = []; + sb.clients.set(cid, cs); + sb.setSseClient(cid, fakeSse()); + return cs; +} + +function storeRecord(id, chat = []) { + const all = sessions.loadSessions(); + all.unshift({ + id, + title: id === "sess-B" ? "Session B" : "New session", + createdAt: Date.now(), + updatedAt: Date.now(), + chat, + workspace: WS, + }); + sessions.saveSessions(all); + return all.find((s) => s.id === id); +} + +function recordBy(id) { + return sessions.loadSessions().find((s) => s && s.id === id) || null; +} + +// §§ marker lines carry a variable duration — filter them for chat +// assertions that pin stable lines only. +const stable = (chat) => + (chat || []).filter((line) => !String(line).startsWith("§§")); + +before(async (t) => { + await setupMocks(t); + sb = await import(absPath("lib/state-bus.js")); + sessions = await import(absPath("lib/sessions.js")); + alerts = await import(absPath("lib/alerts.js")); + const chatMod = await import(absPath("routes/chat.js")); + handleSend = chatMod.handleSend; + const sessionsRoute = await import(absPath("routes/sessions.js")); + handleSwitchSession = sessionsRoute.handleSwitchSession; +}); + +beforeEach(() => { + sb.clients.clear(); + sb.resetCoalesceState(); + try { + rmSync(join(_tmpDataDir, "sessions.json"), { force: true }); + } catch {} + sessions._resetSessionsCacheForTests(); + alerts._resetForTests(); + FakeMcodeAcpClient.reset(); +}); + +after(() => { + try { + rmSync(_tmpDataDir, { recursive: true, force: true }); + } catch {} +}); + +/** + * Shared fixture: one cid bound to an existing session A (one + * completed turn already persisted), and a second session B record to + * switch to. Returns everything the cases need. + */ +async function setupTwoSessions(cid) { + const cs = makeClient(cid, { sessionId: "sess-A" }); + storeRecord("sess-A", []); + storeRecord("sess-B", []); + // Turn 1 — completes immediately; promotes the A record to the + // engine identity (id → mvs_fake_1) and persists ["› hello", "● ok"]. + const res1 = fakeRes(); + const turn1 = handleSend(fakeReq({ content: "hello" }), res1, { cs, cid }); + await waitFor(() => cs.mcodeSessionId, "turn 1 to bind the engine sid"); + FakeMcodeAcpClient.release(); + await turn1; + await waitFor(() => sb.activeRunCount() === 0, "turn 1 to drain"); + const sidA = cs.mcodeSessionId; + assert.deepEqual(stable(cs.chat), ["› hello", "● ok"]); + assert.equal(recordBy(sidA).mcodeSessionId, sidA); + return { cs, cid, sidA }; +} + +// Stream a recognizable multi-line turn into the live prompt, step by +// step (the caller asserts on intermediate buffer states between steps). +function emitThought() { + FakeMcodeAcpClient.emit({ kind: "thought", text: "pondering" }); +} +function emitToolAndAnswer() { + FakeMcodeAcpClient.emit({ kind: "tool_call", update: { toolCallId: "tc-1", name: "Bash", rawInput: { cmd: "ls" } } }); + FakeMcodeAcpClient.emit({ + kind: "tool_update", + update: { toolCallId: "tc-1", status: "completed", rawOutput: { content: [{ type: "text", text: "file.txt" }] } }, + }); + FakeMcodeAcpClient.emit({ kind: "message", text: "part one" }); +} + +describe("run-mirror — mid-run switch keeps views and records isolated", () => { + test("still viewing: live snapshot merges the buffer; finalize drains into cs.chat and the owning record", async () => { + const cid = "cid-mirror-stay"; + const { cs, sidA } = await setupTwoSessions(cid); + + const res2 = fakeRes(); + const turn2 = handleSend(fakeReq({ content: "run A2" }), res2, { cs, cid }); + await waitFor( + () => FakeMcodeAcpClient.pending.length === 1, + "turn 2 prompt to park", + ); + + emitThought(); + let buf = sb.runChatLinesFor(cid, sidA); + assert.ok(buf, "buffer exists for the owning (cid, sid)"); + assert.equal(buf[0], "▲ pondering ▍"); + + emitToolAndAnswer(); + buf = sb.runChatLinesFor(cid, sidA); + assert.equal(buf[0], "▲ pondering", "message stream strips the ▲ cursor"); + assert.equal(buf[1], "→ Bash {\"cmd\":\"ls\"}"); + assert.ok(buf.includes(" [completed]")); + assert.ok(buf.includes(" file.txt")); + assert.match(buf[buf.length - 1], /^● part one/); + + // Live view (owning session on screen): snapshot chat = record chat + // + buffer, and the run indicator is on. + sb.pushStateFor(cid); + const snap = lastSnapshot(cid); + assert.equal(snap.running.active, true); + assert.equal(snap.context.thinkingStatus, "Running"); + assert.deepEqual( + stable(snap.chat).slice(0, 2), + ["› hello", "● ok"], + "record lines come first", + ); + assert.ok(snap.chat.some((l) => String(l).startsWith("● part one"))); + + // Switch away and BACK mid-run: the owning view shows the buffered + // lines so far (switch response) and keeps its running indicator + // even though the switch's resetContext healed cs to idle. + await handleSwitchSession(fakeReq({ id: "sess-B" }), fakeRes(), { cs, cid }); + assert.equal(lastSnapshot(cid).running.active, false); + await handleSwitchSession(fakeReq({ id: sidA }), fakeRes(), { cs, cid }); + const backSnap = lastSnapshot(cid); + assert.equal( + backSnap.running.active, + true, + "owning view keeps its running indicator after switch-back", + ); + // base (3 persisted lines) + the 6 buffered stream lines so far + assert.equal(backSnap.chat.length, 9); + assert.deepEqual( + stable(backSnap.chat).slice(0, 3), + ["› hello", "● ok", "› run A2"], + ); + assert.ok(backSnap.chat.some((l) => String(l).startsWith("● part one"))); + + FakeMcodeAcpClient.release({ answer: "final answer" }); + await turn2; + await waitFor(() => sb.activeRunCount() === 0, "turn 2 to drain"); + + // View: full turn, exactly one final ● line, no leaked ▍ cursor. + const view = stable(cs.chat); + assert.deepEqual(view, [ + "› hello", + "● ok", + "› run A2", + "▲ pondering", + "→ Bash {\"cmd\":\"ls\"}", + " [completed]", + " file.txt", + "● final answer", + ]); + assert.equal( + cs.chat.filter((l) => String(l).startsWith("● ")).length, + 2, + "no duplicate ● line", + ); + assert.ok(cs.chat.every((l) => !String(l).endsWith("▍")), "cursors stripped"); + + // Record: identical content; buffer fully drained. + assert.deepEqual(stable(recordBy(sidA).chat), view); + assert.equal(sb.runChatLinesFor(cid, sidA), null); + assert.equal(sb.hasRunChat(cid, sidA), false); + + // The other session's record never saw any of it. + assert.deepEqual(recordBy("sess-B").chat, []); + }); + + test("switch away mid-run: other view stays clean; finalize writes the OWNING record and never re-points the viewed session", async () => { + const cid = "cid-mirror-switch"; + const { cs, sidA } = await setupTwoSessions(cid); + + const res2 = fakeRes(); + const turn2 = handleSend(fakeReq({ content: "run A2" }), res2, { cs, cid }); + await waitFor(() => FakeMcodeAcpClient.pending.length === 1, "prompt parked"); + emitThought(); + emitToolAndAnswer(); + + // Mid-run switch to B — through the REAL switch route. + const swRes = fakeRes(); + await handleSwitchSession(fakeReq({ id: "sess-B" }), swRes, { cs, cid }); + assert.equal(cs.sessionId, "sess-B"); + assert.equal(cs.mcodeSessionId, null); + + // B's view: no A lines, no running claim, idle thinking status. + const snap = lastSnapshot(cid); + assert.equal(snap.sessionId, "sess-B"); + assert.deepEqual(snap.chat, [], "T2 view must not contain T1 lines"); + assert.equal(snap.running.active, false, "T2 indicator idle"); + assert.equal(snap.context.thinkingStatus, "Idle"); + + // The buffer keeps accumulating for the OWNING session. + FakeMcodeAcpClient.emit({ kind: "message", text: " continued" }); + const buf = sb.runChatLinesFor(cid, sidA); + assert.match(buf[buf.length - 1], /^● part one continued/); + + // Finalize while viewing B. + FakeMcodeAcpClient.release({ answer: "switched answer" }); + await turn2; + await waitFor(() => sb.activeRunCount() === 0, "turn to drain"); + + // B's live view untouched. + assert.deepEqual(cs.chat, []); + assert.equal(cs.mcodeSessionId, null, "viewed session keeps its own binding"); + // B's record untouched. + assert.deepEqual(recordBy("sess-B").chat, []); + + // A's record received the full turn, including the final ● line. + const recA = stable(recordBy(sidA).chat); + assert.deepEqual(recA, [ + "› hello", + "● ok", + "› run A2", + "▲ pondering", + "→ Bash {\"cmd\":\"ls\"}", + " [completed]", + " file.txt", + "● switched answer", + ]); + // Binding + buffer finalize correctly on the owning side. + assert.equal(recordBy(sidA).mcodeSessionId, sidA); + assert.equal(sb.runChatLinesFor(cid, sidA), null); + + // Switching back shows the full conversation. + const backRes = fakeRes(); + await handleSwitchSession(fakeReq({ id: sidA }), backRes, { cs, cid }); + assert.equal(cs.sessionId, sidA); + assert.deepEqual(stable(cs.chat), recA); + }); + + test("first turn on a draft with a pre-bind switch: the DRAFT record is promoted and receives the turn, not the switched-to session", async () => { + const cid = "cid-draft-switch"; + let releaseNewSession; + FakeMcodeAcpClient.newSessionGate = () => + new Promise((r) => { + releaseNewSession = r; + }); + + const cs = makeClient(cid); // brand-new: no sessionId, no sid + storeRecord("sess-B", []); + + const res1 = fakeRes(); + const turn1 = handleSend(fakeReq({ content: "first!" }), res1, { cs, cid }); + // handleSend created the draft record; the turn is parked BEFORE the + // engine session id exists (inside session/new). + await waitFor(() => cs.sessionId, "draft record created"); + const draftId = cs.sessionId; + assert.ok(draftId && draftId !== "sess-B"); + await waitFor(() => releaseNewSession, "turn parked inside session/new"); + + // User switches away BEFORE the bind ran. + await handleSwitchSession(fakeReq({ id: "sess-B" }), fakeRes(), { cs, cid }); + assert.equal(cs.sessionId, "sess-B"); + assert.equal(cs.mcodeSessionId, null); + + releaseNewSession(); + + // The DRAFT record (found via its preserved `›` line) got the engine + // binding and was promoted (id = mvs sid); the switched-to record was + // NOT renamed, merged, or bound. + const promoted = await waitFor( + () => + sessions + .loadSessions() + .find( + (s) => + s && + s.mcodeSessionId && + s.mcodeSessionId === s.id && + Array.isArray(s.chat) && + s.chat.includes("› first!"), + ) || null, + "draft record promoted to the engine identity", + ); + assert.ok(promoted.id.startsWith("mvs_fake_")); + assert.deepEqual(stable(promoted.chat), ["› first!"]); + assert.equal(recordBy(draftId), null, "the uuid draft is gone (promoted)"); + assert.deepEqual(recordBy("sess-B").chat, []); + assert.equal(recordBy("sess-B").mcodeSessionId, undefined, "B unbound"); + + // Stream lines land in the buffer keyed by the NEW engine sid, and + // the promoted record is untouched until finalize. + await waitFor(() => FakeMcodeAcpClient.pending.length === 1, "prompt parked"); + FakeMcodeAcpClient.emit({ kind: "message", text: "draft answer" }); + assert.ok( + sb.runChatLinesFor(cid, promoted.mcodeSessionId), + "buffer keyed by the real engine sid", + ); + + FakeMcodeAcpClient.release({ answer: "draft answer" }); + await turn1; + await waitFor(() => sb.activeRunCount() === 0, "turn to drain"); + + // Turn persisted to the OWNING (promoted draft) record only. + assert.deepEqual(stable(promoted.chat), ["› first!", "● draft answer"]); + assert.deepEqual(cs.chat, [], "B's live view still clean"); + assert.equal(cs.sessionId, "sess-B"); + assert.equal(cs.mcodeSessionId, null, "B's cs never inherited the engine sid"); + assert.deepEqual(recordBy("sess-B").chat, []); + assert.equal(sb.runChatLinesFor(cid, promoted.mcodeSessionId), null); + }); +}); diff --git a/packages/webui/test/server/stream-cumulative-render.test.js b/packages/webui/test/server/stream-cumulative-render.test.js index cb7d561e..7f79dc27 100644 --- a/packages/webui/test/server/stream-cumulative-render.test.js +++ b/packages/webui/test/server/stream-cumulative-render.test.js @@ -56,6 +56,7 @@ class FakeMcodeAcpClient { let mcodeAcp; let sessions; +let stateBus; before(async (t) => { // Mock acp.mjs FIRST so the SUT's `new McodeAcpClient()` imports @@ -73,6 +74,10 @@ before(async (t) => { // "../../acp.mjs"`. mcodeAcp = await import(absPath("lib/mcode-acp.js")); sessions = await import(absPath("lib/sessions.js")); + // session-isolation/02 (run-mirror): stream writes land in the + // per-(cid, owning sid) runChat buffer, not cs.chat — the harness + // reads the routed lines through the same accessor the snapshots use. + stateBus = await import(absPath("lib/state-bus.js")); const chatLine = await import(absPath("lib/chat-line.js")); await setupMocks(t, { acp: { @@ -149,7 +154,16 @@ async function runPrompt(chunks) { // Force-clear any leftover timers / state so the watchdog cannot // keep the test runner alive past this test. cs.running = { active: false, lastDeltaAt: null }; - return cs.chat.map((line) => + // session-isolation/02 (run-mirror): the stream writes went to the + // runChat buffer keyed by (cid, engine sid) — this harness runs + // runMcodeAcp directly (no handleSend finalize drain), so collect + // the routed lines the way a view would see them: the viewed chat + // plus the turn's buffered lines. + const buffered = + (cs.mcodeSessionId && + stateBus.runChatLinesFor("cid-test", cs.mcodeSessionId)) || + []; + return [...(cs.chat || []), ...buffered].map((line) => typeof line === "string" && line.endsWith(" ▍") ? line.slice(0, -2) : line, diff --git a/release/public-source.json b/release/public-source.json index ae9f9ede..e35b5bfa 100644 --- a/release/public-source.json +++ b/release/public-source.json @@ -3526,6 +3526,7 @@ "packages/webui/test/routes/alerts.check.mjs", "packages/webui/test/routes/chat-failed-send.check.mjs", "packages/webui/test/routes/chat-first-turn-session-guard.check.mjs", + "packages/webui/test/routes/chat-run-mirror.check.mjs", "packages/webui/test/routes/chat.check.mjs", "packages/webui/test/routes/debug.check.mjs", "packages/webui/test/routes/export.check.mjs",