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

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);
  });