feat(runtime): meter codex turns and add a fail-closed wake fuse

src/pi/cliEngineSpawn.test.ts

@@ -5,7 +5,7 @@ import path from "node:path";
 import test from "node:test";
 
 import { readChild } from "./cliSession.js";
-import { GROK_STRICT_SANDBOX_PROFILE, renderGrokSandboxArgs, spawnEngine } from "./cliEngineSpawn.js";
+import { GROK_STRICT_SANDBOX_PROFILE, renderCodexArgs, renderGrokSandboxArgs, spawnEngine } from "./cliEngineSpawn.js";
 
 test("autonomous Codex and Grok launches omit wall-clock and turn caps", async () => {
   const root = await mkdtemp(path.join(os.tmpdir(), "daimon-unbounded-cli-"));
@@ -33,6 +33,18 @@ test("autonomous Codex and Grok launches omit wall-clock and turn caps", async (
   } finally { await rm(root, { recursive: true, force: true }); }
 });
 
+test("Codex output, sandbox, config, and cwd boundaries reject caller overrides", () => {
+  for (const injected of [
+    "--json", "--sandbox", "--sandbox=read-only", "--dangerously-bypass-approvals-and-sandbox",
+    "--output-last-message", "-c", "--config", "--skip-git-repo-check", "--color", "-C", "--cd"
+  ]) {
+    assert.throws(() => renderCodexArgs({ commandArgs: [injected] }, "/workspace", undefined), /Daimon-owned/u);
+  }
+  const args = renderCodexArgs({ commandArgs: ["--effort", "high"] }, "/workspace", undefined);
+  assert.deepEqual(args.slice(0, 2), ["--effort", "high"]);
+  assert.equal(args.includes("--json"), true);
+});
+
 test("Grok's kernel sandbox authority cannot be weakened by injected CLI arguments", async () => {
   const root = await mkdtemp(path.join(os.tmpdir(), "daimon-grok-boundary-"));
   try {

src/pi/cliEngineSpawn.ts

@@ -11,12 +11,18 @@ export const renderGrokSandboxArgs = (
   profile: string
 ): string[] => [...assertSafeGrokCommandArgs(commandArgs), "--sandbox", profile];
 
+/**
+ * Codex's `--json` stream is unconditional: it is the only output shape that
+ * carries `turn.completed.usage`, and an unmetered Codex turn is one whose
+ * subscription cost is invisible. The guarded arguments make this Daimon's
+ * security and metering boundary rather than a caller-controlled format.
+ */
 export const renderCodexArgs = (
   options: Pick<CliEngineOptions, "commandArgs">,
   cwd: string,
   endpoint: string | undefined,
   sandbox: string = process.env.DAIMON_CODEX_SANDBOX ?? "danger-full-access"
-): string[] => [...(options.commandArgs ?? []), "exec", "--sandbox", sandbox, "--skip-git-repo-check", "--color", "never", "-C", cwd,
+): string[] => [...assertSafeCodexCommandArgs(options.commandArgs), "exec", "--sandbox", sandbox, "--skip-git-repo-check", "--color", "never", "--json", "-C", cwd,
   "-c", `mcp_servers.daimon.url=${endpoint}`, "-"];
 
 /**
@@ -105,6 +111,15 @@ const assertSafeAgyCommandArgs = (args: readonly string[] | undefined): readonly
   return values;
 };
 
+/** Caller arguments cannot reopen Codex's sandbox, output, cwd, or config boundary. */
+const assertSafeCodexCommandArgs = (args: readonly string[] | undefined): readonly string[] => {
+  const values = args ?? [];
+  if (values.some((value) => /^(?:--json|--sandbox|--dangerously-bypass-approvals-and-sandbox|--output-last-message|--config|--skip-git-repo-check|--color|--cd|-c|-C)(?:=|$)/u.test(value))) {
+    throw new Error("Codex security-boundary arguments are Daimon-owned");
+  }
+  return values;
+};
+
 const assertSafeGrokCommandArgs = (args: readonly string[] | undefined): readonly string[] => {
   const values = args ?? [];
   if (values.some((value) => /^(?:--sandbox|--always-approve|--permission-mode|--leader-socket)(?:=|$)/u.test(value))) {

src/pi/cliSession.test.ts

@@ -66,7 +66,8 @@ test("CLI adapter mounts the harness tool objects and preserves the causal wake
     "const observe = await client.callTool({ name: 'world_observe', arguments: { sense: 'world://proof/sense' } });",
     "const act = await client.callTool({ name: 'world_act', arguments: { affordance: 'world://proof/act', target: 'world://proof/target', input: { ok: true } } });",
     "const refused = await client.callTool({ name: 'world_status', arguments: {} });",
-    `process.stdout.write(JSON.stringify({ listed: listed.tools.map((tool) => tool.name), observe, act, refused, bearer: process.env.${tokenEnv} ?? null, argv: process.argv.join('\\n'), prompt }));`,
+    "const codex = (value) => [{ type: 'item.completed', item: { type: 'agent_message', text: value } }, { type: 'turn.completed' }].map(JSON.stringify).join('\\n');",
+    `process.stdout.write(codex(JSON.stringify({ listed: listed.tools.map((tool) => tool.name), observe, act, refused, bearer: process.env.${tokenEnv} ?? null, argv: process.argv.join('\\n'), prompt })));`,
     "await client.close();"
   ].join("\n"));
   const captured: Parameters<PiSessionFactory>[0][] = [];
@@ -257,7 +258,7 @@ test("protected host control variables never reach Codex, Grok, or AGY children"
   process.env[unrelatedEnv] = "must-never-reach-engine";
   process.env[modelEnv] = "must-never-reach-engine";
   const probe = path.join(root, "probe.mjs");
-  await writeFile(probe, `#!/usr/bin/env node\nconst text = [process.env.${controlEnv} ?? "absent", process.env.${unrelatedEnv} ?? "absent", process.env.${modelEnv} ?? "absent", process.env.CODEX_HOME ?? process.env.GROK_HOME ?? process.env.ANTIGRAVITY_CLI_HOME ?? "missing", process.env.DAIMON_WAKE_ID ?? "absent"].join("|"); const stream = (value) => [{ type: "assistant", parent_tool_use_id: null, session_id: "fake", message: { role: "assistant", stop_reason: "end_turn", content: [{ type: "text", text: value }] } }, { type: "result", subtype: "success", is_error: false, result: value, stop_reason: "end_turn", session_id: "fake" }].map(JSON.stringify).join("\\n"); const agy = (value) => JSON.stringify({ event: "result", result: { conversation_id: "fake", status: "SUCCESS", response: value, num_turns: 1, usage: { input_tokens: 11, output_tokens: 2, thinking_tokens: 1, cache_read_tokens: 0, total_tokens: 13 } } }); process.stdout.write(process.argv.includes("--single") ? stream(text) : process.argv.includes("--output-format") ? agy(text) : text);`);
+  await writeFile(probe, `#!/usr/bin/env node\nconst text = [process.env.${controlEnv} ?? "absent", process.env.${unrelatedEnv} ?? "absent", process.env.${modelEnv} ?? "absent", process.env.CODEX_HOME ?? process.env.GROK_HOME ?? process.env.ANTIGRAVITY_CLI_HOME ?? "missing", process.env.DAIMON_WAKE_ID ?? "absent"].join("|"); const stream = (value) => [{ type: "assistant", parent_tool_use_id: null, session_id: "fake", message: { role: "assistant", stop_reason: "end_turn", content: [{ type: "text", text: value }] } }, { type: "result", subtype: "success", is_error: false, result: value, stop_reason: "end_turn", session_id: "fake" }].map(JSON.stringify).join("\\n"); const codex = (value) => [{ type: "item.completed", item: { type: "agent_message", text: value } }, { type: "turn.completed" }].map(JSON.stringify).join("\\n"); const agy = (value) => JSON.stringify({ event: "result", result: { conversation_id: "fake", status: "SUCCESS", response: value, num_turns: 1, usage: { input_tokens: 11, output_tokens: 2, thinking_tokens: 1, cache_read_tokens: 0, total_tokens: 13 } } }); process.stdout.write(process.argv.includes("--single") ? stream(text) : process.argv.includes("--output-format") ? agy(text) : process.argv.includes("--json") ? codex(text) : text);`);
   await chmod(probe, 0o700);
   try {
     for (const engine of ["codex", "grok", "agy"] as const) {

src/pi/cliSession.ts

@@ -30,6 +30,7 @@ import {
   type CliMcpRegistration
 } from "./cliMcpRegistration.js";
 import { decodeAgyHeadlessTurn, type AgyTurnUsage } from "./agyHeadlessResult.js";
+import { decodeCodexHeadlessTurn, type CodexTurnUsage } from "./codexHeadlessResult.js";
 import { decodeGrokHeadlessResult } from "./grokHeadlessResult.js";
 import { terminateChild, trackCliChild } from "./cliProcess.js";
 import type { PiSessionLike } from "./piAgentHandle.js";
@@ -82,7 +83,7 @@ export type CliEngineOptions = {
    * reports token usage and that do not run behind the Grok engine broker
    * (which meters its own turns). It never fails a turn that published.
    */
-  readonly onTurnUsage?: (usage: AgyTurnUsage) => Promise<void>;
+  readonly onTurnUsage?: (usage: AgyTurnUsage | CodexTurnUsage) => Promise<void>;
 } & ({
   readonly engine: "codex" | "grok";
 } | {
@@ -228,7 +229,7 @@ class CliSession implements PiSessionLike {
     const secretValues = [...environmentSecretValues, ...stagedCredentialSecrets];
     let mount: { endpoint: string; close: () => Promise<void> } | undefined;
     let registration: CliMcpRegistration | undefined;
-    let turnUsage: AgyTurnUsage | undefined;
+    let turnUsage: AgyTurnUsage | CodexTurnUsage | undefined;
     let child: ChildProcess | undefined;
     let output: string | undefined;
     let cleanupFailure: unknown;
@@ -295,6 +296,11 @@ class CliSession implements PiSessionLike {
       await this.options.verifyExecutable?.();
       const childOutput = await outputPromise;
       if (this.options.engine === "grok") output = decodeGrokHeadlessResult(childOutput);
+      else if (this.options.engine === "codex") {
+        const decoded = decodeCodexHeadlessTurn(childOutput);
+        output = decoded.text;
+        turnUsage = decoded.usage;
+      }
       else if (this.options.engine === "agy") {
         const decoded = decodeAgyHeadlessTurn(childOutput);
         output = decoded.text;

src/pi/codexHeadlessResult.test.ts

@@ -0,0 +1,144 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { decodeCodexHeadlessResult, decodeCodexHeadlessTurn } from "./codexHeadlessResult.js";
+
+const frame = (value: unknown): string => JSON.stringify(value);
+const usage = (overrides: Record<string, unknown> = {}) => ({
+  input_tokens: 18_110, cached_input_tokens: 11_008, cache_write_input_tokens: 0,
+  output_tokens: 5, reasoning_output_tokens: 0, ...overrides
+});
+const stream = (...values: unknown[]): string => values.map(frame).join("\n");
+
+test("decodes the captured Codex 0.151.0 stream and reconciled subset accounting", () => {
+  const output = stream(
+    { type: "thread.started", thread_id: "01a053f6-4a8d-7851-93e2-7d7fb1853849" },
+    { type: "turn.started" },
+    { type: "item.completed", item: { id: "item_0", type: "agent_message", text: "ok" } },
+    { type: "turn.completed", usage: usage() }
+  );
+  assert.deepEqual(decodeCodexHeadlessTurn(output), {
+    text: "ok",
+    usage: { input: 18_110, output: 5, cacheRead: 11_008, cacheWrite: 0, total: 18_115, calls: 0, notionalUsd: 0, complete: true }
+  });
+  assert.equal(decodeCodexHeadlessResult(output), "ok");
+});
+
+test("cacheRead is a subset of input and is never added to total", () => {
+  const decoded = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } },
+    { type: "turn.completed", usage: usage() }
+  ));
+  assert.equal(decoded.usage?.total, 18_110 + 5);
+  assert.notEqual(decoded.usage?.total, 18_110 + 11_008 + 5);
+
+  const invalidCacheSubset = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } },
+    { type: "turn.completed", usage: usage({ cached_input_tokens: 18_111 }) }
+  ));
+  assert.equal(invalidCacheSubset.usage?.complete, false);
+
+  const invalidReasoningSubset = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } },
+    { type: "turn.completed", usage: usage({ reasoning_output_tokens: 6 }) }
+  ));
+  assert.equal(invalidReasoningSubset.usage?.complete, false);
+});
+
+test("tool frames are counted while the last agent message is published", () => {
+  const decoded = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "draft" } },
+    { type: "item.started", item: { type: "command_execution" } },
+    { type: "item.completed", item: { type: "command_execution" } },
+    { type: "item.completed", item: { type: "mcp_tool_call" } },
+    { type: "item.completed", item: { type: "agent_message", text: "final" } },
+    { type: "turn.completed", usage: usage() }
+  ));
+  assert.equal(decoded.text, "final");
+  assert.equal(decoded.usage?.calls, 2);
+});
+
+test("unknown future envelope and item types are skipped", () => {
+  assert.equal(decodeCodexHeadlessResult(stream(
+    { type: "future.envelope", payload: true },
+    { type: "item.completed", item: { type: "agent_message", text: "right" } },
+    { type: "item.completed", item: { type: "future_item", text: "wrong" } },
+    { type: "turn.completed", usage: usage() }
+  )), "right");
+});
+
+test("absent or malformed usage remains advisory", () => {
+  const base = [{ type: "item.completed", item: { type: "agent_message", text: "ok" } }];
+  assert.equal(decodeCodexHeadlessTurn(stream(...base, { type: "turn.completed" })).usage, undefined);
+  for (const field of ["input_tokens", "cached_input_tokens", "cache_write_input_tokens", "output_tokens", "reasoning_output_tokens"]) {
+    for (const replacement of ["1", -1, 1.5]) {
+      assert.equal(decodeCodexHeadlessTurn(stream(...base, { type: "turn.completed", usage: usage({ [field]: replacement }) })).usage, undefined);
+    }
+  }
+});
+
+test("invalid subset relationships clear complete without failing publication", () => {
+  const decoded = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } },
+    { type: "turn.completed", usage: usage({ reasoning_output_tokens: 6 }) }
+  ));
+  assert.equal(decoded.usage?.complete, false);
+});
+
+test("a frame after turn.completed preserves the reply and usage", () => {
+  const decoded = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } },
+    { type: "turn.completed", usage: usage() },
+    { type: "future.envelope", payload: true }
+  ));
+  assert.equal(decoded.text, "ok");
+  assert.equal(decoded.usage?.total, 18_115);
+});
+
+test("two turn.completed frames preserve the reply without ambiguous usage", () => {
+  const decoded = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } },
+    { type: "turn.completed", usage: usage() },
+    { type: "turn.completed", usage: usage() }
+  ));
+  assert.equal(decoded.text, "ok");
+  assert.equal(decoded.usage, undefined);
+});
+
+test("a reply without turn.completed is published without usage", () => {
+  const decoded = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } }
+  ));
+  assert.deepEqual(decoded, { text: "ok" });
+});
+
+test("the last non-blank agent message wins", () => {
+  const decoded = decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } },
+    { type: "item.completed", item: { type: "agent_message", text: "   " } },
+    { type: "turn.completed", usage: usage() }
+  ));
+  assert.equal(decoded.text, "ok");
+  assert.equal(decoded.usage?.total, 18_115);
+});
+
+test("only a blank agent message rejects", () => {
+  assert.throws(() => decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "   " } }
+  )), /empty response/u);
+});
+
+test("turn.failed after a reply publishes text without usage", () => {
+  assert.deepEqual(decodeCodexHeadlessTurn(stream(
+    { type: "item.completed", item: { type: "agent_message", text: "ok" } },
+    { type: "turn.failed", error: "no" }
+  )), { text: "ok" });
+});
+
+test("turn.failed without a reply rejects", () => {
+  assert.throws(() => decodeCodexHeadlessTurn(stream({ type: "turn.failed", error: "no" })), /failed turn/u);
+});
+
+test("empty streams reject", () => {
+  assert.throws(() => decodeCodexHeadlessTurn(""), /empty stream/u);
+});