noopolis · apresmoi · Aug 30, 2026
diff --git a/src/pi/codexHeadlessResult.test.ts b/src/pi/codexHeadlessResult.test.ts
@@ -142,3 +142,32 @@ test("turn.failed without a reply rejects", () => {
 test("empty streams reject", () => {
   assert.throws(() => decodeCodexHeadlessTurn(""), /empty stream/u);
 });
+
+/**
+ * Captured verbatim from a live `codex exec --json` turn inside the Daimon
+ * runtime image (codex-cli 0.142.3) and matched by 0.151.0 on the host: Codex
+ * emits no `cache_write_input_tokens` at all. Requiring it made every real
+ * turn decode to `usage: undefined`, which is why the organization's usage
+ * ledger stayed empty across six paid codex wakes.
+ */
+test("a live Codex turn.completed omits cache_write_input_tokens and is still metered", () => {
+  const decoded = decodeCodexHeadlessTurn(stream(
+    { type: "thread.started", thread_id: "01a05507-5624-7522-bb2d-c944ee87c85b" },
+    { type: "turn.started" },
+    { type: "item.completed", item: { id: "item_0", type: "mcp_tool_call", server: "daimon", tool: "probe_ping", status: "completed" } },
+    { type: "item.completed", item: { id: "item_1", type: "agent_message", text: "PROBE-OK:hello-world" } },
+    { type: "turn.completed", usage: { input_tokens: 36_631, cached_input_tokens: 32_896, output_tokens: 85, reasoning_output_tokens: 24 } }
+  ));
+  assert.deepEqual(decoded.usage, {
+    input: 36_631, output: 85, cacheRead: 32_896, cacheWrite: 0, total: 36_716, calls: 1, notionalUsd: 0, complete: true
+  });
+});
+
+test("a malformed cache_write_input_tokens still degrades the whole usage block", () => {
+  const decoded = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } },
+    { type: "turn.completed", usage: usage({ cache_write_input_tokens: -1 }) }
+  ));
+  assert.equal(decoded.usage, undefined);
+  assert.equal(decoded.text, "ok");
+});
diff --git a/src/pi/codexHeadlessResult.ts b/src/pi/codexHeadlessResult.ts
@@ -36,14 +36,27 @@ export type CodexTurnUsage = Readonly<{
 
 export type CodexHeadlessTurn = Readonly<{ text: string; usage?: CodexTurnUsage }>;
 
+/**
+ * Every field Codex actually emits on `turn.completed.usage`.
+ *
+ * `cache_write_input_tokens` is NOT in this list: Codex 0.142.3 and 0.151.0
+ * both emit exactly `{input_tokens, cached_input_tokens, output_tokens,
+ * reasoning_output_tokens}`, verified against a live turn inside the runtime
+ * image and against `token_count` in a session rollout. Requiring the cache
+ * write field made `decodeCodexTurnUsage` return `undefined` for *every* real
+ * turn, so the codex half of the usage ledger recorded nothing at all while
+ * every test passed against a fixture that invented the field.
+ */
 const USAGE_TOKEN_FIELDS = [
   "input_tokens",
   "cached_input_tokens",
-  "cache_write_input_tokens",
   "output_tokens",
   "reasoning_output_tokens"
 ] as const;
 
+/** Absent means zero; present-but-malformed still degrades the whole block. */
+const CACHE_WRITE_FIELD = "cache_write_input_tokens" as const;
+
 const tokenCount = (value: unknown): number | undefined =>
   typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : undefined;
 
@@ -59,7 +72,9 @@ const decodeCodexTurnUsage = (frame: JsonRecord, calls: number): CodexTurnUsage
   const usage = frame.usage;
   const counts = USAGE_TOKEN_FIELDS.map((field) => tokenCount(usage[field]));
   if (counts.some((count) => count === undefined)) return undefined;
-  const [input, cacheRead, cacheWrite, output, reasoningOutput] = counts as [number, number, number, number, number];
+  const cacheWrite = usage[CACHE_WRITE_FIELD] === undefined ? 0 : tokenCount(usage[CACHE_WRITE_FIELD]);
+  if (cacheWrite === undefined) return undefined;
+  const [input, cacheRead, output, reasoningOutput] = counts as [number, number, number, number];
   const total = input + output;
   return {
     input,

diff --git a/src/runtime/fixtures/testMoltnetMachine.mjs b/src/runtime/fixtures/testMoltnetMachine.mjs
@@ -0,0 +1,50 @@
+#!/usr/bin/env node
+// A faithful stand-in for `moltnet machine`, reproducing the two behaviours the
+// real CLI has and an "echo everything back on EOF" fake did not:
+//
+//   1. An identifier containing ":" parses as a scoped agent id and is refused
+//      with `error: invalid request` on stderr and a non-zero exit.
+//   2. End-of-input cancels whatever is still in flight, so a caller that ends
+//      stdin with its request gets `{"error":{"code":"canceled"}}`.
+let buffer = "";
+let pending;
+
+const respond = (request) => {
+  process.stdout.write(`${JSON.stringify({
+    version: "moltnet.machine.v1",
+    correlation_id: request.correlation_id,
+    operation: "send_nudge",
+    send_nudge: { accepted: true, message_id: "msg-1", event_id: "evt-1", thread_created: false, dm_created: false }
+  })}\n`);
+};
+
+const cancel = (request) => {
+  process.stdout.write(`${JSON.stringify({
+    version: "moltnet.machine.v1",
+    correlation_id: request.correlation_id,
+    operation: "send_nudge",
+    error: { code: "canceled" }
+  })}\n`);
+};
+
+process.stdin.on("data", (chunk) => {
+  buffer += chunk;
+  let newline;
+  while ((newline = buffer.indexOf("\n")) >= 0) {
+    const line = buffer.slice(0, newline).trim();
+    buffer = buffer.slice(newline + 1);
+    if (line.length === 0) continue;
+    const request = JSON.parse(line);
+    const identifiers = [request.correlation_id, request.send_nudge?.delivery_id];
+    if (identifiers.some((value) => typeof value === "string" && value.includes(":"))) {
+      process.stderr.write("error: invalid request\n");
+      process.exit(1);
+    }
+    pending = { request, timer: setTimeout(() => { pending = undefined; respond(request); }, 25) };
+  }
+});
+
+process.stdin.on("end", () => {
+  if (pending !== undefined) { clearTimeout(pending.timer); cancel(pending.request); pending = undefined; }
+  process.exit(0);
+});
diff --git a/src/runtime/productionAgentTools.test.ts b/src/runtime/productionAgentTools.test.ts
@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
 import { chmod, mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
+import url from "node:url";
 import test from "node:test";
 
 import { createProductionAgentTools } from "./productionAgentTools.js";
@@ -24,7 +25,7 @@ test("production cognition mounts only declared MCP tools and records a bounded
 test("production Moltnet tool enforces compiled scope and records accepted message receipt", async () => {
   const root = await mkdtemp(path.join(os.tmpdir(), "daimon-production-moltnet-"));
   try {
-    const cli = path.join(root, "moltnet"); await writeFile(cli, "#!/usr/bin/env node\nlet s='';process.stdin.on('data',c=>s+=c);process.stdin.on('end',()=>{const q=JSON.parse(s);process.stdout.write(JSON.stringify({version:'moltnet.machine.v1',correlation_id:q.correlation_id,operation:'send_nudge',send_nudge:{accepted:true,message_id:'msg-1'}})+'\\n')})\n"); await chmod(cli, 0o755);
+    const cli = path.join(root, "moltnet"); await writeFile(cli, `#!/usr/bin/env node\nimport(${JSON.stringify(url.pathToFileURL(path.resolve("src/runtime/fixtures/testMoltnetMachine.mjs")).href)});\n`); await chmod(cli, 0o755);
     const agent: OrganizationRuntimeAgentConfig = { id: "alpha", name: "Alpha", instructions: "work", workspacePath: root, runtimeHomePath: root, engine: { kind: "codex" }, moltnet: { cliPath: cli, configPath: path.join(root, "config.json"), networks: [{ id: "news", rooms: ["desk"], dms: false }] } };
     const tool = (await createProductionAgentTools(agent, { current: "schedule:occurrence" }))[0]!;
     await assert.rejects(tool.execute("call", { network: "news", target: "dm:beta", text: "no" } as never, undefined, undefined, {} as never), /not declared/u);
@@ -34,8 +35,22 @@ test("production Moltnet tool enforces compiled scope and records accepted messa
     await tool.execute("call", { network: "news", target: "room:desk", text: "hello" } as never, undefined, undefined, {} as never);
     const replacement = (await createProductionAgentTools(agent, { current: "schedule:occurrence" }))[0]!;
     await replacement.execute("call", { network: "news", target: "room:desk", text: "hello" } as never, undefined, undefined, {} as never);
-    const stored = await receipts(root); assert.match(stored.join(""), /"kind":"moltnet".*"delivery_id":"daimon:.*"message_id":"msg-1"/u); assert.equal(stored.length, 2);
+    const stored = await receipts(root); assert.match(stored.join(""), /"kind":"moltnet".*"delivery_id":"daimon-.*"message_id":"msg-1"/u); assert.equal(stored.length, 2);
     await writeFile(path.join(root, "tool-state", "unrelated-torn.json"), "{");
     await replacement.execute("call", { network: "news", target: "room:desk", text: "hello" } as never, undefined, undefined, {} as never);
   } finally { await rm(root, { recursive: true, force: true }); }
 });
+
+test("Moltnet identifiers are local ids, never a colon-scoped agent id", async () => {
+  const root = await mkdtemp(path.join(os.tmpdir(), "daimon-production-moltnet-id-"));
+  try {
+    const cli = path.join(root, "moltnet"); await writeFile(cli, `#!/usr/bin/env node\nimport(${JSON.stringify(url.pathToFileURL(path.resolve("src/runtime/fixtures/testMoltnetMachine.mjs")).href)});\n`); await chmod(cli, 0o755);
+    const agent: OrganizationRuntimeAgentConfig = { id: "alpha", name: "Alpha", instructions: "work", workspacePath: root, runtimeHomePath: root, engine: { kind: "codex" }, moltnet: { cliPath: cli, configPath: path.join(root, "config.json"), networks: [{ id: "news", rooms: ["desk"], dms: false }] } };
+    const tool = (await createProductionAgentTools(agent, { current: "schedule:occurrence" }))[0]!;
+    await tool.execute("call", { network: "news", target: "room:desk", text: "hello" } as never, undefined, undefined, {} as never);
+    const stored = (await receipts(root)).join("");
+    const deliveryId = (JSON.parse(stored) as { delivery_id: string }).delivery_id;
+    assert.ok(!deliveryId.includes(":"), `delivery id must be a local id, got ${deliveryId}`);
+    assert.match(deliveryId, /^daimon-[0-9a-f]{64}$/u);
+  } finally { await rm(root, { recursive: true, force: true }); }
+});
diff --git a/src/runtime/productionAgentTools.ts b/src/runtime/productionAgentTools.ts
@@ -16,6 +16,17 @@ import type { PiWakeEnvironmentContextRef } from "../pi/piAgentWakeSupport.js";
 
 const MAX_RESULT = 65_536; const TIMEOUT = 10_000;
 
+/**
+ * Action and delivery identifiers are prefixed with a hyphen, never a colon.
+ *
+ * Moltnet's machine protocol rejects any identifier that parses as a scoped
+ * agent id, and `ParseScopedAgentID` treats *every* `left:right` string as one
+ * (`moltnet/pkg/protocol/identity.go`). A `daimon:<digest>` delivery id
+ * therefore failed `delivery_id must be a local id`, and the CLI answered
+ * `error: invalid request` for every send an agent ever attempted.
+ */
+const DAIMON_ACTION_ID_PREFIX = "daimon-";
+
 export async function createProductionAgentTools(agent: OrganizationRuntimeAgentConfig, wakeContext: PiWakeEnvironmentContextRef = {}): Promise<ToolDefinition[]> {
   await mkdir(path.join(agent.runtimeHomePath, "tool-state"), { recursive: true, mode: 0o700 });
   const tools = [...await Promise.all((agent.mcp ?? []).map((server) => mcpTools(agent, server, wakeContext)))].flat();
@@ -41,7 +52,7 @@ async function mcpTools(agent: OrganizationRuntimeAgentConfig, server: Organizat
       parameters: declared.inputSchema as ToolDefinition["parameters"],
       async execute(_id, params) {
         if (!wakeContext.current) throw new Error("MCP call requires an active wake");
-        const actionId = `daimon:${createHash("sha256").update(JSON.stringify([wakeContext.current, agent.id, server.name, name, params])).digest("hex")}`;
+        const actionId = `${DAIMON_ACTION_ID_PREFIX}${createHash("sha256").update(JSON.stringify([wakeContext.current, agent.id, server.name, name, params])).digest("hex")}`;
         const prior = await priorReceipt(agent, actionId); if (prior !== undefined) return { content: [{ type: "text", text: JSON.stringify(prior) }], details: prior };
         const active = await connect(agent, server);
         try {
@@ -64,7 +75,7 @@ function moltnetTool(agent: OrganizationRuntimeAgentConfig, wakeContext: PiWakeE
       if (network === undefined || Buffer.byteLength(input.text) > 2_048) throw new Error("Moltnet action exceeds declared scope");
       const [kind, target] = input.target.split(":", 2); if ((kind === "room" && !network.rooms.includes(target ?? "")) || (kind === "dm" && !network.dms) || !target || !["room", "dm"].includes(kind ?? "")) throw new Error("Moltnet target is not declared");
       if (!wakeContext.current) throw new Error("Moltnet send requires an active wake");
-      const deliveryId = `daimon:${createHash("sha256").update(JSON.stringify([wakeContext.current, agent.id, input.network, input.target, input.text])).digest("hex")}`;
+      const deliveryId = `${DAIMON_ACTION_ID_PREFIX}${createHash("sha256").update(JSON.stringify([wakeContext.current, agent.id, input.network, input.target, input.text])).digest("hex")}`;
       const prior = await priorReceipt(agent, deliveryId); if (prior !== undefined) return { content: [{ type: "text", text: JSON.stringify(prior) }], details: prior };
       const response = await machine(agent.moltnet!.cliPath, agent.moltnet!.configPath, input.network, { version: "moltnet.machine.v1", correlation_id: deliveryId, operation: "send_nudge", send_nudge: { delivery_id: deliveryId, target: { kind, id: target }, body: input.text } });
       const result = response.send_nudge as { accepted?: boolean; message_id?: string } | undefined; if (result?.accepted !== true || typeof result.message_id !== "string") throw new Error("Moltnet send was not accepted");
@@ -87,7 +98,38 @@ function receiptPath(agent: OrganizationRuntimeAgentConfig, deliveryId: string):
 async function receipt(agent: OrganizationRuntimeAgentConfig, value: Record<string, unknown>): Promise<void> { const deliveryId = String(value.delivery_id); const file = receiptPath(agent, deliveryId); const temporary = `${file}.${process.pid}.${Date.now()}.tmp`; const bytes = `${JSON.stringify({ ...value, at: new Date().toISOString() })}\n`; if (Buffer.byteLength(bytes) > MAX_RESULT) throw new Error("tool receipt exceeds bound"); const handle = await open(temporary, constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, 0o600); try { await handle.writeFile(bytes); await handle.sync(); } finally { await handle.close(); } try { await rename(temporary, file); await pruneReceipts(path.dirname(file)); const directory = await open(path.dirname(file), constants.O_RDONLY); try { await directory.sync(); } finally { await directory.close(); } } catch (error) { await unlink(temporary).catch(() => undefined); throw error; } }
 async function pruneReceipts(directory: string): Promise<void> { const candidates = await Promise.all((await readdir(directory)).filter((name) => /^[a-f0-9]{64}\.json$/u.test(name)).map(async (name) => ({ name, info: await lstat(path.join(directory, name)) }))); if (candidates.some(({ info }) => !info.isFile() || info.nlink !== 1)) throw new Error("tool receipt directory contains an unsafe entry"); for (const candidate of candidates.sort((left, right) => right.info.mtimeMs - left.info.mtimeMs || right.name.localeCompare(left.name)).slice(2048)) await unlink(path.join(directory, candidate.name)); }
 async function priorReceipt(agent: OrganizationRuntimeAgentConfig, deliveryId: string): Promise<Record<string, unknown> | undefined> { const file = receiptPath(agent, deliveryId); let handle; try { handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; throw error; } try { const entry = await handle.stat(); if (!entry.isFile() || entry.size > MAX_RESULT) throw new Error("tool receipt is unsafe or exceeds bound"); const value = JSON.parse(await handle.readFile("utf8")) as Record<string, unknown>; if (value.delivery_id !== deliveryId || value.agent_id !== agent.id || value.engine !== agent.engine.kind) throw new Error("tool receipt identity mismatch"); return value; } finally { await handle.close(); } }
-async function machine(cli: string, config: string, network: string, request: unknown): Promise<Record<string, unknown>> { return await new Promise((resolve, reject) => { const child = spawn(cli, ["machine", "--config", config, "--network", network], { stdio: ["pipe", "pipe", "pipe"] }); let output = "", error = ""; const timer = setTimeout(() => { child.kill("SIGKILL"); reject(new Error("Moltnet machine timed out")); }, TIMEOUT); child.stdout.on("data", (chunk) => { output += chunk; if (Buffer.byteLength(output) > MAX_RESULT) child.kill("SIGKILL"); }); child.stderr.on("data", (chunk) => { error += chunk; }); child.once("error", reject); child.once("exit", (code) => { clearTimeout(timer); if (code !== 0) reject(new Error(`Moltnet machine failed: ${error.slice(0, 1024)}`)); else { try { resolve(JSON.parse(output.trim().split("\n")[0]!) as Record<string, unknown>); } catch (cause) { reject(cause); } } }); child.stdin.end(`${JSON.stringify(request)}\n`); }); }
+/**
+ * One request/response exchange with `moltnet machine`.
+ *
+ * Stdin is held open until the response line arrives. `moltnet machine` treats
+ * end-of-input as cancellation of everything still in flight, so ending stdin
+ * with the request — as this did — raced the send and lost: the CLI answered
+ * `{"error":{"code":"canceled"}}` and no message was ever delivered. The
+ * operation is complete once its line is on stdout, so closing stdin there can
+ * no longer cancel it.
+ */
+async function machine(cli: string, config: string, network: string, request: unknown): Promise<Record<string, unknown>> {
+  return await new Promise((resolve, reject) => {
+    const child = spawn(cli, ["machine", "--config", config, "--network", network], { stdio: ["pipe", "pipe", "pipe"] });
+    let output = "", error = "", settled = false;
+    const settle = (action: () => void): void => { if (settled) return; settled = true; clearTimeout(timer); action(); };
+    const timer = setTimeout(() => { child.kill("SIGKILL"); settle(() => reject(new Error("Moltnet machine timed out"))); }, TIMEOUT);
+    child.stdin.on("error", () => undefined);
+    child.stdout.on("data", (chunk) => {
+      output += chunk;
+      if (Buffer.byteLength(output) > MAX_RESULT) { child.kill("SIGKILL"); settle(() => reject(new Error("Moltnet machine response exceeds bound"))); return; }
+      const newline = output.indexOf("\n"); if (newline < 0) return;
+      const line = output.slice(0, newline).trim(); if (line.length === 0) return;
+      child.stdin.end();
+      settle(() => { try { resolve(JSON.parse(line) as Record<string, unknown>); } catch (cause) { reject(cause); } });
+      child.kill("SIGTERM");
+    });
+    child.stderr.on("data", (chunk) => { error += chunk; });
+    child.once("error", (cause) => settle(() => reject(cause)));
+    child.once("exit", () => settle(() => reject(new Error(`Moltnet machine failed: ${(error.slice(0, 1024) || "no response").trim()}`))));
+    child.stdin.write(`${JSON.stringify(request)}\n`);
+  });
+}
 function digest(value: string): string { return `sha256:${createHash("sha256").update(value).digest("hex")}`; }
 function safe(value: string): string { return value.replace(/[^A-Za-z0-9_]/gu, "_").slice(0, 48); }
 function stringEnvironment(value: NodeJS.ProcessEnv): Record<string, string> { return Object.fromEntries(Object.entries(value).filter((entry): entry is [string, string] => entry[1] !== undefined)); }