// 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>; /** 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; /** * 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; /** Open a TCP connection from there, so `127.0.0.1` is that machine's own. */ connect(hostname: string, port: number, context: Context): Promise; /** Fetch from this machine, with streaming responses and cancellation through `init.signal`. */ fetch(input: string | URL, init?: RequestInit): Promise; } export type SpawnOptions = { readonly cwd?: string; readonly env?: Record; /** 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; readonly writable: WritableStream; resize(cols: number, rows: number): void; /** Its exit code, or `null` when a signal ended it. */ readonly exited: Promise; /** 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; readonly stdout: ReadableStream; readonly stderr: ReadableStream; /** Its exit code, or `null` when a signal ended it. */ readonly exited: Promise; kill(signal?: "SIGTERM" | "SIGKILL"): void; } /** A TCP connection from an environment's machine. */ export interface Socket { readonly readable: ReadableStream; readonly writable: WritableStream; 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 { 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 { return connectHere(hostname, port, context); } fetch(input: string | URL, init?: RequestInit): Promise { return fetch(input, init); } pty(argv: readonly string[], options: PtyOptions, context: Context): Promise { // 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 { 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 { 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((resolve) => child.once("exit", resolve)); void exited.then(() => context.abortSignal?.removeEventListener("abort", abort)); try { await new Promise((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, 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 { 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 { 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; }