fix(server): per-segment stream buffers, DB-authoritative switch healing, wedged sync recovery

packages/webui/server/lib/mcode-acp.js

@@ -403,6 +403,16 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = [])
       durationMs: null,
       stopReason: null,
       tps: null,
+      // session-isolation/06: per-segment accumulator reset.
+      // `lastChunkKind` is the kind of the chunk that last wrote a
+      // `▲` or `●` line; when the new chunk is the same kind we
+      // append to the existing buffer (normal streaming growth), when
+      // it is different we reset so the new line contains only the
+      // new segment. Reset on every turn (streamAcpPrompt is called
+      // once per prompt), so a new message after stop/save/resume
+      // starts with lastChunkKind === null and the first chunk of
+      // any kind triggers a clean accumulator.
+      lastChunkKind: null,
     };
     const t0 = Date.now();
     cs.running = {
@@ -699,11 +709,22 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = [])
             );
           }
         } else if (c.kind === "thought" && typeof c.text === "string") {
-          r.thinking = (r.thinking || "") + c.text;
+          // session-isolation/06 (stream cumulative-render): each
+          // agent_message segment starts fresh — the bug was that
+          // r.thinking was turn-long, so after a tool_call line broke
+          // streamUpdateLine's same-prefix chain, the next message
+          // chunk appended a NEW `▲` line containing every prior
+          // segment. Reset the buffer when the previous chunk kind
+          // was something other than a thought.
+          if (r.lastChunkKind !== "thought") r.thinking = "";
+          r.thinking += c.text;
+          r.lastChunkKind = "thought";
           const oneLine = r.thinking.replace(/\n+/g, " ").trim();
           streamUpdateLine(cs.chat, "▲", oneLine);
         } else if (c.kind === "message" && typeof c.text === "string") {
-          r.answer = (r.answer || "") + c.text;
+          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);
         } else if (c.kind === "tool_call" && c.update) {
@@ -716,6 +737,11 @@ function streamAcpPrompt(client, sid, content, label, cs, cid, attachments = [])
           // 记下这行在 chat 里的位置（之后 tool_update 用来在它后面插输出）
           if (!r.toolIndexById) r.toolIndexById = new Map();
           r.toolIndexById.set(u.toolCallId, cs.chat.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
+          // see lastChunkKind === "message" and skip the reset.
+          r.lastChunkKind = "tool_call";
         } else if (c.kind === "tool_update" && c.update) {
           applyToolUpdate(r, cs, c.update);
         } else if (c.kind === "plan_update" && c.update) {

packages/webui/server/lib/transcript-sync.js

@@ -35,6 +35,31 @@ export const TRANSCRIPT_SYNC_MS =
 /** Only real mcode session ids have a transcript to read. */
 const MVS_SESSION_ID = /^mvs_[a-f0-9]{32}$/;
 
+// session-isolation/06 (Item 3 — wedge healing): a tab can sit with
+// `cs.running.active=true` forever if the backend received SIGTERM
+// mid-stream — see the graceful-shutdown ticket for the cause. The
+// transcript-sync poller must NOT skip such a tab forever, otherwise
+// the polluted buffer persists across view reloads. Define a stuck-
+// run threshold (5 minutes — comfortably longer than any realistic
+// model latency) and let the DB-rebuild path run when the last
+// delta is older than that AND there is no live ACP child to write
+// the next line. The threshold is configurable for tests.
+const DEFAULT_WEDGED_RUN_MS = 5 * 60 * 1000;
+// Local binding first — the alias below only re-exports this same
+// value. Without the local binding the wedge branch below would
+// throw "TRANSCRIPT_SYNC_WEDGED_MS is not defined" on every tick
+// that hits a stale tab, which (a) prevents the wedge healing from
+// running AND (b) aborts the whole sync pass so even tabs that
+// would normally sync cleanly stop syncing while any tab is
+// mid-turn. The earlier commit shipped that bug (the unit tests
+// passed because they read the exported alias; the internal
+// reference was the broken one).
+const TRANSCRIPT_SYNC_WEDGED_MS = (() => {
+  const env = Number(process.env.MCODE_WEBUI_TRANSCRIPT_WEDGED_MS);
+  return Number.isFinite(env) && env >= 0 ? env : DEFAULT_WEDGED_RUN_MS;
+})();
+export { TRANSCRIPT_SYNC_WEDGED_MS as wedgedRunMs };
+
 /**
  * Did the stored transcript move?
  *
@@ -68,8 +93,35 @@ export function syncTranscriptsOnce({ dbPath = MCODE_RUNTIME_DB } = {}) {
     if (!MVS_SESSION_ID.test(cs.mcodeSessionId || "")) continue;
     // A local turn owns `cs.chat` until it finishes: streaming writes lines the
     // DB does not have yet, and a concurrent read would roll them back.
-    if (cs.running && cs.running.active) continue;
-    if (getActiveChild(cid)) continue;
+    //
+    // session-isolation/06 (wedge healing): the `active` flag can stick
+    // `true` forever if the backend received SIGTERM mid-stream and
+    // the active-child registry was not cleared (see the
+    // graceful-shutdown ticket). Without the wedge exception below,
+    // a wedged tab keeps the polluted chat and transcript-sync never
+    // recovers it. The exception fires when:
+    //   - cs.running.active is true (looks wedged)
+    //   - AND the last delta is older than TRANSCRIPT_SYNC_WEDGED_MS
+    //     (no stream activity for >5 min by default)
+    //   - AND there is no live ACP child to write the next line
+    // An active real run (lastDeltaAt recent) still skips; a wedged
+    // run (stale lastDeltaAt, no active child) heals.
+    if (cs.running && cs.running.active) {
+      const lastDelta =
+        (cs.running && cs.running.lastDeltaAt) ||
+        (cs.running && cs.running.startedAt) ||
+        0;
+      const stale = lastDelta
+        ? Date.now() - lastDelta > TRANSCRIPT_SYNC_WEDGED_MS
+        : true;
+      if (stale && !getActiveChild(cid)) {
+        // fall through to the heal path below
+      } else {
+        continue;
+      }
+    } else if (getActiveChild(cid)) {
+      continue;
+    }
 
     let read;
     try {

packages/webui/server/routes/sessions.js

@@ -42,6 +42,53 @@ import { append as _eventsAppend } from "../lib/events.js";
 // workspace write lands on the same boundary.
 import { assertWorkspacePath } from "../lib/workspace.js";
 
+/**
+ * Detect the cumulative-render pollution pattern in a stored chat
+ * buffer (session-isolation/06). When the engine emits each segment
+ * of an `agent_message`, streamUpdateLine writes a new `●` line; a
+ * non-cumulative buffer has each line containing only its own
+ * segment's text. A cumulative buffer — the bug — has at least one
+ * later `●` line whose text is a strict superset of an earlier
+ * `●` line (because the accumulator never reset between segments and
+ * every later line re-wrote every prior segment's text). This
+ * predicate is O(n^2) in the number of `●` lines but a single
+ * session's `chat` is bounded (~400 lines by the transcript cap) so
+ * the worst case is a few thousand substring checks per switch —
+ * cheap enough.
+ *
+ * Returns true when the buffer is clearly cumulative (an earlier
+ * `●` line is a strict substring of a later one AND the longer line
+ * strictly extends the shorter). Conservative on both sides:
+ *   - a single-`●`-line buffer is never cumulative;
+ *   - non-`●` lines (system, tool, ▲ thought) are ignored — only
+ *     `●` rows matter, since the cumulative bug only affects message
+ *     segments;
+ *   - ties (equal-length `●` lines) are NOT cumulative — same
+ *     length, no superset relation.
+ */
+function chatLooksCumulative(chat) {
+  if (!Array.isArray(chat) || chat.length === 0) return false;
+  const dots = [];
+  for (const line of chat) {
+    if (typeof line !== "string") continue;
+    // Match the same prefix the streamer writes: `● ` then text.
+    // Also accept bare `●` at end-of-line (transcript-sync appends
+    // stripped-down `●` markers in some paths).
+    if (line.startsWith("● ")) dots.push(line.slice(2));
+    else if (line === "●") continue;
+    else continue;
+  }
+  for (let i = 0; i < dots.length; i += 1) {
+    for (let j = i + 1; j < dots.length; j += 1) {
+      const a = dots[i];
+      const b = dots[j];
+      if (b.length <= a.length) continue; // strict superset ⇒ longer
+      if (b.includes(a)) return true;
+    }
+  }
+  return false;
+}
+
 // _auditFail — shared failure sink for audit writes. events.js#append
 // THROWS on write failure; a governance action must not complete with
 // a missing audit trail, so every route-level append is wrapped and
@@ -297,34 +344,73 @@ export async function handleSwitchSession(req, res, ctx) {
   // grammar BEFORE responding, so response session.chat and cs.chat
   // carry history. Caps inside (last 400 lines / 200KB) keep the SSE
   // state push bounded; a 1000+-message session must not balloon it.
+  //
+  // session-isolation/06 (persist hygiene): the original rule only
+  // backfilled when target.chat was empty, so a polluted buffer
+  // (the cumulative-render bug from Item 1, before its fix) would
+  // persist via saveSessions and win forever. The new rule is:
+  //   - if stored chat is empty → backfill (unchanged).
+  //   - if stored chat looks cumulative → prefer DB read and re-persist.
+  //     "cumulative" = at least two `●` lines whose text is a strict
+  //     superset of an earlier `●` line (the engine emits each
+  //     segment's full text per line, so a non-cumulative buffer has
+  //     no such inclusion pair).
+  //   - otherwise → keep stored chat. DB-authoritative: transcript-sync
+  //     overwrites the stored chat from the engine DB on the next tick
+  //     (~4s later), so any stored-only lines a user typed into the
+  //     composer but never sent will be lost. The rule above does not
+  //     promise draft preservation; it promises to NOT clobber a
+  //     clean stored buffer with the DB read on every switch. Draft
+  //     preservation is a separate concern (the composer keeps its
+  //     own draft in its own state, see composer-draft.test.ts).
   // FAILURE MUST NOT BREAK SWITCHING: any error logs and continues
-  // with chat: [] — the switch itself always succeeds.
+  // with the original chat — the switch itself always succeeds.
   if (
     target.mcodeSessionId &&
-    /^mvs_[a-f0-9]{32}$/.test(target.mcodeSessionId) &&
-    (!Array.isArray(target.chat) || target.chat.length === 0)
+    /^mvs_[a-f0-9]{32}$/.test(target.mcodeSessionId)
   ) {
-    try {
-      const r = loadTranscriptChatLines(target.mcodeSessionId, {
-        dbPath: MCODE_RUNTIME_DB,
-      });
-      if (r.ok && r.lines.length > 0) {
-        target.chat = r.lines;
-        target.updatedAt = Date.now();
-        saveSessions(all); // persist the populated wrapper (updatedAt bumped)
-        console.log(
-          `[switch] cid=${cid} transcript backfill ${target.id.substring(0, 8)}… mcode=${target.mcodeSessionId.substring(0, 12)}… lines=${r.lines.length} msgs=${r.messageCount} probe=${r.probe}${r.truncated ? " (capped)" : ""}`,
-        );
-      } else if (!r.ok) {
-        console.log(
-          `[switch] cid=${cid} transcript unavailable for ${target.mcodeSessionId.substring(0, 12)}… reason=${r.reason || "unknown"}`,
+    const storedHasChat = Array.isArray(target.chat) && target.chat.length > 0;
+    const storedCumulative = storedHasChat && chatLooksCumulative(target.chat);
+    const shouldBackfill =
+      !storedHasChat || storedCumulative;
+    if (shouldBackfill) {
+      try {
+        const r = loadTranscriptChatLines(target.mcodeSessionId, {
+          dbPath: MCODE_RUNTIME_DB,
+        });
+        if (r.ok && r.lines.length > 0) {
+          const dbEmpty = target.chat.length === 0;
+          const dbShrinks = r.lines.length < target.chat.length;
+          const reason = dbEmpty
+            ? "empty"
+            : storedCumulative
+              ? "stored_cumulative"
+              : "stored_shrinks";
+          target.chat = r.lines;
+          target.updatedAt = Date.now();
+          saveSessions(all); // persist the populated wrapper (updatedAt bumped)
+          console.log(
+            `[switch] cid=${cid} transcript backfill ${target.id.substring(0, 8)}… mcode=${target.mcodeSessionId.substring(0, 12)}… reason=${reason} lines=${r.lines.length} msgs=${r.messageCount} probe=${r.probe}${r.truncated ? " (capped)" : ""}`,
+          );
+        } else if (!r.ok) {
+          console.log(
+            `[switch] cid=${cid} transcript unavailable for ${target.mcodeSessionId.substring(0, 12)}… reason=${r.reason || "unknown"}`,
+          );
+        } else if (storedCumulative) {
+          // Cumulative buffer + DB read came back empty — preserve
+          // the stored chat (which is at least the user's last view)
+          // and log the discrepancy so a post-mortem can see what
+          // happened.
+          console.log(
+            `[switch] cid=${cid} stored chat looked cumulative but DB read returned no lines; preserving stored chat for ${target.mcodeSessionId.substring(0, 12)}…`,
+          );
+        }
+      } catch (e) {
+        console.warn(
+          `[switch] cid=${cid} transcript backfill failed for ${target.mcodeSessionId.substring(0, 12)}… (continuing with stored chat):`,
+          e && e.message ? e.message : e,
         );
       }
-    } catch (e) {
-      console.warn(
-        `[switch] cid=${cid} transcript backfill failed for ${target.mcodeSessionId.substring(0, 12)}… (continuing with empty chat):`,
-        e && e.message ? e.message : e,
-      );
     }
   }
   const prevSid = cs.sessionId;

packages/webui/test/integration/upload-limits.test.js

@@ -148,8 +148,19 @@ function multipartBody(parts) {
 // POST /api/upload with a chunked writer loop that yields to the event
 // loop, so the 413-early-response path can preempt a large body mid-
 // send. Resolves { status, json, bytesWritten } once the response ends.
-function postUpload({ port, body, contentType, timeoutMs = 15000 }) {
-  return new Promise((resolve, reject) => {
+//
+// macOS CI runners (PR #30) hit a fixed 15s window roughly 1 run in 3
+// on the "oversized single file → 413" path — the bottleneck is the
+// kernel-side accept / write race between our 2 MB body upload and the
+// server's read-and-close-on-cap path, not the test logic. Rather than
+// lengthen every deadline (which just pushes the problem down the CI
+// queue), we retry the request ONCE on a clean timeout, then surface
+// the timeout if both attempts fail. Each attempt's deadline is still
+// 15 s, but the test as a whole gets a full ~30 s window — bounded by
+// `retries * timeoutMs`. Other errors propagate immediately so the
+// 413 / leftover assertions still pin the actual behaviour.
+function postUpload({ port, body, contentType, timeoutMs = 15000, retries = 1 }) {
+  const attempt = () => new Promise((resolve, reject) => {
     let settled = false;
     let bytesWritten = 0;
     const req = http.request(
@@ -238,6 +249,15 @@ function postUpload({ port, body, contentType, timeoutMs = 15000 }) {
       if (!req.destroyed && !settled) req.end();
     })().catch(() => {});
   });
+  // Bounded retry on a clean timeout — see the function-level
+  // comment above. Other errors propagate immediately so the 413 /
+  // leftover assertions still pin the actual behaviour.
+  return attempt().catch((e) => {
+    if (retries <= 0 || !(e && /no response within/.test(e.message))) {
+      throw e;
+    }
+    return attempt();
+  });
 }
 
 function listUploads(dir) {