char-slop/berry

git clone https://git.t4t.associates/char-slop/berry

Charlotte Somcleanup: hang up PTYs when their process exits05756372026-10-18

// Where a conversation's files and commands are. Server-side only: web bundles import nothing from here.

import { homedir } from "node:os";
import { resolve } from "node:path";
import { spawn } from "node:child_process";
import process from "node:process";
import { type Readable, Writable } from "node:stream";
import type { Context } from "@earendil-works/chord";
import { type ExecutionEnv, getOrThrow } from "@earendil-works/pi-durable/env";
import { NodeExecutionEnv } from "@earendil-works/pi-durable/env/node";
import { ptyHere } from "./pty.ts";

/** The name of the machine berry itself runs on. */
export const LOCAL = "local";

/** A machine a conversation is bound to, at its working directory (`cwd`). */
export interface Environment extends ExecutionEnv {
  /** `local`, or the name settings give it. */
  readonly name: string;
  /** The machine's `process.platform`, such as `linux`, `darwin`, or `win32`. */
  readonly platform: string;
  readonly home: string;
  /** The process environment commands start from there. */
  readonly variables: Readonly<Record<string, string>>;
  /** Aborted when the remote connection is lost; local environments have no connection lifetime. */
  readonly lifetime?: AbortSignal;
  /**
   * Start a program there, `argv[0]` found on its PATH, and talk to it while it runs: a desktop's helper, or an MCP
   * server. `env` adds to the variables it starts from; aborting `context` kills it. Rejects when it cannot start.
   */
  spawn(argv: readonly string[], options: SpawnOptions, context: Context): Promise<Process>;
  /**
   * Start a program there on a terminal of its own, in a session it leads, as a login shell wants. Only on linux, for
   * now. Aborting `context` hangs the terminal up, and so does the program exiting, whatever it leaves running.
   */
  pty(argv: readonly string[], options: PtyOptions, context: Context): Promise<Pty>;
  /** Open a TCP connection from there, so `127.0.0.1` is that machine's own. */
  connect(hostname: string, port: number, context: Context): Promise<Socket>;
  /** Fetch from this machine, with streaming responses and cancellation through `init.signal`. */
  fetch(input: string | URL, init?: RequestInit): Promise<Response>;
}

export type SpawnOptions = {
  readonly cwd?: string;
  readonly env?: Record<string, string>;
  /** Own a POSIX process group; remote workers also stop descendants when its leader exits. */
  readonly group?: boolean;
};

export type PtyOptions = SpawnOptions & { readonly cols: number; readonly rows: number };

/** A program on a pseudo-terminal: what it writes to the terminal, what it reads as typed there, and its size. */
export interface Pty {
  readonly readable: ReadableStream<Uint8Array>;
  readonly writable: WritableStream<Uint8Array>;
  resize(cols: number, rows: number): void;
  /** Its exit code, or `null` when a signal ended it. */
  readonly exited: Promise<number | null>;
  /** Hang the terminal up, as closing a terminal window does. */
  kill(): void;
}

/** A program running on an environment's machine. */
export interface Process {
  readonly stdin: WritableStream<Uint8Array>;
  readonly stdout: ReadableStream<Uint8Array>;
  readonly stderr: ReadableStream<Uint8Array>;
  /** Its exit code, or `null` when a signal ended it. */
  readonly exited: Promise<number | null>;
  kill(signal?: "SIGTERM" | "SIGKILL"): void;
}

/** A TCP connection from an environment's machine. */
export interface Socket {
  readonly readable: ReadableStream<Uint8Array>;
  readonly writable: WritableStream<Uint8Array>;
  close(): void;
}

class LocalEnvironment extends NodeExecutionEnv implements Environment {
  readonly name = LOCAL;
  readonly platform = process.platform;
  readonly home = homedir();
  get variables() {
    return Deno.env.toObject();
  }
  spawn(argv: readonly string[], options: SpawnOptions, context: Context): Promise<Process> {
    return Promise.resolve(spawnHere(argv, {
      ...options,
      cwd: options.cwd === undefined ? this.cwd : resolve(this.cwd, options.cwd),
    }, context));
  }
  connect(hostname: string, port: number, context: Context): Promise<Socket> {
    return connectHere(hostname, port, context);
  }
  fetch(input: string | URL, init?: RequestInit): Promise<Response> {
    return fetch(input, init);
  }
  pty(argv: readonly string[], options: PtyOptions, context: Context): Promise<Pty> {
    // failing to start rejects, rather than throws
    return new Promise((done) =>
      done(ptyHere(argv, {
        ...options,
        cwd: options.cwd === undefined ? this.cwd : resolve(this.cwd, options.cwd),
      }, context))
    );
  }
}

// Readable.toWeb cancels pipes with an AbortError that Deno's node adapter can leave unhandled.
function processOutput(pipe: Readable): ReadableStream<Uint8Array> {
  let closed = false;
  return new ReadableStream({
    start(controller) {
      pipe.on("data", (chunk: Uint8Array) => {
        if (closed) return;
        controller.enqueue(chunk);
        if ((controller.desiredSize ?? 0) <= 0) pipe.pause();
      });
      pipe.on("error", (error) => {
        if (closed) return;
        closed = true;
        controller.error(error);
      });
      pipe.once("close", () => {
        if (closed) return;
        closed = true;
        controller.close();
      });
      pipe.once("end", () => {
        if (closed) return;
        closed = true;
        controller.close();
      });
    },
    pull: () => void pipe.resume(),
    cancel() {
      closed = true;
      pipe.destroy();
    },
  });
}

/** A program on this machine, `cwd` as given. */
export async function spawnHere(
  argv: readonly string[],
  { cwd, env, group }: SpawnOptions,
  context: Context,
): Promise<Process> {
  context.abortSignal?.throwIfAborted();
  if (group && process.platform !== "win32") {
    const child = spawn(argv[0], argv.slice(1), {
      cwd,
      env: { ...process.env, ...env },
      detached: true,
      stdio: "pipe",
    });
    const kill = (signal: "SIGTERM" | "SIGKILL" = "SIGTERM") => {
      // the group may still have descendants after its leader has exited
      if (child.pid !== undefined) {
        try {
          process.kill(-child.pid, signal);
        } catch {
          child.kill(signal);
        }
      } else child.kill(signal);
    };
    const abort = () => kill();
    context.abortSignal?.addEventListener("abort", abort, { once: true });
    if (context.abortSignal?.aborted) abort();
    const exited = new Promise<number | null>((resolve) => child.once("exit", resolve));
    void exited.then(() => context.abortSignal?.removeEventListener("abort", abort));
    try {
      await new Promise<void>((resolve, reject) => {
        child.once("spawn", resolve);
        child.once("error", reject);
      });
    } catch (error) {
      context.abortSignal?.removeEventListener("abort", abort);
      throw error;
    }
    child.on("error", () => {});
    child.stdin!.on("error", () => {});
    return {
      stdin: Writable.toWeb(child.stdin!) as WritableStream<Uint8Array>,
      stdout: processOutput(child.stdout!),
      stderr: processOutput(child.stderr!),
      exited,
      kill,
    };
  }
  const child = new Deno.Command(argv[0], {
    args: argv.slice(1),
    ...(cwd === undefined ? {} : { cwd }),
    ...(env === undefined ? {} : { env }),
    stdin: "piped",
    stdout: "piped",
    stderr: "piped",
  }).spawn();
  const kill = (signal: "SIGTERM" | "SIGKILL" = "SIGTERM") => {
    try {
      child.kill(signal);
    } catch {
      // already gone
    }
  };
  const abort = () => kill();
  context.abortSignal?.addEventListener("abort", abort, { once: true });
  if (context.abortSignal?.aborted) abort();
  const exited = child.status.then((status) => (status.signal ? null : status.code));
  void exited.finally(() => context.abortSignal?.removeEventListener("abort", abort));
  return { stdin: child.stdin, stdout: child.stdout, stderr: child.stderr, exited, kill };
}

/** A TCP connection from this machine. */
export async function connectHere(
  hostname: string,
  port: number,
  context: Context,
): Promise<Socket> {
  context.abortSignal?.throwIfAborted();
  const connection = await Deno.connect({ hostname, port });
  return {
    readable: connection.readable,
    writable: connection.writable,
    close() {
      try {
        connection.close();
      } catch {
        // already closed
      }
    },
  };
}

/** The machine berry runs on, at `cwd`. */
export function localEnvironment(cwd: string): Environment {
  return new LocalEnvironment({ cwd });
}

/** The directory `path` names from the environment's `cwd`, links resolved; an empty path is home. */
export async function directory(env: Environment, path: string, context: Context): Promise<string> {
  const target = getOrThrow(await env.canonicalPath(path.trim() || env.home, context));
  if (getOrThrow(await env.fileInfo(target, context)).kind !== "directory") {
    throw new Error(`Not a directory: ${target}`);
  }
  return target;
}