fix(terminal): bun-safe exec attach, box self-heal, surfaced errors

The workspace terminal died everywhere: bun's node:http client never
emits the client 'upgrade' event (it surfaces the 101 as a plain
response), so docker-modem's exec.start({hijack:true}) waited forever —
a connected WebSocket with no bytes flowing (prod), or an instant close
when the socket was unreachable (dev). The pre-Phase-1 bridge avoided
dockerode attach for exactly this reason; the real-TTY-resize rewrite
reintroduced it.

- runtime/exec-stream.ts: speak the exec-start HTTP upgrade over a raw
  net socket (docker and podman both answer 101 UPGRADED). Works under
  bun and node; honors the modem's socketPath/host:port. The stream is
  returned paused so the first screen paint can't race the caller's
  data handler. Exec create/resize stay on dockerode (plain POSTs).
- user-container.ts: acquireUserBox re-ensures the container on every
  acquire instead of trusting a cached "initialized" flag, so a box
  removed behind the worker's back (manual rm, another instance's
  reaper) is recreated instead of bricking terminals and agent turns
  until restart.
- terminal.ts: bridge failures are logged and close the WebSocket with
  a clamped human-readable reason instead of a silent catch;
  xterm-session.tsx displays that reason instead of a mute
  "Session ended". Also roomier terminal line spacing (1.2 -> 1.35).

apps/agent-worker/src/runtime/exec-stream.ts

@@ -0,0 +1,131 @@
+import net from 'node:net';
+import type { Duplex } from 'node:stream';
+
+import { getDockerClient } from './docker';
+
+// Hand-rolled replacement for docker-modem's hijacked `exec.start`. Bun's
+// node:http client never emits the `'upgrade'` event (it surfaces the 101 as a
+// plain response), so dockerode's `exec.start({ hijack: true })` waits forever
+// under bun — the terminal connects but no bytes ever flow. Docker and podman
+// both answer the exec-start upgrade with `101 UPGRADED` followed by the raw
+// TTY byte stream on the same connection, so we speak that one exchange
+// directly over a net socket and hand the socket to the caller.
+
+export type UpgradeParse =
+  | { complete: false }
+  | { complete: true; statusCode: number; remainder: Buffer };
+
+const HEADER_TERMINATOR = Buffer.from('\r\n\r\n');
+
+// Parse an accumulating exec-start response buffer. Incomplete until the CRLF
+// CRLF header terminator arrives; after it, `remainder` is the start of the raw
+// stream (or the error body) and must be preserved byte-exact.
+export const parseUpgradeResponse = (data: Buffer): UpgradeParse => {
+  const headerEnd = data.indexOf(HEADER_TERMINATOR);
+  if (headerEnd === -1) return { complete: false };
+  const statusLine =
+    data.subarray(0, headerEnd).toString('latin1').split('\r\n', 1)[0] ?? '';
+  const match = /^HTTP\/1\.[01] (\d{3})/.exec(statusLine);
+  return {
+    complete: true,
+    statusCode: match?.[1] ? Number.parseInt(match[1], 10) : 0,
+    remainder: data.subarray(headerEnd + HEADER_TERMINATOR.length),
+  };
+};
+
+type ModemTarget = {
+  socketPath?: string;
+  host?: string;
+  port?: number | string;
+  protocol?: string;
+};
+
+/**
+ * Start a created exec and return the raw bidirectional TTY stream. Writes are
+ * stdin; reads are the terminal output (Tty:true, so no stdcopy framing). The
+ * socket is returned PAUSED: output that arrives while the caller is still
+ * wiring handlers stays buffered instead of being dropped (the first tmux
+ * screen paint often lands within the same tick as the upgrade). Callers must
+ * `stream.resume()` once their `data` handler is attached.
+ */
+export const openExecStream = (
+  execId: string,
+  { timeoutMs = 15_000 }: { timeoutMs?: number } = {},
+): Promise<Duplex> =>
+  new Promise((resolve, reject) => {
+    const modem = getDockerClient().modem as ModemTarget;
+    if (modem.protocol && modem.protocol !== 'http' && !modem.socketPath) {
+      reject(
+        new Error(
+          `Exec streaming supports unix sockets and plain tcp docker hosts; got protocol "${modem.protocol}".`,
+        ),
+      );
+      return;
+    }
+    const socket = modem.socketPath
+      ? net.connect({ path: modem.socketPath })
+      : net.connect({
+          host: modem.host ?? 'localhost',
+          port: Number(modem.port ?? 2375),
+        });
+
+    let buffered = Buffer.alloc(0);
+    let settled = false;
+    const fail = (error: Error) => {
+      if (settled) return;
+      settled = true;
+      clearTimeout(timer);
+      socket.destroy();
+      reject(error);
+    };
+    const timer = setTimeout(
+      () => fail(new Error(`Exec start timed out after ${timeoutMs}ms.`)),
+      timeoutMs,
+    );
+
+    socket.on('error', (error: Error) =>
+      fail(new Error(`Exec stream connection failed: ${error.message}`)),
+    );
+    socket.on('close', () =>
+      fail(
+        new Error(
+          `Exec stream closed before the upgrade completed${buffered.length ? `: ${buffered.toString('utf8').slice(0, 300)}` : '.'}`,
+        ),
+      ),
+    );
+
+    socket.on('connect', () => {
+      const body = JSON.stringify({ Detach: false, Tty: true });
+      socket.write(
+        `POST /exec/${encodeURIComponent(execId)}/start HTTP/1.1\r\n` +
+          'Host: docker\r\n' +
+          'Content-Type: application/json\r\n' +
+          `Content-Length: ${Buffer.byteLength(body)}\r\n` +
+          'Connection: Upgrade\r\n' +
+          'Upgrade: tcp\r\n' +
+          '\r\n' +
+          body,
+      );
+    });
+
+    const onData = (chunk: Buffer) => {
+      buffered = Buffer.concat([buffered, chunk]);
+      const parsed = parseUpgradeResponse(buffered);
+      if (!parsed.complete) return;
+      if (parsed.statusCode !== 101 && parsed.statusCode !== 200) {
+        fail(
+          new Error(
+            `Exec start failed: HTTP ${parsed.statusCode || 'unparseable'} ${parsed.remainder.toString('utf8').slice(0, 300)}`.trim(),
+          ),
+        );
+        return;
+      }
+      settled = true;
+      clearTimeout(timer);
+      socket.off('data', onData);
+      socket.pause();
+      if (parsed.remainder.length > 0) socket.unshift(parsed.remainder);
+      resolve(socket);
+    };
+    socket.on('data', onData);
+  });

apps/agent-worker/src/terminal.ts

@@ -7,6 +7,7 @@ import type { BoxHandle } from './user-container';
 import { boxHomePaths } from './box';
 import { env } from './env';
 import { getDockerClient } from './runtime/docker';
+import { openExecStream } from './runtime/exec-stream';
 import {
   parseBoxTokenUsername,
   verifyBoxTerminalToken,
@@ -43,6 +44,12 @@ export const parseResizeMessage = (
   }
 };
 
+// Close with a reason the browser can display. The WebSocket close-frame
+// payload caps the reason at 123 bytes — `ws` throws past that — so clamp.
+const closeWithReason = (ws: WebSocket, code: number, reason: string) => {
+  ws.close(code, Buffer.from(reason, 'utf8').subarray(0, 120).toString('utf8'));
+};
+
 // The login-shell command shared by the job and box terminals: prefer a
 // resumable tmux session, else fall back to a plain interactive login shell.
 const SHELL_CMD = [
@@ -118,7 +125,9 @@ const runShellBridge = async (ws: WebSocket, opts: ShellBridgeOptions) => {
     handleHolder.current = handle;
     boxName = handle.boxName;
   } catch (error) {
-    ws.close(
+    console.error('Terminal box acquire failed:', error);
+    closeWithReason(
+      ws,
       1011,
       `Failed to start terminal: ${error instanceof Error ? error.message : 'unknown error'}`,
     );
@@ -131,18 +140,38 @@ const runShellBridge = async (ws: WebSocket, opts: ShellBridgeOptions) => {
     return;
   }
 
-  const docker = getDockerClient();
-  const container = docker.getContainer(boxName);
-  const exec = await container.exec({
-    AttachStdin: true,
-    AttachStdout: true,
-    AttachStderr: true,
-    Tty: true,
-    Cmd: SHELL_CMD,
-    Env: ['TERM=xterm-256color', ...opts.envFlags],
-    WorkingDir: opts.cwd,
-  });
-  const stream = await exec.start({ hijack: true, stdin: true, Tty: true });
+  // The exec create is a plain API call, but the attach is NOT dockerode's
+  // `exec.start({ hijack: true })`: docker-modem's hijack waits for node's http
+  // client `'upgrade'` event, which bun never emits, so under bun it hangs
+  // forever with a connected-but-dead terminal. `openExecStream` speaks the
+  // upgrade exchange over a raw socket instead (works under bun and node,
+  // against docker and podman). `exec.resize` is a plain POST and stays on
+  // dockerode.
+  let exec: import('dockerode').Exec;
+  let stream: Duplex;
+  try {
+    const docker = getDockerClient();
+    const container = docker.getContainer(boxName);
+    exec = await container.exec({
+      AttachStdin: true,
+      AttachStdout: true,
+      AttachStderr: true,
+      Tty: true,
+      Cmd: SHELL_CMD,
+      Env: ['TERM=xterm-256color', ...opts.envFlags],
+      WorkingDir: opts.cwd,
+    });
+    stream = await openExecStream(exec.id);
+  } catch (error) {
+    console.error('Terminal exec attach failed:', error);
+    closeWithReason(
+      ws,
+      1011,
+      `Failed to attach terminal: ${error instanceof Error ? error.message : 'unknown error'}`,
+    );
+    cleanup();
+    return;
+  }
   execHolder.current = { exec, stream };
 
   if (isReleased()) {
@@ -153,10 +182,6 @@ const runShellBridge = async (ws: WebSocket, opts: ShellBridgeOptions) => {
 
   await exec.resize({ h: rows, w: cols }).catch(() => undefined);
 
-  // Replay any keystrokes the client sent before the process was ready.
-  for (const buffered of pendingInput) stream.write(buffered);
-  pendingInput.length = 0;
-
   stream.on('data', (chunk: Buffer) => {
     if (ws.readyState === ws.OPEN) ws.send(chunk, { binary: true });
   });
@@ -166,6 +191,12 @@ const runShellBridge = async (ws: WebSocket, opts: ShellBridgeOptions) => {
   stream.on('error', () => {
     if (ws.readyState === ws.OPEN) ws.close();
   });
+
+  // Replay any keystrokes the client sent before the process was ready, then
+  // start the (paused) stream now that the output handler is attached.
+  for (const buffered of pendingInput) stream.write(buffered);
+  pendingInput.length = 0;
+  stream.resume();
 };
 
 // Job terminal: opens the shell at the repo checkout, with the job's selected
@@ -231,8 +262,15 @@ export const attachTerminalServer = (server: Server) => {
         return;
       }
       wss.handleUpgrade(request, socket, head, (ws) => {
-        void bridgeBox(ws, username).catch(() => {
-          if (ws.readyState === ws.OPEN) ws.close();
+        void bridgeBox(ws, username).catch((error: unknown) => {
+          console.error(`Box terminal bridge failed (${username}):`, error);
+          if (ws.readyState === ws.OPEN) {
+            closeWithReason(
+              ws,
+              1011,
+              `Terminal bridge failed: ${error instanceof Error ? error.message : 'unknown error'}`,
+            );
+          }
         });
       });
       return;
@@ -250,8 +288,15 @@ export const attachTerminalServer = (server: Server) => {
       return;
     }
     wss.handleUpgrade(request, socket, head, (ws) => {
-      void bridge(ws, jobId).catch(() => {
-        if (ws.readyState === ws.OPEN) ws.close();
+      void bridge(ws, jobId).catch((error: unknown) => {
+        console.error(`Job terminal bridge failed (${jobId}):`, error);
+        if (ws.readyState === ws.OPEN) {
+          closeWithReason(
+            ws,
+            1011,
+            `Terminal bridge failed: ${error instanceof Error ? error.message : 'unknown error'}`,
+          );
+        }
       });
     });
   });

apps/agent-worker/src/user-container.ts

@@ -13,7 +13,6 @@ export type BoxHandle = { boxName: string; release: () => void };
 
 type Box = {
   name: string;
-  initialized: boolean;
   refs: Set<symbol>;
   idleTimer?: ReturnType<typeof setTimeout>;
 };
@@ -79,7 +78,6 @@ export const acquireUserBox = (args: {
     if (!box) {
       box = {
         name: userContainerName(args.username),
-        initialized: false,
         refs: new Set(),
       };
       boxes.set(args.username, box);
@@ -88,10 +86,13 @@ export const acquireUserBox = (args: {
     // release it without leaking registry state.
     const handle = makeHandle(args.username, box);
     try {
-      if (!box.initialized) {
-        box.name = await ensureUserContainer(args);
-        box.initialized = true;
-      }
+      // Re-verify with the container runtime on EVERY acquire, not just the
+      // first: the registry goes stale when the box is removed behind the
+      // worker's back (manual `docker rm`, another worker's idle reaper), and a
+      // cached "initialized" flag then hands out handles to a dead container
+      // until restart. ensureUserContainer is idempotent — an inspect when the
+      // box is running, a recreate when it isn't.
+      box.name = await ensureUserContainer(args);
       return handle;
     } catch (error) {
       handle.release();
@@ -109,7 +110,7 @@ export const reconcileExistingBoxes = async (): Promise<void> => {
   for (const name of names) {
     const username = name.replace(/^spoon-box-/, '');
     if (boxes.has(username)) continue;
-    const box: Box = { name, initialized: false, refs: new Set() };
+    const box: Box = { name, refs: new Set() };
     boxes.set(username, box);
     scheduleReapIfIdle(username);
   }

apps/agent-worker/tests/unit/exec-stream.test.ts

@@ -0,0 +1,72 @@
+import { describe, expect, it } from 'vitest';
+
+// exec-stream imports the docker runtime, whose env module requires these at
+// import time — set them before the (hoist-free) dynamic import.
+process.env.SPOON_WORKER_TOKEN ??= 'test-worker-token';
+process.env.GITHUB_APP_ID ??= '123';
+process.env.GITHUB_APP_PRIVATE_KEY ??= 'test-key';
+const { parseUpgradeResponse } = await import('../../src/runtime/exec-stream');
+
+describe('parseUpgradeResponse', () => {
+  it('reports incomplete until the header terminator arrives', () => {
+    expect(parseUpgradeResponse(Buffer.from('')).complete).toBe(false);
+    expect(
+      parseUpgradeResponse(Buffer.from('HTTP/1.1 101 UPGRADED\r\n')).complete,
+    ).toBe(false);
+    expect(
+      parseUpgradeResponse(
+        Buffer.from('HTTP/1.1 101 UPGRADED\r\nConnection: Upgrade\r\n'),
+      ).complete,
+    ).toBe(false);
+  });
+
+  it('parses a 101 upgrade and preserves stream bytes after the headers', () => {
+    const payload = Buffer.concat([
+      Buffer.from(
+        'HTTP/1.1 101 UPGRADED\r\n' +
+          'Content-Type: application/vnd.docker.raw-stream\r\n' +
+          'Connection: Upgrade\r\n' +
+          'Upgrade: tcp\r\n' +
+          '\r\n',
+      ),
+      Buffer.from([0x1b, 0x5b, 0x48, 0x00, 0xff]),
+    ]);
+    const parsed = parseUpgradeResponse(payload);
+    expect(parsed).toMatchObject({ complete: true, statusCode: 101 });
+    if (!parsed.complete) throw new Error('unreachable');
+    expect([...parsed.remainder]).toEqual([0x1b, 0x5b, 0x48, 0x00, 0xff]);
+  });
+
+  it('parses a 200 stream response (docker without upgrade) as success-shaped', () => {
+    const parsed = parseUpgradeResponse(
+      Buffer.from('HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nhi'),
+    );
+    expect(parsed).toMatchObject({ complete: true, statusCode: 200 });
+    if (!parsed.complete) throw new Error('unreachable');
+    expect(parsed.remainder.toString('utf8')).toBe('hi');
+  });
+
+  it('parses error statuses with their body so callers can report them', () => {
+    const parsed = parseUpgradeResponse(
+      Buffer.from(
+        'HTTP/1.1 404 Not Found\r\nContent-Length: 26\r\n\r\n{"message":"no such exec"}',
+      ),
+    );
+    expect(parsed).toMatchObject({ complete: true, statusCode: 404 });
+    if (!parsed.complete) throw new Error('unreachable');
+    expect(parsed.remainder.toString('utf8')).toBe(
+      '{"message":"no such exec"}',
+    );
+  });
+
+  it('returns statusCode 0 for an unparseable status line', () => {
+    const parsed = parseUpgradeResponse(Buffer.from('garbage\r\n\r\n'));
+    expect(parsed).toMatchObject({ complete: true, statusCode: 0 });
+  });
+
+  it('does not treat a bare LF-LF as a header terminator', () => {
+    expect(
+      parseUpgradeResponse(Buffer.from('HTTP/1.1 101 UPGRADED\n\n')).complete,
+    ).toBe(false);
+  });
+});

apps/agent-worker/tests/unit/user-container.test.ts

@@ -22,9 +22,39 @@ describe('box registry', () => {
     vi.clearAllMocks();
   });
 
-  test('serializes concurrent acquire into a single docker run', async () => {
+  test('re-ensures the container on every acquire so an externally removed box self-heals', async () => {
     const module = await load();
     const docker = await import('../../src/runtime/docker');
+    const first = await module.acquireUserBox({
+      username: 'zoe',
+      workdir: '/w',
+      containerHome: '/home/zoe',
+    });
+    first.release();
+    // Simulate `docker rm` behind the worker's back: the registry entry still
+    // exists, but acquire must go back to the runtime rather than trust it.
+    const second = await module.acquireUserBox({
+      username: 'zoe',
+      workdir: '/w',
+      containerHome: '/home/zoe',
+    });
+    expect(docker.ensureUserContainer).toHaveBeenCalledTimes(2);
+    expect(second.boxName).toBe('spoon-box-zoe');
+  });
+
+  test('serializes concurrent acquires so ensure calls never overlap', async () => {
+    const module = await load();
+    const docker = await import('../../src/runtime/docker');
+    let inFlight = 0;
+    vi.mocked(docker.ensureUserContainer).mockImplementation(
+      async (args: { username: string }) => {
+        inFlight += 1;
+        expect(inFlight).toBe(1);
+        await Promise.resolve();
+        inFlight -= 1;
+        return `spoon-box-${args.username}`;
+      },
+    );
     const [a, b] = await Promise.all([
       module.acquireUserBox({
         username: 'alice',
@@ -37,7 +67,6 @@ describe('box registry', () => {
         containerHome: '/home/alice',
       }),
     ]);
-    expect(docker.ensureUserContainer).toHaveBeenCalledTimes(1);
     expect(a.boxName).toBe('spoon-box-alice');
     a.release();
     await vi.advanceTimersByTimeAsync(2000);
@@ -83,7 +112,7 @@ describe('box registry', () => {
     );
   });
 
-  test('ensures an adopted box once before returning acquire handles', async () => {
+  test('acquiring an adopted box ensures it and returns valid handles', async () => {
     const module = await load();
     const docker = await import('../../src/runtime/docker');
     vi.mocked(docker.listWorkspaceContainerNames).mockResolvedValueOnce([
@@ -102,7 +131,7 @@ describe('box registry', () => {
       containerHome: '/home/casey',
     });
 
-    expect(docker.ensureUserContainer).toHaveBeenCalledTimes(1);
+    expect(docker.ensureUserContainer).toHaveBeenCalled();
     expect(first.boxName).toBe('spoon-box-casey');
     expect(second.boxName).toBe('spoon-box-casey');
   });