diff --git a/src/pi/cliEngineSpawn.test.ts b/src/pi/cliEngineSpawn.test.ts index 4282571..037bcc5 100644 --- a/src/pi/cliEngineSpawn.test.ts +++ b/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 { diff --git a/src/pi/cliEngineSpawn.ts b/src/pi/cliEngineSpawn.ts index f6af688..70cebb5 100644 --- a/src/pi/cliEngineSpawn.ts +++ b/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, 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))) { diff --git a/src/pi/cliSession.test.ts b/src/pi/cliSession.test.ts index 69f9e52..d887922 100644 --- a/src/pi/cliSession.test.ts +++ b/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[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) { diff --git a/src/pi/cliSession.ts b/src/pi/cliSession.ts index 45b14aa..e414613 100644 --- a/src/pi/cliSession.ts +++ b/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; + readonly onTurnUsage?: (usage: AgyTurnUsage | CodexTurnUsage) => Promise; } & ({ readonly engine: "codex" | "grok"; } | { @@ -228,7 +229,7 @@ class CliSession implements PiSessionLike { const secretValues = [...environmentSecretValues, ...stagedCredentialSecrets]; let mount: { endpoint: string; close: () => Promise } | 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; diff --git a/src/pi/codexHeadlessResult.test.ts b/src/pi/codexHeadlessResult.test.ts new file mode 100644 index 0000000..cbcb164 --- /dev/null +++ b/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 = {}) => ({ + 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); +}); diff --git a/src/pi/codexHeadlessResult.ts b/src/pi/codexHeadlessResult.ts new file mode 100644 index 0000000..e19bb63 --- /dev/null +++ b/src/pi/codexHeadlessResult.ts @@ -0,0 +1,117 @@ +type JsonRecord = Readonly>; + +const isRecord = (value: unknown): value is JsonRecord => + typeof value === "object" && value !== null && !Array.isArray(value); + +const invalidResult = (detail: string): Error => + new Error(`Codex CLI returned no publishable terminal response: ${detail}`); + +/** + * Token accounting for one Codex turn, from the terminal frame emitted by + * `codex exec --json`: + * + * {"type":"turn.completed","usage":{"input_tokens":…, + * "cached_input_tokens":…,"cache_write_input_tokens":…, + * "output_tokens":…,"reasoning_output_tokens":…}} + * + * Codex's reconciled session total proves `cached_input_tokens` is a subset of + * `input_tokens`: total is `input + output`, unlike AGY's disjoint cache-read + * bucket. Adding cacheRead would over-count every cached turn. Likewise, + * `reasoning_output_tokens` is a subset of `output_tokens`; it is read only to + * validate that relationship, never added. Codex reports cache writes, but no + * cost, so `cacheWrite` is measured while `notionalUsd: 0` means absent, not + * free. `calls` counts completed command/MCP tools; Codex exposes no AGY-like + * model-step count, so this is explicitly a tool-call count. + */ +export type CodexTurnUsage = Readonly<{ + input: number; + output: number; + cacheRead: number; + cacheWrite: number; + total: number; + calls: number; + notionalUsd: number; + complete: boolean; +}>; + +export type CodexHeadlessTurn = Readonly<{ text: string; usage?: CodexTurnUsage }>; + +const USAGE_TOKEN_FIELDS = [ + "input_tokens", + "cached_input_tokens", + "cache_write_input_tokens", + "output_tokens", + "reasoning_output_tokens" +] as const; + +const tokenCount = (value: unknown): number | undefined => + typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : undefined; + +/** + * Advisory usage decoding: an absent or malformed block is `undefined`, never + * a failed published turn and never a zero-filled record. Every field must be + * a non-negative safe integer because a substituted zero is byte-identical to + * a real zero. `complete` requires a non-zero derived total and valid subset + * relationships. As with AGY, every emitted count remains a lower bound. + */ +const decodeCodexTurnUsage = (frame: JsonRecord, calls: number): CodexTurnUsage | undefined => { + if (!isRecord(frame.usage)) return undefined; + 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 total = input + output; + return { + input, + output, + cacheRead, + cacheWrite, + total, + calls, + notionalUsd: 0, + complete: total > 0 && cacheRead <= input && reasoningOutput <= output + }; +}; + +/** + * Decode Codex's NDJSON stream without mistaking tool/progress items for the + * reply. Unknown envelopes and item types are skipped because Codex owns the + * vocabulary and a future frame must not fail a turn that published. + * A Codex turn that spent subscription money and produced a reply must reach + * the organization: uncertainty about accounting degrades to `usage: + * undefined` and never discards publishable text. Unlike AGY's stricter + * envelope, Codex may omit, repeat, or follow `turn.completed` with another + * frame, so only exactly one completion frame supplies trustworthy usage. + */ +export const decodeCodexHeadlessTurn = (output: string): CodexHeadlessTurn => { + const lines = output.split(/\r?\n/).filter((line) => line.trim().length > 0); + if (lines.length === 0) throw invalidResult("empty stream"); + let text: string | undefined; + const completed: JsonRecord[] = []; + let failed = false; + let calls = 0; + for (const line of lines) { + let frame: unknown; + try { frame = JSON.parse(line); } catch { throw invalidResult("invalid JSON"); } + if (!isRecord(frame) || typeof frame.type !== "string") throw invalidResult("invalid event"); + if (frame.type === "turn.failed") { + failed = true; + continue; + } + if (frame.type === "turn.completed") { + completed.push(frame); + continue; + } + if (frame.type !== "item.completed" || !isRecord(frame.item) || typeof frame.item.type !== "string") continue; + if (frame.item.type === "command_execution" || frame.item.type === "mcp_tool_call") calls += 1; + if (frame.item.type === "agent_message" && typeof frame.item.text === "string" && frame.item.text.trim().length > 0) { + text = frame.item.text; + } + } + if (typeof text !== "string") throw invalidResult(failed ? "failed turn" : "empty response"); + const usage = !failed && completed.length === 1 ? decodeCodexTurnUsage(completed[0], calls) : undefined; + return usage === undefined ? { text: text.trim() } : { text: text.trim(), usage }; +}; + +/** Text-only view of {@link decodeCodexHeadlessTurn}, for callers that do not meter. */ +export const decodeCodexHeadlessResult = (output: string): string => decodeCodexHeadlessTurn(output).text; diff --git a/src/runtime/cli.test.ts b/src/runtime/cli.test.ts index 4c6e273..82ab4c2 100644 --- a/src/runtime/cli.test.ts +++ b/src/runtime/cli.test.ts @@ -47,7 +47,7 @@ test("CLI strictly authenticates and routes a production Daimon engine", async ( const readinessReceipt = path.join(root, "state", "runtime-readiness.json"); await writeFile(inboundAuth, JSON.stringify({ tokens: { access_token: "test-access", refresh_token: "test-refresh" } }), { mode: 0o600 }); await chmod(inboundAuth, 0o600); - await writeFile(program, `#!/usr/bin/env node\nif (process.argv.includes('--version')) process.stdout.write('test'); else { process.stdin.resume(); process.stdin.on('end', () => process.stdout.write(process.env.${tokenEnv} ?? 'absent')); }`); + await writeFile(program, `#!/usr/bin/env node\nif (process.argv.includes('--version')) process.stdout.write('test'); else { process.stdin.resume(); process.stdin.on('end', () => { const text = process.env.${tokenEnv} ?? 'absent'; process.stdout.write([{ type: 'item.completed', item: { type: 'agent_message', text } }, { type: 'turn.completed' }].map(JSON.stringify).join('\\n')); }); }`); await chmod(program, 0o700); await writeFile(configPath, JSON.stringify({ version: ORGANIZATION_RUNTIME_VERSION, diff --git a/src/runtime/engineDispatcher.test.ts b/src/runtime/engineDispatcher.test.ts index 8b811b7..d9a19b1 100644 --- a/src/runtime/engineDispatcher.test.ts +++ b/src/runtime/engineDispatcher.test.ts @@ -1,5 +1,5 @@ import assert from "node:assert/strict"; -import { access, chmod, mkdir, mkdtemp, rm, unlink, writeFile } from "node:fs/promises"; +import { access, chmod, mkdir, mkdtemp, readFile, rm, unlink, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import test from "node:test"; @@ -43,7 +43,9 @@ test("production dispatcher starts each closed engine intent through Daimon", as const root = await mkdtemp(path.join(os.tmpdir(), "daimon-dispatcher-")); const priorPath = process.env.PATH; const priorRun = process.env.NOOPOLIS_RUN_ID; - const stub = `#!/usr/bin/env node\nconst args = process.argv.slice(2); const text = process.env.DAIMON_DISPATCH_CONTROL ?? "absent"; 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 } } }); if (args.includes("mcp")) process.stdout.write("ok"); else process.stdout.write(args.includes("--single") ? stream(text) : args.includes("--output-format") ? agy(text) : text);`; + const priorLedger = process.env.DAIMON_TURN_USAGE_LEDGER_PATH; + const ledger = path.join(root, "usage.jsonl"); + const stub = `#!/usr/bin/env node\nconst args = process.argv.slice(2); const text = process.env.DAIMON_DISPATCH_CONTROL ?? "absent"; 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", usage: { input_tokens: 11, cached_input_tokens: 3, cache_write_input_tokens: 0, output_tokens: 2, reasoning_output_tokens: 1 } }].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 } } }); if (args.includes("mcp")) process.stdout.write("ok"); else process.stdout.write(args.includes("--single") ? stream(text) : args.includes("--output-format") ? agy(text) : args.includes("--json") ? codex(text) : text);`; try { for (const name of ["codex", "grok", "agy"]) { const file = path.join(root, name); @@ -53,6 +55,7 @@ test("production dispatcher starts each closed engine intent through Daimon", as } process.env.PATH = `${root}${path.delimiter}${priorPath ?? ""}`; process.env.NOOPOLIS_RUN_ID = "dispatcher-test"; + process.env.DAIMON_TURN_USAGE_LEDGER_PATH = ledger; process.env.DAIMON_DISPATCH_CONTROL = "host-only"; for (const kind of ["codex", "grok", "agy"] as const) { const handle = await startOrganizationRuntimeEngine(rootConfig(root, kind), "DAIMON_DISPATCH_CONTROL", undefined, kind === "agy" ? "unix:path=/private/realm/bus" : undefined); @@ -60,11 +63,18 @@ test("production dispatcher starts each closed engine intent through Daimon", as assert.equal(result.text, "absent"); await handle.stop(); } + const metered = (await readFile(ledger, "utf8")).trim().split("\n").map((line) => JSON.parse(line)); + assert.deepEqual(metered.map(({ engine, wake, total }) => ({ engine, wake, total })), [ + { engine: "codex", wake: "codex-wake", total: 13 }, + { engine: "agy", wake: "agy-wake", total: 13 } + ], "Codex and AGY carry advisory session meters while unbrokered Grok remains unchanged"); } finally { if (priorPath === undefined) delete process.env.PATH; else process.env.PATH = priorPath; if (priorRun === undefined) delete process.env.NOOPOLIS_RUN_ID; else process.env.NOOPOLIS_RUN_ID = priorRun; + if (priorLedger === undefined) delete process.env.DAIMON_TURN_USAGE_LEDGER_PATH; + else process.env.DAIMON_TURN_USAGE_LEDGER_PATH = priorLedger; delete process.env.DAIMON_DISPATCH_CONTROL; await rm(root, { recursive: true, force: true }); } @@ -74,7 +84,7 @@ test("engine dispatcher threads a declared memory bank into the Pi harness", asy const root = await mkdtemp(path.join(os.tmpdir(), "daimon-dispatcher-memory-")); const priorPath = process.env.PATH; const priorRun = process.env.NOOPOLIS_RUN_ID; - const stub = `#!/usr/bin/env node\nconst args = process.argv.slice(2); const text = process.env.DAIMON_DISPATCH_CONTROL ?? "absent"; 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 } } }); if (args.includes("mcp")) process.stdout.write("ok"); else process.stdout.write(args.includes("--single") ? stream(text) : args.includes("--output-format") ? agy(text) : text);`; + const stub = `#!/usr/bin/env node\nconst args = process.argv.slice(2); const text = process.env.DAIMON_DISPATCH_CONTROL ?? "absent"; 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", usage: { input_tokens: 11, cached_input_tokens: 3, cache_write_input_tokens: 0, output_tokens: 2, reasoning_output_tokens: 1 } }].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 } } }); if (args.includes("mcp")) process.stdout.write("ok"); else process.stdout.write(args.includes("--single") ? stream(text) : args.includes("--output-format") ? agy(text) : args.includes("--json") ? codex(text) : text);`; try { const file = path.join(root, "codex"); await writeFile(file, stub); @@ -175,7 +185,7 @@ test("Daimon frames one escaped identity envelope for every production engine", "if (args.includes('mcp')) process.stdout.write('ok');", "else if (args.includes('--single')) { const text = args[args.indexOf('--single') + 1]; process.stdout.write(stream(text)); }", "else if (args.includes('--print')) process.stdout.write(JSON.stringify({ event: 'result', result: { conversation_id: 'fake', status: 'SUCCESS', response: args[args.indexOf('--print') + 1], num_turns: 1 } }));", - "else { const chunks = []; for await (const chunk of process.stdin) chunks.push(chunk); process.stdout.write(Buffer.concat(chunks).toString('utf8')); }" + "else { const chunks = []; for await (const chunk of process.stdin) chunks.push(chunk); const text = Buffer.concat(chunks).toString('utf8'); process.stdout.write([{ type: 'item.completed', item: { type: 'agent_message', text } }, { type: 'turn.completed' }].map(JSON.stringify).join('\\n')); }" ].join("\n"); try { for (const kind of ["codex", "grok", "agy"] as const) { diff --git a/src/runtime/engineDispatcher.ts b/src/runtime/engineDispatcher.ts index 63b6fc5..9543869 100644 --- a/src/runtime/engineDispatcher.ts +++ b/src/runtime/engineDispatcher.ts @@ -94,6 +94,11 @@ function adapterFor(agent: OrganizationRuntimeAgentConfig, controlTokenEnv: stri // appends to. `recordTurnUsage` is advisory and never rejects. onTurnUsage: (usage) => recordTurnUsage(resolveTurnUsageLedgerPath(), { agent: agent.id, wake: wakeEnvironmentContext.current ?? "wake", engine: "agy", usage }) } : { engine, redactedEnvironmentNames: [controlTokenEnv], identityPrompt: identityEnvelope(agent), command: executablePath, engineHomePath, verifyExecutable, verifyRuntimePaths, + ...(engine === "codex" ? { + // Codex has no broker to meter it, so publish terminal-frame usage + // to the shared advisory ledger after the session publishes. + onTurnUsage: (usage: import("../pi/codexHeadlessResult.js").CodexTurnUsage) => recordTurnUsage(resolveTurnUsageLedgerPath(), { agent: agent.id, wake: wakeEnvironmentContext.current ?? "wake", engine: "codex", usage }) + } : {}), ...(engine==="grok"&&grokBroker!==undefined?{}:{credentialSecretValues: () => readPortableEngineCredentialSecrets(agent.id, engine, engineHomePath)}), ...(engine==="grok"&&grokBroker!==undefined?{grokBrokerTurn:(prompt:string,endpoint:string,signal:AbortSignal)=>grokBroker.turn(agent.id,wakeEnvironmentContext.current??"wake",prompt,endpoint,signal)}:{}), ...(engine === "grok" && verifyGrokSandbox ? { diff --git a/src/runtime/organizationRuntimeControl.ts b/src/runtime/organizationRuntimeControl.ts index bd5cf08..98b14a7 100644 --- a/src/runtime/organizationRuntimeControl.ts +++ b/src/runtime/organizationRuntimeControl.ts @@ -8,6 +8,7 @@ import { type OrganizationRuntimeWakeRequest } from "./organizationRuntime.js"; import { createOrganizationRuntimeHost } from "./organizationRuntimeHost.js"; +import { WakeFuse, type WakeFuseTripReason } from "./wakeFuse.js"; import { createScheduleController, type ScheduleController, type ScheduleControllerOptions } from "./schedule.js"; import { WakeAcceptanceConflictError, WakeAcceptanceStore, WakeExecutionClaimLostError, publicAcceptance, type WakeAcceptanceStoreTestOptions } from "./wakeAcceptanceStore.js"; import { @@ -28,6 +29,10 @@ export type OrganizationRuntimeControlOptions = Readonly<{ acceptanceStorePath: type TestControlOptions = OrganizationRuntimeControlOptions & Readonly<{ scheduleOptions?: Pick; storeOptions?: WakeAcceptanceStoreTestOptions; + fuseEnvironment?: NodeJS.ProcessEnv; + fusePollIntervalMsForTest?: number; + beforeTripTerminalizationForTest?: () => Promise; + afterFuseAdmissionForTest?: () => Promise; }>; type CoreHost = OrganizationRuntimeHost; type AcceptanceRecord = Awaited>["record"]; @@ -45,7 +50,10 @@ export function createOrganizationRuntimeControlHost(config: unknown, options: O /** @internal Test seam; intentionally absent from the public runtime barrel. */ export function createOrganizationRuntimeControlHostWithCoreForTest(config: unknown, host: CoreHost, options: TestControlOptions): OrganizationRuntimeControlHost { - return createControl(parseOrganizationRuntimeConfig(config), host, options, options.storeOptions); + return createControl(parseOrganizationRuntimeConfig(config), host, { + ...options, + fuseEnvironment: options.fuseEnvironment ?? { DAIMON_WAKE_FUSE: "off" } + }, options.storeOptions); } function createControl(config: OrganizationRuntimeConfig, host: CoreHost, options: TestControlOptions, storeOptions?: WakeAcceptanceStoreTestOptions): OrganizationRuntimeControlHost { @@ -56,8 +64,12 @@ function createControl(config: OrganizationRuntimeConfig, host: CoreHost, option const agentTails = new Map>(); const acceptanceTails = new Map>(); const retryWaiters = new Set<() => void>(); + const persistenceInFlight = new Set>(); let store: WakeAcceptanceStore | undefined; let schedules: ScheduleController | undefined; + let fuse: WakeFuse | undefined; + let fusePoll: ReturnType | undefined; + let fuseTrip: Promise | undefined; let started = false; let stopping = false; @@ -133,18 +145,60 @@ function createControl(config: OrganizationRuntimeConfig, host: CoreHost, option }; const persistRequest = async (request: OrganizationRuntimeWakeAcceptanceRequest): Promise => { + const persistence: Promise = (async () => { try { const accepted = await store!.accept(request); dispatch(accepted.record); return publicAcceptance(accepted.record); } catch (error) { if (error instanceof WakeAcceptanceConflictError) return { version: "noopolis.daimon.wake-acceptance.v2", state: "rejected", code: "delivery_conflict" }; throw error; } + })(); + persistenceInFlight.add(persistence); + try { return await persistence; } finally { persistenceInFlight.delete(persistence); } }; const acceptRequest = async (request: OrganizationRuntimeWakeAcceptanceRequest): Promise => await serializeAcceptance(request.agent_id, async () => await persistRequest(request)); + const tripFuse = (reason: WakeFuseTripReason): Promise => { + if (fuseTrip !== undefined) return fuseTrip; + fuseTrip = (async () => { + // Order is deliberate: refuse arrivals, terminalize queued work, then + // await already-running turns without stopping the underlying host. + stopping = true; + for (const wake of retryWaiters) wake(); + await options.beforeTripTerminalizationForTest?.(); + if (store !== undefined) { + for (let pass = 0; pass < 16; pass += 1) { + const queued = (await store.activity()).filter((record) => record.state === "accepted"); + await Promise.all(queued.map(async (record) => { await store!.transitionAcceptedToStopped(record.acceptance_id); })); + if (persistenceInFlight.size > 0) await Promise.allSettled([...persistenceInFlight]); + const remaining = (await store.activity()).some((record) => record.state === "accepted"); + if (!remaining && persistenceInFlight.size === 0) break; + // The durable trip marker makes any remainder non-dispatchable on the + // next start if a pathological producer outlives this bounded drain. + } + } + await Promise.allSettled(inFlight.values()); + void reason; + })(); + return fuseTrip; + }; + + const admitAndPersist = async (request: OrganizationRuntimeWakeAcceptanceRequest): Promise => { + const verdict = await fuse!.admit(request.agent_id, request.delivery_id); + if (verdict.state === "tripped") { + await tripFuse(verdict.reason); + return { version: "noopolis.daimon.wake-acceptance.v2", state: "stopped", code: "host_stopping" }; + } + await options.afterFuseAdmissionForTest?.(); + if (stopping) return { version: "noopolis.daimon.wake-acceptance.v2", state: "stopped", code: "host_stopping" }; + return await acceptRequest(request); + }; + return { - wake: async (request) => await host.wake(request), + wake: async (request) => stopping + ? { version: "noopolis.daimon.wake-result.v1", status: "stopped", agentId: request.agentId, wakeId: request.event.id, code: "host_stopping" } + : await host.wake(request), health: async (agentId) => await host.health(agentId), activity: async (request) => await host.activity(request), async start(): Promise { @@ -152,18 +206,35 @@ function createControl(config: OrganizationRuntimeConfig, host: CoreHost, option if (stopping) throw new Error("organization runtime control host has been stopped"); if (expectedToken === undefined || !expectedToken.trim()) throw new Error("required control token is missing or blank"); const opened = await WakeAcceptanceStore.open(options.acceptanceStorePath, storeOptions); + let openedFuse: WakeFuse | undefined; try { + openedFuse = await WakeFuse.open({ organizationKey: [...knownAgents].sort().join("\u0000"), environment: options.fuseEnvironment }); await host.start(); store = opened; + fuse = openedFuse; started = true; - for (const record of await opened.recoverable(knownAgents)) dispatch(record); - if (config.version === "noopolis.daimon.organization-runtime.v2") { + const startupTrip = fuse.tripped(); + if (startupTrip !== undefined) await tripFuse(startupTrip); + else for (const record of await opened.recoverable(knownAgents)) dispatch(record); + fusePoll = setInterval(() => { + void fuse?.pollOperatorStop() + .then((reason) => { if (reason !== undefined) void tripFuse(reason).catch(() => undefined); }) + .catch(() => undefined); + // A partial drain cannot cause an unhandled rejection; the durable + // marker keeps all remaining accepted work stopped on the next start. + }, options.fusePollIntervalMsForTest ?? WAKE_FUSE_OPERATOR_POLL_MS); + fusePoll.unref(); + if (!stopping && config.version === "noopolis.daimon.organization-runtime.v2") { schedules = createScheduleController({ acceptanceStorePath: options.acceptanceStorePath, agents: config.agents, ...options.scheduleOptions, accept: async (occurrence) => await serializeAcceptance(occurrence.agentId, async () => { if (agentTails.has(occurrence.agentId)) return false; - const accepted = await persistRequest({ token: expectedToken, agent_id: occurrence.agentId, delivery_id: occurrence.deliveryId, event: { version: "noopolis.daimon.wake.v2", kind: "schedule", text: occurrence.prompt, occurred_at: occurrence.occurredAt } }); + const request: OrganizationRuntimeWakeAcceptanceRequest = { token: expectedToken, agent_id: occurrence.agentId, delivery_id: occurrence.deliveryId, event: { version: "noopolis.daimon.wake.v2", kind: "schedule", text: occurrence.prompt, occurred_at: occurrence.occurredAt } }; + const verdict = await fuse!.admit(request.agent_id, request.delivery_id); + if (verdict.state === "tripped") { await tripFuse(verdict.reason); return false; } + if (stopping) return false; + const accepted = await persistRequest(request); if (accepted.state !== "accepted") throw new Error(`scheduled wake ${occurrence.deliveryId} was not durably accepted`); return true; }) @@ -172,6 +243,7 @@ function createControl(config: OrganizationRuntimeConfig, host: CoreHost, option } } catch (error) { await host.stop().catch(() => undefined); + await openedFuse?.close().catch(() => undefined); await opened.close().catch(() => undefined); throw error; } @@ -182,7 +254,7 @@ function createControl(config: OrganizationRuntimeConfig, host: CoreHost, option if (!tokensEqual(expectedToken, request.token)) return { version: "noopolis.daimon.wake-acceptance.v2", state: "rejected", code: "unauthorized" }; if (!started || stopping) return { version: "noopolis.daimon.wake-acceptance.v2", state: "stopped", code: stopping ? "host_stopping" : "host_stopped" }; if (!knownAgents.has(request.agent_id)) return { version: "noopolis.daimon.wake-acceptance.v2", state: "rejected", code: "unknown_agent" }; - return await acceptRequest(request); + return await admitAndPersist(request); }, async wakeReceipt(token: string | undefined, acceptanceId: string): Promise { if (!tokensEqual(expectedToken, token) || store === undefined) return undefined; @@ -194,13 +266,16 @@ function createControl(config: OrganizationRuntimeConfig, host: CoreHost, option }, async stop(): Promise { stopping = true; + if (fusePoll !== undefined) clearInterval(fusePoll); for (const wake of retryWaiters) wake(); await schedules?.stop(); const result = await host.stop(); await Promise.allSettled(inFlight.values()); await store?.releaseClaims(ownerId); await store?.close(); + await fuse?.close(); store = undefined; + fuse = undefined; schedules = undefined; started = false; return result; @@ -208,6 +283,8 @@ function createControl(config: OrganizationRuntimeConfig, host: CoreHost, option }; } +export const WAKE_FUSE_OPERATOR_POLL_MS = 1_000; + function tokensEqual(expected: string | undefined, actual: string | undefined): boolean { if (expected === undefined || !expected.trim()) return false; const digest = (value: string): Buffer => createHash("sha256").update(value).digest(); diff --git a/src/runtime/turnUsageLedger.test.ts b/src/runtime/turnUsageLedger.test.ts index 30d92e7..26c5bb2 100644 --- a/src/runtime/turnUsageLedger.test.ts +++ b/src/runtime/turnUsageLedger.test.ts @@ -116,7 +116,18 @@ test("an AGY turn renders the same record shape, labelled agy", () => { input: 44_937, output: 444, cache_read: 0, cache_write: 0, total: 45_381, calls: 1, notional_usd: 0, complete: true }); - assert.deepEqual([...TURN_USAGE_ENGINES], ["agy", "grok"], "codex is uninstrumented and must not claim zero usage"); + assert.deepEqual([...TURN_USAGE_ENGINES], ["agy", "codex", "grok"]); +}); + +test("a Codex turn renders measured cache writes and subset-safe totals", () => { + const parsed = JSON.parse(renderTurnUsageLine(entry({ + engine: "codex", + usage: { input: 18_110, output: 5, cacheRead: 11_008, cacheWrite: 0, total: 18_115, calls: 0, notionalUsd: 0, complete: true } + }))); + assert.equal(parsed.engine, "codex"); + assert.equal(parsed.total, 18_115); + assert.equal(parsed.cache_read, 11_008); + assert.equal(parsed.cache_write, 0); }); test("the ledger path override is honoured only when it is absolute", () => { diff --git a/src/runtime/turnUsageLedger.ts b/src/runtime/turnUsageLedger.ts index 4aaf6ef..ca7a5ee 100644 --- a/src/runtime/turnUsageLedger.ts +++ b/src/runtime/turnUsageLedger.ts @@ -57,10 +57,11 @@ export type TurnUsageMeasurement = Readonly<{ /** * Every engine whose headless stream reports its own token accounting. * - * Codex is deliberately absent: it is uninstrumented, and `spawnfile usage` - * renders a Codex agent as a dashed roster row rather than as zero usage. + * Codex, AGY, and Grok all publish decoded terminal-stream accounting here. + * Spawnfile's reader mirrors this list and is updated separately in packet + * A1b; until then, a reader that rejects `codex` will drop every Codex line. */ -export const TURN_USAGE_ENGINES = ["agy", "grok"] as const; +export const TURN_USAGE_ENGINES = ["agy", "codex", "grok"] as const; export type TurnUsageEntry = Readonly<{ agent: string; diff --git a/src/runtime/wakeAcceptance.test.ts b/src/runtime/wakeAcceptance.test.ts index 4a90b59..a447fed 100644 --- a/src/runtime/wakeAcceptance.test.ts +++ b/src/runtime/wakeAcceptance.test.ts @@ -7,6 +7,7 @@ import test from "node:test"; import { ORGANIZATION_RUNTIME_VERSION, type OrganizationRuntimeHost, type OrganizationRuntimeWakeRequest } from "./organizationRuntime.js"; import { createOrganizationRuntimeControlHostWithCoreForTest } from "./organizationRuntimeControl.js"; +import { WakeFuse } from "./wakeFuse.js"; import { WakeAcceptanceStore, WakeTransitionLockBlockedError } from "./wakeAcceptanceStore.js"; import { MAX_WAKE_COMPLETION_TEXT_BYTES, parseWakeAcceptanceRequest, wakeAcceptanceDigest } from "./wakeAcceptanceTypes.js"; import { TERMINAL_RECEIPT_IDEMPOTENCY_HORIZON } from "./wakeAcceptanceRetention.js"; @@ -75,6 +76,225 @@ test("control accepts before a fake turn finishes, redacts status, and rejects c } finally { await rm(root, { recursive: true, force: true }); } }); +test("a fuse trip terminalizes queued deliveries, refuses arrivals, and awaits running work", async () => { + const root = await privateRoot(); + const usage = await mkdtemp(path.join(os.tmpdir(), "daimon-fuse-usage-")); + const core = new FakeCoreHost(); + const control = createOrganizationRuntimeControlHostWithCoreForTest(config, core, { + acceptanceStorePath: root, controlToken: token, storeOptions: testStoreOptions, + fuseEnvironment: fuseEnvironment(usage, 2) + }); + try { + await control.start(); + await control.accept(request("running")); + await core.waitForWakes(1); + await control.accept(request("queued")); + let settled = false; + const trip = control.accept(request("trip")).then((value) => { settled = true; return value; }); + await new Promise((resolve) => setTimeout(resolve, 50)); + const items = (await control.activityV2(token))?.items ?? []; + assert.equal(items.filter((item) => item.state === "accepted").length, 0, "a trip must leave zero records in state accepted"); + assert.equal(items.find((item) => item.delivery_id === "queued")?.state, "stopped"); + assert.equal(items.find((item) => item.delivery_id === "queued")?.code, "host_stopping"); + assert.equal(items.find((item) => item.delivery_id === "running")?.state, "running"); + assert.equal(settled, false, "the trip waits for a running turn"); + core.release(); + assert.deepEqual(await trip, { version: "noopolis.daimon.wake-acceptance.v2", state: "stopped", code: "host_stopping" }); + assert.deepEqual(await control.accept(request("later")), { version: "noopolis.daimon.wake-acceptance.v2", state: "stopped", code: "host_stopping" }); + const wakeCount = core.wakes.length; + assert.deepEqual(await control.wake({ token, agentId: "alpha", event: { version: "noopolis.daimon.wake.v1", id: "direct-after-trip", kind: "manual", text: "blocked", occurredAt: "2026-08-17T00:00:00.000Z" } }), { + version: "noopolis.daimon.wake-result.v1", status: "stopped", agentId: "alpha", wakeId: "direct-after-trip", code: "host_stopping" + }); + assert.equal(core.wakes.length, wakeCount); + await control.stop(); + } finally { core.release(); await control.stop().catch(() => undefined); await rm(root, { recursive: true, force: true }); await rm(usage, { recursive: true, force: true }); } +}); + +test("a persistence that lands after the trip snapshot is terminalized", async () => { + const root = await privateRoot(); + const usage = await mkdtemp(path.join(os.tmpdir(), "daimon-fuse-usage-")); + let reachedTransition!: () => void; + const transitionReached = new Promise((resolve) => { reachedTransition = resolve; }); + let releaseTransition!: () => void; + const transitionRelease = new Promise((resolve) => { releaseTransition = resolve; }); + let blockOnce = true; + const core = new FakeCoreHost(); + const control = createOrganizationRuntimeControlHostWithCoreForTest(config, core, { + acceptanceStorePath: root, controlToken: token, + storeOptions: { ...testStoreOptions, afterFinalLockAssertion: async () => { + if (!blockOnce) return; + blockOnce = false; + reachedTransition(); + await transitionRelease; + } }, + fuseEnvironment: fuseEnvironment(usage, 2) + }); + try { + await control.start(); + const first = await control.accept(request("claim-blocked")); + assert.equal(first.state, "accepted"); + await transitionReached; + const latePersistence = control.accept(request("persist-after-snapshot")); + const trip = control.accept(request("trip-after-snapshot")); + releaseTransition(); + assert.equal((await latePersistence).state, "accepted"); + core.release(); + assert.deepEqual(await trip, { version: "noopolis.daimon.wake-acceptance.v2", state: "stopped", code: "host_stopping" }); + const items = (await control.activityV2(token))?.items ?? []; + assert.equal(items.filter((item) => item.state === "accepted").length, 0); + assert.equal(items.find((item) => item.delivery_id === "persist-after-snapshot")?.state, "stopped"); + await control.stop(); + } finally { releaseTransition(); core.release(); await control.stop().catch(() => undefined); await rm(root, { recursive: true, force: true }); await rm(usage, { recursive: true, force: true }); } +}); + +test("a request admitted before a concurrent trip is refused before persistence", async () => { + const root = await privateRoot(); + const usage = await mkdtemp(path.join(os.tmpdir(), "daimon-fuse-usage-")); + let admissionReached!: () => void; + const reachedAdmission = new Promise((resolve) => { admissionReached = resolve; }); + let releaseAdmission!: () => void; + const admissionRelease = new Promise((resolve) => { releaseAdmission = resolve; }); + let tripReached!: () => void; + const reachedTrip = new Promise((resolve) => { tripReached = resolve; }); + let releaseTrip!: () => void; + const tripRelease = new Promise((resolve) => { releaseTrip = resolve; }); + const control = createOrganizationRuntimeControlHostWithCoreForTest(config, new FakeCoreHost(), { + acceptanceStorePath: root, controlToken: token, storeOptions: testStoreOptions, + fuseEnvironment: fuseEnvironment(usage, 10), fusePollIntervalMsForTest: 1, + afterFuseAdmissionForTest: async () => { admissionReached(); await admissionRelease; }, + beforeTripTerminalizationForTest: async () => { tripReached(); await tripRelease; } + }); + try { + await control.start(); + const pending = control.accept(request("admitted-before-trip")); + await reachedAdmission; + await writeFile(path.join(usage, "fuse.stop"), ""); + await Promise.race([reachedTrip, new Promise((_, reject) => setTimeout(() => reject(new Error("concurrent trip timed out")), 1_000))]); + releaseAdmission(); + assert.deepEqual(await pending, { version: "noopolis.daimon.wake-acceptance.v2", state: "stopped", code: "host_stopping" }); + const items = (await control.activityV2(token))?.items ?? []; + assert.equal(items.some((item) => item.delivery_id === "admitted-before-trip"), false, "a request caught by the admit/persist trip window must not create a store record"); + } finally { + releaseAdmission(); releaseTrip(); + await control.stop().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + await rm(usage, { recursive: true, force: true }); + } +}); + +test("invalid authorities consume no admission and duplicate delivery is fuse-idempotent", async () => { + const root = await privateRoot(); + const usage = await mkdtemp(path.join(os.tmpdir(), "daimon-fuse-usage-")); + const core = new FakeCoreHost(); + const control = createOrganizationRuntimeControlHostWithCoreForTest(config, core, { + acceptanceStorePath: root, controlToken: token, storeOptions: testStoreOptions, + fuseEnvironment: fuseEnvironment(usage, 1) + }); + try { + await control.start(); + assert.deepEqual(await control.accept({ nope: true }), { version: "noopolis.daimon.wake-acceptance.v2", state: "rejected", code: "invalid_request" }); + assert.deepEqual(await control.accept({ ...request("unauthorized"), token: "wrong" }), { version: "noopolis.daimon.wake-acceptance.v2", state: "rejected", code: "unauthorized" }); + assert.deepEqual(await control.accept({ ...request("unknown"), agent_id: "missing" }), { version: "noopolis.daimon.wake-acceptance.v2", state: "rejected", code: "unknown_agent" }); + const first = await control.accept(request("paid-once")); + assert.equal(first.state, "accepted"); + await waitFor(() => core.wakes.length === 1); + core.release(); + if (first.state === "accepted") await waitFor(async () => (await control.wakeReceipt(token, first.acceptance_id))?.state === "completed"); + assert.equal((await control.accept(request("paid-once"))).state, "accepted"); + const tripped = control.accept(request("second-unique")); + assert.deepEqual(await tripped, { version: "noopolis.daimon.wake-acceptance.v2", state: "stopped", code: "host_stopping" }); + await control.stop(); + } finally { await rm(root, { recursive: true, force: true }); await rm(usage, { recursive: true, force: true }); } +}); + +test("startup into an operator-tripped fuse terminalizes recovery without dispatch", async () => { + const root = await privateRoot(); + const usage = await mkdtemp(path.join(os.tmpdir(), "daimon-fuse-usage-")); + try { + const setup = await WakeAcceptanceStore.open(root, testStoreOptions); + const parked = await setup.accept(parseWakeAcceptanceRequest(request("parked"))); + await setup.close(); + await writeFile(path.join(usage, "fuse.stop"), ""); + const core = new FakeCoreHost(); + const control = createOrganizationRuntimeControlHostWithCoreForTest(config, core, { + acceptanceStorePath: root, controlToken: token, storeOptions: testStoreOptions, + fuseEnvironment: fuseEnvironment(usage, 10) + }); + await control.start(); + assert.equal(core.wakes.length, 0); + assert.equal((await control.wakeReceipt(token, parked.record.acceptance_id))?.state, "stopped"); + await control.stop(); + } finally { await rm(root, { recursive: true, force: true }); await rm(usage, { recursive: true, force: true }); } +}); + +test("startup into a ceiling-tripped fuse terminalizes accepted recovery without dispatch", async () => { + const root = await privateRoot(); + const usage = await mkdtemp(path.join(os.tmpdir(), "daimon-fuse-usage-")); + try { + const setup = await WakeAcceptanceStore.open(root, testStoreOptions); + const parked = await setup.accept(parseWakeAcceptanceRequest(request("ceiling-parked"))); + await setup.close(); + const fuse = await WakeFuse.open({ organizationKey: "alpha", environment: fuseEnvironment(usage, 1) }); + await fuse.admit("alpha", "paid"); + assert.deepEqual(await fuse.admit("alpha", "trip"), { state: "tripped", reason: "wake_ceiling" }); + await fuse.close(); + const core = new FakeCoreHost(); + const control = createOrganizationRuntimeControlHostWithCoreForTest(config, core, { + acceptanceStorePath: root, controlToken: token, storeOptions: testStoreOptions, + fuseEnvironment: fuseEnvironment(usage, 1) + }); + await control.start(); + const items = (await control.activityV2(token))?.items ?? []; + assert.equal(core.wakes.length, 0); + assert.equal(items.filter((item) => item.state === "accepted").length, 0); + assert.equal((await control.wakeReceipt(token, parked.record.acceptance_id))?.state, "stopped"); + await control.stop(); + } finally { await rm(root, { recursive: true, force: true }); await rm(usage, { recursive: true, force: true }); } +}); + +test("startup fails before the core host when the fuse cannot open", async () => { + const root = await privateRoot(); + let starts = 0; + const core = new FakeCoreHost(); core.start = async () => { starts += 1; }; + const control = createOrganizationRuntimeControlHostWithCoreForTest(config, core, { + acceptanceStorePath: root, controlToken: token, storeOptions: testStoreOptions, + fuseEnvironment: fuseEnvironment(path.join(root, "missing"), 10) + }); + try { + await assert.rejects(control.start(), /ENOENT/); + assert.equal(starts, 0); + } finally { await rm(root, { recursive: true, force: true }); } +}); + +test("a rejecting polled trip does not become an unhandled rejection", async () => { + const root = await privateRoot(); + const usage = await mkdtemp(path.join(os.tmpdir(), "daimon-fuse-usage-")); + const unhandled: unknown[] = []; + const listener = (reason: unknown): void => { unhandled.push(reason); }; + process.on("unhandledRejection", listener); + let tripAttempted!: () => void; + const attempted = new Promise((resolve) => { tripAttempted = resolve; }); + const core = new FakeCoreHost(); + const control = createOrganizationRuntimeControlHostWithCoreForTest(config, core, { + acceptanceStorePath: root, controlToken: token, storeOptions: testStoreOptions, + fuseEnvironment: fuseEnvironment(usage, 10), fusePollIntervalMsForTest: 1, + beforeTripTerminalizationForTest: async () => { tripAttempted(); throw new Error("injected terminalization failure"); } + }); + try { + await control.start(); + await writeFile(path.join(usage, "fuse.stop"), ""); + await Promise.race([attempted, new Promise((_, reject) => setTimeout(() => reject(new Error("trip poll timed out")), 1_000))]); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(unhandled, []); + assert.deepEqual(await control.accept(request("after-partial-trip")), { version: "noopolis.daimon.wake-acceptance.v2", state: "stopped", code: "host_stopping" }); + } finally { + process.off("unhandledRejection", listener); + await control.stop().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + await rm(usage, { recursive: true, force: true }); + } +}); + test("schedule acceptance preserves its WakeEvent kind through the durable FIFO", async () => { const root = await privateRoot(); const core = new FakeCoreHost(); @@ -367,6 +587,9 @@ class FakeCoreHost implements Pick { const root = await mkdtemp(path.join(os.tmpdir(), "daimon-acceptance-")); await chmod(root, 0o700); return root; } +function fuseEnvironment(directory: string, maxWakes: number): NodeJS.ProcessEnv { + return { DAIMON_WAKE_FUSE_DIRECTORY: directory, DAIMON_TURN_USAGE_LEDGER_PATH: path.join(directory, "usage.jsonl"), DAIMON_WAKE_FUSE_EPOCH: "integration", DAIMON_WAKE_FUSE_MAX_WAKES: String(maxWakes), DAIMON_WAKE_FUSE_MAX_TOKENS: "1000000" }; +} async function waitFor(predicate: () => boolean | Promise, timeoutMs = 5_000): Promise { const deadline = Date.now() + timeoutMs; do { diff --git a/src/runtime/wakeAcceptanceStore.ts b/src/runtime/wakeAcceptanceStore.ts index 6ab9a5d..beb8f14 100644 --- a/src/runtime/wakeAcceptanceStore.ts +++ b/src/runtime/wakeAcceptanceStore.ts @@ -121,6 +121,18 @@ export class WakeAcceptanceStore { return this.serialize(async () => await this.transitionNow(acceptanceId, state, code)); } + /** Fuse race seam: never stops a record that was claimed after enumeration. */ + transitionAcceptedToStopped(acceptanceId: string): Promise { + return this.serialize(async () => { + const target = await this.pathForAcceptanceId(acceptanceId); + const prior = await this.read(target); + if (prior.state !== "accepted") return prior; + const record: Stored = { ...prior, state: "stopped", code: "host_stopping", updated_at: new Date().toISOString() }; + await this.replace(target, record); + return record; + }); + } + acquireClaim(acceptanceId: string, ownerId: string): Promise { return this.serialize(async () => { if (!uuid(ownerId)) throw new Error("wake execution owner is invalid"); diff --git a/src/runtime/wakeFuse.test.ts b/src/runtime/wakeFuse.test.ts new file mode 100644 index 0000000..f71b8fc --- /dev/null +++ b/src/runtime/wakeFuse.test.ts @@ -0,0 +1,165 @@ +import assert from "node:assert/strict"; +import { mkdir, mkdtemp, readFile, rm, unlink, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import test from "node:test"; + +import { WakeFuse, WAKE_FUSE_VERSION } from "./wakeFuse.js"; + +const withDirectory = async (body: (directory: string) => Promise): Promise => { + const directory = await mkdtemp(path.join(os.tmpdir(), "daimon-wake-fuse-")); + try { await body(directory); } finally { await rm(directory, { recursive: true, force: true }); } +}; +const environment = (directory: string, values: NodeJS.ProcessEnv = {}): NodeJS.ProcessEnv => ({ + DAIMON_WAKE_FUSE_DIRECTORY: directory, + DAIMON_WAKE_FUSE_EPOCH: "test-epoch", + DAIMON_WAKE_FUSE_MAX_WAKES: "2", + DAIMON_WAKE_FUSE_MAX_TOKENS: "1000", + DAIMON_TURN_USAGE_LEDGER_PATH: path.join(directory, "usage.jsonl"), + ...values +}); +const records = async (directory: string): Promise>> => + (await readFile(path.join(directory, "admissions.jsonl"), "utf8")).trim().split("\n").map((line) => JSON.parse(line) as Record); + +test("admission below the ceiling appends exactly one admission", async () => await withDirectory(async (directory) => { + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory) }); + assert.deepEqual(await fuse.admit("alpha", "one"), { state: "admitted" }); + assert.equal((await records(directory)).filter((record) => record.kind === "admission").length, 1); +})); + +test("the n+1 admission trips before append", async () => await withDirectory(async (directory) => { + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory) }); + await fuse.admit("alpha", "one"); await fuse.admit("alpha", "two"); + assert.deepEqual(await fuse.admit("alpha", "three"), { state: "tripped", reason: "wake_ceiling" }); + assert.equal((await records(directory)).filter((record) => record.kind === "admission").length, 2); +})); + +test("concurrent organization admissions cannot overshoot the wake ceiling", async () => await withDirectory(async (directory) => { + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_MAX_WAKES: "8" }) }); + const verdicts = await Promise.all(Array.from({ length: 9 }, (_, index) => fuse.admit(`agent-${index % 4}`, `delivery-${index}`))); + assert.equal(verdicts.filter((verdict) => verdict.state === "admitted").length, 8); + assert.ok(verdicts.some((verdict) => verdict.state === "tripped")); + assert.equal((await records(directory)).filter((record) => record.kind === "admission").length, 8); +})); + +test("token accounting excludes pre-epoch and skips malformed usage", async () => await withDirectory(async (directory) => { + const times = [new Date("2026-08-30T00:00:00.000Z"), new Date("2026-08-30T00:00:00.000Z")]; + await writeFile(path.join(directory, "usage.jsonl"), [ + JSON.stringify({ at: "2026-08-29T23:59:59.999Z", total: 5000 }), + "not-json", + JSON.stringify({ at: "2026-08-30T00:00:00.000Z", total: 1000 }) + ].join("\n") + "\n"); + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory), now: () => times.shift() ?? new Date("2026-08-30T00:00:00.000Z") }); + assert.deepEqual(await fuse.admit("alpha", "one"), { state: "tripped", reason: "token_ceiling" }); +})); + +test("pre-epoch token rows alone cannot trip the ceiling", async () => await withDirectory(async (directory) => { + await writeFile(path.join(directory, "usage.jsonl"), [ + JSON.stringify({ at: "2026-08-29T23:59:59.999Z", total: 5000 }), + JSON.stringify({ at: "2026-08-30T00:00:00.001Z", total: 10 }) + ].join("\n") + "\n"); + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_MAX_TOKENS: "100" }), now: () => new Date("2026-08-30T00:00:00.000Z") }); + assert.deepEqual(await fuse.admit("alpha", "one"), { state: "admitted" }); +})); + +test("same-epoch open reloads prior admissions", async () => await withDirectory(async (directory) => { + const first = await WakeFuse.open({ organizationKey: "org", environment: environment(directory) }); + await first.admit("alpha", "one"); + await first.admit("alpha", "two"); + await first.close(); + const reopened = await WakeFuse.open({ organizationKey: "org", environment: environment(directory) }); + assert.deepEqual(await reopened.admit("alpha", "three"), { state: "tripped", reason: "wake_ceiling" }); +})); + +test("a wake ceiling trip is restored with its reason in the same epoch", async () => await withDirectory(async (directory) => { + const first = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_MAX_WAKES: "1" }) }); + await first.admit("alpha", "one"); + assert.deepEqual(await first.admit("alpha", "two"), { state: "tripped", reason: "wake_ceiling" }); + const reopened = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_MAX_WAKES: "1" }) }); + assert.equal(reopened.tripped(), "wake_ceiling"); +})); + +test("a corrupt trip marker fails closed", async () => await withDirectory(async (directory) => { + await writeFile(path.join(directory, "fuse.trip.json"), "not-json\n"); + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory) }); + assert.equal(fuse.tripped(), "ledger_unavailable"); +})); + +test("a valid trip marker from a previous epoch does not trip a new epoch", async () => await withDirectory(async (directory) => { + const first = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_EPOCH: "old", DAIMON_WAKE_FUSE_MAX_WAKES: "1" }) }); + await first.admit("alpha", "one"); + await first.admit("alpha", "two"); + const fresh = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_EPOCH: "new", DAIMON_WAKE_FUSE_MAX_WAKES: "1" }) }); + assert.equal(fresh.tripped(), undefined); +})); + +test("token accounting follows the relocated usage ledger", async () => await withDirectory(async (directory) => { + const elsewhere = path.join(directory, "elsewhere.jsonl"); + await writeFile(elsewhere, `${JSON.stringify({ at: "2026-08-30T00:00:00.000Z", total: 1000 })}\n`); + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_TURN_USAGE_LEDGER_PATH: elsewhere }), now: () => new Date("2026-08-30T00:00:00.000Z") }); + assert.deepEqual(await fuse.admit("alpha", "one"), { state: "tripped", reason: "token_ceiling" }); +})); + +test("a malformed usage line alone is skipped", async () => await withDirectory(async (directory) => { + await writeFile(path.join(directory, "usage.jsonl"), "broken\n"); + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory) }); + assert.deepEqual(await fuse.admit("alpha", "one"), { state: "admitted" }); +})); + +test("admissions from another epoch do not count", async () => await withDirectory(async (directory) => { + await writeFile(path.join(directory, "admissions.jsonl"), `${JSON.stringify({ v: WAKE_FUSE_VERSION, kind: "admission", epoch: "old", at: new Date().toISOString(), agent: "alpha", delivery: "old" })}\n`); + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_MAX_WAKES: "1" }) }); + assert.deepEqual(await fuse.admit("alpha", "new"), { state: "admitted" }); +})); + +test("operator stop trips on the next admission", async () => await withDirectory(async (directory) => { + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory) }); + await writeFile(path.join(directory, "fuse.stop"), "ignored"); + assert.deepEqual(await fuse.admit("alpha", "one"), { state: "tripped", reason: "operator_stop" }); +})); + +test("an append failure refuses admission and fails closed", async () => await withDirectory(async (directory) => { + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory) }); + await unlink(path.join(directory, "admissions.jsonl")); + await mkdir(path.join(directory, "admissions.jsonl")); + assert.deepEqual(await fuse.admit("alpha", "one"), { state: "tripped", reason: "ledger_unavailable" }); +})); + +test("invalid wake ceilings throw instead of applying defaults", async () => await withDirectory(async (directory) => { + for (const value of ["0", "-1", "1.5", "abc"]) { + await assert.rejects(WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_MAX_WAKES: value }) }), /positive integer/); + } +})); + +test("off admits unconditionally without storage and every other setting is rejected", async () => await withDirectory(async (directory) => { + const missing = path.join(directory, "missing"); + const warnings: string[] = []; + const original = console.error; + console.error = (...values: unknown[]) => { warnings.push(values.join(" ")); }; + try { + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(missing, { DAIMON_WAKE_FUSE: "off", DAIMON_WAKE_FUSE_MAX_WAKES: "1" }) }); + await WakeFuse.open({ organizationKey: "org", environment: environment(missing, { DAIMON_WAKE_FUSE: "off" }) }); + assert.deepEqual(await fuse.admit("alpha", "one"), { state: "admitted" }); + assert.deepEqual(await fuse.admit("alpha", "two"), { state: "admitted" }); + } finally { console.error = original; } + assert.deepEqual(warnings, ["DAIMON WAKE FUSE IS OFF: wake admission is unbounded"]); + await assert.rejects(WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE: "on" }) }), /exactly 'off'/); +})); + +test("once tripped the fuse never admits again", async () => await withDirectory(async (directory) => { + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_MAX_WAKES: "1" }) }); + await fuse.admit("alpha", "one"); + assert.equal((await fuse.admit("alpha", "two")).state, "tripped"); + assert.equal((await fuse.admit("alpha", "one")).state, "tripped"); +})); + +test("a failed trip-marker write is retried without reopening admission", async () => await withDirectory(async (directory) => { + const fuse = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_MAX_WAKES: "1" }) }); + await fuse.admit("alpha", "one"); + await mkdir(path.join(directory, "fuse.trip.json")); + assert.deepEqual(await fuse.admit("alpha", "two"), { state: "tripped", reason: "wake_ceiling" }); + await rm(path.join(directory, "fuse.trip.json"), { recursive: true }); + assert.deepEqual(await fuse.admit("alpha", "three"), { state: "tripped", reason: "wake_ceiling" }); + const reopened = await WakeFuse.open({ organizationKey: "org", environment: environment(directory, { DAIMON_WAKE_FUSE_MAX_WAKES: "1" }) }); + assert.equal(reopened.tripped(), "wake_ceiling"); +})); diff --git a/src/runtime/wakeFuse.ts b/src/runtime/wakeFuse.ts new file mode 100644 index 0000000..0b3c80c --- /dev/null +++ b/src/runtime/wakeFuse.ts @@ -0,0 +1,221 @@ +import { constants } from "node:fs"; +import { open, readFile, readdir, rename, stat, unlink } from "node:fs/promises"; +import path from "node:path"; +import { createHash, randomUUID } from "node:crypto"; + +import { resolveTurnUsageLedgerPath, TURN_USAGE_LEDGER, TURN_USAGE_MAX_IDENTIFIER_CHARS, TURN_USAGE_ROTATE_BYTES } from "./turnUsageLedger.js"; + +export const WAKE_FUSE_VERSION = "noopolis.daimon.wake-fuse.v1" as const; +export const WAKE_FUSE_DIRECTORY_ENV = "DAIMON_WAKE_FUSE_DIRECTORY" as const; +export const DEFAULT_WAKE_FUSE_MAX_WAKES = 100; +export const DEFAULT_WAKE_FUSE_MAX_TOKENS = 5_000_000; + +export type WakeFuseTripReason = "wake_ceiling" | "token_ceiling" | "operator_stop" | "ledger_unavailable"; +export type WakeFuseVerdict = Readonly<{ state: "admitted" }> | Readonly<{ state: "tripped"; reason: WakeFuseTripReason }>; +export type WakeFuseOptions = Readonly<{ + organizationKey: string; + environment?: NodeJS.ProcessEnv; + now?: () => Date; +}>; + +type Admission = Readonly<{ v: typeof WAKE_FUSE_VERSION; kind: "admission"; epoch: string; at: string; agent: string; delivery: string }>; +type EpochStart = Readonly<{ v: typeof WAKE_FUSE_VERSION; kind: "epoch_start"; epoch: string; at: string }>; +type TripMarker = Readonly<{ v: typeof WAKE_FUSE_VERSION; kind: "trip"; epoch: string; at: string; reason: WakeFuseTripReason }>; +const TRIP_MARKER = "fuse.trip.json"; +let warnedDisarmed = false; + +/** + * Durable catastrophic-wake admission fuse for one organization/container. + * + * Operator trip: `docker exec touch /var/lib/spawnfile/daimon/usage/fuse.stop`. + * Unlike `recordTurnUsage` ("advisory: never rejects"), every storage or + * evaluation failure here rejects admission: this safety boundary fails closed. + */ +export class WakeFuse { + private serial: Promise = Promise.resolve(); + private reason: WakeFuseTripReason | undefined; + private tripMarkerMissing = false; + + private constructor( + private readonly armed: boolean, + private readonly directory: string, + private readonly epoch: string, + private readonly epochStartedAt: string, + private readonly maxWakes: number, + private readonly maxTokens: number, + private readonly admissions: Set, + private readonly usageLedgerPath: string, + private readonly now: () => Date + ) {} + + static async open(options: WakeFuseOptions): Promise { + const environment = options.environment ?? process.env; + const setting = environment.DAIMON_WAKE_FUSE; + if (setting !== undefined && setting !== "off") throw new Error("DAIMON_WAKE_FUSE must be exactly 'off' when set"); + if (environment.DAIMON_WAKE_FUSE_EPOCH !== undefined && nonBlank(environment.DAIMON_WAKE_FUSE_EPOCH) === undefined) throw new Error("DAIMON_WAKE_FUSE_EPOCH must be non-blank"); + const epoch = nonBlank(environment.DAIMON_WAKE_FUSE_EPOCH) + // Agent ids are the stable organization identity available to Daimon. + ?? `organization-${createHash("sha256").update(options.organizationKey).digest("hex")}`; + const maxWakes = ceiling(environment.DAIMON_WAKE_FUSE_MAX_WAKES, DEFAULT_WAKE_FUSE_MAX_WAKES, "DAIMON_WAKE_FUSE_MAX_WAKES"); + const maxTokens = ceiling(environment.DAIMON_WAKE_FUSE_MAX_TOKENS, DEFAULT_WAKE_FUSE_MAX_TOKENS, "DAIMON_WAKE_FUSE_MAX_TOKENS"); + const directory = resolveWakeFuseDirectory(environment); + const now = options.now ?? (() => new Date()); + const usageLedgerPath = resolveTurnUsageLedgerPath(environment); + if (setting === "off") { + if (!warnedDisarmed) { + warnedDisarmed = true; + console.error("DAIMON WAKE FUSE IS OFF: wake admission is unbounded"); + } + return new WakeFuse(false, directory, epoch, now().toISOString(), maxWakes, maxTokens, new Set(), usageLedgerPath, now); + } + + // Unbounded defaults would reproduce exactly the failure this module prevents. + await readdir(directory); + const records = await readFuseRecords(directory); + const existingStart = records.filter((record): record is EpochStart => record.kind === "epoch_start" && record.epoch === epoch).at(-1); + const epochStartedAt = existingStart?.at ?? now().toISOString(); + if (existingStart === undefined) await append(directory, { v: WAKE_FUSE_VERSION, kind: "epoch_start", epoch, at: epochStartedAt }); + const admissions = new Set(records + .filter((record): record is Admission => record.kind === "admission" && record.epoch === epoch) + .map((record) => key(record.agent, record.delivery))); + const fuse = new WakeFuse(true, directory, epoch, epochStartedAt, maxWakes, maxTokens, admissions, usageLedgerPath, now); + if (await exists(path.join(directory, "fuse.stop"))) fuse.reason = "operator_stop"; + else fuse.reason = await readTripMarker(directory, epoch); + return fuse; + } + + admit(agentId: string, deliveryId: string): Promise { + if (!this.armed) return Promise.resolve({ state: "admitted" }); + // One organization is one container/control-host process. This promise + // chain is organization-wide in-process serialization, not cross-process locking. + const result = this.serial.catch(() => undefined).then(async () => await this.admitNow(agentId, deliveryId)); + this.serial = result.then(() => undefined, () => undefined); + return result; + } + + async pollOperatorStop(): Promise { + if (!this.armed || this.reason !== undefined) return this.reason; + try { + if (await exists(path.join(this.directory, "fuse.stop"))) await this.trip("operator_stop"); + } catch { await this.trip("ledger_unavailable"); } + return this.reason; + } + + tripped(): WakeFuseTripReason | undefined { return this.reason; } + async close(): Promise { await this.serial; } + + private async admitNow(agentId: string, deliveryId: string): Promise { + if (this.reason !== undefined) return await this.trip(this.reason); + try { + if (await exists(path.join(this.directory, "fuse.stop"))) return await this.trip("operator_stop"); + const admissionKey = key(bounded(agentId), bounded(deliveryId)); + if (this.admissions.has(admissionKey)) return { state: "admitted" }; + if (this.admissions.size >= this.maxWakes) return await this.trip("wake_ceiling"); + // Lagging indicator: usage is written after turn completion, so in-flight + // spend can overshoot by one concurrent round. The wake ceiling bounds it; + // design §5.2 owns the later pre-spawn reservation. + if (await sumTokens(this.usageLedgerPath, this.epochStartedAt) >= this.maxTokens) return await this.trip("token_ceiling"); + const record: Admission = { v: WAKE_FUSE_VERSION, kind: "admission", epoch: this.epoch, at: this.now().toISOString(), agent: bounded(agentId), delivery: bounded(deliveryId) }; + await append(this.directory, record); + this.admissions.add(admissionKey); + return { state: "admitted" }; + } catch { return await this.trip("ledger_unavailable"); } + } + + private async trip(reason: WakeFuseTripReason): Promise { + if (this.reason === undefined) { + this.reason = reason; + this.tripMarkerMissing = true; + } + if (this.tripMarkerMissing) { + try { + await writeTripMarker(this.directory, { v: WAKE_FUSE_VERSION, kind: "trip", epoch: this.epoch, at: this.now().toISOString(), reason: this.reason }); + this.tripMarkerMissing = false; + } catch { + // Keep the in-memory fuse closed and retry durability on every later + // admit/trip. Marker I/O must never turn a tripped verdict into an + // unhandled rejection at the HTTP boundary. + } + } + return { state: "tripped", reason: this.reason }; + } +} + +export function resolveWakeFuseDirectory(environment: NodeJS.ProcessEnv = process.env): string { + const override = environment[WAKE_FUSE_DIRECTORY_ENV]?.trim(); + return override !== undefined && override.startsWith("/") ? override : TURN_USAGE_LEDGER.directoryPath; +} + +function ceiling(value: string | undefined, fallback: number, name: string): number { + if (value === undefined) return fallback; + const parsed = Number(value); + if (!Number.isSafeInteger(parsed) || parsed <= 0) throw new Error(`${name} must be a positive integer`); + return parsed; +} +function nonBlank(value: string | undefined): string | undefined { const trimmed = value?.trim(); return trimmed ? trimmed : undefined; } +function bounded(value: string): string { return [...value].slice(0, TURN_USAGE_MAX_IDENTIFIER_CHARS).join(""); } +function key(agent: string, delivery: string): string { return `${agent}\u0000${delivery}`; } +function admissionsPath(directory: string): string { return path.join(directory, "admissions.jsonl"); } + +async function append(directory: string, record: Admission | EpochStart): Promise { + const file = admissionsPath(directory); + try { if ((await stat(file)).size >= TURN_USAGE_ROTATE_BYTES) await rename(file, `${file}.1`); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; } + const bytes = Buffer.from(`${JSON.stringify(record)}\n`, "utf8"); + const handle = await open(file, constants.O_WRONLY | constants.O_APPEND | constants.O_CREAT | constants.O_NOFOLLOW, TURN_USAGE_LEDGER.fileMode); + try { const result = await handle.write(bytes, 0, bytes.length); if (result.bytesWritten !== bytes.length) throw new Error("wake fuse admission append was torn"); } + finally { await handle.close(); } +} + +async function readFuseRecords(directory: string): Promise> { + const records: Array = []; + for (const file of [`${admissionsPath(directory)}.1`, admissionsPath(directory)]) { + for (const line of await lines(file)) { + try { + const value = JSON.parse(line) as Partial; + if (value.v !== WAKE_FUSE_VERSION || typeof value.epoch !== "string" || typeof value.at !== "string" || Number.isNaN(Date.parse(value.at))) continue; + if (value.kind === "epoch_start") records.push(value as EpochStart); + else if (value.kind === "admission" && typeof value.agent === "string" && typeof value.delivery === "string") records.push(value as Admission); + } catch { /* malformed historical fuse lines do not count */ } + } + } + return records; +} + +async function sumTokens(ledgerPath: string, since: string): Promise { + let total = 0; + for (const file of [`${ledgerPath}.1`, ledgerPath]) { + for (const line of await lines(file)) { + try { + const value = JSON.parse(line) as { at?: unknown; total?: unknown }; + if (typeof value.at === "string" && !Number.isNaN(Date.parse(value.at)) && value.at >= since && typeof value.total === "number" && Number.isFinite(value.total) && value.total >= 0) total += value.total; + } catch { /* usage accounting is advisory input; malformed lines are skipped */ } + } + } + return total; +} +async function writeTripMarker(directory: string, marker: TripMarker): Promise { + const target = path.join(directory, TRIP_MARKER); + const temporary = path.join(directory, `.${TRIP_MARKER}.${process.pid}.${randomUUID()}`); + const bytes = Buffer.from(`${JSON.stringify(marker)}\n`, "utf8"); + try { + const handle = await open(temporary, constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, TURN_USAGE_LEDGER.fileMode); + try { const result = await handle.write(bytes, 0, bytes.length); if (result.bytesWritten !== bytes.length) throw new Error("wake fuse trip marker write was torn"); } + finally { await handle.close(); } + await rename(temporary, target); + } finally { await unlink(temporary).catch(() => undefined); } +} +async function readTripMarker(directory: string, epoch: string): Promise { + try { + const value = JSON.parse(await readFile(path.join(directory, TRIP_MARKER), "utf8")) as Partial; + if (value.v !== WAKE_FUSE_VERSION || value.kind !== "trip" || typeof value.epoch !== "string" || typeof value.at !== "string" || Number.isNaN(Date.parse(value.at)) || !isTripReason(value.reason)) return "ledger_unavailable"; + // A valid marker belongs to one explicit counting window. Selecting a new + // epoch deliberately clears it; a corrupt marker cannot prove that and fails closed. + return value.epoch === epoch ? value.reason : undefined; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; + return "ledger_unavailable"; + } +} +function isTripReason(value: unknown): value is WakeFuseTripReason { return value === "wake_ceiling" || value === "token_ceiling" || value === "operator_stop" || value === "ledger_unavailable"; } +async function lines(file: string): Promise { try { return (await readFile(file, "utf8")).split("\n").filter(Boolean); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return []; throw error; } } +async function exists(file: string): Promise { try { await stat(file); return true; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; throw error; } }