Flush live shell tails on a timer and isolate per-call feeds

A rate-limited second write could sit until the next chunk or settle, and a shared feed mixed parallel run_shell tails. Hydrate and rollback also dropped older member ids and truncated call snapshots.

Author: TheGreatAxios

docs/TUI.md

@@ -100,7 +100,7 @@ The past tense is a map keyed by raw tool name (`src/tui/tool-formatter.ts`:
 name.
 
 Because a lane's row identity moves to the newest call, a lane also carries
-the call ids it absorbed (`memberIds`, newest appended, last 32 kept). A
+the call ids it absorbed (`memberIds`, newest appended). A
 result resolves its lane when its call id is the row's own id **or** one of
 its members — this is what pairs a resumed transcript's parallel batch
 (call, call, result, result) correctly. An id matching nothing still answers

src/agent/posix-tool-plugins.ts

@@ -30,7 +30,7 @@ import {
 import type { PermissionGate } from "../permission/gate.js";
 import { createWorktreeRootsProvider } from "../permission/worktree-roots.js";
 import type { CompactionArchive } from "../session/compaction-archive.js";
-import type { ShellOutputFeed } from "../session/shell-output-feed.js";
+import type { ShellOutputFeedMap } from "../session/shell-output-feed.js";
 
 export interface CorePosixToolPluginsArgs {
   cwd: string;
@@ -48,9 +48,9 @@ export interface CorePosixToolPluginsArgs {
   // Live getter for the background-shell registry (run_shell background:true).
   // Omitted makes background runs fail closed in shell-guard.
   getBackgroundShellRegistry?: () => BackgroundShellRegistry | undefined;
-  // Live getter for the bounded shell-output feed the transcript polls for a
-  // running command's live tail. Omitted leaves the tail unwired.
-  getShellOutputFeed?: () => ShellOutputFeed | undefined;
+  // Live getter for the per-call bounded shell-output feeds the transcript
+  // polls for a running command's live tail. Omitted leaves the tail unwired.
+  getShellOutputFeeds?: () => ShellOutputFeedMap | undefined;
   /** Primary-only evidence archive; workers omit this getter. */
   getEvidenceArchive?: () => CompactionArchive | undefined;
 }
@@ -90,7 +90,7 @@ export function buildCorePosixToolPlugins(
     getContextDir,
     shellEnv,
     getBackgroundShellRegistry,
-    getShellOutputFeed,
+    getShellOutputFeeds,
     getEvidenceArchive,
   } = args;
   // Pre-gate sandboxes honor yolo mode so outside-workspace path tools and shell
@@ -124,7 +124,7 @@ export function buildCorePosixToolPlugins(
       ...(getBackgroundShellRegistry !== undefined
         ? { getBackgroundShellRegistry }
         : {}),
-      ...(getShellOutputFeed !== undefined ? { getShellOutputFeed } : {}),
+      ...(getShellOutputFeeds !== undefined ? { getShellOutputFeeds } : {}),
     }),
     ...(getEvidenceArchive !== undefined
       ? [evidenceArchiveSearchPlugin(getEvidenceArchive)]

src/agent/tools.ts

@@ -90,7 +90,7 @@ import {
   createBackgroundShellRegistry,
   type BackgroundShellExit,
 } from "../shell/background-shell.js";
-import { createShellOutputFeed } from "../session/shell-output-feed.js";
+import { createShellOutputFeedMap } from "../session/shell-output-feed.js";
 import { createListDirTool } from "../util/list-dir.js";
 import {
   createExaMCPWebFetchTool,
@@ -290,9 +290,9 @@ export interface AgentToolset {
     callbacks: MCPConnectCallbacks,
     signal?: AbortSignal,
   ) => Promise<void>;
-  // Bounded live-output tail of the session's foreground shell, polled by the
-  // transcript for the pending run_shell row's live lines.
-  shellOutputFeed: ReturnType<typeof createShellOutputFeed>;
+  // Per-call bounded live-output tails of foreground shells, polled by the
+  // transcript for each pending run_shell row's live lines.
+  shellOutputFeed: ReturnType<typeof createShellOutputFeedMap>;
   // Connect one newly persisted server through the same lifecycle as startup MCP.
   connectMCPServer: (
     config: MCPServerConfig,
@@ -364,10 +364,10 @@ export async function createAgentToolset(
       : {}),
   });
   const shellCollect = createShellCollectTool(backgroundShells);
-  // Bounded live-output tail of the foreground shell, polled by the TUI for
-  // the pending run_shell row's live lines. Workers get one too; nothing
+  // Per-call bounded live-output tails of foreground shells, polled by the TUI
+  // for each pending run_shell row's live lines. Workers get a map too; nothing
   // reads it unless a transcript polls it (silent degradation).
-  const shellOutputFeed = createShellOutputFeed();
+  const shellOutputFeed = createShellOutputFeedMap();
   const sessionBlobReader =
     getBlobReader !== undefined
       ? createLazyBlobReader(getBlobReader)
@@ -455,7 +455,7 @@ export async function createAgentToolset(
       ...(getEvidenceArchive !== undefined ? { getEvidenceArchive } : {}),
       ...(shellEnv !== undefined ? { shellEnv } : {}),
       getBackgroundShellRegistry: () => backgroundShells,
-      getShellOutputFeed: () => shellOutputFeed,
+      getShellOutputFeeds: () => shellOutputFeed,
     }),
   });
 

src/plugins/shell-guard-plugin.test.ts

@@ -15,6 +15,7 @@ import { createShellOutputFeed } from "../session/shell-output-feed.js";
 import {
   BoundedShellOutput,
   MAX_SHELL_OUTPUT_BYTES,
+  SHELL_FEED_EMIT_MS,
   advertiseShellGuardTimeout,
   resolveShellTimeoutMs,
   reapLiveChildren,
@@ -39,6 +40,35 @@ describe("runGuardedShell", () => {
     expect(output).toContain("hello");
   });
 
+  test("a rate-limited second write reaches the feed before the process exits", async () => {
+    const feed = createShellOutputFeed();
+    let finished = false;
+    const running = runGuardedShell(
+      { command: "echo first; sleep 0.02; echo second; sleep 0.4" },
+      neverAbort(),
+      undefined,
+      undefined,
+      (text) => {
+        feed.append(text);
+      },
+    ).then((result) => {
+      finished = true;
+      return result;
+    });
+    const deadline = Date.now() + SHELL_FEED_EMIT_MS + 80;
+    while (
+      !feed.snapshot().includes("second") &&
+      Date.now() < deadline &&
+      !finished
+    ) {
+      await Bun.sleep(10);
+    }
+    expect(finished).toBe(false);
+    expect(feed.snapshot()).toContain("second");
+    const result = await running;
+    expect(result.exitCode).toBe(0);
+  });
+
   test("omitted timeout does not arm a timer", async () => {
     const start = Date.now();
     const { exitCode, timedOut, output } = await runGuardedShell(
@@ -329,15 +359,20 @@ describe("background run_shell (shellGuardPlugin)", () => {
     let emits = 0;
     const handler = defined(
       shellGuardPlugin(process.cwd(), undefined, undefined, {
-        getShellOutputFeed: () => {
-          return {
-            append: (text) => {
+        getShellOutputFeeds: () => {
+          const wrapped = {
+            append: (text: string) => {
               emits += 1;
               feed.append(text);
             },
             snapshot: () => feed.snapshot(),
             clear: () => feed.clear(),
           };
+          return {
+            forCall: () => wrapped,
+            get: () => wrapped,
+            drop: () => undefined,
+          };
         },
       }).middleware,
     )(fallback);

src/plugins/shell-guard-plugin.ts

@@ -2,7 +2,7 @@ import { spawn, type ChildProcess } from "node:child_process";
 import { realpathSync } from "node:fs";
 import { StringDecoder } from "node:string_decoder";
 import type { ToolPlugin } from "@intx/tools-posix";
-import type { ShellOutputFeed } from "../session/shell-output-feed.js";
+import type { ShellOutputFeedMap } from "../session/shell-output-feed.js";
 import {
   killProcessTree,
   type BackgroundShellRegistry,
@@ -321,20 +321,43 @@ export async function runGuardedShell(
 
     let settled = false;
     let timer: ReturnType<typeof setTimeout> | undefined;
+    let emitTimer: ReturnType<typeof setTimeout> | undefined;
 
     // Live-output cadence for the transcript's shell tail (when wired).
     const decoder = new StringDecoder("utf8");
     let pendingOutput = "";
     let lastEmitAt = 0;
+    const clearEmitTimer = (): void => {
+      if (emitTimer !== undefined) {
+        clearTimeout(emitTimer);
+        emitTimer = undefined;
+      }
+    };
     const emitPendingOutput = (final: boolean): void => {
-      if (onOutput === undefined || pendingOutput.length === 0) return;
-      if (!final && Date.now() - lastEmitAt < SHELL_FEED_EMIT_MS) return;
+      if (onOutput === undefined || pendingOutput.length === 0) {
+        if (final) clearEmitTimer();
+        return;
+      }
+      if (!final) {
+        const wait = SHELL_FEED_EMIT_MS - (Date.now() - lastEmitAt);
+        if (wait > 0) {
+          if (emitTimer === undefined) {
+            emitTimer = setTimeout(() => {
+              emitTimer = undefined;
+              emitPendingOutput(false);
+            }, wait);
+          }
+          return;
+        }
+      }
+      clearEmitTimer();
       lastEmitAt = Date.now();
       onOutput(pendingOutput);
       pendingOutput = "";
     };
     const clearTimer = () => {
       if (timer !== undefined) clearTimeout(timer);
+      clearEmitTimer();
     };
 
     const settle = (err?: Error) => {
@@ -436,9 +459,9 @@ export interface ShellGuardPluginOptions {
   // Live getter for the background-shell registry. Unwired (undefined result)
   // makes `background: true` fail closed: nothing spawns, no handle returns.
   getBackgroundShellRegistry?: () => BackgroundShellRegistry | undefined;
-  // Live getter for the bounded shell-output feed the transcript polls for a
-  // running command's live tail. Unwired, the tail is simply not painted.
-  getShellOutputFeed?: () => ShellOutputFeed | undefined;
+  // Live getter for the per-call bounded shell-output feeds the transcript
+  // polls for a running command's live tail. Unwired, the tail is simply not painted.
+  getShellOutputFeeds?: () => ShellOutputFeedMap | undefined;
 }
 
 function resolveAllowOutsideCwd(
@@ -569,8 +592,8 @@ export function shellGuardPlugin(
             };
           }
           const wrappedCommand = wrapCommandWithPwdProbe(command);
-          const feed = options.getShellOutputFeed?.();
-          if (feed !== undefined) feed.clear();
+          const feeds = options.getShellOutputFeeds?.();
+          const feed = feeds?.forCall(call.id);
           try {
             const { output, exitCode, timedOut, outputTruncated } =
               await runGuardedShell(
@@ -627,6 +650,8 @@ export function shellGuardPlugin(
               content: err instanceof Error ? err.message : String(err),
               isError: true,
             };
+          } finally {
+            feeds?.drop(call.id);
           }
         }).catch((err: unknown) => ({
           callId: call.id,

src/session/shell-output-feed.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test";
 
 import {
   createShellOutputFeed,
+  createShellOutputFeedMap,
   SHELL_FEED_LIMIT_BYTES,
 } from "./shell-output-feed.js";
 
@@ -43,3 +44,23 @@ describe("shell output feed", () => {
     expect(feed.snapshot()).toBe("");
   });
 });
+
+describe("shell output feed map", () => {
+  test("isolates tails per call and keeps the 8 KiB bound on each", () => {
+    const feeds = createShellOutputFeedMap();
+    const a = feeds.forCall("a");
+    const b = feeds.forCall("b");
+    a.append("alpha\n");
+    b.append("beta\n");
+    expect(feeds.get("a")?.snapshot()).toBe("alpha\n");
+    expect(feeds.get("b")?.snapshot()).toBe("beta\n");
+    a.append("x".repeat(SHELL_FEED_LIMIT_BYTES + 1));
+    expect(new TextEncoder().encode(a.snapshot()).length).toBeLessThanOrEqual(
+      SHELL_FEED_LIMIT_BYTES,
+    );
+    expect(feeds.get("b")?.snapshot()).toBe("beta\n");
+    feeds.drop("a");
+    expect(feeds.get("a")).toBeUndefined();
+    expect(feeds.get("b")?.snapshot()).toBe("beta\n");
+  });
+});

src/session/shell-output-feed.ts

@@ -3,8 +3,8 @@
  *
  * Pure and synchronous: the shell-guard plugin appends output chunks, the TUI
  * polls `snapshot()` on its sticky tick and paints the tail onto the pending
- * `run_shell` row. Nothing here is persisted; the whole feed is a display
- * affordance for the wait.
+ * `run_shell` row that owns that call. Nothing here is persisted; the whole
+ * feed is a display affordance for the wait.
  */
 
 /** Tail of output a feed keeps before the oldest chunk is dropped. */
@@ -14,10 +14,17 @@ export interface ShellOutputFeed {
   append(chunk: string): void;
   /** The retained tail, oldest first. Empty string when nothing has landed. */
   snapshot(): string;
-  /** Drop everything (a new foreground shell is about to run). */
+  /** Drop everything retained in this feed. */
   clear(): void;
 }
 
+/** Per-call live tails so parallel run_shells cannot share or wipe a sibling. */
+export interface ShellOutputFeedMap {
+  forCall(callId: string): ShellOutputFeed;
+  get(callId: string): ShellOutputFeed | undefined;
+  drop(callId: string): void;
+}
+
 function byteLength(text: string): number {
   return new TextEncoder().encode(text).length;
 }
@@ -60,3 +67,23 @@ export function createShellOutputFeed(
     },
   };
 }
+
+export function createShellOutputFeedMap(): ShellOutputFeedMap {
+  const feeds = new Map<string, ShellOutputFeed>();
+  return {
+    forCall(callId: string): ShellOutputFeed {
+      let feed = feeds.get(callId);
+      if (feed === undefined) {
+        feed = createShellOutputFeed();
+        feeds.set(callId, feed);
+      }
+      return feed;
+    },
+    get(callId: string): ShellOutputFeed | undefined {
+      return feeds.get(callId);
+    },
+    drop(callId: string): void {
+      feeds.delete(callId);
+    },
+  };
+}

src/subagent/run.ts

@@ -22,7 +22,7 @@ import {
   workerPermissionGate,
 } from "../permission/reactor-authorize.js";
 import { createSessionStores } from "../session/optimized-context-store.js";
-import { createShellOutputFeed } from "../session/shell-output-feed.js";
+import { createShellOutputFeedMap } from "../session/shell-output-feed.js";
 import { createAgentWithLiveToolDispatch } from "../agent/live-tool-dispatch.js";
 import { type } from "arktype";
 import { createPosixTools } from "@intx/tools-posix";
@@ -598,9 +598,9 @@ async function runSubAgentInner(
   let childBlobReader: BlobReader | undefined;
   let childBlobWriter: SpillBlobWriter | undefined;
   let childContextDir: string | undefined;
-  // Live shell tail for this worker's transcript; nothing polls it unless a
-  // transcript host does (silent degradation), but the run_shell path shares it.
-  const childShellOutputFeed = createShellOutputFeed();
+  // Per-call live shell tails for this worker's transcript; nothing polls them
+  // unless a transcript host does (silent degradation).
+  const childShellOutputFeed = createShellOutputFeedMap();
   const sessionBlobReader = createCompositeBlobReader(
     () => childBlobReader,
     params.getBlobReader,
@@ -617,7 +617,7 @@ async function runSubAgentInner(
       ...(params.shellEnv !== undefined ? { shellEnv: params.shellEnv } : {}),
       readFileGuard: { blobReader: sessionBlobReader },
       getBackgroundShellRegistry: () => backgroundShells,
-      getShellOutputFeed: () => childShellOutputFeed,
+      getShellOutputFeeds: () => childShellOutputFeed,
       getBlobWriter: () => childBlobWriter,
       getContextDir: () => childContextDir,
       extraToolPlugins: [

src/tui/product-host.ts

@@ -179,11 +179,12 @@ export interface ProductHostConfig {
    */
   readonly subAgentSessions?: () => readonly TaskProgressSession[];
   /**
-   * The session's bounded shell-output feed, polled on the same sticky tick to
-   * paint a running command's live output tail onto its pending row. Omitted
-   * hosts (tests, the demo shell) paint pending shell rows without a tail.
+   * Per-call bounded shell-output feeds, polled on the same sticky tick to
+   * paint a running command's live output tail onto the pending row that owns
+   * that call. Omitted hosts (tests, the demo shell) paint pending shell rows
+   * without a tail.
    */
-  readonly shellOutputFeed?: () => ShellOutputFeed | undefined;
+  readonly shellOutputFeed?: (callId: string) => ShellOutputFeed | undefined;
   /**
    * Renderer factory override for headless mounting in tests.
    * Defaults to the real `createCliRenderer`; tests inject a
@@ -391,7 +392,7 @@ export async function mountProductHost(
       }
       // Live shell tail: deduped in the bridge, so an unchanged snapshot is a
       // no-op and this poll cadence (200 ms) is the paint cadence.
-      bridge.syncShellOutputs(config.shellOutputFeed?.());
+      bridge.syncShellOutputs(config.shellOutputFeed);
       // Elapsed clock, stall flip, and post-finish linger are wall-time — repaint
       // the strip on this tick while sticky is needed. paintChromeZones re-enters
       // setChromeZones (which may paintChrome again on an unchanged-zone path),