char-slop/berry

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

Charlotte Sommcp: add streaming fetch on conversation-local and remote environmentse5ec3802026-10-10

// berry's side of an environment on another machine: the worker it starts there, and the ExecutionEnv that speaks to
// it. Nothing is retried: a call the connection lost may or may not have happened, so it fails.

import { Buffer } from "node:buffer";
import { dirname, fromFileUrl } from "@std/path";
import type { Context } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, withAbortSignal } from "@earendil-works/chord/context";
import {
  type BinaryReader,
  type DirReader,
  ExecutionError,
  FileError,
  type FileInfo,
  type FileWatcher,
  type Result,
  type ShellExecOptions,
  type ShellExecResult,
  type ShellOutputInfo,
  type TextLineReader,
  type WatchChange,
  type WatchTarget,
} from "@earendil-works/pi-durable/env";
import type {
  Environment,
  Process,
  Pty,
  PtyOptions,
  Socket,
  SpawnOptions,
} from "@berry/api/environment.ts";
import { denoExecutable } from "../deno.ts";
import {
  decode,
  encode,
  type FetchOptions,
  type FetchResponse,
  type Hello,
  lines,
  PROTOCOL,
  type Request,
  type Response,
} from "./codec.ts";

/**
 * Reads a length-prefixed module off stdin and imports it, handing it the rest of stdin. Remote shells (sh, cmd, and
 * powershell alike) pass it through double quotes untouched, as it holds none of `"$\`\\%!`.
 */
export const BOOTSTRAP = [
  "const r=Deno.stdin.readable.getReader();let b=new Uint8Array(0);",
  "const take=async(n)=>{while(b.length<n){const c=await r.read();if(c.done)Deno.exit(70);",
  "const m=new Uint8Array(b.length+c.value.length);m.set(b);m.set(c.value,b.length);b=m}",
  "const o=b.slice(0,n);b=b.slice(n);return o};",
  "const n=new DataView((await take(4)).buffer).getUint32(0);",
  "const s=new TextDecoder().decode(await take(n));",
  "globalThis.berryStdin={reader:r,buffered:b};",
  "await import('data:text/javascript;base64,'+s)",
].join("");

/** How berry starts the worker on `destination`, with `deno` as the far side calls it. */
export function sshCommand(destination: string, deno = "deno"): string[] {
  return [
    "ssh",
    "-T",
    "-o",
    "BatchMode=yes",
    "-o",
    "ConnectTimeout=15",
    "-o",
    "ServerAliveInterval=15",
    "-o",
    "ServerAliveCountMax=3",
    "--",
    destination,
    `${deno} eval --quiet "${BOOTSTRAP}"`,
  ];
}

let bundled: Promise<string> | undefined;

/** The worker as a single module that needs nothing but deno, in base64 for the bootstrap. */
function workerSource(): Promise<string> {
  bundled ??= (async () => {
    const worker = fromFileUrl(new URL("./worker.ts", import.meta.url));
    const out = await Deno.makeTempDir({ prefix: "berry-worker-" });
    try {
      const { success, stderr } = await new Deno.Command(denoExecutable(), {
        args: ["bundle", "--quiet", "--platform", "deno", "--output", `${out}/worker.js`, worker],
        cwd: dirname(worker),
        env: { NO_COLOR: "1" },
        stdout: "null",
        stderr: "piped",
      }).output();
      if (!success) {
        throw new Error(`could not bundle the worker:\n${new TextDecoder().decode(stderr)}`);
      }
      return Buffer.from(await Deno.readFile(`${out}/worker.js`)).toString("base64");
    } finally {
      await Deno.remove(out, { recursive: true });
    }
  })();
  bundled.catch(() => (bundled = undefined));
  return bundled;
}

const HELLO_TIMEOUT_MS = 60_000;
const STDERR_KEPT = 4096;

const messageOf = (error: unknown) => (error instanceof Error ? error.message : String(error));

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

function abortable<T>(promise: Promise<T>, signal: AbortSignal): Promise<T> {
  return new Promise((resolve, reject) => {
    const abort = () => {
      signal.removeEventListener("abort", abort);
      reject(signal.reason);
    };
    signal.addEventListener("abort", abort, { once: true });
    void promise.then(
      (value) => {
        signal.removeEventListener("abort", abort);
        resolve(value);
      },
      (error) => {
        signal.removeEventListener("abort", abort);
        reject(error);
      },
    );
    if (signal.aborted) abort();
  });
}

type CallOptions = {
  readonly cwd?: string;
  readonly context: Context;
  /** Receives what the call streams, until it settles or `until` does. */
  readonly listen?: (event: unknown) => void;
  readonly until?: Promise<void>;
};

/** One worker, and the calls in flight to it. */
export class Connection {
  readonly #name: string;
  readonly #process: Deno.ChildProcess;
  readonly #writer: WritableStreamDefaultWriter<Uint8Array>;
  readonly #pending = new Map<number, PromiseWithResolvers<unknown>>();
  readonly #listeners = new Map<number, (event: unknown) => void>();
  readonly #gone = Promise.withResolvers<Error>();
  readonly #lifetime = new AbortController();
  #next = 0;
  #lost: Error | undefined;

  private constructor(
    name: string,
    child: Deno.ChildProcess,
    writer: WritableStreamDefaultWriter<Uint8Array>,
    readonly hello: Hello,
  ) {
    this.#name = name;
    this.#process = child;
    this.#writer = writer;
  }

  /** Start the worker with `argv`, which runs the bootstrap on the far side. */
  static async open(name: string, argv: readonly string[]): Promise<Connection> {
    const source = await workerSource();
    const child = new Deno.Command(argv[0], {
      args: argv.slice(1),
      stdin: "piped",
      stdout: "piped",
      stderr: "piped",
    }).spawn();
    let stderr = "";
    const stderrReader = child.stderr.getReader();
    const errors = (async () => {
      const decoder = new TextDecoder();
      try {
        while (true) {
          const { done, value } = await stderrReader.read();
          if (done) break;
          stderr = (stderr + decoder.decode(value, { stream: true })).slice(-STDERR_KEPT);
        }
        stderr = (stderr + decoder.decode()).slice(-STDERR_KEPT);
      } finally {
        stderrReader.releaseLock();
      }
    })().catch(() => {});
    const stopErrors = async () => {
      await settleWithin(errors, 100);
      await settleWithin(Promise.allSettled([stderrReader.cancel().catch(() => {}), errors]), 100);
    };
    const failed = (why: string) =>
      new Error(`${name}: ${why}${stderr.trim() ? `\n${stderr.trim()}` : ""}`);

    const writer = child.stdin.getWriter();
    const length = new Uint8Array(4);
    new DataView(length.buffer).setUint32(0, source.length);
    const incoming = lines(child.stdout.getReader());
    let timer: ReturnType<typeof setTimeout> | undefined;
    try {
      await writer.write(length);
      await writer.write(new TextEncoder().encode(source));
      const first = await Promise.race([
        incoming.next(),
        new Promise<never>((_, reject) => {
          timer = setTimeout(() => reject(new Error("timed out")), HELLO_TIMEOUT_MS);
        }),
      ]);
      if (first.done) throw new Error("closed");
      const hello = (decode(first.value) as { hello?: Hello }).hello;
      if (hello?.protocol !== PROTOCOL) {
        throw new Error(`speaks protocol ${hello?.protocol}, not ${PROTOCOL}`);
      }
      const connection = new Connection(name, child, writer, hello);
      void connection.#receive(incoming, stopErrors, failed);
      return connection;
    } catch (error) {
      try {
        child.kill();
      } catch {
        // already gone
      }
      await child.status;
      await stopErrors();
      throw failed(`could not start its worker: ${messageOf(error)}`);
    } finally {
      clearTimeout(timer);
    }
  }

  /** Why calls fail, once the worker cannot be reached. */
  get lost(): Error | undefined {
    return this.#lost;
  }

  /** Settles with why, once the worker cannot be reached. */
  get gone(): Promise<Error> {
    return this.#gone.promise;
  }

  get lifetime(): AbortSignal {
    return this.#lifetime.signal;
  }

  async #receive(
    incoming: AsyncGenerator<string>,
    stopErrors: () => Promise<void>,
    failed: (why: string) => Error,
  ) {
    try {
      for await (const line of incoming) {
        const message = decode(line) as Response;
        if ("hello" in message) continue;
        if ("o" in message) {
          this.#listeners.get(message.i)?.(message.o);
          continue;
        }
        const pending = this.#pending.get(message.i);
        this.#pending.delete(message.i);
        if ("e" in message) pending?.reject(new Error(`${this.#name}: ${message.e}`));
        else pending?.resolve(message.r);
      }
    } catch (error) {
      this.#lost = failed(`lost: ${messageOf(error)}`);
    }
    await this.#process.status;
    await stopErrors();
    this.#lost ??= failed("lost");
    this.#lifetime.abort(this.#lost);
    this.#gone.resolve(this.#lost);
    for (const pending of this.#pending.values()) pending.reject(this.#lost);
    this.#pending.clear();
    this.#listeners.clear();
  }

  #send(request: Request) {
    // a failed write means the worker is gone, which #receive reports
    void this.#writer.write(new TextEncoder().encode(`${encode(request)}\n`)).catch(() => {});
  }

  async call(method: string, args: readonly unknown[], options: CallOptions): Promise<unknown> {
    if (this.#lost) throw this.#lost;
    const i = ++this.#next;
    const pending = Promise.withResolvers<unknown>();
    this.#pending.set(i, pending);
    if (options.listen) {
      const listen = options.listen;
      this.#listeners.set(i, listen);
      void (options.until ?? pending.promise).catch(() => {}).finally(() => {
        if (this.#listeners.get(i) === listen) this.#listeners.delete(i);
      });
    }
    const signal = options.context.abortSignal;
    const abort = () => this.#send({ c: i });
    signal?.addEventListener("abort", abort, { once: true });
    this.#send({
      i,
      m: method,
      a: args,
      ...(options.cwd === undefined ? {} : { cwd: options.cwd }),
    });
    if (signal?.aborted) abort();
    try {
      return await pending.promise;
    } finally {
      signal?.removeEventListener("abort", abort);
    }
  }

  /** End the worker, and with it everything it runs. */
  async close(): Promise<void> {
    await this.#writer.close().catch(() => {});
    const timer = setTimeout(() => {
      try {
        this.#process.kill();
      } catch {
        // already gone
      }
    }, 5_000);
    await this.#process.status;
    clearTimeout(timer);
    await this.#gone.promise;
  }
}

type Calls = (method: string, args: readonly unknown[], options: CallOptions) => Promise<unknown>;

const fileFailure = (error: unknown) => ({
  ok: false as const,
  error: new FileError("unknown", messageOf(error)),
});
// a call already aborted must not happen at all, which only this side can be sure of
const fileAborted = { ok: false as const, error: new FileError("aborted", "aborted") };

/** A machine berry reaches through a worker there, at `cwd`. */
export class RemoteEnvironment implements Environment {
  readonly id: string;
  readonly platform: string;
  readonly home: string;
  readonly variables: Readonly<Record<string, string>>;
  readonly lifetime: AbortSignal;
  readonly #connection: Connection;
  /** What `cleanup` stops: the calls made through this object that still run. */
  readonly #running = new Set<AbortController>();

  constructor(readonly name: string, connection: Connection, public cwd: string) {
    this.id = `ssh:${name}`;
    this.platform = connection.hello.platform;
    this.home = connection.hello.home;
    this.variables = connection.hello.variables;
    this.lifetime = connection.lifetime;
    this.#connection = connection;
  }

  /** A call that `cleanup` can stop, failing as the environment's calls fail. */
  #call: Calls = async (method, args, options) => {
    const stop = new AbortController();
    this.#running.add(stop);
    const signals = [
      stop.signal,
      ...(options.context.abortSignal ? [options.context.abortSignal] : []),
    ];
    try {
      return await this.#connection.call(method, args, {
        ...options,
        context: withAbortSignal(AbortSignal.any(signals), options.context),
      });
    } finally {
      this.#running.delete(stop);
    }
  };

  async #file<T>(method: string, args: readonly unknown[], context: Context) {
    if (context.abortSignal?.aborted) return fileAborted;
    try {
      return await this.#call(method, args, { cwd: this.cwd, context }) as Result<T, FileError>;
    } catch (error) {
      return fileFailure(error);
    }
  }

  /** Calls on something the worker holds open for this one. */
  #held(handle: number) {
    return async <T>(method: string, args: readonly unknown[], context: Context) => {
      if (context.abortSignal?.aborted) return fileAborted as T;
      try {
        return await this.#call("handle", [handle, method, ...args], { context }) as T;
      } catch (error) {
        return fileFailure(error) as T;
      }
    };
  }

  absolutePath = (path: string, context: Context) =>
    this.#file<string>("absolutePath", [path], context);
  joinPath = (parts: string[], context: Context) =>
    this.#file<string>("joinPath", [parts], context);
  readTextFile = (path: string, context: Context) =>
    this.#file<string>("readTextFile", [path], context);
  readTextLines = (path: string, options: { maxLines?: number } | undefined, context: Context) =>
    this.#file<string[]>("readTextLines", [path, options], context);
  readBinaryFile = (path: string, context: Context) =>
    this.#file<Uint8Array>("readBinaryFile", [path], context);
  writeFile = (path: string, content: string | Uint8Array, context: Context) =>
    this.#file<void>("writeFile", [path, content], context);
  appendFile = (path: string, content: string | Uint8Array, context: Context) =>
    this.#file<void>("appendFile", [path, content], context);
  truncateFile = (path: string, size: number, context: Context) =>
    this.#file<void>("truncateFile", [path, size], context);
  flushFile = (path: string, context: Context) => this.#file<void>("flushFile", [path], context);
  renameFile = (from: string, to: string, context: Context) =>
    this.#file<void>("renameFile", [from, to], context);
  fileInfo = (path: string, context: Context) => this.#file<FileInfo>("fileInfo", [path], context);
  listDir = (path: string, context: Context) => this.#file<FileInfo[]>("listDir", [path], context);
  canonicalPath = (path: string, context: Context) =>
    this.#file<string>("canonicalPath", [path], context);
  exists = (path: string, context: Context) => this.#file<boolean>("exists", [path], context);
  createDir = (path: string, options: { recursive?: boolean } | undefined, context: Context) =>
    this.#file<void>("createDir", [path, options], context);
  remove = (
    path: string,
    options: { recursive?: boolean; force?: boolean } | undefined,
    context: Context,
  ) => this.#file<void>("remove", [path, options], context);
  createTempDir = (prefix: string | undefined, context: Context) =>
    this.#file<string>("createTempDir", [prefix], context);
  createTempFile = (
    options: { prefix?: string; suffix?: string } | undefined,
    context: Context,
  ) => this.#file<string>("createTempFile", [options], context);

  async openTextLineReader(
    path: string,
    context: Context,
  ): Promise<Result<TextLineReader, FileError>> {
    const opened = await this.#file<{ handle: number }>("openTextLineReader", [path], context);
    if (!opened.ok) return opened;
    const held = this.#held(opened.value.handle);
    return {
      ok: true,
      value: {
        readLine: (context) => held("readLine", [], context),
        close: (context) => held("close", [], context),
      },
    };
  }

  async openBinaryReader(
    path: string,
    options: { noFollow?: boolean } | undefined,
    context: Context,
  ): Promise<Result<BinaryReader, FileError>> {
    const opened = await this.#file<{ handle: number }>(
      "openBinaryReader",
      [path, options],
      context,
    );
    if (!opened.ok) return opened;
    const held = this.#held(opened.value.handle);
    return {
      ok: true,
      value: {
        info: (context) => held("info", [], context),
        read: (offset, length, context) => held("read", [offset, length], context),
        scanLines: (options, context) => held("scanLines", [options], context),
        close: (context) => held("close", [], context),
      },
    };
  }

  async openDirReader(path: string, context: Context): Promise<Result<DirReader, FileError>> {
    const opened = await this.#file<{ handle: number }>("openDirReader", [path], context);
    if (!opened.ok) return opened;
    const held = this.#held(opened.value.handle);
    return {
      ok: true,
      value: {
        next: (count, context) => held("next", [count], context),
        close: (context) => held("close", [], context),
      },
    };
  }

  async watch(
    targets: readonly WatchTarget[],
    onChange: (change: WatchChange) => void,
    context: Context,
  ): Promise<Result<FileWatcher, FileError>> {
    const closed = Promise.withResolvers<void>();
    let opened;
    try {
      opened = await this.#connection.call("watch", [targets], {
        cwd: this.cwd,
        context,
        listen: (change) => onChange(change as WatchChange),
        until: closed.promise,
      }) as Result<{ handle: number; mode: FileWatcher["mode"] }, FileError>;
    } catch (error) {
      opened = fileFailure(error);
    }
    if (!opened.ok) {
      closed.resolve();
      return opened;
    }
    const held = this.#held(opened.value.handle);
    return {
      ok: true,
      value: {
        mode: opened.value.mode,
        async close(context) {
          await held("close", [], context);
          closed.resolve();
        },
      },
    };
  }

  async exec(
    command: string | readonly string[],
    options: ShellExecOptions | undefined,
    context: Context,
  ): Promise<Result<ShellExecResult, ExecutionError>> {
    if (context.abortSignal?.aborted) {
      return { ok: false, error: new ExecutionError("aborted", "aborted") };
    }
    const { onOutput, ...rest } = options ?? {};
    // a throwing callback stops the command, as it would here
    const failing = new AbortController();
    let failure: unknown;
    const listen = onOutput &&
      ((event: unknown) => {
        const [text, info] = event as [string, ShellOutputInfo];
        try {
          onOutput(text, context, info);
        } catch (error) {
          failure ??= error;
          failing.abort();
        }
      });
    try {
      const result = await this.#call("exec", [command, { ...rest, output: !!onOutput }], {
        cwd: this.cwd,
        context: withAbortSignal(
          AbortSignal.any([failing.signal, ...(context.abortSignal ? [context.abortSignal] : [])]),
          context,
        ),
        ...(listen ? { listen } : {}),
      }) as Result<ShellExecResult, ExecutionError>;
      if (failure === undefined) return result;
      return { ok: false, error: new ExecutionError("callback_error", messageOf(failure)) };
    } catch (error) {
      return { ok: false, error: new ExecutionError("unknown", messageOf(error)) };
    }
  }

  async fetch(input: string | URL, init?: RequestInit): Promise<globalThis.Response> {
    init?.signal?.throwIfAborted();
    const request = new globalThis.Request(input, init);
    const stop = new AbortController();
    this.#running.add(stop);
    const signal = AbortSignal.any([stop.signal, request.signal]);
    const during = withAbortSignal(signal, BACKGROUND_CONTEXT);
    let opened: FetchResponse | undefined;
    let controller: ReadableStreamDefaultController<Uint8Array> | undefined;
    let ended = false;
    const finish = () => {
      ended = true;
      this.#running.delete(stop);
      signal.removeEventListener("abort", abort);
    };
    const release = async (handle: number | undefined) => {
      if (handle !== undefined) {
        await abortable(
          this.#connection.call("handle", [handle, "close"], {
            context: BACKGROUND_CONTEXT,
          }),
          AbortSignal.timeout(1000),
        ).catch(() => {});
      }
    };
    const close = async () => {
      if (ended) return;
      finish();
      await release(opened?.handle);
    };
    const abort = () => {
      if (ended) return;
      controller?.error(signal.reason);
      void close();
    };
    try {
      const options: FetchOptions = {
        method: request.method,
        headers: [...request.headers],
        ...(request.body
          ? { body: new Uint8Array(await abortable(request.arrayBuffer(), signal)) }
          : {}),
        redirect: request.redirect,
        credentials: request.credentials,
        cache: request.cache,
        referrer: request.referrer,
        referrerPolicy: request.referrerPolicy,
        integrity: request.integrity,
      };
      signal.throwIfAborted();
      const opening = this.#connection.call("fetch", [request.url, options], {
        context: during,
      }).then((value) => {
        opened = value as FetchResponse;
        // a canceled caller must still release a handle returned late by the worker
        if (ended) void release(opened.handle);
        return opened;
      });
      opened = await abortable(opening, signal);
      signal.throwIfAborted();
      const body = opened.handle === undefined ? null : new ReadableStream<Uint8Array>({
        start: (stream) => {
          controller = stream;
          signal.addEventListener("abort", abort, { once: true });
        },
        pull: async (stream) => {
          try {
            const next = await abortable(
              this.#connection.call("handle", [opened!.handle, "read"], {
                context: during,
              }),
              signal,
            ) as ReadableStreamReadResult<Uint8Array>;
            if (ended) return;
            if (next.done) {
              finish();
              stream.close();
            } else stream.enqueue(next.value);
          } catch (error) {
            if (!ended) {
              stream.error(signal.aborted ? signal.reason : error);
              void close();
            }
          }
        },
        cancel: close,
      });
      if (body === null) finish();
      const response = new globalThis.Response(body, opened);
      Object.defineProperties(response, {
        url: { value: opened.url },
        redirected: { value: opened.redirected },
      });
      return response;
    } catch (error) {
      void close();
      if (signal.aborted) throw signal.reason;
      throw new TypeError(`Fetch failed on ${this.name}: ${messageOf(error)}`, { cause: error });
    }
  }

  async spawn(argv: readonly string[], options: SpawnOptions, context: Context): Promise<Process> {
    return (await this.#pipe("spawn", [argv, options], context)).process;
  }

  async pty(argv: readonly string[], options: PtyOptions, context: Context): Promise<Pty> {
    const { process, operate } = await this.#pipe("pty", [argv, options], context);
    return {
      readable: process.stdout,
      writable: process.stdin,
      resize: (cols, rows) => void operate("resize", cols, rows).catch(() => {}),
      exited: process.exited,
      kill: process.kill,
    };
  }

  async connect(hostname: string, port: number, context: Context): Promise<Socket> {
    const { stdin, stdout, kill } =
      (await this.#pipe("connect", [hostname, port], context)).process;
    return { readable: stdout, writable: stdin, close: kill };
  }

  /** Something the worker runs or holds open, spoken to through its pipes until it exits. */
  async #pipe(
    method: string,
    args: readonly unknown[],
    context: Context,
  ): Promise<{ process: Process; operate(operation: string, ...rest: unknown[]): Promise<void> }> {
    context.abortSignal?.throwIfAborted();
    const streams: Record<string, ReadableStreamDefaultController<Uint8Array>> = {};
    const readable = (name: string) =>
      new ReadableStream<Uint8Array>({
        start: (controller) => void (streams[name] = controller),
        cancel: () => void delete streams[name],
      });
    const stdout = readable("stdout");
    const stderr = readable("stderr");
    const exited = Promise.withResolvers<number | null>();
    const finished = Promise.withResolvers<void>();
    const end = (error?: Error) => {
      for (const stream of Object.values(streams)) {
        try {
          if (error) stream.error(error);
          else stream.close();
        } catch {
          // already ended
        }
      }
    };
    const listen = (event: unknown) => {
      const [name, chunk] = event as [string, Uint8Array | number | null];
      if (name === "exit") exited.resolve(chunk as number | null);
      else if (name === "end") {
        end();
        finished.resolve();
      } else if (chunk === null) {
        streams[name]?.close();
        delete streams[name];
      } else streams[name]?.enqueue(chunk as Uint8Array);
    };
    let opened: { pipe: number };
    try {
      opened = await this.#connection.call(method, args, {
        cwd: this.cwd,
        context,
        listen,
        until: finished.promise,
      }) as { pipe: number };
    } catch (error) {
      exited.resolve(null);
      finished.resolve();
      throw error;
    }
    // what berry cannot reach any more has ended, as far as anyone here can tell
    void Promise.race([this.#connection.gone, finished.promise]).then((outcome) => {
      if (outcome instanceof Error) {
        end(outcome);
        exited.resolve(null);
        finished.resolve();
      }
    });
    const operate = (operation: string, ...rest: unknown[]) =>
      this.#connection.call("pipe", [opened.pipe, operation, ...rest], {
        context: BACKGROUND_CONTEXT,
      }).then(() => {});
    const kill = (signal: "SIGTERM" | "SIGKILL" = "SIGTERM") =>
      void operate("kill", signal).catch(() => {});
    const abort = () => kill();
    context.abortSignal?.addEventListener("abort", abort, { once: true });
    if (context.abortSignal?.aborted) abort();
    void exited.promise.then(() => context.abortSignal?.removeEventListener("abort", abort));
    const process = {
      stdin: new WritableStream<Uint8Array>({
        write: (chunk) => operate("write", chunk),
        close: () => operate("end"),
        abort: () => operate("kill"),
      }),
      stdout,
      stderr,
      exited: exited.promise,
      kill,
    };
    return { process, operate };
  }

  cleanup(_context: Context): Promise<void> {
    for (const stop of this.#running) stop.abort();
    return Promise.resolve();
  }
}