char-slop/berry

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

Charlotte Somwindows: add terminal support via ConPTY + PowerShell0853dde2026-10-15

// Runs on the machine an environment names. berry starts it over ssh and bootstraps it on stdin; from then on, every
// line in is a call on this machine's environment, and every line out what came of one.

import { homedir } from "node:os";
import process from "node:process";
import { BACKGROUND_CONTEXT, withAbortSignal } from "@earendil-works/chord/context";
import type { Context } from "@earendil-works/chord";
import { FileError, type ShellExecOptions } from "@earendil-works/pi-durable/env";
import { NodeExecutionEnv } from "@earendil-works/pi-durable/env/node";
import { resolve } from "node:path";
import {
  connectHere,
  type Process,
  type Pty,
  type PtyOptions,
  spawnHere,
  type SpawnOptions,
} from "@berry/api/environment.ts";
import { ptyHere } from "@berry/api/pty.ts";
import {
  decode,
  encode,
  type FetchOptions,
  lines,
  PROTOCOL,
  type Request,
  type Response,
} from "./codec.ts";

/** What the bootstrap read of stdin before handing it on. */
type Input = { reader: ReadableStreamDefaultReader<Uint8Array>; buffered: Uint8Array };
const input = (globalThis as { berryStdin?: Input }).berryStdin ??
  { reader: Deno.stdin.readable.getReader(), buffered: new Uint8Array() };

const output = Deno.stdout.writable.getWriter();
const encoder = new TextEncoder();
let shuttingDown = false;
// berry gone: nothing to tell
const send = (message: Response) => {
  if (shuttingDown) return;
  void output.write(encoder.encode(`${encode(message)}\n`)).catch(() => {});
};

/** Methods that take paths and return a result, and nothing else. */
const PLAIN = new Set([
  "absolutePath",
  "joinPath",
  "readTextFile",
  "readTextLines",
  "readBinaryFile",
  "writeFile",
  "appendFile",
  "truncateFile",
  "flushFile",
  "renameFile",
  "fileInfo",
  "listDir",
  "canonicalPath",
  "exists",
  "createDir",
  "remove",
  "createTempDir",
  "createTempFile",
]);

/** Readers and watchers berry holds open, by number. */
const handles = new Map<number, Record<string, (...args: unknown[]) => Promise<unknown>>>();
let handled = 0;
const running = new Map<number, AbortController>();
const settling = new Set<Promise<unknown>>();

/** Programs and sockets berry talks to through their pipes, by number. */
const pipes = new Map<
  number,
  {
    stdin: WritableStreamDefaultWriter<Uint8Array>;
    kill(signal?: "SIGTERM" | "SIGKILL"): void;
    resize?(cols: number, rows: number): void;
    close(): Promise<void>;
  }
>();
let piped = 0;

async function settleWithin(promise: Promise<unknown>, ms: number) {
  let timer: ReturnType<typeof setTimeout> | undefined;
  try {
    await Promise.race([
      promise.catch(() => {}),
      new Promise<void>((resolve) => timer = setTimeout(resolve, ms)),
    ]);
  } finally {
    clearTimeout(timer);
  }
}

/** Forward what `process` says to call `i`, until it exits. */
function pipe(
  i: number,
  process: Process & Partial<Pick<Pty, "resize">>,
  { group = false, socket = false } = {},
) {
  const id = ++piped;
  const stdin = process.stdin.getWriter();
  const readers = [process.stdout.getReader(), process.stderr.getReader()];
  const forward = async (reader: ReadableStreamDefaultReader<Uint8Array>, name: string) => {
    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        send({ i, o: [name, value] });
      }
    } finally {
      send({ i, o: [name, null] });
      reader.releaseLock();
    }
  };
  const streaming = Promise.allSettled([
    forward(readers[0], "stdout"),
    forward(readers[1], "stderr"),
  ]);
  let closing: Promise<void> | undefined;
  const close = () =>
    closing ??= (async () => {
      process.kill("SIGKILL");
      await settleWithin(
        Promise.allSettled([
          stdin.abort(),
          ...readers.map((reader) => reader.cancel().catch(() => {})),
          streaming,
        ]),
        1000,
      );
    })();
  pipes.set(id, { stdin, kill: process.kill, resize: process.resize, close });
  const complete = (async () => {
    if (socket) await streaming;
    const code = await process.exited;
    send({ i, o: ["exit", code] });
    if (group) {
      process.kill();
      await settleWithin(streaming, 1000);
      // descendants cannot keep an exited leader's pipes or process group alive indefinitely
      process.kill("SIGKILL");
      await close();
    } else {
      await streaming;
      void stdin.abort().catch(() => {});
    }
    pipes.delete(id);
    send({ i, o: ["end", null] });
  })();
  void complete.catch(() => {
    send({ i, o: ["exit", null] });
    void close().finally(() => {
      pipes.delete(id);
      send({ i, o: ["end", null] });
    });
  });
  return { pipe: id };
}

type Opened = { ok: true; value: object } | { ok: false; error: unknown };

function keep(result: Opened, extra: (value: object) => object = () => ({})) {
  if (!result.ok) return result;
  const handle = ++handled;
  handles.set(handle, result.value as Record<string, (...args: unknown[]) => Promise<unknown>>);
  return { ok: true, value: { handle, ...extra(result.value) } };
}

async function dispatch(
  i: number,
  m: string,
  a: readonly unknown[],
  cwd: string,
  context: Context,
) {
  const env = new NodeExecutionEnv({ cwd });
  switch (m) {
    case "exec": {
      const [command, options] = a as [string | string[], ShellExecOptions & { output: boolean }];
      const { output: streamed, ...rest } = options;
      return await env.exec(command, {
        ...rest,
        ...(streamed ? { onOutput: (text, _, info) => send({ i, o: [text, info] }) } : {}),
      }, context);
    }
    case "openBinaryReader":
      return keep(
        await env.openBinaryReader(a[0] as string, a[1] as { noFollow?: boolean }, context),
      );
    case "openTextLineReader":
      return keep(await env.openTextLineReader(a[0] as string, context));
    case "openDirReader":
      return keep(await env.openDirReader(a[0] as string, context));
    case "fetch": {
      const [url, options] = a as [string, FetchOptions];
      const lifetime = new AbortController();
      const abort = () => lifetime.abort(context.abortSignal?.reason);
      context.abortSignal?.addEventListener("abort", abort, { once: true });
      if (context.abortSignal?.aborted) abort();
      let response: globalThis.Response;
      try {
        response = await fetch(url, {
          ...options,
          body: options.body === undefined ? undefined : new Uint8Array(options.body),
          signal: lifetime.signal,
        });
      } finally {
        context.abortSignal?.removeEventListener("abort", abort);
      }
      const metadata = {
        status: response.status,
        statusText: response.statusText,
        headers: [...response.headers],
        url: response.url,
        redirected: response.redirected,
      };
      if (!response.body) return metadata;
      const reader = response.body.getReader();
      const handle = ++handled;
      const close = async () => {
        handles.delete(handle);
        lifetime.abort();
        await reader.cancel().catch(() => {});
      };
      handles.set(handle, {
        async read(context: Context) {
          const abort = () => lifetime.abort(context.abortSignal?.reason);
          context.abortSignal?.addEventListener("abort", abort, { once: true });
          if (context.abortSignal?.aborted) abort();
          try {
            const next = await reader.read();
            if (next.done) handles.delete(handle);
            return next;
          } catch (error) {
            await close();
            throw error;
          } finally {
            context.abortSignal?.removeEventListener("abort", abort);
          }
        },
        close,
      } as Record<string, (...args: unknown[]) => Promise<unknown>>);
      return { ...metadata, handle };
    }
    case "spawn": {
      const [argv, options] = a as [string[], SpawnOptions];
      const at = options.cwd === undefined ? cwd : resolve(cwd, options.cwd);
      // it outlives this call: berry kills it through its pipe
      return pipe(i, await spawnHere(argv, { ...options, cwd: at }, BACKGROUND_CONTEXT), {
        group: options.group,
      });
    }
    case "pty": {
      const [argv, options] = a as [string[], PtyOptions];
      const at = options.cwd === undefined ? cwd : resolve(cwd, options.cwd);
      const pty = await ptyHere(argv, { ...options, cwd: at }, BACKGROUND_CONTEXT);
      return pipe(i, {
        stdin: pty.writable,
        stdout: pty.readable,
        stderr: new ReadableStream({ start: (controller) => controller.close() }),
        exited: pty.exited,
        kill: pty.kill,
        resize: pty.resize,
      });
    }
    case "connect": {
      const socket = await connectHere(a[0] as string, a[1] as number, context);
      return pipe(i, {
        stdin: socket.writable,
        stdout: socket.readable,
        stderr: new ReadableStream({ start: (controller) => controller.close() }),
        // done once the far end closes it
        exited: Promise.resolve(0),
        kill: socket.close,
      }, { socket: true });
    }
    case "pipe": {
      const [id, operation, ...rest] = a as [
        number,
        "write" | "end" | "kill" | "resize",
        ...unknown[],
      ];
      const target = pipes.get(id);
      // already ended
      if (target === undefined) return;
      if (operation === "write") await target.stdin.write(rest[0] as Uint8Array);
      else if (operation === "end") await target.stdin.close();
      else if (operation === "resize") target.resize?.(rest[0] as number, rest[1] as number);
      else target.kill(rest[0] as "SIGTERM" | "SIGKILL" | undefined);
      return;
    }
    case "watch":
      return keep(
        await env.watch(a[0] as never, (change) => send({ i, o: change }), context),
        (watcher) => ({ mode: (watcher as { mode: string }).mode }),
      );
    case "handle": {
      const [handle, method, ...rest] = a as [number, string, ...unknown[]];
      const target = handles.get(handle);
      // as a reader refuses use after it is closed, and takes another close as done
      if (target === undefined) {
        return method === "close"
          ? undefined
          : { ok: false, error: new FileError("invalid", `handle ${handle} is closed`) };
      }
      if (method === "close") handles.delete(handle);
      return await target[method].call(target, ...rest, context);
    }
  }
  if (!PLAIN.has(m)) throw new Error(`No method ${m}`);
  const method = env[m as keyof NodeExecutionEnv] as (...args: unknown[]) => Promise<unknown>;
  return await method.call(env, ...a, context);
}

send({
  hello: {
    protocol: PROTOCOL,
    platform: process.platform,
    home: homedir(),
    variables: Object.fromEntries(
      Object.entries(process.env).filter((entry): entry is [string, string] =>
        entry[1] !== undefined
      ),
    ),
  },
});

for await (const line of lines(input.reader, input.buffered)) {
  if (line === "") continue;
  const request = decode(line) as Request;
  if ("c" in request) {
    running.get(request.c)?.abort();
    continue;
  }
  const { i, m, a, cwd = homedir() } = request;
  const controller = new AbortController();
  running.set(i, controller);
  const settled = dispatch(i, m, a, cwd, withAbortSignal(controller.signal, BACKGROUND_CONTEXT))
    .then(
      (r) => send({ i, r }),
      (error) => send({ i, e: error instanceof Error ? error.message : String(error) }),
    )
    .finally(() => {
      running.delete(i);
      settling.delete(settled);
    });
  settling.add(settled);
}

// berry has gone: so does everything it started
shuttingDown = true;
for (const controller of running.values()) controller.abort();
const owned = [...pipes.values()];
for (const { kill } of owned) kill();
await settleWithin(Promise.allSettled(settling), 1000);
await settleWithin(
  Promise.allSettled([
    ...owned.map((pipe) => pipe.close()),
    ...[...handles.values()].map((handle) => handle.close?.call(handle, BACKGROUND_CONTEXT)),
  ]),
  1000,
);
await settleWithin(output.close(), 1000);
Deno.exit(0);