Publish quixos-nix-helpers from abc91ab1cbbd61ffd7f0a49dbc5768598ac8745c

.quixos-subtree-source.json

@@ -1,7 +1,7 @@
 {
   "version": 1,
   "sourceRepo": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos",
-  "sourceCommit": "039f055a7dcff2a29d59a8c37d7433b269b50f7a",
+  "sourceCommit": "abc91ab1cbbd61ffd7f0a49dbc5768598ac8745c",
   "sourcePath": "quixos-instance/quixos-nix-helpers",
   "exportName": "quixos-nix-helpers",
   "mirrorRemote": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos-nix-helpers.git"

restricted-execution/service.mjs

@@ -1,6 +1,8 @@
 import { spawn } from "node:child_process";
+import { randomUUID } from "node:crypto";
 import { createServer, request as httpRequest } from "node:http";
-import { mkdtemp, readFile, writeFile, rm } from "node:fs/promises";
+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";
 
@@ -13,9 +15,68 @@ const absolute = (value) =>
   !/[\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 () => {
+          try {
+            await resource.release?.();
+          } finally {
+            this.#leases.delete(lease);
+          }
+        })());
+      release.attach = resource.attach;
+      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.
- * Per-frame process replacement prevents retained globals/handles crossing calls. */
+ * 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) ||
@@ -47,6 +108,9 @@ export async function serveRestrictedRegistry(config, io = process) {
     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) ||
@@ -60,18 +124,31 @@ export async function serveRestrictedRegistry(config, io = process) {
     limits.deadlineMs > 60000 ||
     !Number.isSafeInteger(limits.maxQueued) ||
     limits.maxQueued < 0 ||
-    limits.maxQueued > 64
+    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 readFile(config.module, "utf8");
-  if (Buffer.byteLength(packageSource) > 1024 * 1024) throw fixedError("RESTRICTED_SOURCE_LIMIT");
-  const workerSource = await readFile(config.worker, "utf8");
-  if (Buffer.byteLength(workerSource) > 65536) throw fixedError("RESTRICTED_ENTRY_LIMIT");
-  const sdk = await Promise.all(config.sdk.map(async (m) => ({ ...m, source: await readFile(m.path, "utf8") })));
-  if (sdk.some((m) => Buffer.byteLength(m.source) > 2 * 1024 * 1024)) throw fixedError("RESTRICTED_SDK_LIMIT");
-  let active,
-    buffered = Buffer.alloc(0),
-    stopped = false;
+  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;
@@ -83,11 +160,42 @@ export async function serveRestrictedRegistry(config, io = process) {
       throw fixedError("RESTRICTED_BACKPRESSURE");
     io.stdout.write(line);
   };
+  // Linux USER_HZ is 100 on the supported NixOS architectures. Read process
+  // CPU outside the engine; authored code cannot forge these usage messages.
+  const sampleCPU = async (call) => {
+    if (!call.child?.pid) return;
+    let text;
+    try {
+      text = await readFile(`/proc/${call.child.pid}/stat`, "utf8");
+    } catch (error) {
+      if (error.code === "ENOENT" || error.code === "ESRCH") return;
+      throw error;
+    }
+    const fields = text.slice(text.lastIndexOf(")") + 2).split(" ");
+    const total = (Number(fields[11]) + Number(fields[12])) * 10000;
+    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);
@@ -95,29 +203,47 @@ export async function serveRestrictedRegistry(config, io = process) {
       // Cancellation can arrive during async directory/socket preparation.
       // Wait for it to settle before claiming every process/handle is gone.
       await call.preparing?.catch(() => {});
-      if (call.child?.pid) {
-        try {
-          process.kill(-call.child.pid, "SIGKILL");
-        } catch (error) {
-          if (error.code !== "ESRCH") throw error;
-        }
-        await call.exited;
+      clearTimeout(call.cpuTimer);
+      await call.cpuRead?.catch(() => {});
+      let usageFailure;
+      try {
+        await sampleCPU(call);
+      } catch (error) {
+        usageFailure = error;
       }
-      call.server?.closeAllConnections();
-      if (call.server) await new Promise((resolve) => call.server.close(resolve));
-      if (call.directory) await rm(call.directory, { recursive: true, force: true });
-      if (active === call) active = undefined;
+      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 status = await readFile(`/proc/${engine.child.pid}/status`, "utf8").catch(() => "");
+          const rss = Number(status.match(/^VmRSS:\s+(\d+)/m)?.[1] ?? Infinity) * 1024;
+          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 = async (error) => {
-    if (stopped) return;
+  const fail = (error) => {
+    if (failure) return failure;
     stopped = true;
-    io.stdin.destroy();
-    for (const call of [...queued]) await retire(call, error);
-    if (active) await retire(active, error);
-    io.stdout.end();
-    io.exitCode = 1;
+    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 = [];
@@ -129,17 +255,188 @@ export async function serveRestrictedRegistry(config, io = process) {
     }
     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) {
+        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;
+    engine.child = spawn(config.workerd, ["serve", "--experimental", file], {
+      detached: true,
+      env: { TZ: "UTC", LANG: "C.UTF-8" },
+      cwd: engine.directory,
+      stdio: ["ignore", "ignore", "pipe"],
+    });
+    metrics.starts++;
+    engine.exited = new Promise((resolve) => {
+      engine.child.once("exit", resolve);
+      engine.child.once("error", resolve);
+    });
+    call.child = engine.child;
+    await engine.release.attach?.(engine.child.pid);
+    watchCPU(call);
+    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);
+      }
+    });
+    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 = () => {
-    if (!stopped && !active && queued.length) void invoke(queued.shift()).catch(fail);
+    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) => {
-    await retire(call, reason);
-    send(reply);
-    pump();
+    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;
-    active = call;
+    call.startedAt = performance.now();
+    active.set(message.invocation, call);
     const terminate = (code) => {
       if (!call.open) return;
       void finish(call, fixedError(code), {
@@ -150,119 +447,46 @@ export async function serveRestrictedRegistry(config, io = process) {
     };
     call.terminate = terminate;
     try {
-      let engine;
-      call.preparing = (async () => {
-        call.directory = await mkdtemp(join(tmpdir(), "qx-frame-"));
-        if (!call.open) return;
-        const broker = join(call.directory, "broker.sock");
-        engine = join(call.directory, "engine.sock");
-        call.server = createServer(async (request, response) => {
-          try {
-            if (!call.open || request.method !== "POST" || request.url !== "/port") 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 > 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) => {
-          call.server.once("error", reject);
-          call.server.listen(broker, resolve);
-        });
-        await writeFile(join(call.directory, "package.mjs"), packageSource);
-        await writeFile(join(call.directory, "entry.mjs"), workerSource);
-        const modules = [
-          { name: "entry.mjs", esModule: 'embed "entry.mjs"' },
-          { name: "package.mjs", esModule: 'embed "package.mjs"' },
-        ];
-        for (const [index, module] of sdk.entries()) {
-          const file = `sdk-${index}.mjs`;
-          await writeFile(join(call.directory, file), module.source);
-          modules.push({ name: module.name, esModule: "embed " + JSON.stringify(file) });
-        }
-        const setting = { registryExport: config.registryExport, exports: config.exports, maxBytes: limits.maxBytes };
-        const configuration = `using Workerd = import "/workerd/workerd.capnp"; const config :Workerd.Config = (
-        services = [(name="main", worker=(modules=[${modules.map((m) => `(name=${JSON.stringify(m.name)}, esModule=${m.esModule})`).join(",")}],
-          compatibilityDate="2026-10-05", globalOutbound="deny",
-          bindings=[(name="BROKER", service="broker"),(name="CONFIG",json=${JSON.stringify(JSON.stringify(setting))})])),
-          (name="broker",external=(address=${JSON.stringify("unix:" + broker)},http=())), (name="deny",network=(allow=[]))],
-        sockets=[(name="frame",address=${JSON.stringify("unix:" + engine)},http=(),service="main")]);`;
-        const file = join(call.directory, "config.capnp");
-        await writeFile(file, configuration);
-        if (!call.open) return;
-        call.child = spawn(config.workerd, ["serve", file], {
-          detached: true,
-          env: { TZ: "UTC", LANG: "C.UTF-8" },
-          cwd: call.directory,
-          stdio: ["ignore", "ignore", "pipe"],
-        });
-        call.exited = new Promise((resolve) => {
-          call.child.once("exit", resolve);
-          call.child.once("error", resolve);
-        });
-        let diagnostics = 0;
-        call.child.stderr.on("data", (chunk) => {
-          if ((diagnostics += chunk.length) > 65536) terminate("RESTRICTED_DIAGNOSTIC_LIMIT");
-        });
-        call.child.once("error", () => terminate("RESTRICTED_START_FAILED"));
-        void call.exited.then(() => {
-          if (call.open) terminate("RESTRICTED_ENGINE_EXITED");
-        });
-      })();
+      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: engine, path: "/", method: "POST" }, async (response) => {
-            try {
-              const result = await body(response);
-              if (response.statusCode !== 200) throw fixedError("RESTRICTED_PACKAGE_FAILED");
-              resolve(result);
-            } catch (error) {
-              reject(error);
-            }
-          });
+          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));
+          request.end(JSON.stringify({ message, frameToken: call.token }));
         };
         attempt();
       });
+      call.roundtripMs = performance.now() - roundtripAt;
       if (!call.open) return;
       if (
         !object(result) ||
@@ -286,6 +510,7 @@ export async function serveRestrictedRegistry(config, io = process) {
               error: { code: result.error.code, message: result.error.code },
             };
       await finish(call, fixedError("INVOCATION_ENDED"), reply);
+      metrics.completed++;
     } catch (error) {
       if (call.open) {
         const code = /^RESTRICTED_[A-Z_]+$/.test(error.code ?? "") ? error.code : "RESTRICTED_PACKAGE_FAILED";
@@ -311,9 +536,37 @@ export async function serveRestrictedRegistry(config, io = process) {
         message.frame.length > 4096
       )
         throw fixedError("INVALID_RESTRICTED_INVOCATION");
-      if ([active, ...queued].some((call) => call?.message.invocation === message.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");
-      if (active && queued.length >= limits.maxQueued) {
+      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",
@@ -321,16 +574,18 @@ export async function serveRestrictedRegistry(config, io = process) {
         });
         return;
       }
-      const call = { message, open: true, pending: new Map(), sequence: 0 };
       // 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);
-      }, limits.deadlineMs);
+      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;
@@ -346,7 +601,7 @@ export async function serveRestrictedRegistry(config, io = process) {
       }).catch(fail);
       return;
     }
-    const call = active;
+    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");
@@ -390,7 +645,16 @@ export async function serveRestrictedRegistry(config, io = process) {
   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")) };
+  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) {

restricted-execution/service.test.mjs

@@ -5,7 +5,7 @@ import { mkdtemp, writeFile, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { fileURLToPath } from "node:url";
-import { serveRestrictedRegistry } from "./service.mjs";
+import { serveRestrictedRegistry, RestrictedEngineBudget } from "./service.mjs";
 
 const workerd = process.env.QUIXOS_TEST_WORKERD;
 if (!workerd) throw Error("Set QUIXOS_TEST_WORKERD to the pinned engine");
@@ -19,6 +19,7 @@ async function fixture(source, options = {}) {
   const io = { stdin: new PassThrough(), stdout: new PassThrough(), exitCode: 0 };
   const waiting = [],
     queued = [];
+  const usage = [];
   let buffer = "";
   io.stdout.on("data", (chunk) => {
     buffer += chunk;
@@ -26,6 +27,10 @@ async function fixture(source, options = {}) {
     while ((index = buffer.indexOf("\n")) >= 0) {
       const value = JSON.parse(buffer.slice(0, index));
       buffer = buffer.slice(index + 1);
+      if (value.kind === "usage") {
+        usage.push(value);
+        continue;
+      }
       if (waiting.length) waiting.shift()(value);
       else queued.push(value);
     }
@@ -50,7 +55,7 @@ async function fixture(source, options = {}) {
   );
   assert.deepEqual(await receive(), { abi, kind: "ready", exports: ["test"] });
   const send = (value) => io.stdin.write(JSON.stringify({ abi, ...value }) + "\n");
-  const start = (input = null) => {
+  const start = (input = null, budget) => {
     const invocation = "invoke-" + ++next,
       frame = "frame-" + next;
     send({
@@ -62,6 +67,7 @@ async function fixture(source, options = {}) {
       receiver: null,
       input,
       effect: "snapshot",
+      ...(budget ? { budget } : {}),
     });
     return { invocation, frame };
   };
@@ -70,6 +76,8 @@ async function fixture(source, options = {}) {
     receive,
     send,
     start,
+    usage,
+    stats: service.stats,
     async close() {
       await service.close();
       await rm(directory, { recursive: true, force: true });
@@ -78,6 +86,94 @@ async function fixture(source, options = {}) {
 }
 const registry = (code) => `export const registry={exports:['test'],async invoke(request,channel){${code}}};`;
 
+test("parallel frames route independently and cancellation cannot retire a neighbor", async () => {
+  const f = await fixture(
+    registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:request.input});"),
+  );
+  try {
+    const first = f.start("a"),
+      second = f.start("b");
+    const ports = new Map([await f.receive(), await f.receive()].map((port) => [port.invocation, port]));
+    assert.equal(ports.get(first.invocation).input, "a");
+    assert.equal(ports.get(second.invocation).input, "b");
+    f.send({ kind: "cancel", ...first });
+    assert.equal((await f.receive()).kind, "cancelled");
+    f.send({
+      kind: "port-result",
+      ...first,
+      sequence: ports.get(first.invocation).sequence,
+      status: "returned",
+      output: "late",
+    });
+    f.send({
+      kind: "port-result",
+      ...second,
+      sequence: ports.get(second.invocation).sequence,
+      status: "returned",
+      output: "survived",
+    });
+    const result = await f.receive();
+    assert.equal(result.invocation, second.invocation);
+    assert.equal(result.output, "survived");
+    assert.equal(f.io.exitCode, 0);
+  } finally {
+    await f.close();
+  }
+});
+
+test("A to B to A uses reserved nested capacity while both root lanes are saturated", async () => {
+  const source = registry(
+    "if(request.input==='leaf')return 'done'; return await channel({portId:'next',action:{kind:'state',primitive:'read'},input:request.input});",
+  );
+  const a = await fixture(source, { maxActive: 1, maxNested: 1, maxQueued: 1 });
+  const b = await fixture(source, { maxActive: 1, maxNested: 1, maxQueued: 1 });
+  const budget = (depth) => ({ root: "root-chain", depth, deadlineMs: 2000, calls: 20 });
+  try {
+    const parent = a.start("B", budget(1)),
+      aPort = await a.receive();
+    const neighbor = b.start("held", { ...budget(1), root: "other-root" });
+    await b.receive();
+    const child = b.start("A", budget(2)),
+      bPort = await b.receive();
+    const leaf = a.start("leaf", budget(3));
+    const leafResult = await a.receive();
+    assert.equal(leafResult.invocation, leaf.invocation);
+    assert.equal(leafResult.output, "done");
+    b.send({ kind: "port-result", ...child, sequence: bPort.sequence, status: "returned", output: leafResult.output });
+    const childResult = await b.receive();
+    a.send({
+      kind: "port-result",
+      ...parent,
+      sequence: aPort.sequence,
+      status: "returned",
+      output: childResult.output,
+    });
+    assert.equal((await a.receive()).output, "done");
+    b.send({ kind: "cancel", ...neighbor });
+    assert.equal((await b.receive()).kind, "cancelled");
+  } finally {
+    await a.close();
+    await b.close();
+  }
+});
+
+test("the remaining root wall budget overrides a longer adapter deadline", async () => {
+  const f = await fixture(
+    registry("return await channel({portId:'held',action:{kind:'state',primitive:'read'},input:null});"),
+    { deadlineMs: 2000 },
+  );
+  try {
+    const started = performance.now();
+    f.start(null, { root: "short-root", depth: 2, deadlineMs: 150, calls: 10 });
+    await f.receive();
+    const result = await f.receive();
+    assert.equal(result.error.code, "RESTRICTED_DEADLINE");
+    assert.ok(performance.now() - started < 1000);
+  } finally {
+    await f.close();
+  }
+});
+
 test("real engine returns results and resets all globals between frames", async () => {
   const f = await fixture(
     registry(
@@ -222,6 +318,10 @@ test("initialization and regex loops are killed, then a fresh frame can start",
         f.start();
         const result = await f.receive();
         assert.equal(result.error.code, "RESTRICTED_DEADLINE");
+        assert.ok(
+          f.usage.some((value) => value.cpuMicroseconds >= 50000),
+          "External watchdog did not account spinning engine CPU",
+        );
       }
     } finally {
       await f.close();
@@ -268,7 +368,7 @@ test("cancellation during asynchronous preparation leaves no running frame", asy
 test("bounded FIFO rejects overload without losing active or queued frames", async () => {
   const f = await fixture(
     registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:request.input});"),
-    { maxQueued: 2 },
+    { maxQueued: 2, maxActive: 1 },
   );
   try {
     const first = f.start(1),
@@ -307,7 +407,7 @@ test("bounded FIFO rejects overload without losing active or queued frames", asy
 test("queue wait consumes the deadline and cancelled work never reaches its ports", async () => {
   const f = await fixture(
     registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"),
-    { deadlineMs: 350 },
+    { deadlineMs: 350, maxActive: 1 },
   );
   try {
     const first = f.start();
@@ -325,3 +425,174 @@ test("queue wait consumes the deadline and cancelled work never reaches its port
     await f.close();
   }
 });
+
+test("warm engines create fresh transitive module state, closures and built-ins", async () => {
+  const directory = await mkdtemp(join(tmpdir(), "qx-transitive-"));
+  const helper = join(directory, "helper.mjs");
+  await writeFile(
+    helper,
+    "let value=0, retained; export function touch(channel){const old=retained; retained=channel; return {count:++value, captured:!!old};}",
+  );
+  const f = await fixture(
+    `import {touch} from 'helper.mjs'; let count=0;
+    export const registry={exports:['test'],async invoke(request,channel){
+      const before={object:Object.prototype.retained??null,array:Array.prototype.retained??null,
+        map:Map.prototype.retained??null, global:globalThis.retained??null};
+      Object.prototype.retained=request.input; Array.prototype.retained=request.input;
+      Map.prototype.retained=request.input; globalThis.retained=request.input;
+      return {count:++count, helper:touch(channel), before};
+    }};`,
+    { sdk: [{ name: "helper.mjs", path: helper }] },
+  );
+  try {
+    for (const value of ["private graph A", "private graph B", "private graph C"]) {
+      f.start(value);
+      assert.deepEqual((await f.receive()).output, {
+        count: 1,
+        helper: { count: 1, captured: false },
+        before: { object: null, array: null, map: null, global: null },
+      });
+    }
+    assert.equal(f.stats().starts, 1, "Qualification must exercise actual engine reuse");
+    assert.equal(f.stats().reused, 2);
+    assert.equal(f.stats().idle, 1);
+  } finally {
+    await f.close();
+    await rm(directory, { recursive: true, force: true });
+  }
+});
+
+test("an expired frame token cannot borrow a later invocation's broker", async () => {
+  const directory = await mkdtemp(join(tmpdir(), "qx-token-"));
+  const worker = join(directory, "token-test.mjs");
+  // Trusted test entrypoint deliberately replays the old token. Authored
+  // registries never receive CONFIG or BROKER directly.
+  await writeFile(
+    worker,
+    `export default {async fetch(request,env) {
+    const message=await request.json();
+    const output=message.input ? (await env.BROKER.fetch('http://frame/port/'+message.input,{
+      method:'POST',body:JSON.stringify({portId:'read',action:{kind:'state',primitive:'read'},input:null})})).status
+      : env.CONFIG.frameToken;
+    return Response.json({status:'returned',output});
+  }};`,
+  );
+  const f = await fixture(registry("return null;"), { worker });
+  try {
+    f.start();
+    const token = (await f.receive()).output;
+    assert.match(token, /^[a-f0-9-]{36}$/);
+    f.start(token);
+    assert.equal((await f.receive()).output, 409);
+    assert.equal(f.stats().starts, 1);
+    assert.equal(f.stats().reused, 1);
+  } finally {
+    await f.close();
+    await rm(directory, { recursive: true, force: true });
+  }
+});
+
+test("idle engines evict without a graph poll and restart with fresh state", async () => {
+  const f = await fixture(registry("return request.input;"), { engineIdleMs: 30 });
+  try {
+    f.start("first");
+    assert.equal((await f.receive()).output, "first");
+    await new Promise((resolve) => setTimeout(resolve, 100));
+    assert.equal(f.stats().engines, 0);
+    f.start("second");
+    assert.equal((await f.receive()).output, "second");
+    assert.equal(f.stats().starts, 2);
+  } finally {
+    await f.close();
+  }
+});
+
+for (const loop of [
+  "while(true) await Promise.resolve();",
+  "const memory=[]; for(let n=0;n<128;n++) memory.push(new Uint8Array(256*1024)); while(true) await Promise.resolve(memory);",
+])
+  test("await/allocation pressure is fenced without delaying a neighboring engine: " + loop.slice(0, 28), async () => {
+    const f = await fixture(registry(`if(request.input==='spin'){${loop}} return request.input;`), {
+      deadlineMs: 1500,
+    });
+    try {
+      const stuck = f.start("spin", { root: "stuck", depth: 1, deadlineMs: 350, calls: 10 });
+      const before = performance.now();
+      const healthy = f.start("healthy");
+      const replies = [await f.receive(), await f.receive()];
+      assert.equal(replies.find((r) => r.invocation === healthy.invocation).output, "healthy");
+      assert.equal(replies.find((r) => r.invocation === stuck.invocation).error.code, "RESTRICTED_DEADLINE");
+      assert.equal(replies[0].invocation, healthy.invocation, "Healthy work must finish before the stuck deadline");
+      assert.ok(performance.now() - before < 1200);
+      f.start("after fault");
+      assert.equal((await f.receive()).output, "after fault");
+      assert.equal(f.stats().starts, 2, "The healthy engine survives its neighbor's termination");
+    } finally {
+      await f.close();
+    }
+  });
+
+test("detached microtask loops cannot escape the invocation wall deadline", async () => {
+  const f = await fixture(
+    registry("void(async()=>{while(true)await Promise.resolve();})(); return 'nominal return';"),
+    { deadlineMs: 200 },
+  );
+  try {
+    const before = performance.now();
+    f.start();
+    assert.equal((await f.receive()).error.code, "RESTRICTED_DEADLINE");
+    assert.ok(performance.now() - before < 1000);
+    assert.equal(f.stats().engines, 0);
+  } finally {
+    await f.close();
+  }
+});
+
+test("detached digest callbacks cannot use a completed frame or contaminate reuse", async () => {
+  const f = await fixture(
+    registry(`
+    void(async()=>{
+      for(let n=0;n<100;n++)await crypto.subtle.digest('SHA-256',new Uint8Array(32768));
+      await channel({portId:'late',action:{kind:'state',primitive:'read'},input:request.input});
+    })().catch(()=>{});
+    return request.input;
+  `),
+  );
+  try {
+    f.start("private A");
+    assert.equal((await f.receive()).output, "private A");
+    await new Promise((resolve) => setTimeout(resolve, 75));
+    f.start("private B");
+    const reply = await f.receive();
+    assert.equal(reply.status, "returned", "No late graph port may appear after completion");
+    assert.equal(reply.output, "private B");
+    assert.equal(f.stats().starts, 1);
+  } finally {
+    await f.close();
+  }
+});
+
+test("the shared engine budget evicts idle code families and rejects busy oversubscription", async () => {
+  const engineBudget = new RestrictedEngineBudget(1);
+  const a = await fixture(registry("return request.input;"), { engineBudget });
+  const b = await fixture(
+    registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"),
+    { engineBudget },
+  );
+  try {
+    a.start("first family");
+    assert.equal((await a.receive()).output, "first family");
+    const busy = b.start();
+    const port = await b.receive();
+    assert.equal(a.stats().engines, 0, "Idle foreign code is evicted before allocating another engine");
+    assert.equal(engineBudget.stats().engines, 1);
+    a.start("overload");
+    assert.equal((await a.receive()).error.code, "RESTRICTED_ENGINE_LIMIT");
+    assert.equal(engineBudget.stats().engines, 1);
+    b.send({ kind: "port-result", ...busy, sequence: port.sequence, status: "returned", output: "survived" });
+    assert.equal((await b.receive()).output, "survived");
+  } finally {
+    await Promise.all([a.close(), b.close()]);
+  }
+  assert.equal(engineBudget.stats().engines, 0);
+});

restricted-execution/worker.mjs

@@ -1,5 +1,7 @@
 // Platform entrypoint, evaluated before loading the authored registry. A fresh
-// engine process owns exactly one frame. BROKER grants only that frame's ports.
+// Worker owns exactly one frame. The parent engine is reusable; BROKER's host
+// token grants only this frame and expires before the next invocation.
+const monotonic = performance.now.bind(performance);
 const deny = () => {
   throw Error("RESTRICTED_AMBIENT_ACCESS");
 };
@@ -48,7 +50,9 @@ export default {
     if (used) return new Response("FRAME_EXPIRED", { status: 409 });
     used = true;
     const input = await request.json();
+    const loading = monotonic();
     const source = await import("./package.mjs");
+    const initializationMs = monotonic() - loading;
     const registry = source[env.CONFIG.registryExport];
     if (
       !registry ||
@@ -59,7 +63,7 @@ export default {
     let open = true;
     const channel = async (value) => {
       if (!open) throw Error("EXPIRED_CONTEXT");
-      const response = await env.BROKER.fetch("http://frame/port", {
+      const response = await env.BROKER.fetch("http://frame/port/" + env.CONFIG.frameToken, {
         method: "POST",
         body: JSON.stringify(value),
       });
@@ -71,6 +75,7 @@ export default {
         });
       return result.output;
     };
+    const running = monotonic();
     try {
       const output = await registry.invoke(
         { exportId: input.exportId, workspace: input.workspace, receiver: input.receiver, input: input.input },
@@ -78,7 +83,13 @@ export default {
       );
       const body = JSON.stringify({ status: "returned", output });
       if (new TextEncoder().encode(body).length > env.CONFIG.maxBytes) throw Error("RESTRICTED_OUTPUT_LIMIT");
-      return new Response(body, { headers: { "content-type": "application/json" } });
+      return new Response(body, {
+        headers: {
+          "content-type": "application/json",
+          "x-quixos-initialization-ms": String(initializationMs),
+          "x-quixos-execution-ms": String(monotonic() - running),
+        },
+      });
     } catch (error) {
       // Only authored/codec rejection codes cross the engine boundary. Engine
       // diagnostics stay private to the supervisor, never become protocol data.

restricted-execution/workerd.nix

@@ -1,5 +1,6 @@
 # Pinned upstream engine. This binary supplies an import/capability boundary;
-# production execution additionally requires a workspace VM and OS limits.
+# production execution additionally requires OS limits and platform approval
+# for shared placement, or a workspace VM for unreviewed application code.
 { pkgs }:
 let
   version = "1.20261005.1";
@@ -37,7 +38,7 @@ pkgs.stdenvNoCC.mkDerivation {
     "$out/bin/workerd" --version | grep -Fx 'workerd 2026-10-05'
   '';
   meta = {
-    description = "Pinned workerd engine for VM-scoped Quixos restricted execution";
+    description = "Pinned workerd engine for Quixos restricted execution";
     homepage = "https://github.com/cloudflare/workerd";
     license = pkgs.lib.licenses.asl20;
     platforms = builtins.attrNames platforms;