quixos-nix-helpers/restricted-execution/service.mjs

import { spawn } from "node:child_process";
import { randomUUID } from "node:crypto";
import { createServer, request as httpRequest } from "node:http";
import { mkdtemp, readFile, writeFile, rm, open } from "node:fs/promises";
import { constants } from "node:fs";
import { tmpdir } from "node:os";
import { join, dirname } from "node:path";

const abi = "quixos-transaction-ports-v1";
const fixedError = (code) => Object.assign(Error(code), { code });
const object = (v) => !!v && typeof v === "object" && !Array.isArray(v);
const absolute = (value) =>
  typeof value === "string" &&
  value.startsWith("/") &&
  !/[\s\u0000-\u001f]/.test(value) &&
  !value.split("/").includes("..");

/** Host-wide engine reservations. Only trusted adapters hold these callbacks;
 * workers cannot alter admission or evict another invocation's engine. */
export class RestrictedEngineBudget {
  #leases = new Set();
  constructor(limit = 8, reserve = async () => ({})) {
    if (!Number.isSafeInteger(limit) || limit < 1 || limit > 64) throw fixedError("INVALID_ENGINE_BUDGET");
    this.limit = limit;
    this.reserve = reserve;
  }
  async acquire(idle, evict) {
    while (this.#leases.size >= this.limit) {
      const victim = [...this.#leases].find((entry) => entry.idle());
      if (!victim) throw fixedError("RESTRICTED_ENGINE_LIMIT");
      await victim.evict();
    }
    const lease = { idle, evict };
    this.#leases.add(lease);
    try {
      const resource = await this.reserve();
      let releasing;
      const release = () =>
        (releasing ??= (async () => {
          await resource.release?.();
          this.#leases.delete(lease);
        })());
      release.attach = resource.attach;
      release.usage = resource.usage;
      release.kill = resource.kill;
      release.uid = resource.uid;
      return release;
    } catch (error) {
      this.#leases.delete(lease);
      throw error;
    }
  }
  stats() {
    return { engines: this.#leases.size, limit: this.limit };
  }
}
async function boundedSource(path, maximum, code) {
  const handle = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
  try {
    const stat = await handle.stat();
    if (!stat.isFile() || stat.size > maximum) throw fixedError(code);
    const bytes = Buffer.alloc(stat.size + 1);
    let offset = 0;
    while (offset < bytes.length) {
      const read = await handle.read(bytes, offset, bytes.length - offset, offset);
      if (!read.bytesRead) break;
      offset += read.bytesRead;
    }
    if (offset !== stat.size) throw fixedError(code);
    return new TextDecoder("utf-8", { fatal: true }).decode(bytes.subarray(0, offset));
  } finally {
    await handle.close();
  }
}

/** Trusted stdio adapter inside one workspace VM. Engine processes never own
 * coordinator credentials. The supervising service must impose VM/OS limits.
 * Each engine runs one fresh Worker per invocation. Host tokens expire before
 * reuse; cancellation and watchdogs destroy the affected engine, not neighbors. */
export async function serveRestrictedRegistry(config, io = process) {
  if (
    !object(config) ||
    !absolute(config.workerd) ||
    !absolute(config.worker) ||
    !absolute(config.module) ||
    (config.sandbox !== undefined && !absolute(config.sandbox)) ||
    (config.sandboxIdentity !== undefined && typeof config.sandboxIdentity !== "boolean") ||
    !/^[$A-Z_a-z][$\w]*$/.test(config.registryExport) ||
    !Array.isArray(config.exports) ||
    !config.exports.length ||
    config.exports.length > 10000 ||
    config.exports.some((x) => typeof x !== "string" || !x || x.length > 4096) ||
    new Set(config.exports).size !== config.exports.length ||
    !Array.isArray(config.sdk) ||
    config.sdk.length > 8 ||
    config.sdk.some(
      (m) =>
        !object(m) ||
        !absolute(m.path) ||
        typeof m.name !== "string" ||
        !m.name ||
        /[\u0000-\u001f]/.test(m.name) ||
        ["entry.mjs", "package.mjs"].includes(m.name),
    ) ||
    new Set(config.sdk.map((m) => m.name)).size !== config.sdk.length
  )
    throw fixedError("INVALID_RESTRICTED_CONFIGURATION");
  const limits = {
    maxBytes: config.maxBytes ?? 1024 * 1024,
    maxPortCalls: config.maxPortCalls ?? 1000,
    deadlineMs: config.deadlineMs ?? 4000,
    maxQueued: config.maxQueued ?? 8,
    maxActive: config.maxActive ?? 2,
    maxNested: config.maxNested ?? 4,
    engineIdleMs: config.engineIdleMs ?? 30000,
  };
  if (
    !Number.isSafeInteger(limits.maxBytes) ||
    limits.maxBytes < 1024 ||
    limits.maxBytes > 4 * 1024 * 1024 ||
    !Number.isSafeInteger(limits.maxPortCalls) ||
    limits.maxPortCalls < 1 ||
    limits.maxPortCalls > 10000 ||
    !Number.isSafeInteger(limits.deadlineMs) ||
    limits.deadlineMs < 1 ||
    limits.deadlineMs > 60000 ||
    !Number.isSafeInteger(limits.maxQueued) ||
    limits.maxQueued < 0 ||
    limits.maxQueued > 64 ||
    !Number.isSafeInteger(limits.maxActive) ||
    limits.maxActive < 1 ||
    limits.maxActive > 16 ||
    !Number.isSafeInteger(limits.maxNested) ||
    limits.maxNested < 1 ||
    limits.maxNested > 16 ||
    !Number.isSafeInteger(limits.engineIdleMs) ||
    limits.engineIdleMs < 1 ||
    limits.engineIdleMs > 60000
  )
    throw fixedError("INVALID_RESTRICTED_LIMITS");
  const packageSource = await boundedSource(config.module, 1024 * 1024, "RESTRICTED_SOURCE_LIMIT");
  const workerSource = await boundedSource(config.worker, 65536, "RESTRICTED_ENTRY_LIMIT");
  const sdk = [];
  for (const m of config.sdk)
    sdk.push({ ...m, source: await boundedSource(m.path, 2 * 1024 * 1024, "RESTRICTED_SDK_LIMIT") });
  const engineBudget = config.engineBudget ?? new RestrictedEngineBudget(limits.maxActive + limits.maxNested);
  const active = new Map(),
    engines = new Set(),
    idleEngines = [];
  const metrics = { starts: 0, reused: 0, completed: 0, peakRSSBytes: 0 };
  let buffered = Buffer.alloc(0),
    stopped = false,
    failure;
  const queued = [];
  const send = (value) => {
    if (stopped) return;
    const line = JSON.stringify({ abi, ...value }) + "\n";
    if (
      Buffer.byteLength(line) > limits.maxBytes ||
      io.stdout.writableLength + Buffer.byteLength(line) > limits.maxBytes * 2
    )
      throw fixedError("RESTRICTED_BACKPRESSURE");
    io.stdout.write(line);
  };
  // A sandbox has namespace helpers as well as workerd. Charge the complete
  // delegated cgroup when available; the qualification fallback follows the
  // bounded process tree, never trusting a worker's self-reported counters.
  const engineUsage = async (engine) => {
    if (engine.release?.usage) return engine.release.usage();
    engine.processTicks ??= new Map();
    const pending = [engine.child.pid];
    let memoryBytes = 0,
      visited = 0;
    while (pending.length) {
      if (++visited > 64) throw fixedError("RESTRICTED_PROCESS_LIMIT");
      const pid = pending.pop();
      try {
        const stat = await readFile(`/proc/${pid}/stat`, "utf8");
        const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" ");
        engine.processTicks.set(`${pid}:${fields[19]}`, Number(fields[11]) + Number(fields[12]));
        const status = await readFile(`/proc/${pid}/status`, "utf8");
        memoryBytes += Number(status.match(/^VmRSS:\s+(\d+)/m)?.[1] ?? 0) * 1024;
        const children = await readFile(`/proc/${pid}/task/${pid}/children`, "utf8");
        pending.push(...children.trim().split(/\s+/).filter(Boolean).map(Number));
      } catch (error) {
        if (!["ENOENT", "ESRCH"].includes(error.code)) throw error;
      }
    }
    return {
      cpuMicroseconds: [...engine.processTicks.values()].reduce((sum, ticks) => sum + ticks * 10000, 0),
      memoryBytes,
    };
  };
  const sampleCPU = async (call) => {
    if (!call.child?.pid) return;
    const { cpuMicroseconds: total } = await engineUsage(call.engine);
    const cpuMicroseconds = total - (call.cpuBaseline ?? 0);
    if (call.engine) call.engine.cpuTotal = total;
    if (!Number.isSafeInteger(cpuMicroseconds) || cpuMicroseconds < (call.cpuMicroseconds ?? 0))
      throw fixedError("RESTRICTED_CPU_ACCOUNTING");
    if (cpuMicroseconds !== (call.cpuMicroseconds ?? 0)) {
      call.cpuMicroseconds = cpuMicroseconds;
      send({ kind: "usage", invocation: call.message.invocation, frame: call.message.frame, cpuMicroseconds });
    }
  };
  const watchCPU = (call) => {
    call.cpuTimer = setTimeout(() => {
      call.cpuRead = sampleCPU(call).then(() => {
        if (call.open) watchCPU(call);
      });
      void call.cpuRead.catch(fail);
    }, 20);
  };
  const retire = async (call, reason) => {
    if (call.retiring) return call.retiring;
    call.open = false;
    call.retiring = (async () => {
      clearTimeout(call.timer);
      clearTimeout(call.cpuTimer);
      const index = queued.indexOf(call);
      if (index >= 0) queued.splice(index, 1);
      for (const pending of call.pending.values()) pending.reject(reason);
      call.pending.clear();
      // Cancellation can arrive during async directory/socket preparation.
      // Wait for it to settle before claiming every process/handle is gone.
      await call.preparing?.catch(() => {});
      clearTimeout(call.cpuTimer);
      await call.cpuRead?.catch(() => {});
      let usageFailure;
      try {
        await sampleCPU(call);
      } catch (error) {
        usageFailure = error;
      }
      const engine = call.engine;
      if (engine) {
        // Revoke the frame token before returning an engine to the idle pool.
        // A delayed child request can never acquire the next call's authority.
        if (engine.current === call) engine.current = null;
        if (!stopped && reason.code === "INVOCATION_ENDED" && !usageFailure && !engine.destroying) {
          const rss = (await engineUsage(engine)).memoryBytes;
          metrics.peakRSSBytes = Math.max(metrics.peakRSSBytes, rss);
          // Bound retained engine heaps as well as lifetime. Hard aggregate
          // physical memory remains the supervising process/cgroup's contract.
          if (rss <= 128 * 1024 * 1024) {
            idleEngines.push(engine);
            engine.idleTimer = setTimeout(() => void destroyEngine(engine).catch(fail), limits.engineIdleMs);
          } else await destroyEngine(engine);
        } else await destroyEngine(engine);
      }
      active.delete(call.message.invocation);
      if (usageFailure) throw usageFailure;
    })();
    return call.retiring;
  };
  const fail = (error) => {
    if (failure) return failure;
    stopped = true;
    return (failure = (async () => {
      io.stdin.destroy();
      for (const call of [...queued]) await retire(call, error);
      await Promise.all([...active.values()].map((call) => retire(call, error)));
      await Promise.all([...engines].map(destroyEngine));
      io.stdout.end();
      io.exitCode = 1;
    })());
  };
  const body = async (stream) => {
    const chunks = [];
    let bytes = 0;
    for await (const chunk of stream) {
      bytes += chunk.length;
      if (bytes > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT");
      chunks.push(chunk);
    }
    return JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(Buffer.concat(chunks)));
  };
  const destroyEngine = async (engine) => {
    if (engine.destroying) return engine.destroying;
    engine.destroying = (async () => {
      clearTimeout(engine.idleTimer);
      const index = idleEngines.indexOf(engine);
      if (index >= 0) idleEngines.splice(index, 1);
      engine.current = null;
      if (engine.child?.pid) {
        if (engine.release.kill) await engine.release.kill();
        else {
          try {
            process.kill(-engine.child.pid, "SIGKILL");
          } catch (error) {
            if (error.code !== "ESRCH") throw error;
          }
        }
        await engine.exited;
      }
      engine.server?.closeAllConnections();
      if (engine.server) await new Promise((resolve) => engine.server.close(resolve));
      try {
        if (engine.directory) await rm(engine.directory, { recursive: true, force: true });
      } finally {
        engines.delete(engine);
        await engine.release?.();
      }
    })();
    return engine.destroying;
  };
  const prepareEngine = async (call) => {
    const cached = idleEngines.pop();
    if (cached) {
      clearTimeout(cached.idleTimer);
      cached.current = call;
      call.engine = cached;
      call.child = cached.child;
      call.cpuBaseline = cached.cpuTotal ?? 0;
      metrics.reused++;
      watchCPU(call);
      return;
    }
    const engine = { current: call, cpuTotal: 0 };
    engines.add(engine);
    call.engine = engine;
    engine.release = await engineBudget.acquire(
      () => !engine.current && !engine.destroying,
      () => destroyEngine(engine),
    );
    if (!call.open) return;
    engine.directory = await mkdtemp(join(tmpdir(), "qx-engine-"));
    if (!call.open) return;
    const broker = join(engine.directory, "broker.sock");
    engine.socket = join(engine.directory, "engine.sock");
    engine.server = createServer(async (request, response) => {
      const call = engine.current;
      const message = call?.message;
      try {
        if (!call?.open || request.method !== "POST" || request.url !== "/port/" + call.token)
          throw fixedError("EXPIRED_CONTEXT");
        const port = await body(request);
        if (
          !call.open ||
          !object(port) ||
          Object.keys(port).some((key) => !["portId", "action", "input", "key"].includes(key)) ||
          typeof port.portId !== "string" ||
          !object(port.action) ||
          !Object.hasOwn(port, "input")
        )
          throw fixedError("INVALID_RESTRICTED_PORT");
        if (
          ++call.sequence > Math.min(limits.maxPortCalls, message.budget?.calls ?? limits.maxPortCalls) ||
          call.pending.size >= limits.maxPortCalls
        )
          throw fixedError("RESTRICTED_PORT_LIMIT");
        const sequence = call.sequence;
        const result = new Promise((resolve, reject) => call.pending.set(sequence, { resolve, reject }));
        void result.catch(() => {});
        send({
          kind: "port",
          invocation: message.invocation,
          frame: message.frame,
          sequence,
          portId: port.portId,
          action: { ...port.action, ...(port.key === undefined ? {} : { key: port.key }) },
          input: port.input,
        });
        const value = await result;
        if (!call.open) throw fixedError("EXPIRED_CONTEXT");
        const encoded = JSON.stringify(value);
        if (Buffer.byteLength(encoded) > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT");
        response.writeHead(200, { "content-type": "application/json" });
        response.end(encoded);
      } catch {
        response.writeHead(409);
        response.end('{"status":"failed","error":{"code":"RESTRICTED_PORT_FAILED"}}');
      }
    });

    await new Promise((resolve, reject) => {
      engine.server.once("error", reject);
      engine.server.listen(broker, resolve);
    });
    const modules = { "entry.mjs": { js: workerSource }, "package.mjs": { js: packageSource } };
    for (const module of sdk) modules[module.name] = { js: module.source };
    const setting = { registryExport: config.registryExport, exports: config.exports, maxBytes: limits.maxBytes };
    // The parent contains immutable checked source only. It never imports the
    // authored module or keeps a child stub, request, result or frame handle.
    const parent = `const modules=${JSON.stringify(modules)}, setting=${JSON.stringify(setting)};
      export default {async fetch(request,env) {
        const input=await request.json();
        const child=env.LOADER.load({compatibilityDate:"2026-10-05",mainModule:"entry.mjs",modules,
          globalOutbound:null,env:{BROKER:env.BROKER,CONFIG:{...setting,frameToken:input.frameToken}}});
        const response=await child.getEntrypoint().fetch(new Request("http://invocation/",{
          method:"POST",body:JSON.stringify(input.message)}));
        const body=await response.arrayBuffer();
        return new Response(body,{status:response.status,headers:response.headers});
      }};`;
    await writeFile(join(engine.directory, "parent.mjs"), parent);
    const configuration = `using Workerd = import "/workerd/workerd.capnp"; const config :Workerd.Config = (
      services=[(name="main",worker=(modules=[(name="parent.mjs",esModule=embed "parent.mjs")],
        compatibilityDate="2026-10-05",compatibilityFlags=["experimental"],globalOutbound="deny",
        bindings=[(name="LOADER",workerLoader=()),(name="BROKER",service="broker")])),
        (name="broker",external=(address=${JSON.stringify("unix:" + broker)},http=())),
        (name="deny",network=(allow=[]))],
      sockets=[(name="frame",address=${JSON.stringify("unix:" + engine.socket)},http=(),service="main")]);`;
    const file = join(engine.directory, "config.capnp");
    await writeFile(file, configuration);
    if (!call.open) return;
    if (config.sandboxIdentity && !Number.isSafeInteger(engine.release.uid))
      throw fixedError("RESTRICTED_IDENTITY_UNAVAILABLE");
    engine.child = spawn(
      config.sandbox ?? config.workerd,
      config.sandbox
        ? [engine.directory, ...(config.sandboxIdentity ? [String(engine.release.uid)] : [])]
        : ["serve", "--experimental", file],
      {
        detached: true,
        env: { TZ: "UTC", LANG: "C.UTF-8" },
        cwd: engine.directory,
        stdio: [config.sandbox ? "pipe" : "ignore", "ignore", "pipe"],
      },
    );
    metrics.starts++;
    engine.exited = new Promise((resolve) => {
      engine.child.once("exit", resolve);
      engine.child.once("error", resolve);
    });
    engine.child.stdin?.on("error", () => engine.current?.terminate("RESTRICTED_START_FAILED"));
    call.child = engine.child;
    let diagnostics = 0;
    engine.child.stderr.on("data", (chunk) => {
      if ((diagnostics += chunk.length) > 65536) {
        if (engine.current?.open) engine.current.terminate("RESTRICTED_DIAGNOSTIC_LIMIT");
        else void destroyEngine(engine).catch(fail);
      } else config.diagnostics?.(chunk);
    });
    await engine.release.attach?.(engine.child.pid);
    if (config.sandbox && call.open) engine.child.stdin.end(Buffer.from([1]));
    watchCPU(call);
    engine.child.once("error", () => engine.current?.terminate("RESTRICTED_START_FAILED"));
    void engine.exited.then(() => {
      if (engine.current?.open) engine.current.terminate("RESTRICTED_ENGINE_EXITED");
      else void destroyEngine(engine).catch(fail);
    });
  };
  const available = (call) =>
    [...active.values()].filter((entry) => entry.nested === call.nested).length <
    (call.nested ? limits.maxNested : limits.maxActive);
  const pump = () => {
    while (!stopped) {
      const index = queued.findIndex(available);
      if (index < 0) return;
      void invoke(queued.splice(index, 1)[0]).catch(fail);
    }
  };
  const finish = async (call, reason, reply) => {
    if (reply.kind === "cancelled") call.cancellationReply = reply;
    return (call.finished ??= (async () => {
      await retire(call, reason);
      metrics.lastInvocation = {
        queueMs: (call.startedAt ?? performance.now()) - call.receivedAt,
        preparationMs: call.preparationMs ?? 0,
        roundtripMs: call.roundtripMs ?? 0,
        initializationMs: call.initializationMs ?? null,
        executionMs: call.executionMs ?? null,
        totalMs: performance.now() - call.receivedAt,
      };
      send(call.cancellationReply ?? reply);
      pump();
    })());
  };
  const invoke = async (call) => {
    const { message } = call;
    call.startedAt = performance.now();
    active.set(message.invocation, call);
    const terminate = (code) => {
      if (!call.open) return;
      void finish(call, fixedError(code), {
        invocation: message.invocation,
        status: "failed",
        error: { code, message: code },
      }).catch(fail);
    };
    call.terminate = terminate;
    try {
      call.token = randomUUID();
      call.preparing = prepareEngine(call);
      await call.preparing;
      call.preparationMs = performance.now() - call.startedAt;
      if (!call.open) return;
      const roundtripAt = performance.now();
      const result = await new Promise((resolve, reject) => {
        const attempt = () => {
          if (!call.open) {
            reject(fixedError("EXPIRED_CONTEXT"));
            return;
          }
          const request = httpRequest(
            { socketPath: call.engine.socket, path: "/", method: "POST", agent: false },
            async (response) => {
              try {
                const result = await body(response);
                for (const [header, key] of [
                  ["x-quixos-initialization-ms", "initializationMs"],
                  ["x-quixos-execution-ms", "executionMs"],
                ]) {
                  const value = Number(response.headers[header]);
                  if (Number.isFinite(value) && value >= 0 && value <= limits.deadlineMs) call[key] = value;
                }
                if (response.statusCode !== 200) throw fixedError("RESTRICTED_PACKAGE_FAILED");
                resolve(result);
              } catch (error) {
                reject(error);
              }
            },
          );
          request.once("error", (error) => {
            if (call.open && ["ENOENT", "ECONNREFUSED"].includes(error.code)) setTimeout(attempt, 5);
            else reject(error);
          });
          request.end(JSON.stringify({ message, frameToken: call.token }));
        };
        attempt();
      });
      call.roundtripMs = performance.now() - roundtripAt;
      if (!call.open) return;
      if (
        !object(result) ||
        !["returned", "failed"].includes(result.status) ||
        Object.keys(result).some(
          (key) => !(result.status === "returned" ? ["status", "output"] : ["status", "error"]).includes(key),
        ) ||
        (result.status === "returned"
          ? !Object.hasOwn(result, "output")
          : !object(result.error) || !/^[A-Z0-9_]{1,128}$/.test(result.error.code))
      )
        throw fixedError("RESTRICTED_PACKAGE_FAILED");
      // Authored code can alter serialization prototypes. Treat even the
      // platform worker's JSON as untrusted; never accept routing/ABI fields.
      const reply =
        result.status === "returned"
          ? { invocation: message.invocation, status: "returned", output: result.output }
          : {
              invocation: message.invocation,
              status: "failed",
              error: { code: result.error.code, message: result.error.code },
            };
      await finish(call, fixedError("INVOCATION_ENDED"), reply);
      metrics.completed++;
    } catch (error) {
      config.diagnostics?.(Buffer.from(`RESTRICTED_HOST_ERROR:${error.code ?? error.name}\n`));
      if (call.open) {
        const code = /^RESTRICTED_[A-Z_]+$/.test(error.code ?? "") ? error.code : "RESTRICTED_PACKAGE_FAILED";
        await finish(call, error, { invocation: message.invocation, status: "failed", error: { code, message: code } });
      }
    }
  };
  const receive = (message) => {
    if (!object(message) || message.abi !== abi) throw fixedError("RESTRICTED_ABI_MISMATCH");
    if (message.kind === "shutdown") {
      void fail(fixedError("SERVICE_SHUTDOWN"));
      return;
    }
    if (message.kind === "invoke") {
      if (
        !config.exports.includes(message.exportId) ||
        !["snapshot", "transaction"].includes(message.effect) ||
        typeof message.invocation !== "string" ||
        !message.invocation ||
        message.invocation.length > 4096 ||
        typeof message.frame !== "string" ||
        !message.frame ||
        message.frame.length > 4096
      )
        throw fixedError("INVALID_RESTRICTED_INVOCATION");
      if (
        message.budget !== undefined &&
        (!object(message.budget) ||
          Object.keys(message.budget).sort().join(",") !== "calls,deadlineMs,depth,root" ||
          typeof message.budget.root !== "string" ||
          !message.budget.root ||
          message.budget.root.length > 4096 ||
          !Number.isSafeInteger(message.budget.depth) ||
          message.budget.depth < 0 ||
          message.budget.depth > 32 ||
          !Number.isSafeInteger(message.budget.calls) ||
          message.budget.calls < 1 ||
          message.budget.calls > 1000000 ||
          !Number.isSafeInteger(message.budget.deadlineMs) ||
          message.budget.deadlineMs < 1 ||
          message.budget.deadlineMs > 60000)
      )
        throw fixedError("INVALID_RESTRICTED_BUDGET");
      if (active.has(message.invocation) || queued.some((call) => call.message.invocation === message.invocation))
        throw fixedError("RESTRICTED_DUPLICATE_INVOCATION");
      const call = {
        message,
        receivedAt: performance.now(),
        open: true,
        pending: new Map(),
        sequence: 0,
        nested: (message.budget?.depth ?? 0) > 1,
      };
      // Descendants cannot wait behind ancestors that hold all nested capacity.
      // Reject excess fan-out/depth rather than deadlock an admitted call tree.
      if (!available(call) && (call.nested || queued.length >= limits.maxQueued)) {
        send({
          invocation: message.invocation,
          status: "failed",
          error: { code: "RESTRICTED_QUEUE_FULL", message: "RESTRICTED_QUEUE_FULL" },
        });
        return;
      }
      // Deadline includes queue wait: overload cannot extend a transaction frame.
      call.timer = setTimeout(
        () => {
          if (!call.open) return;
          void finish(call, fixedError("RESTRICTED_DEADLINE"), {
            invocation: message.invocation,
            status: "failed",
            error: { code: "RESTRICTED_DEADLINE", message: "RESTRICTED_DEADLINE" },
          }).catch(fail);
        },
        Math.min(limits.deadlineMs, message.budget?.deadlineMs ?? limits.deadlineMs),
      );
      queued.push(call);
      pump();
      return;
    }
    const waiting = queued.find(
      (call) => call.message.invocation === message.invocation && call.message.frame === message.frame,
    );
    if (waiting && message.kind === "cancel") {
      void finish(waiting, fixedError("INVOCATION_CANCELLED"), {
        kind: "cancelled",
        invocation: message.invocation,
        frame: message.frame,
      }).catch(fail);
      return;
    }
    const call = active.get(message.invocation);
    if (!call || message.invocation !== call.message.invocation || message.frame !== call.message.frame) {
      if (message.kind === "cancel") send({ kind: "cancelled", invocation: message.invocation, frame: message.frame });
      else if (message.kind !== "port-result") throw fixedError("RESTRICTED_UNKNOWN_INVOCATION");
      return;
    }
    if (message.kind === "cancel") {
      void finish(call, fixedError("INVOCATION_CANCELLED"), {
        kind: "cancelled",
        invocation: message.invocation,
        frame: message.frame,
      }).catch(fail);
      return;
    }
    if (message.kind !== "port-result" || !["returned", "failed"].includes(message.status))
      throw fixedError("INVALID_RESTRICTED_RESULT");
    const pending = call.pending.get(message.sequence);
    if (!pending) throw fixedError("RESTRICTED_UNKNOWN_PORT");
    call.pending.delete(message.sequence);
    pending.resolve(message);
  };
  io.stdin.on("data", (chunk) => {
    if (stopped) return;
    try {
      let offset = 0;
      while (offset < chunk.length) {
        const newline = chunk.indexOf(10, offset),
          end = newline < 0 ? chunk.length : newline;
        if (buffered.length + end - offset > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT");
        buffered = Buffer.concat([buffered, chunk.subarray(offset, end)]);
        if (newline < 0) break;
        receive(JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(buffered)));
        buffered = Buffer.alloc(0);
        offset = newline + 1;
        if (stopped) break;
      }
    } catch (error) {
      void fail(error);
    }
  });
  io.stdin.on("end", () => void fail(fixedError("COORDINATOR_DISCONNECTED")));
  io.stdin.on("error", (error) => void fail(error));
  io.stdout.on("error", (error) => void fail(error));
  send({ kind: "ready", exports: config.exports });
  return {
    close: () => fail(fixedError("SERVICE_SHUTDOWN")),
    stats: () => ({
      ...metrics,
      active: active.size,
      queued: queued.length,
      engines: engines.size,
      idle: idleEngines.length,
    }),
  };
}

if (process.argv[1] === new URL(import.meta.url).pathname) {
  const config = JSON.parse(await readFile(process.argv[2], "utf8"));
  await serveRestrictedRegistry({
    ...config,
    workerd: process.env.QUIXOS_RESTRICTED_WORKERD,
    worker: process.env.QUIXOS_RESTRICTED_WORKER,
    sandbox: process.env.QUIXOS_RESTRICTED_SANDBOX,
    module: join(dirname(process.argv[2]), config.module),
  });
}