From a640b5cb9495186f9bf68f68f4263d6cca09d364 Mon Sep 17 00:00:00 2001 From: Gabriel Brown Date: Sun, 12 Jul 2026 22:14:37 -0400 Subject: [PATCH] fix(terminal): bun-safe exec attach, box self-heal, surfaced errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 131 ++++++++++++++++++ apps/agent-worker/src/terminal.ts | 87 +++++++++--- apps/agent-worker/src/user-container.ts | 15 +- .../tests/unit/exec-stream.test.ts | 72 ++++++++++ .../tests/unit/user-container.test.ts | 37 ++++- .../agent-workspace/xterm-session.tsx | 15 +- 6 files changed, 321 insertions(+), 36 deletions(-) create mode 100644 apps/agent-worker/src/runtime/exec-stream.ts create mode 100644 apps/agent-worker/tests/unit/exec-stream.test.ts diff --git a/apps/agent-worker/src/runtime/exec-stream.ts b/apps/agent-worker/src/runtime/exec-stream.ts new file mode 100644 index 0000000..c568842 --- /dev/null +++ b/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 => + 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); + }); diff --git a/apps/agent-worker/src/terminal.ts b/apps/agent-worker/src/terminal.ts index b3e9a19..e7e8350 100644 --- a/apps/agent-worker/src/terminal.ts +++ b/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'}`, + ); + } }); }); }); diff --git a/apps/agent-worker/src/user-container.ts b/apps/agent-worker/src/user-container.ts index f5ac6d5..d56ed0d 100644 --- a/apps/agent-worker/src/user-container.ts +++ b/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; idleTimer?: ReturnType; }; @@ -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 => { 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); } diff --git a/apps/agent-worker/tests/unit/exec-stream.test.ts b/apps/agent-worker/tests/unit/exec-stream.test.ts new file mode 100644 index 0000000..d26c8ee --- /dev/null +++ b/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); + }); +}); diff --git a/apps/agent-worker/tests/unit/user-container.test.ts b/apps/agent-worker/tests/unit/user-container.test.ts index 7f6a7a3..930ae21 100644 --- a/apps/agent-worker/tests/unit/user-container.test.ts +++ b/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'); }); diff --git a/apps/next/src/components/agent-workspace/xterm-session.tsx b/apps/next/src/components/agent-workspace/xterm-session.tsx index 13b26a6..e4415d3 100644 --- a/apps/next/src/components/agent-workspace/xterm-session.tsx +++ b/apps/next/src/components/agent-workspace/xterm-session.tsx @@ -189,7 +189,7 @@ export const XtermSession = ({ const term = new Terminal({ fontFamily: TERMINAL_FONT, fontSize: 13, - lineHeight: 1.2, + lineHeight: 1.35, cursorBlink: true, theme: themeIsLight ? lightTheme : darkTheme, allowProposedApi: true, @@ -257,8 +257,13 @@ export const XtermSession = ({ if (typeof event.data === 'string') term.write(event.data); else term.write(new Uint8Array(event.data)); }; - ws.onclose = () => { - if (!isAborted()) setStatus('closed'); + ws.onclose = (event: CloseEvent) => { + if (isAborted()) return; + setStatus('closed'); + // The worker attaches a human-readable reason when the bridge fails + // (e.g. container runtime unreachable) — show it instead of a mute + // "Session ended". + if (event.reason) setErrorText(event.reason); }; ws.onerror = () => { if (!isAborted()) setStatus('error'); @@ -323,7 +328,9 @@ export const XtermSession = ({ {status === 'connected' ? `Connected · ${label}` : status === 'idle' - ? (waitingLabel ? 'Waiting…' : 'Idle') + ? waitingLabel + ? 'Waiting…' + : 'Idle' : status === 'connecting' ? 'Connecting…' : status === 'closed'