/* oxlint-disable executor/no-try-catch-or-throw, executor/no-promise-reject, executor/no-instanceof-tagged-error -- subprocess boundary normalizes JavaScript errors into typed runtime errors */import { spawn, spawnSync, type ChildProcessByStdio } from "node:child_process";import { access, mkdir, rm, writeFile } from "node:fs/promises";import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";import { createServer as createNetServer, type Server as NetServer } from "node:net";import type { Readable } from "node:stream";import { tmpdir } from "node:os";import { dirname, join } from "node:path";import { fileURLToPath } from "node:url"; import {  recoverExecutionBody,  stripTypeScript,  type CodeExecutor,  type ExecuteResult,  type SandboxToolInvoker,} from "@executor-js/codemode-core";import * as Cause from "effect/Cause";import * as Data from "effect/Data";import * as Effect from "effect/Effect"; export const WORKERD_VERSION = "1.20260708.1"; const DEFAULT_TIMEOUT_MS = 5 * 60_000;const DEFAULT_HOST_TIMEOUT_GRACE_MS = 30_000;export class WorkerdSubprocessError extends Data.TaggedError("WorkerdSubprocessError")<{  readonly message: string;  readonly cause?: unknown;}> {} export type WorkerdToolBridge = {  readonly call: (toolPath: string, args: unknown) => Promise<unknown>;}; export type WorkerdModuleSource =  | string  | {      readonly kind: "esModule";      readonly source: string;    }  | {      readonly kind: "wasm";      readonly bytes: Uint8Array;    }; export type WorkerdModuleRunnerOptions = {  readonly modules: Readonly<Record<string, WorkerdModuleSource>>;  readonly mainModule: string;  readonly compatibilityDate?: string;  readonly compatibilityFlags?: readonly string[];  readonly toolBridge?: WorkerdToolBridge;  readonly hostToken?: string;  readonly unsafeEval?: boolean;  readonly globalOutbound?: "blocked" | "internet";  readonly workerdBin?: string;  readonly startupTimeoutMs?: number;  readonly restartBackoffMs?: number;}; export type WorkerdRunResult<T = unknown> = {  readonly status: number;  readonly body: T;  readonly elapsedMs: number;}; export type WorkerdModuleRunner = {  readonly pid: () => number | undefined;  readonly ensureStarted: () => Promise<void>;  readonly run: <T = unknown>(payload: unknown, timeoutMs?: number) => Promise<WorkerdRunResult<T>>;  readonly dispose: () => Promise<void>;  readonly crashForTest: () => void;  readonly tempDirForTest: () => string | undefined;}; type PendingRequest = {  readonly reject: (error: Error) => void;}; const packageDir = dirname(dirname(fileURLToPath(import.meta.url))); const normalizeError = (cause: unknown): Error =>  cause instanceof Error ? cause : new Error(String(cause)); const renderCause = (cause: Cause.Cause<unknown>): string => {  const squashed = Cause.squash(cause);  if (squashed instanceof Error) return squashed.message;  return String(squashed);}; const readRequestBody = (request: IncomingMessage): Promise<unknown> =>  new Promise((resolve, reject) => {    let body = "";    request.setEncoding("utf8");    request.on("data", (chunk) => {      body += chunk;    });    request.on("error", reject);    request.on("end", () => {      try {        resolve(body.length > 0 ? JSON.parse(body) : {});      } catch (cause) {        reject(cause);      }    });  }); const writeJson = (response: ServerResponse, status: number, body: unknown): void => {  response.writeHead(status, { "content-type": "application/json" });  response.end(JSON.stringify(body));}; const closeServer = (server: Server): Promise<void> =>  new Promise((resolve) => {    server.close(() => resolve());  }); const listenLocal = (server: Server): Promise<number> =>  new Promise((resolve, reject) => {    server.on("error", reject);    server.listen(0, "127.0.0.1", () => {      const address = server.address();      if (typeof address === "object" && address !== null) {        resolve(address.port);        return;      }      reject(new Error("server did not expose a TCP port"));    });  }); const reserveLocalPort = (): Promise<{  readonly port: number;  readonly close: () => Promise<void>;}> =>  new Promise((resolve, reject) => {    const server: NetServer = createNetServer();    server.on("error", reject);    server.listen(0, "127.0.0.1", () => {      const address = server.address();      if (typeof address === "object" && address !== null) {        resolve({          port: address.port,          close: () =>            new Promise((closeResolve) => {              server.close(() => closeResolve());            }),        });        return;      }      reject(new Error("reservation did not expose a TCP port"));    });  }); const findWorkerdBin = async (configured?: string): Promise<string> => {  if (configured) return configured;   const candidates = [    join(packageDir, "node_modules", "workerd", "bin", "workerd"),    join(packageDir, "..", "..", "..", "node_modules", "workerd", "bin", "workerd"),    join(process.cwd(), "node_modules", "workerd", "bin", "workerd"),  ];   for (const candidate of candidates) {    try {      await access(candidate);      return candidate;    } catch {      // Try the next package-manager layout.    }  }   return "workerd";}; export const isWorkerdAvailable = (workerdBin?: string): boolean => {  const configured = workerdBin ?? "workerd";  const result = spawnSync(configured, ["--version"], {    stdio: "ignore",    timeout: 5000,  });  if (result.error === undefined && result.status === 0) return true;   const local = spawnSync(    join(packageDir, "node_modules", "workerd", "bin", "workerd"),    ["--version"],    {      stdio: "ignore",      timeout: 5000,    },  );  if (local.error === undefined && local.status === 0) return true;   const workspace = spawnSync(    join(packageDir, "..", "..", "..", "node_modules", "workerd", "bin", "workerd"),    ["--version"],    {      stdio: "ignore",      timeout: 5000,    },  );  return workspace.error === undefined && workspace.status === 0;}; const capnpString = (value: string): string => JSON.stringify(value); const buildConfig = (input: {  readonly listenPort: number;  readonly hostPort: number | null;  readonly modules: Readonly<Record<string, WorkerdModuleSource>>;  readonly mainModule: string;  readonly compatibilityDate: string;  readonly compatibilityFlags: readonly string[];  readonly unsafeEval: boolean;  readonly globalOutbound: "blocked" | "internet";}): string => {  const moduleNames = [    input.mainModule,    ...Object.keys(input.modules)      .filter((name) => name !== input.mainModule)      .sort(),  ];  const moduleEntries = moduleNames    .map((name) => {      const source = input.modules[name];      const field = typeof source === "object" && source.kind === "wasm" ? "wasm" : "esModule";      return `( name = ${capnpString(name)}, ${field} = embed ${capnpString(name)} )`;    })    .join(",\n    ");  const services = [    '( name = "main", worker = .mainWorker )',    '( name = "blocked", worker = .blockedWorker )',    ...(input.hostPort === null      ? []      : [`( name = "host", external = ( address = "127.0.0.1:${input.hostPort}", http = () ) )`]),  ].join(",\n    ");  const bindingEntries = [    ...(input.hostPort === null ? [] : ['( name = "HOST", service = "host" )']),    ...(input.unsafeEval ? ['( name = "UNSAFE_EVAL", unsafeEval = void )'] : []),  ];  const bindings =    bindingEntries.length === 0 ? "" : `\n  bindings = [${bindingEntries.join(", ")}],`;  const flags =    input.compatibilityFlags.length === 0      ? ""      : `\n  compatibilityFlags = [${input.compatibilityFlags.map(capnpString).join(", ")}],`;   return `using Workerd = import "/workerd/workerd.capnp"; const config :Workerd.Config = (  services = [    ${services},  ],  sockets = [    ( name = "http", address = "127.0.0.1:${input.listenPort}", http = (), service = "main" ),  ],); const mainWorker :Workerd.Worker = (  modules = [    ${moduleEntries},  ],  compatibilityDate = ${capnpString(input.compatibilityDate)},${flags}${bindings}  globalOutbound = ${capnpString(input.globalOutbound)},); const blockedWorker :Workerd.Worker = (  serviceWorkerScript = "addEventListener('fetch', event => { event.respondWith(new Response('Outbound fetch is blocked.', { status: 403 })); })",  compatibilityDate = ${capnpString(input.compatibilityDate)},);`;}; const startToolServer = async (  bridge: WorkerdToolBridge | undefined,  token: string,): Promise<{ readonly server: Server; readonly port: number }> => {  const server = createServer((request, response) => {    void (async () => {      if (request.method !== "POST" || request.url !== "/tool") {        writeJson(response, 404, { ok: false, error: "not found" });        return;      }      if (request.headers["x-executor-token"] !== token) {        writeJson(response, 401, { ok: false, error: "unauthorized" });        return;      }      if (!bridge) {        writeJson(response, 403, { ok: false, error: "tool bridge is not bound" });        return;      }      try {        const body = (await readRequestBody(request)) as { toolPath?: unknown; args?: unknown };        if (typeof body.toolPath !== "string" || body.toolPath.length === 0) {          writeJson(response, 400, { ok: false, error: "toolPath is required" });          return;        }        const result = await bridge.call(body.toolPath, body.args);        writeJson(response, 200, { ok: true, result });      } catch (cause) {        writeJson(response, 200, { ok: false, error: normalizeError(cause).message });      }    })();  });  const port = await listenLocal(server);  return { server, port };}; const waitReady = async (port: number, token: string, timeoutMs: number): Promise<void> => {  const startedAt = performance.now();  let lastError: unknown;  while (performance.now() - startedAt < timeoutMs) {    try {      const response = await fetch(`http://127.0.0.1:${port}/__health`, {        method: "POST",        headers: { "x-executor-token": token },      });      if (response.ok) {        const body = (await response.json().catch(() => null)) as {          readonly runnerToken?: unknown;        } | null;        if (body?.runnerToken === token) return;        lastError = new Error("health check did not echo the runner token");      } else {        lastError = new Error(`health check returned ${response.status}`);      }    } catch (cause) {      lastError = cause;    }    await new Promise((resolve) => setTimeout(resolve, 20));  }  throw new Error(`workerd did not become ready: ${normalizeError(lastError).message}`);}; export const createWorkerdModuleRunner = (  options: WorkerdModuleRunnerOptions,): WorkerdModuleRunner => {  const token = options.hostToken ?? crypto.randomUUID();  const startupTimeoutMs = Math.max(100, options.startupTimeoutMs ?? 30_000);  const restartBackoffMs = Math.max(0, options.restartBackoffMs ?? 100);  const compatibilityDate = options.compatibilityDate ?? "2025-06-01";  const compatibilityFlags = options.compatibilityFlags ?? ["nodejs_compat"];  const pending = new Set<PendingRequest>();  let processState:    | {        readonly proc: ChildProcessByStdio<null, Readable, Readable>;        readonly tmp: string;        readonly listenPort: number;        readonly hostServer: Server;      }    | undefined;  let starting: Promise<void> | undefined;  let disposed = false;  let lastExitAt = 0;   const failPending = (message: string) => {    const error = new Error(message);    for (const request of pending) {      request.reject(error);    }    pending.clear();  };   const disposeState = async (kill: boolean) => {    const state = processState;    processState = undefined;    if (!state) return;    if (kill && state.proc.exitCode === null && state.proc.signalCode === null) {      // workerd serve ignores SIGTERM in this version. SIGKILL is the verified immediate shutdown path.      const exited = new Promise<void>((resolve) => state.proc.once("exit", () => resolve()));      state.proc.kill("SIGKILL");      await exited;    }    await closeServer(state.hostServer).catch(() => undefined);    await rm(state.tmp, { recursive: true, force: true }).catch(() => undefined);  };   const start = async () => {    if (disposed) throw new Error("workerd runner has been disposed");    if (processState) return;    if (starting) return starting;     starting = (async () => {      const now = Date.now();      const delayMs = Math.max(0, restartBackoffMs - (now - lastExitAt));      if (delayMs > 0) {        await new Promise((resolve) => setTimeout(resolve, delayMs));      }       const tmp = join(        tmpdir(),        `executor-workerd-${process.pid}-${Date.now()}-${crypto.randomUUID()}`,      );      await mkdir(tmp, { recursive: true, mode: 0o700 });      for (const [name, source] of Object.entries(options.modules)) {        await mkdir(dirname(join(tmp, name)), { recursive: true, mode: 0o700 });        if (typeof source === "string") {          await writeFile(join(tmp, name), source, { mode: 0o600 });        } else if (source.kind === "wasm") {          await writeFile(join(tmp, name), source.bytes, { mode: 0o600 });        } else {          await writeFile(join(tmp, name), source.source, { mode: 0o600 });        }      }       const hostServer = await startToolServer(options.toolBridge, token);      const reservation = await reserveLocalPort();      const listenPort = reservation.port;      await writeFile(        join(tmp, "config.capnp"),        buildConfig({          listenPort,          hostPort: options.toolBridge ? hostServer.port : null,          modules: options.modules,          mainModule: options.mainModule,          compatibilityDate,          compatibilityFlags,          unsafeEval: options.unsafeEval ?? false,          globalOutbound: options.globalOutbound ?? "blocked",        }),        { mode: 0o600 },      );      await reservation.close();       const workerdBin = await findWorkerdBin(options.workerdBin);      const proc = spawn(workerdBin, ["serve", join(tmp, "config.capnp"), "--experimental"], {        stdio: ["ignore", "pipe", "pipe"],      });      proc.stderr.setEncoding("utf8");      proc.stdout.resume();      proc.stderr.resume();      proc.on("exit", (code, signal) => {        lastExitAt = Date.now();        if (processState?.proc === proc) {          processState = undefined;        }        failPending(          `workerd subprocess exited while requests were in flight (code=${String(code)} signal=${String(signal)})`,        );      });      proc.on("error", (cause) => {        failPending(`workerd subprocess error: ${normalizeError(cause).message}`);      });      processState = { proc, tmp, listenPort, hostServer: hostServer.server };      try {        await waitReady(listenPort, token, startupTimeoutMs);      } catch (cause) {        await disposeState(true);        throw cause;      }    })();     try {      await starting;    } finally {      starting = undefined;    }  };   const run = async <T = unknown>(    payload: unknown,    timeoutMs?: number,  ): Promise<WorkerdRunResult<T>> => {    await start();    const state = processState;    if (!state) throw new Error("workerd subprocess is not running");     const controller = new AbortController();    const effectiveTimeoutMs = Math.max(100, timeoutMs ?? DEFAULT_TIMEOUT_MS);    const timer = setTimeout(      () => controller.abort(),      effectiveTimeoutMs + DEFAULT_HOST_TIMEOUT_GRACE_MS,    );    const startedAt = performance.now();    return await new Promise<WorkerdRunResult<T>>((resolve, reject) => {      const pendingRequest = { reject };      pending.add(pendingRequest);      fetch(`http://127.0.0.1:${state.listenPort}/run`, {        method: "POST",        headers: { "content-type": "application/json" },        body: JSON.stringify(payload),        signal: controller.signal,      })        .then(async (response) => {          const text = await response.text();          const body = text.length > 0 ? (JSON.parse(text) as T) : (null as T);          resolve({ status: response.status, body, elapsedMs: performance.now() - startedAt });        })        .catch((cause) => {          reject(normalizeError(cause));        })        .finally(() => {          clearTimeout(timer);          pending.delete(pendingRequest);        });    });  };   return {    pid: () => processState?.proc.pid,    ensureStarted: start,    run,

Lines 1–500 of 645. Load more