paseo/packages/server/src/terminal/terminal-worker-process.ts

import { createTerminalManager } from "./terminal-manager.js";
import { randomUUID } from "node:crypto";
import { constants } from "node:os";
import { captureTerminalLines } from "./terminal-capture.js";
import { TerminalOutputCoalescer } from "./terminal-output-coalescer.js";
import type {
  TerminalPtyProcess,
  TerminalPtyLaunchInput,
  TerminalSession,
  TerminalStateSnapshotOptions,
} from "./terminal.js";
import type {
  TerminalWorkerRequest,
  TerminalWorkerParentMessage,
  TerminalWorkerPtyEvent,
  TerminalWorkerMessage,
  TerminalWorkerStateResult,
  WorkerTerminalInfo,
} from "./terminal-worker-protocol.js";

type TerminalCreateRequest = Extract<TerminalWorkerRequest, { type: "createTerminal" }>;

const ptyProcesses = new Map<string, RuntimePtyProxy>();
const pendingPtySpawns = new Map<
  string,
  { resolve: (process: TerminalPtyProcess | null) => void; reject: (error: Error) => void }
>();
const manager = createTerminalManager({
  resolvePtyLauncher: () => launchRuntimePty,
});
const unsubscribeByTerminalId = new Map<string, Array<() => void>>();
const outputCoalescerByTerminalId = new Map<string, TerminalOutputCoalescer>();
let ipcClosing = false;

interface InFlightTerminalCreateRequest {
  requestId: string;
  errorReported: boolean;
}

let inFlightTerminalCreateRequest: InFlightTerminalCreateRequest | null = null;

// The conpty failure signal is process-scoped, not request-scoped. Serializing
// creates keeps an async spawn failure attributable to exactly one request.
let createTerminalQueue: Promise<void> = Promise.resolve();

// node-pty completes its Windows conpty spawn asynchronously on a separate
// conout worker thread. When that spawn fails (bad cwd, missing command, etc.)
// it throws an exception there that cannot be caught at the call site and would
// otherwise crash this worker process and sever every existing terminal.
process.on("uncaughtException", (error) => {
  console.error("Terminal worker uncaught exception (kept alive):", error);
  reportInFlightTerminalCreateFailure(error);
});

function sendToParent(message: TerminalWorkerMessage): void {
  if (ipcClosing || !process.connected || !process.send) {
    closePtyBridge(new Error("Terminal parent IPC is not connected"));
    return;
  }
  try {
    process.send(message, (error) => {
      if (error) closePtyBridge(error);
    });
  } catch (error) {
    closePtyBridge(error);
  }
}

function closePtyBridge(error: unknown): void {
  ipcClosing = true;
  const failure = error instanceof Error ? error : new Error(String(error));
  for (const pending of pendingPtySpawns.values()) pending.reject(failure);
  pendingPtySpawns.clear();
  for (const proxy of ptyProcesses.values()) {
    proxy.emitExit({ code: 1, signal: null });
  }
  ptyProcesses.clear();
}

interface RuntimePtyProxy extends TerminalPtyProcess {
  emitData(data: string): void;
  emitExit(exit: { code: number | null; signal: NodeJS.Signals | null }): void;
}

function launchRuntimePty(
  input: TerminalPtyLaunchInput,
): TerminalPtyProcess | null | Promise<TerminalPtyProcess | null> {
  return new Promise<TerminalPtyProcess | null>((resolve, reject) => {
    const processId = randomUUID();
    pendingPtySpawns.set(processId, { resolve, reject });
    sendToParent({ type: "ptySpawn", processId, input });
  });
}

function createRuntimePtyProxy(processId: string): RuntimePtyProxy {
  const dataListeners = new Set<(data: string) => void>();
  const exitListeners = new Set<(event: { exitCode: number; signal?: number }) => void>();
  let pendingData = "";
  let pendingExit: { exitCode: number; signal?: number } | null = null;
  return {
    pid: -1,
    write(data) {
      sendToParent({ type: "ptyWrite", processId, data });
    },
    resize(cols, rows) {
      sendToParent({ type: "ptyResize", processId, cols, rows });
    },
    kill(signal) {
      sendToParent({ type: "ptyKill", processId, signal });
    },
    onData(listener) {
      dataListeners.add(listener);
      if (pendingData) {
        const data = pendingData;
        pendingData = "";
        listener(data);
      }
      return { dispose: () => dataListeners.delete(listener) };
    },
    onExit(listener) {
      exitListeners.add(listener);
      const exit = pendingExit;
      if (exit) queueMicrotask(() => listener(exit));
      return { dispose: () => exitListeners.delete(listener) };
    },
    emitData(data) {
      if (dataListeners.size === 0) {
        pendingData += data;
        return;
      }
      for (const listener of dataListeners) listener(data);
    },
    emitExit(exit) {
      const signal = exit.signal ? constants.signals[exit.signal] : undefined;
      const event = { exitCode: exit.code ?? 0, signal };
      if (exitListeners.size === 0) pendingExit = event;
      else for (const listener of exitListeners) listener(event);
      dataListeners.clear();
      if (exitListeners.size > 0) exitListeners.clear();
    },
  };
}

function handlePtyEvent(message: TerminalWorkerPtyEvent): void {
  if (message.type === "ptySpawnResult") {
    const pending = pendingPtySpawns.get(message.processId);
    if (!pending) return;
    pendingPtySpawns.delete(message.processId);
    if (message.mode === "legacy") {
      pending.resolve(null);
      return;
    }
    const proxy = createRuntimePtyProxy(message.processId);
    ptyProcesses.set(message.processId, proxy);
    pending.resolve(proxy);
    return;
  }
  if (message.type === "ptySpawnError") {
    const pending = pendingPtySpawns.get(message.processId);
    pendingPtySpawns.delete(message.processId);
    pending?.reject(new Error(message.error));
    return;
  }
  const proxy = ptyProcesses.get(message.processId);
  if (!proxy) return;
  if (message.type === "ptyData") proxy.emitData(message.data);
  else {
    ptyProcesses.delete(message.processId);
    setImmediate(() => proxy.emitExit(message.exit));
  }
}

function isPtyEvent(message: TerminalWorkerParentMessage): message is TerminalWorkerPtyEvent {
  return (
    message.type === "ptySpawnResult" ||
    message.type === "ptySpawnError" ||
    message.type === "ptyData" ||
    message.type === "ptyExit"
  );
}

function buildTerminalStateResult(
  session: TerminalSession | undefined,
  options?: TerminalStateSnapshotOptions,
): TerminalWorkerStateResult {
  if (!session) {
    return null;
  }
  return { ...session.getStateSnapshot(options), replayPreamble: session.getReplayPreamble() };
}

function toTerminalInfo(session: TerminalSession): WorkerTerminalInfo {
  return {
    id: session.id,
    name: session.name,
    cwd: session.cwd,
    workspaceId: session.workspaceId,
    ...(session.getTitle() ? { title: session.getTitle() } : {}),
    activity: session.getActivity(),
  };
}

function terminalWorkerErrorMessage(error: unknown): string {
  return error instanceof Error ? error.message : "Terminal worker request failed";
}

function reportInFlightTerminalCreateFailure(error: unknown): void {
  if (!inFlightTerminalCreateRequest || inFlightTerminalCreateRequest.errorReported) {
    return;
  }
  inFlightTerminalCreateRequest.errorReported = true;
  sendToParent({
    type: "response",
    requestId: inFlightTerminalCreateRequest.requestId,
    ok: false,
    error: terminalWorkerErrorMessage(error),
  });
}

function clearTerminalSubscriptions(terminalId: string): void {
  const subscriptions = unsubscribeByTerminalId.get(terminalId);
  if (subscriptions) {
    for (const unsubscribe of subscriptions) {
      try {
        unsubscribe();
      } catch {
        // no-op
      }
    }
  }
  unsubscribeByTerminalId.delete(terminalId);
  const coalescer = outputCoalescerByTerminalId.get(terminalId);
  if (coalescer) {
    coalescer.dispose();
    outputCoalescerByTerminalId.delete(terminalId);
  }
}

function watchTerminal(session: TerminalSession): void {
  clearTerminalSubscriptions(session.id);

  // Coalesce pty output chunks into a single IPC message per ~5ms window so a
  // burst of small chunks no longer costs one process.send each. The batch
  // carries the LAST chunk's revision (the highest) so downstream snapshot
  // replay dedup stays correct.
  let pendingOutputRevision: number | undefined;
  const outputCoalescer = new TerminalOutputCoalescer({
    timers: { setTimeout, clearTimeout },
    onFlush: ({ payload }) => {
      const revision = pendingOutputRevision;
      pendingOutputRevision = undefined;
      sendToParent({
        type: "terminalMessage",
        terminalId: session.id,
        message: { type: "output", data: payload.toString("utf8"), revision },
      });
    },
  });
  outputCoalescerByTerminalId.set(session.id, outputCoalescer);

  const unsubscribeMessage = session.subscribe(
    (message) => {
      if (message.type === "output") {
        pendingOutputRevision = message.revision;
        outputCoalescer.handle(message.data);
        return;
      }
      // Non-output messages (snapshot/snapshotReady/titleChange) must not jump
      // ahead of buffered output: flush the coalescer first, then forward.
      outputCoalescer.flush();
      sendToParent({
        type: "terminalMessage",
        terminalId: session.id,
        message,
      });
    },
    // Creation already sends an authoritative state in terminalCreated. A second
    // asynchronous state snapshot can arrive after the create response and
    // overwrite a resize issued immediately by the caller.
    { initialSnapshot: "ready" },
  );
  const unsubscribeExit = session.onExit((info) => {
    outputCoalescer.flush();
    clearTerminalSubscriptions(session.id);
    sendToParent({
      type: "terminalExit",
      terminalId: session.id,
      info,
    });
  });
  const unsubscribeTitle = session.onTitleChange((title) => {
    outputCoalescer.flush();
    sendToParent({
      type: "terminalTitleChange",
      terminalId: session.id,
      title,
    });
  });
  const unsubscribeCommandFinished = session.onCommandFinished((info) => {
    outputCoalescer.flush();
    sendToParent({
      type: "terminalCommandFinished",
      terminalId: session.id,
      info,
    });
  });
  const unsubscribeActivity = session.onActivityChange((transition) => {
    sendToParent({
      type: "terminalActivityChange",
      terminalId: session.id,
      activity: transition.activity,
      previous: transition.previous,
    });
  });

  unsubscribeByTerminalId.set(session.id, [
    unsubscribeMessage,
    unsubscribeExit,
    unsubscribeTitle,
    unsubscribeCommandFinished,
    unsubscribeActivity,
  ]);
}

function enqueueCreateTerminalRequest(message: TerminalCreateRequest): Promise<void> {
  const nextRequest = createTerminalQueue.then(() => handleCreateTerminalRequest(message));
  createTerminalQueue = nextRequest.catch(() => {});
  return nextRequest;
}

async function handleCreateTerminalRequest(message: TerminalCreateRequest): Promise<void> {
  const request: InFlightTerminalCreateRequest = {
    requestId: message.requestId,
    errorReported: false,
  };
  inFlightTerminalCreateRequest = request;
  try {
    const { workspaceId } = message.options;
    if (!workspaceId) {
      throw new Error("workspaceId is required");
    }
    const session = await manager.createTerminal({ ...message.options, workspaceId });
    if (request.errorReported) {
      session.kill();
      return;
    }
    watchTerminal(session);
    const initialSnapshot = session.getStateSnapshot();
    sendToParent({
      type: "terminalCreated",
      terminal: toTerminalInfo(session),
      state: initialSnapshot.state,
    });
    sendToParent({
      type: "response",
      requestId: message.requestId,
      ok: true,
      result: {
        terminal: toTerminalInfo(session),
        state: initialSnapshot.state,
      },
    });
  } catch (error) {
    reportInFlightTerminalCreateFailure(error);
  } finally {
    if (inFlightTerminalCreateRequest === request) {
      inFlightTerminalCreateRequest = null;
    }
  }
}

async function handleRequest(message: TerminalWorkerRequest): Promise<void> {
  switch (message.type) {
    case "createTerminal": {
      await enqueueCreateTerminalRequest(message);
      return;
    }

    case "registerCwdEnv": {
      manager.registerCwdEnv({ cwd: message.cwd, env: message.env });
      sendToParent({ type: "response", requestId: message.requestId, ok: true });
      return;
    }

    case "setActivity": {
      await manager.setTerminalActivity(message.terminalId, message.state);
      sendToParent({ type: "response", requestId: message.requestId, ok: true });
      return;
    }

    case "clearAttention": {
      await manager.clearTerminalAttention(message.terminalId);
      sendToParent({ type: "response", requestId: message.requestId, ok: true });
      return;
    }

    case "killTerminal": {
      manager.killTerminal(message.terminalId);
      // Removal is owned by session.onExit -> terminalExit; the parent mirror
      // clears contribution and emits terminalsChanged from that single path.
      clearTerminalSubscriptions(message.terminalId);
      sendToParent({ type: "response", requestId: message.requestId, ok: true });
      return;
    }

    case "killTerminalAndWait": {
      await manager.killTerminalAndWait(message.terminalId, message.options);
      clearTerminalSubscriptions(message.terminalId);
      sendToParent({ type: "response", requestId: message.requestId, ok: true });
      return;
    }

    case "getTerminalState": {
      // Flush buffered output before snapshotting: the headless state already includes it,
      // so if the coalescer emitted it afterward (in a batch carrying a revision past the
      // snapshot's) the controller's revision dedup wouldn't drop it and the client would
      // see the bytes twice. Flushing first sends them with a revision <= the snapshot's.
      outputCoalescerByTerminalId.get(message.terminalId)?.flush();
      sendToParent({
        type: "response",
        requestId: message.requestId,
        ok: true,
        result: buildTerminalStateResult(manager.getTerminal(message.terminalId), message.options),
      });
      return;
    }

    case "captureTerminal": {
      const session = manager.getTerminal(message.terminalId);
      const result = session
        ? captureTerminalLines(session, {
            start: message.start,
            end: message.end,
            stripAnsi: message.stripAnsi,
          })
        : { lines: [], totalLines: 0 };
      sendToParent({
        type: "response",
        requestId: message.requestId,
        ok: true,
        result,
      });
      return;
    }

    case "killAll": {
      await manager.killAll();
      for (const terminalId of Array.from(unsubscribeByTerminalId.keys())) {
        clearTerminalSubscriptions(terminalId);
      }
      sendToParent({ type: "response", requestId: message.requestId, ok: true });
      return;
    }

    case "send": {
      const session = manager.getTerminal(message.terminalId);
      session?.send(message.message);
      sendToParent({ type: "response", requestId: message.requestId, ok: true });
      return;
    }
  }
}

process.on("message", (message: TerminalWorkerParentMessage) => {
  if (isPtyEvent(message)) {
    handlePtyEvent(message);
    return;
  }
  void handleRequest(message).catch((error: unknown) => {
    sendToParent({
      type: "response",
      requestId: message.requestId,
      ok: false,
      error: terminalWorkerErrorMessage(error),
    });
  });
});

process.once("disconnect", () => {
  closePtyBridge(new Error("Terminal parent IPC disconnected"));
  void manager.killAll();
});