diff --git a/src/pi/codexHeadlessResult.test.ts b/src/pi/codexHeadlessResult.test.ts index cbcb164..69f509d 100644 --- 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 index e19bb63..fac9edc 100644 --- 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 new file mode 100755 index 0000000..dafcd8c --- /dev/null +++ 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 index b7b4058..8844f54 100644 --- 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 index 69f20b4..7ffa55b 100644 --- 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:` 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 { 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): Promise { 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 { 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 | 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; 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> { 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); } 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> { + 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); } 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 { return Object.fromEntries(Object.entries(value).filter((entry): entry is [string, string] => entry[1] !== undefined)); }