Publish quixos-nix-helpers from 98cd7bc237027368fd2c77f294f59281ef3df160

Show per-export placement, static eligibility blockers, bounded metadata and live pool/idle/timing diagnostics. Attach inspector updates through one private shard stream and the existing authenticated SSE hub without waking graph views or VMs. Fix packed agent preparation to use OpenSSH and the worker’s own private systemd credential.

.quixos-subtree-source.json

@@ -1,7 +1,7 @@
 {
   "version": 1,
   "sourceRepo": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos",
-  "sourceCommit": "932fe7714761fb8b5db4bd6401357cdff762a428",
+  "sourceCommit": "98cd7bc237027368fd2c77f294f59281ef3df160",
   "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/default.nix

@@ -37,7 +37,7 @@ let
     mkdir -p "$out"
     cat > entry.mjs <<'JS'
     export {definePortableRegistry,QxReject} from '${core}/dist/portable.js';
-    export {encodeArgument,decodeArgument,canonicalJson,parseJson} from '${core}/dist/value.js';
+    export {encodeArgument,decodeArgument,canonicalJson,parseJson,hashJson} from '${core}/dist/value.js';
     export {workspaceId,operationId,objectId,entryId,hashIdentity} from '${core}/dist/identity.js';
     export {reference} from '${core}/dist/references.js';
     export {CodecError,ExecutionError} from '${core}/dist/limits.js';

restricted-execution/service.mjs

@@ -148,6 +148,11 @@ export async function serveRestrictedRegistry(config, io = process) {
     engines = new Set(),
     idleEngines = [];
   const metrics = { starts: 0, reused: 0, completed: 0, peakRSSBytes: 0 };
+  const changed = () => {
+    try {
+      config.onChange?.();
+    } catch {}
+  };
   let buffered = Buffer.alloc(0),
     stopped = false,
     failure;
@@ -234,16 +239,19 @@ export async function serveRestrictedRegistry(config, io = process) {
       }
       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) {
+          // Keep the lease busy through every await. Another artifact can evict
+          // an idle lease immediately; never publish an engine still retiring
+          // or one whose process died while its final usage read was pending.
+          if (rss <= 128 * 1024 * 1024 && !stopped && !engine.destroying && engine.current === call) {
+            // Revoke the frame token atomically with publishing idle capacity.
+            engine.current = null;
             idleEngines.push(engine);
+            engine.idleUntil = Date.now() + limits.engineIdleMs;
             engine.idleTimer = setTimeout(() => void destroyEngine(engine).catch(fail), limits.engineIdleMs);
           } else await destroyEngine(engine);
         } else await destroyEngine(engine);
@@ -300,6 +308,7 @@ export async function serveRestrictedRegistry(config, io = process) {
       } finally {
         engines.delete(engine);
         await engine.release?.();
+        changed();
       }
     })();
     return engine.destroying;
@@ -418,6 +427,7 @@ export async function serveRestrictedRegistry(config, io = process) {
       },
     );
     metrics.starts++;
+    changed();
     engine.exited = new Promise((resolve) => {
       engine.child.once("exit", resolve);
       engine.child.once("error", resolve);
@@ -462,6 +472,9 @@ export async function serveRestrictedRegistry(config, io = process) {
         executionMs: call.executionMs ?? null,
         totalMs: performance.now() - call.receivedAt,
       };
+      metrics.lastFailure =
+        reply.status === "failed" && /^[A-Z0-9_]{1,128}$/.test(reply.error?.code) ? reply.error.code : null;
+      changed();
       send(call.cancellationReply ?? reply);
       pump();
     })());
@@ -470,6 +483,7 @@ export async function serveRestrictedRegistry(config, io = process) {
     const { message } = call;
     call.startedAt = performance.now();
     active.set(message.invocation, call);
+    changed();
     const terminate = (code) => {
       if (!call.open) return;
       void finish(call, fixedError(code), {
@@ -687,6 +701,9 @@ export async function serveRestrictedRegistry(config, io = process) {
       queued: queued.length,
       engines: engines.size,
       idle: idleEngines.length,
+      idleDeadline: idleEngines.length
+        ? new Date(Math.min(...idleEngines.map((engine) => engine.idleUntil))).toISOString()
+        : null,
     }),
   };
 }

restricted-execution/service.test.mjs

@@ -307,6 +307,47 @@ test("the checked portable SDK keeps codecs and nested graph ports inside the fr
     await f.close();
   }
 });
+test("checked handlers hash canonical values without ambient capabilities", async () => {
+  const sdk = process.env.QUIXOS_TEST_SDK;
+  const C = await import(sdk);
+  const workspace = C.workspaceId("workspace:restricted");
+  const text = { kind: "scalar", name: "string" };
+  const value = {
+    kind: "record",
+    fields: {
+      text,
+      options: {
+        kind: "record",
+        fields: {
+          level: { kind: "scalar", name: "int32" },
+          enabled: { kind: "scalar", name: "bool" },
+        },
+      },
+    },
+  };
+  const declarations = [
+    { id: "test", name: "test", effect: "snapshot", receiver: null, input: value, output: text, methods: [] },
+  ];
+  const source = `import {definePortableRegistry,hashJson} from 'qx:core'; export const registry=definePortableRegistry(${JSON.stringify(declarations)},{test:ctx=>hashJson('test-content',ctx.input)});`;
+  const f = await fixture(source, { sdk: [{ name: "qx:core", path: sdk }] });
+  try {
+    const input = { text: "雨后的书店", options: { level: 3, enabled: true } };
+    let expected;
+    for (const candidate of [input, { options: input.options, text: input.text }, { ...input, text: "別の文" }]) {
+      f.start(C.encodeArgument(value, candidate, { workspace }));
+      const result = await f.receive();
+      assert.equal(result.status, "returned");
+      const actual = C.decodeArgument(text, result.output, { workspace });
+      assert.equal(actual, await C.hashJson("test-content", candidate));
+      if (candidate.text === input.text) {
+        expected ??= actual;
+        assert.equal(actual, expected, "Object key order must not invalidate content caches");
+      } else assert.notEqual(actual, expected, "Content changes must invalidate the key");
+    }
+  } finally {
+    await f.close();
+  }
+});
 test("ambient clock, entropy, network and native imports are denied", async () => {
   for (const code of [
     "Date.now()",
@@ -529,12 +570,25 @@ test("an expired frame token cannot borrow a later invocation's broker", async (
 });
 
 test("idle engines evict without a graph poll and restart with fresh state", async () => {
-  const f = await fixture(registry("return request.input;"), { engineIdleMs: 30 });
+  const updates = [];
+  const f = await fixture(registry("return request.input;"), {
+    engineIdleMs: 30,
+    onChange: () => updates.push(f.stats()),
+  });
   try {
     f.start("first");
     assert.equal((await f.receive()).output, "first");
     await new Promise((resolve) => setTimeout(resolve, 100));
     assert.equal(f.stats().engines, 0);
+    assert.ok(
+      updates.some((s) => s.active > 0),
+      "Inspectors observe active invocation admission",
+    );
+    assert.ok(
+      updates.some((s) => s.idle > 0 && s.idleDeadline),
+      "Idle eviction deadline is visible",
+    );
+    assert.equal(updates.at(-1).engines, 0, "Eviction pushes an update without a graph poll");
     f.start("second");
     assert.equal((await f.receive()).output, "second");
     assert.equal(f.stats().starts, 2);
@@ -632,3 +686,39 @@ test("the shared engine budget evicts idle code families and rejects busy oversu
   }
   assert.equal(engineBudget.stats().engines, 0);
 });
+
+test("engine retirement accounting keeps its admission lease busy", async () => {
+  const observations = [];
+  let idle;
+  const engineBudget = new RestrictedEngineBudget(1, async () => ({
+    usage: async () => {
+      observations.push(idle());
+      await new Promise((done) => setImmediate(done));
+      return { cpuMicroseconds: 0, memoryBytes: 1024 };
+    },
+    release: async () => {},
+  }));
+  const acquire = engineBudget.acquire.bind(engineBudget);
+  engineBudget.acquire = (isIdle, evict) => {
+    idle = isIdle;
+    return acquire(isIdle, evict);
+  };
+  const f = await fixture(registry("return request.input;"), { engineBudget });
+  try {
+    f.start("first");
+    assert.equal((await f.receive()).output, "first");
+    assert.ok(observations.length > 0);
+    assert.ok(
+      observations.every((value) => !value),
+      "An engine must not become evictable during asynchronous retirement accounting",
+    );
+    assert.equal(f.stats().idle, 1);
+    assert.equal(idle(), true);
+    f.start("second");
+    assert.equal((await f.receive()).output, "second");
+    assert.equal(f.stats().starts, 1);
+  } finally {
+    await f.close();
+  }
+  assert.equal(engineBudget.stats().engines, 0);
+});