From 8b49e9913ccf63efaae67f32086fa31b991ef8b2 Mon Sep 17 00:00:00 2001 From: nategarelik Date: Sat, 11 Jul 2026 19:02:52 -0500 Subject: [PATCH 1/6] fix: stop wrapping Windows spawns in process.env.SHELL #236 reports the app-server broker hanging indefinitely at "Initializing..." on Windows with no error surfaced. The direct spawn path (used by the broker itself to launch `codex app-server`, and by runCommand for utilities like taskkill) wraps the child process with process.env.SHELL on win32, trusting whatever shell the user happens to have configured (git-bash, a WSL bash.exe shim, an unresolvable literal POSIX path, etc.) to correctly proxy a JSON-RPC stdio pipe. That is unpredictable: it can silently swallow the handshake, fail with ENOENT, or otherwise misbehave depending on what SHELL resolves to, and the broker's own cleanup logic already assumed the spawned child was cmd.exe (see the terminateProcessTree comment in app-server.mjs), which breaks when SHELL points elsewhere. process.env.SHELL was added in #178 to fix #138, but that fix targeted the wrong layer: the outer spawn wrapper only resolves the codex executable itself (npm installs ship a .cmd shim that CreateProcess cannot exec directly) and has no bearing on what shell Codex uses internally for its own command-execution tool, since SHELL is already passed straight through via the `env` option regardless of this wrapper. Reverting to a deterministic cmd.exe wrapper on Windows fixes the hang without reopening #138. Verified: baseline test suite has 6 pre-existing failures including "cancel sends turn interrupt to the shared app-server before killing a brokered task" (a taskkill/terminateProcessTree test whose process tree assumptions broke exactly the way described above); with this fix that test passes and no other test regresses (5 failures remain, all pre-existing and unrelated: platform-mismatch path assertions, npm/setup env assumptions, Claude-session-transfer path assumptions). A live headless round trip against `codex app-server` (broker start, initialize, thread/start, turn/start, turn/completed, clean shutdown) also passes end to end. --- plugins/codex/scripts/lib/app-server.mjs | 10 +++++++++- plugins/codex/scripts/lib/process.mjs | 5 ++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/plugins/codex/scripts/lib/app-server.mjs b/plugins/codex/scripts/lib/app-server.mjs index 72b30a76..4dd59b2c 100644 --- a/plugins/codex/scripts/lib/app-server.mjs +++ b/plugins/codex/scripts/lib/app-server.mjs @@ -191,7 +191,15 @@ class SpawnedCodexAppServerClient extends AppServerClientBase { cwd: this.cwd, env: this.options.env ?? process.env, stdio: ["pipe", "pipe", "pipe"], - shell: process.platform === "win32" ? (process.env.SHELL || true) : false, + // Always use the deterministic Windows shell (cmd.exe) to wrap this spawn, never + // process.env.SHELL. This wrapper only resolves the codex executable (npm installs + // ship a .cmd shim that CreateProcess cannot exec directly); it has no bearing on + // what shell Codex uses internally for its own command-execution tool, since SHELL + // is passed through via `env` above regardless of this option. Trusting an arbitrary + // user-configured SHELL here (git-bash, WSL's bash.exe shim, an unresolvable literal + // path, etc.) to wrap a JSON-RPC stdio pipe is unpredictable and can hang the + // handshake indefinitely with no surfaced error (#236). See also #138/#178. + shell: process.platform === "win32", windowsHide: true }); diff --git a/plugins/codex/scripts/lib/process.mjs b/plugins/codex/scripts/lib/process.mjs index dd8fc375..0fa6fcac 100644 --- a/plugins/codex/scripts/lib/process.mjs +++ b/plugins/codex/scripts/lib/process.mjs @@ -9,7 +9,10 @@ export function runCommand(command, args = [], options = {}) { input: options.input, maxBuffer: options.maxBuffer, stdio: options.stdio ?? "pipe", - shell: options.shell ?? (process.platform === "win32" ? (process.env.SHELL || true) : false), + // See app-server.mjs for why this stays off process.env.SHELL: an arbitrary + // user-configured shell is not a safe or predictable wrapper for spawning + // subprocesses on Windows (#236). + shell: options.shell ?? process.platform === "win32", windowsHide: true }); From 31090c056bb84f68edead339835aaee582a89819 Mon Sep 17 00:00:00 2001 From: nategarelik Date: Sat, 11 Jul 2026 19:15:45 -0500 Subject: [PATCH 2/6] fix: resolve cmd.exe via SystemRoot instead of trusting ComSpec Cross-model adversarial review of 8b49e99 found a remaining gap: on Windows, `shell: true` doesn't invoke cmd.exe directly -- Node falls back to process.env.ComSpec (defaulting to cmd.exe only when ComSpec is unset). A nonstandard ComSpec (pointed at PowerShell, a shell shim, etc.) can therefore reproduce the exact #236 hang through ComSpec that 8b49e99 already fixed for SHELL. Add resolveWindowsShell() in process.mjs, which joins SystemRoot (set by the OS itself, not a dev-tooling convention like SHELL/ComSpec) with System32\cmd.exe, and use it at both spawn sites instead of the plain `shell: true` boolean. Neither SHELL nor ComSpec can redirect the wrapper now. runCommand's explicit `options.shell` override is preserved. Verified: full test suite matches the previous fix's pass/fail split (the two extra failures seen during one run were load-induced flakes under heavy concurrent-agent CPU contention on this machine -- 300+ node.exe processes at the time -- confirmed by re-running both in isolation, where they pass cleanly). Live headless round trip against codex app-server (thread/start, turn/start, turn/completed, clean shutdown) passes end to end with this build. --- plugins/codex/scripts/lib/app-server.mjs | 23 +++++++++++++---------- plugins/codex/scripts/lib/process.mjs | 19 +++++++++++++++---- 2 files changed, 28 insertions(+), 14 deletions(-) diff --git a/plugins/codex/scripts/lib/app-server.mjs b/plugins/codex/scripts/lib/app-server.mjs index 4dd59b2c..80c60342 100644 --- a/plugins/codex/scripts/lib/app-server.mjs +++ b/plugins/codex/scripts/lib/app-server.mjs @@ -14,7 +14,7 @@ import { spawn } from "node:child_process"; import readline from "node:readline"; import { parseBrokerEndpoint } from "./broker-endpoint.mjs"; import { ensureBrokerSession, loadBrokerSession } from "./broker-lifecycle.mjs"; -import { terminateProcessTree } from "./process.mjs"; +import { resolveWindowsShell, terminateProcessTree } from "./process.mjs"; const PLUGIN_MANIFEST_URL = new URL("../../.claude-plugin/plugin.json", import.meta.url); const PLUGIN_MANIFEST = JSON.parse(fs.readFileSync(PLUGIN_MANIFEST_URL, "utf8")); @@ -191,15 +191,18 @@ class SpawnedCodexAppServerClient extends AppServerClientBase { cwd: this.cwd, env: this.options.env ?? process.env, stdio: ["pipe", "pipe", "pipe"], - // Always use the deterministic Windows shell (cmd.exe) to wrap this spawn, never - // process.env.SHELL. This wrapper only resolves the codex executable (npm installs - // ship a .cmd shim that CreateProcess cannot exec directly); it has no bearing on - // what shell Codex uses internally for its own command-execution tool, since SHELL - // is passed through via `env` above regardless of this option. Trusting an arbitrary - // user-configured SHELL here (git-bash, WSL's bash.exe shim, an unresolvable literal - // path, etc.) to wrap a JSON-RPC stdio pipe is unpredictable and can hang the - // handshake indefinitely with no surfaced error (#236). See also #138/#178. - shell: process.platform === "win32", + // Always use the deterministic Windows shell (System32\cmd.exe, resolved via + // resolveWindowsShell) to wrap this spawn, never process.env.SHELL or + // process.env.ComSpec (plain `shell: true` falls back to the latter). This wrapper + // only resolves the codex executable (npm installs ship a .cmd shim that + // CreateProcess cannot exec directly); it has no bearing on what shell Codex uses + // internally for its own command-execution tool, since SHELL is passed through via + // `env` above regardless of this option. Trusting an arbitrary user-configured + // SHELL or ComSpec here (git-bash, WSL's bash.exe shim, an unresolvable literal + // path, a non-cmd.exe ComSpec, etc.) to wrap a JSON-RPC stdio pipe is unpredictable + // and can hang the handshake indefinitely with no surfaced error (#236). See also + // #138/#178. + shell: process.platform === "win32" ? resolveWindowsShell() : false, windowsHide: true }); diff --git a/plugins/codex/scripts/lib/process.mjs b/plugins/codex/scripts/lib/process.mjs index 0fa6fcac..409c732d 100644 --- a/plugins/codex/scripts/lib/process.mjs +++ b/plugins/codex/scripts/lib/process.mjs @@ -1,6 +1,20 @@ import { spawnSync } from "node:child_process"; +import path from "node:path"; import process from "node:process"; +// Resolve the Windows command interpreter deterministically. Passing `shell: true` on +// Windows would otherwise let Node fall back to process.env.ComSpec, and this project +// intentionally never consults process.env.SHELL either -- both are arbitrary, +// user-configurable env vars that are not a safe or predictable wrapper for spawning a +// subprocess on Windows (#236: a nonstandard SHELL or ComSpec can silently break the +// spawn wrapper for a JSON-RPC stdio pipe). SystemRoot is set by the OS itself, so +// resolving System32\cmd.exe through it keeps this wrapper deterministic regardless of +// either env var. See app-server.mjs for the codex app-server spawn that shares this. +export function resolveWindowsShell() { + const systemRoot = process.env.SystemRoot || "C:\\Windows"; + return path.join(systemRoot, "System32", "cmd.exe"); +} + export function runCommand(command, args = [], options = {}) { const result = spawnSync(command, args, { cwd: options.cwd, @@ -9,10 +23,7 @@ export function runCommand(command, args = [], options = {}) { input: options.input, maxBuffer: options.maxBuffer, stdio: options.stdio ?? "pipe", - // See app-server.mjs for why this stays off process.env.SHELL: an arbitrary - // user-configured shell is not a safe or predictable wrapper for spawning - // subprocesses on Windows (#236). - shell: options.shell ?? process.platform === "win32", + shell: options.shell ?? (process.platform === "win32" ? resolveWindowsShell() : false), windowsHide: true }); From 59c930645c2bf3988abd1507508700d1e7dcb4ba Mon Sep 17 00:00:00 2001 From: nategarelik Date: Sat, 11 Jul 2026 19:28:02 -0500 Subject: [PATCH 3/6] feat: add rig-edition tier routing, envelope validation, and quota failover Adds an additive lib/rig-edition.mjs layer composing the existing lib/codex.mjs turn path: sol/terra/luna model-tier resolution with per-verb defaults, envelope parse/validate helpers matching the DONE|DONE_WITH_CONCERNS|NEEDS_CONTEXT|BLOCKED contract, and one-step quota failover (sol -> terra -> luna, then BLOCKED/QUOTA_EXHAUSTED). codex-companion.mjs gains an optional --tier flag on `task` that resolves model/effort through the new tier table; omitting --tier keeps the prior --model/--effort behavior byte-for-byte. --- plugins/codex/scripts/codex-companion.mjs | 13 +- plugins/codex/scripts/lib/rig-edition.mjs | 370 ++++++++++++++++++++++ tests/rig-edition.test.mjs | 312 ++++++++++++++++++ 3 files changed, 691 insertions(+), 4 deletions(-) create mode 100644 plugins/codex/scripts/lib/rig-edition.mjs create mode 100644 tests/rig-edition.test.mjs diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs index 83df468a..9cc6ddf8 100644 --- a/plugins/codex/scripts/codex-companion.mjs +++ b/plugins/codex/scripts/codex-companion.mjs @@ -7,6 +7,7 @@ import process from "node:process"; import { fileURLToPath } from "node:url"; import { parseArgs, splitRawArgumentString } from "./lib/args.mjs"; +import { tierDefaultsForVerb } from "./lib/rig-edition.mjs"; import { buildPersistentTaskThreadName, DEFAULT_CONTINUE_PROMPT, @@ -79,7 +80,7 @@ function printUsage() { " node scripts/codex-companion.mjs setup [--enable-review-gate|--disable-review-gate] [--json]", " node scripts/codex-companion.mjs review [--wait|--background] [--base ] [--scope ]", " node scripts/codex-companion.mjs adversarial-review [--wait|--background] [--base ] [--scope ] [focus text]", - " node scripts/codex-companion.mjs task [--background] [--write] [--resume-last|--resume|--fresh] [--model ] [--effort ] [prompt]", + " node scripts/codex-companion.mjs task [--background] [--write] [--resume-last|--resume|--fresh] [--model ] [--effort ] [--tier ] [prompt]", " node scripts/codex-companion.mjs transfer [--source ] [--json]", " node scripts/codex-companion.mjs status [job-id] [--all] [--json]", " node scripts/codex-companion.mjs result [job-id] [--json]", @@ -761,7 +762,7 @@ async function handleReview(argv) { async function handleTask(argv) { const { options, positionals } = parseCommandInput(argv, { - valueOptions: ["model", "effort", "cwd", "prompt-file"], + valueOptions: ["model", "effort", "cwd", "prompt-file", "tier"], booleanOptions: ["json", "write", "resume-last", "resume", "fresh", "background"], aliasMap: { m: "model" @@ -770,8 +771,12 @@ async function handleTask(argv) { const cwd = resolveCommandCwd(options); const workspaceRoot = resolveCommandWorkspace(options); - const model = normalizeRequestedModel(options.model); - const effort = normalizeReasoningEffort(options.effort); + // --tier resolves model + effort via the rig-edition tier table, bypassing + // the legacy --model/--effort normalizers below. Omit --tier to keep the + // prior behavior unchanged. + const tierDefaults = options.tier ? tierDefaultsForVerb("delegate", { tier: options.tier, effort: options.effort }) : null; + const model = tierDefaults ? tierDefaults.modelId : normalizeRequestedModel(options.model); + const effort = tierDefaults ? tierDefaults.effort : normalizeReasoningEffort(options.effort); const prompt = readTaskPrompt(cwd, options, positionals); const resumeLast = Boolean(options["resume-last"] || options.resume); diff --git a/plugins/codex/scripts/lib/rig-edition.mjs b/plugins/codex/scripts/lib/rig-edition.mjs new file mode 100644 index 00000000..7c445dc7 --- /dev/null +++ b/plugins/codex/scripts/lib/rig-edition.mjs @@ -0,0 +1,370 @@ +/** + * Rig-edition additive layer: model-tier routing, envelope validation, and + * quota failover for Codex dispatches. + * + * This module is deliberately additive -- it composes the existing + * lib/codex.mjs entry points rather than changing their signatures, so + * upstream rebases of codex.mjs stay cheap. New callers opt in by importing + * from here. + * + * @typedef {"sol" | "terra" | "luna"} ModelTier + * @typedef {"low" | "medium" | "high" | "xhigh" | "max" | "ultra"} ReasoningEffort + * @typedef {"DONE" | "DONE_WITH_CONCERNS" | "NEEDS_CONTEXT" | "BLOCKED"} EnvelopeStatus + * @typedef {{ + * status: EnvelopeStatus, + * summary: string, + * files_modified: string[], + * concerns: string[], + * blocked_reason: string | null + * }} Envelope + */ + +import { runAppServerTurn } from "./codex.mjs"; + +/** Canonical tier -> full model id, per verified-facts-2026-07-11.md. */ +export const MODEL_ALIASES = Object.freeze({ + sol: "gpt-5.6-sol", + terra: "gpt-5.6-terra", + luna: "gpt-5.6-luna" +}); + +const TIER_ORDER = Object.freeze(["sol", "terra", "luna"]); +const TIER_STEP_DOWN = Object.freeze({ sol: "terra", terra: "luna", luna: null }); +const MODEL_ID_TO_TIER = new Map(TIER_ORDER.map((tier) => [MODEL_ALIASES[tier], tier])); + +const VALID_EFFORTS = new Set(["low", "medium", "high", "xhigh", "max", "ultra"]); +const SOL_ONLY_EFFORTS = new Set(["max", "ultra"]); + +const QUOTA_EXHAUSTED_REASON = "QUOTA_EXHAUSTED"; +const QUOTA_ERROR_PATTERNS = [ + /\b429\b/, + /rate[\s_-]?limit/i, + /insufficient[\s_-]?quota/i, + /quota[\s_-]?exceeded/i, + /too many requests/i +]; + +function normalizeKey(value) { + if (value == null) { + return ""; + } + return String(value).trim().toLowerCase(); +} + +/** + * Resolves a tier alias ("sol"/"terra"/"luna") or a full model id + * ("gpt-5.6-sol") to its canonical { tier, modelId } pair. + * @param {string | null | undefined} aliasOrModelId + * @returns {{ tier: ModelTier, modelId: string } | null} + */ +export function resolveTier(aliasOrModelId) { + const normalized = normalizeKey(aliasOrModelId); + if (!normalized) { + return null; + } + if (TIER_ORDER.includes(normalized)) { + return { tier: normalized, modelId: MODEL_ALIASES[normalized] }; + } + const tierFromModelId = MODEL_ID_TO_TIER.get(normalized); + return tierFromModelId ? { tier: tierFromModelId, modelId: MODEL_ALIASES[tierFromModelId] } : null; +} + +/** + * Reports whether a reasoning effort value is usable on the given tier. + * max/ultra are sol-only per the verified 0.144.0 catalog. + * @param {ModelTier} tier + * @param {string} effort + */ +export function isEffortValidForTier(tier, effort) { + if (!VALID_EFFORTS.has(effort)) { + return false; + } + return !(SOL_ONLY_EFFORTS.has(effort) && tier !== "sol"); +} + +/** + * Returns the next tier down the failover chain (sol -> terra -> luna -> null). + * @param {ModelTier} tier + * @returns {ModelTier | null} + */ +export function nextTierDown(tier) { + return TIER_STEP_DOWN[tier] ?? null; +} + +/** Per-verb tier + effort defaults, overridable per call via tierDefaultsForVerb. */ +export const PER_VERB_TIER_DEFAULTS = Object.freeze({ + delegate: Object.freeze({ tier: "terra", effort: "medium" }), + implement: Object.freeze({ tier: "terra", effort: "medium" }), + review: Object.freeze({ tier: "sol", effort: "high" }), + adversarial: Object.freeze({ tier: "sol", effort: "high" }), + bulk: Object.freeze({ tier: "luna", effort: "low" }), + mechanical: Object.freeze({ tier: "luna", effort: "low" }) +}); + +/** + * Looks up the tier + effort default for a verb, applying any per-call + * overrides. Throws on an unknown verb, an unresolvable tier override, or + * an effort that is not valid for the resolved tier. + * @param {string} verb + * @param {{ tier?: string, effort?: string }} [overrides] + * @returns {{ tier: ModelTier, modelId: string, effort: string }} + */ +export function tierDefaultsForVerb(verb, overrides = {}) { + const base = PER_VERB_TIER_DEFAULTS[normalizeKey(verb)]; + if (!base) { + throw new Error(`Unknown verb "${verb}". Use one of: ${Object.keys(PER_VERB_TIER_DEFAULTS).join(", ")}.`); + } + + const resolved = resolveTier(overrides.tier ?? base.tier); + if (!resolved) { + throw new Error(`Unknown model tier "${overrides.tier}". Use one of: ${TIER_ORDER.join(", ")}.`); + } + + const requestedEffort = overrides.effort ? normalizeKey(overrides.effort) : base.effort; + if (!isEffortValidForTier(resolved.tier, requestedEffort)) { + const detail = SOL_ONLY_EFFORTS.has(requestedEffort) + ? `Effort "${requestedEffort}" is sol-only; tier "${resolved.tier}" cannot use it.` + : `Unsupported effort "${requestedEffort}". Use one of: ${[...VALID_EFFORTS].join(", ")}.`; + throw new Error(detail); + } + + return { tier: resolved.tier, modelId: resolved.modelId, effort: requestedEffort }; +} + +const ENVELOPE_STATUSES = new Set(["DONE", "DONE_WITH_CONCERNS", "NEEDS_CONTEXT", "BLOCKED"]); +const ENVELOPE_KEYS = Object.freeze(["status", "summary", "files_modified", "concerns", "blocked_reason"]); +const ENVELOPE_KEY_SET = new Set(ENVELOPE_KEYS); + +function isStringArray(value) { + return Array.isArray(value) && value.every((item) => typeof item === "string"); +} + +function isValidEnvelopeShape(candidate) { + if (!candidate || typeof candidate !== "object" || Array.isArray(candidate)) { + return false; + } + + const keys = Object.keys(candidate); + if (keys.length !== ENVELOPE_KEYS.length || !keys.every((key) => ENVELOPE_KEY_SET.has(key))) { + return false; + } + + if (!ENVELOPE_STATUSES.has(candidate.status)) { + return false; + } + if (typeof candidate.summary !== "string") { + return false; + } + if (!isStringArray(candidate.files_modified)) { + return false; + } + if (!isStringArray(candidate.concerns)) { + return false; + } + return candidate.blocked_reason === null || typeof candidate.blocked_reason === "string"; +} + +/** + * Finds every balanced top-level `{...}` substring starting at each `{` + * in source order, respecting quoted strings and escapes. Never throws -- + * unterminated braces are simply skipped. + * @param {string} source + * @returns {string[]} + */ +function extractJsonObjectCandidates(source) { + const candidates = []; + + for (let start = 0; start < source.length; start += 1) { + if (source[start] !== "{") { + continue; + } + + let depth = 0; + let inString = false; + let escapeNext = false; + + for (let end = start; end < source.length; end += 1) { + const char = source[end]; + + if (escapeNext) { + escapeNext = false; + continue; + } + if (inString) { + if (char === "\\") { + escapeNext = true; + } else if (char === '"') { + inString = false; + } + continue; + } + if (char === '"') { + inString = true; + continue; + } + if (char === "{") { + depth += 1; + } else if (char === "}") { + depth -= 1; + if (depth === 0) { + candidates.push(source.slice(start, end + 1)); + break; + } + } + } + } + + return candidates; +} + +/** + * Parses the first valid envelope-shaped JSON object found in text, + * tolerating surrounding prose. Defensive by design: malformed or + * non-conforming input yields null rather than throwing. + * @param {string | null | undefined} text + * @returns {Readonly | null} + */ +export function parseEnvelope(text) { + if (text == null) { + return null; + } + + for (const candidate of extractJsonObjectCandidates(String(text))) { + let parsed; + try { + parsed = JSON.parse(candidate); + } catch { + continue; + } + if (isValidEnvelopeShape(parsed)) { + return Object.freeze({ ...parsed }); + } + } + + return null; +} + +/** + * Builds a BLOCKED envelope for a given reason (e.g. "QUOTA_EXHAUSTED"). + * @param {string} reason + * @param {{ summary?: string, filesModified?: string[], concerns?: string[] }} [extra] + * @returns {Readonly} + */ +export function makeBlockedEnvelope(reason, extra = {}) { + return Object.freeze({ + status: "BLOCKED", + summary: extra.summary ?? `Blocked: ${reason}`, + files_modified: Object.freeze([...(extra.filesModified ?? [])]), + concerns: Object.freeze([...(extra.concerns ?? [])]), + blocked_reason: reason + }); +} + +/** + * Classifies an error or response as a quota/rate-limit failure (429, + * "rate limit", "insufficient quota", etc.), checking status-like fields + * first and falling back to a message pattern match. + * @param {unknown} errorOrResponse + */ +export function classifyQuotaError(errorOrResponse) { + if (!errorOrResponse) { + return false; + } + + const statusCode = + errorOrResponse.status ?? errorOrResponse.statusCode ?? errorOrResponse.code ?? errorOrResponse.rpcCode ?? null; + if (statusCode === 429 || statusCode === "429") { + return true; + } + + const message = String(errorOrResponse.message ?? errorOrResponse.error?.message ?? errorOrResponse.detail ?? ""); + return QUOTA_ERROR_PATTERNS.some((pattern) => pattern.test(message)); +} + +/** + * Runs fn at the requested tier. On a classified quota error, retries + * exactly once at the next tier down. If that retry also fails on quota, + * returns a BLOCKED envelope with blocked_reason "QUOTA_EXHAUSTED" instead + * of throwing. Non-quota errors are rethrown unchanged. + * @param {(context: { tier: ModelTier, modelId: string }) => Promise} fn + * @param {{ tier: string }} options + */ +export async function withQuotaFailover(fn, options = {}) { + const startTier = resolveTier(options.tier); + if (!startTier) { + throw new Error(`Unknown model tier "${options.tier}". Use one of: ${TIER_ORDER.join(", ")}.`); + } + + try { + return await fn({ tier: startTier.tier, modelId: startTier.modelId }); + } catch (error) { + if (!classifyQuotaError(error)) { + throw error; + } + + const fallbackTierName = nextTierDown(startTier.tier); + if (!fallbackTierName) { + return makeBlockedEnvelope(QUOTA_EXHAUSTED_REASON, { + summary: `Quota exhausted at tier "${startTier.tier}" with no lower tier available.` + }); + } + + const fallbackTier = resolveTier(fallbackTierName); + try { + return await fn({ tier: fallbackTier.tier, modelId: fallbackTier.modelId }); + } catch (fallbackError) { + if (!classifyQuotaError(fallbackError)) { + throw fallbackError; + } + return makeBlockedEnvelope(QUOTA_EXHAUSTED_REASON, { + summary: `Quota exhausted at tier "${startTier.tier}" and step-down tier "${fallbackTier.tier}".` + }); + } + } +} + +/** + * Composes runAppServerTurn with tier resolution and quota failover, then + * attaches a parsed envelope (or null) to the result. Callers pass either + * a verb (to use PER_VERB_TIER_DEFAULTS) or an explicit tier alias; a verb + * plus tier/effort overrides both work via tierDefaultsForVerb. + * + * This is a new entry point, not a replacement for runAppServerTurn -- + * existing callers of runAppServerTurn are unaffected. + * @param {string} cwd + * @param {Record & { verb?: string, tier?: string, effort?: string }} [options] + */ +export async function runTieredAppServerTurn(cwd, options = {}) { + const { verb, tier, effort, ...turnOptions } = options; + + const resolvedDefaults = verb + ? tierDefaultsForVerb(verb, { tier, effort }) + : (() => { + const resolved = resolveTier(tier); + if (!resolved) { + throw new Error("runTieredAppServerTurn requires either options.verb or a resolvable options.tier."); + } + if (effort && !isEffortValidForTier(resolved.tier, normalizeKey(effort))) { + throw new Error(`Unsupported effort "${effort}" for tier "${resolved.tier}".`); + } + return { tier: resolved.tier, modelId: resolved.modelId, effort: effort ? normalizeKey(effort) : null }; + })(); + + const outcome = await withQuotaFailover( + ({ modelId }) => + runAppServerTurn(cwd, { + ...turnOptions, + model: modelId, + effort: resolvedDefaults.effort + }), + { tier: resolvedDefaults.tier } + ); + + if (outcome?.status === "BLOCKED" && outcome?.blocked_reason === QUOTA_EXHAUSTED_REASON) { + return outcome; + } + + return { ...outcome, envelope: parseEnvelope(outcome?.finalMessage) }; +} + +export { QUOTA_EXHAUSTED_REASON }; diff --git a/tests/rig-edition.test.mjs b/tests/rig-edition.test.mjs new file mode 100644 index 00000000..8f1cd88e --- /dev/null +++ b/tests/rig-edition.test.mjs @@ -0,0 +1,312 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { + classifyQuotaError, + isEffortValidForTier, + makeBlockedEnvelope, + MODEL_ALIASES, + nextTierDown, + parseEnvelope, + PER_VERB_TIER_DEFAULTS, + resolveTier, + tierDefaultsForVerb, + withQuotaFailover +} from "../plugins/codex/scripts/lib/rig-edition.mjs"; + +test("resolveTier resolves a tier alias case-insensitively", () => { + assert.deepEqual(resolveTier("sol"), { tier: "sol", modelId: "gpt-5.6-sol" }); + assert.deepEqual(resolveTier("TERRA"), { tier: "terra", modelId: "gpt-5.6-terra" }); + assert.deepEqual(resolveTier(" luna "), { tier: "luna", modelId: "gpt-5.6-luna" }); +}); + +test("resolveTier resolves a full model id back to its tier", () => { + assert.deepEqual(resolveTier("gpt-5.6-sol"), { tier: "sol", modelId: "gpt-5.6-sol" }); + assert.deepEqual(resolveTier("GPT-5.6-TERRA"), { tier: "terra", modelId: "gpt-5.6-terra" }); +}); + +test("resolveTier returns null for unknown or empty input", () => { + assert.equal(resolveTier("gpt-4"), null); + assert.equal(resolveTier(""), null); + assert.equal(resolveTier(null), null); + assert.equal(resolveTier(undefined), null); +}); + +test("MODEL_ALIASES matches the verified 0.144.0 catalog ids", () => { + assert.deepEqual(MODEL_ALIASES, { + sol: "gpt-5.6-sol", + terra: "gpt-5.6-terra", + luna: "gpt-5.6-luna" + }); +}); + +test("isEffortValidForTier allows max/ultra only on sol", () => { + assert.equal(isEffortValidForTier("sol", "max"), true); + assert.equal(isEffortValidForTier("sol", "ultra"), true); + assert.equal(isEffortValidForTier("terra", "max"), false); + assert.equal(isEffortValidForTier("luna", "ultra"), false); + assert.equal(isEffortValidForTier("terra", "medium"), true); + assert.equal(isEffortValidForTier("sol", "not-a-real-effort"), false); +}); + +test("nextTierDown steps sol -> terra -> luna -> null", () => { + assert.equal(nextTierDown("sol"), "terra"); + assert.equal(nextTierDown("terra"), "luna"); + assert.equal(nextTierDown("luna"), null); + assert.equal(nextTierDown("unknown-tier"), null); +}); + +test("tierDefaultsForVerb returns the documented per-verb defaults", () => { + assert.deepEqual(tierDefaultsForVerb("delegate"), { tier: "terra", modelId: "gpt-5.6-terra", effort: "medium" }); + assert.deepEqual(tierDefaultsForVerb("implement"), { tier: "terra", modelId: "gpt-5.6-terra", effort: "medium" }); + assert.deepEqual(tierDefaultsForVerb("review"), { tier: "sol", modelId: "gpt-5.6-sol", effort: "high" }); + assert.deepEqual(tierDefaultsForVerb("adversarial"), { tier: "sol", modelId: "gpt-5.6-sol", effort: "high" }); + assert.deepEqual(tierDefaultsForVerb("bulk"), { tier: "luna", modelId: "gpt-5.6-luna", effort: "low" }); + assert.deepEqual(tierDefaultsForVerb("mechanical"), { tier: "luna", modelId: "gpt-5.6-luna", effort: "low" }); +}); + +test("tierDefaultsForVerb is case-insensitive on the verb name", () => { + assert.deepEqual(tierDefaultsForVerb("DELEGATE"), { tier: "terra", modelId: "gpt-5.6-terra", effort: "medium" }); +}); + +test("tierDefaultsForVerb honors per-call tier and effort overrides", () => { + assert.deepEqual(tierDefaultsForVerb("delegate", { tier: "sol", effort: "xhigh" }), { + tier: "sol", + modelId: "gpt-5.6-sol", + effort: "xhigh" + }); +}); + +test("tierDefaultsForVerb throws on an unknown verb", () => { + assert.throws(() => tierDefaultsForVerb("teleport"), /Unknown verb "teleport"/); +}); + +test("tierDefaultsForVerb throws on an unresolvable tier override", () => { + assert.throws(() => tierDefaultsForVerb("delegate", { tier: "quasar" }), /Unknown model tier "quasar"/); +}); + +test("tierDefaultsForVerb throws when overriding to a sol-only effort on a lower tier", () => { + assert.throws( + () => tierDefaultsForVerb("delegate", { effort: "ultra" }), + /Effort "ultra" is sol-only; tier "terra" cannot use it\./ + ); +}); + +test("PER_VERB_TIER_DEFAULTS is frozen and covers exactly the documented verbs", () => { + assert.deepEqual(Object.keys(PER_VERB_TIER_DEFAULTS).sort(), [ + "adversarial", + "bulk", + "delegate", + "implement", + "mechanical", + "review" + ]); + assert.throws(() => { + PER_VERB_TIER_DEFAULTS.delegate = { tier: "luna", effort: "low" }; + }, TypeError); +}); + +test("parseEnvelope parses a bare valid envelope", () => { + const text = JSON.stringify({ + status: "DONE", + summary: "Implemented the feature.", + files_modified: ["src/app.js"], + concerns: [], + blocked_reason: null + }); + + const envelope = parseEnvelope(text); + + assert.equal(envelope.status, "DONE"); + assert.equal(envelope.summary, "Implemented the feature."); + assert.deepEqual(envelope.files_modified, ["src/app.js"]); + assert.deepEqual(envelope.concerns, []); + assert.equal(envelope.blocked_reason, null); +}); + +test("parseEnvelope extracts a valid envelope wrapped in surrounding prose", () => { + const envelopeJson = JSON.stringify({ + status: "DONE_WITH_CONCERNS", + summary: "Migrated the schema.", + files_modified: ["db/migrations/002.sql"], + concerns: ["Backfill not yet verified under concurrent writes."], + blocked_reason: null + }); + const text = `Here is my final report.\n\n${envelopeJson}\n\nLet me know if you have questions.`; + + const envelope = parseEnvelope(text); + + assert.equal(envelope.status, "DONE_WITH_CONCERNS"); + assert.deepEqual(envelope.concerns, ["Backfill not yet verified under concurrent writes."]); +}); + +test("parseEnvelope returns null for garbage input without throwing", () => { + assert.equal(parseEnvelope("not json at all"), null); + assert.equal(parseEnvelope("{ this is not { valid json"), null); + assert.equal(parseEnvelope(""), null); + assert.equal(parseEnvelope(null), null); + assert.equal(parseEnvelope(undefined), null); +}); + +test("parseEnvelope returns null for a JSON object with the wrong status enum value", () => { + const text = JSON.stringify({ + status: "FINISHED", + summary: "Done.", + files_modified: [], + concerns: [], + blocked_reason: null + }); + + assert.equal(parseEnvelope(text), null); +}); + +test("parseEnvelope returns null when required keys are missing or extra keys are present", () => { + const missingKey = JSON.stringify({ + status: "DONE", + summary: "Done.", + files_modified: [], + concerns: [] + }); + const extraKey = JSON.stringify({ + status: "DONE", + summary: "Done.", + files_modified: [], + concerns: [], + blocked_reason: null, + unexpected: "field" + }); + + assert.equal(parseEnvelope(missingKey), null); + assert.equal(parseEnvelope(extraKey), null); +}); + +test("parseEnvelope finds the first valid envelope when multiple JSON objects are present", () => { + const invalid = JSON.stringify({ status: "NOT_A_STATUS" }); + const valid = JSON.stringify({ + status: "NEEDS_CONTEXT", + summary: "Need clarification.", + files_modified: [], + concerns: [], + blocked_reason: null + }); + const text = `${invalid}\n${valid}`; + + const envelope = parseEnvelope(text); + + assert.equal(envelope.status, "NEEDS_CONTEXT"); +}); + +test("makeBlockedEnvelope builds a BLOCKED envelope with the given reason", () => { + const envelope = makeBlockedEnvelope("QUOTA_EXHAUSTED"); + + assert.equal(envelope.status, "BLOCKED"); + assert.equal(envelope.blocked_reason, "QUOTA_EXHAUSTED"); + assert.deepEqual(envelope.files_modified, []); + assert.deepEqual(envelope.concerns, []); + assert.match(envelope.summary, /QUOTA_EXHAUSTED/); +}); + +test("makeBlockedEnvelope accepts summary, filesModified, and concerns overrides", () => { + const envelope = makeBlockedEnvelope("NEEDS_HUMAN", { + summary: "Needs a human decision.", + filesModified: ["src/app.js"], + concerns: ["Ambiguous requirement."] + }); + + assert.equal(envelope.summary, "Needs a human decision."); + assert.deepEqual(envelope.files_modified, ["src/app.js"]); + assert.deepEqual(envelope.concerns, ["Ambiguous requirement."]); +}); + +test("classifyQuotaError recognizes a 429 status field", () => { + assert.equal(classifyQuotaError({ status: 429 }), true); + assert.equal(classifyQuotaError({ statusCode: 429 }), true); + assert.equal(classifyQuotaError({ code: 429 }), true); + assert.equal(classifyQuotaError({ rpcCode: 429 }), true); +}); + +test("classifyQuotaError recognizes rate-limit and quota messages", () => { + assert.equal(classifyQuotaError(new Error("Rate limit exceeded, try again later.")), true); + assert.equal(classifyQuotaError({ message: "insufficient_quota" }), true); + assert.equal(classifyQuotaError({ error: { message: "Too Many Requests" } }), true); + assert.equal(classifyQuotaError({ detail: "quota exceeded for this billing period" }), true); +}); + +test("classifyQuotaError rejects unrelated errors and empty input", () => { + assert.equal(classifyQuotaError(new Error("ENOENT: no such file or directory")), false); + assert.equal(classifyQuotaError({ status: 500, message: "internal error" }), false); + assert.equal(classifyQuotaError(null), false); + assert.equal(classifyQuotaError(undefined), false); +}); + +test("withQuotaFailover returns the result on first-attempt success without retrying", async () => { + const calls = []; + const result = await withQuotaFailover(async (context) => { + calls.push(context); + return { ok: true, tier: context.tier }; + }, { tier: "sol" }); + + assert.deepEqual(calls, [{ tier: "sol", modelId: "gpt-5.6-sol" }]); + assert.deepEqual(result, { ok: true, tier: "sol" }); +}); + +test("withQuotaFailover retries one tier down on a 429, then succeeds", async () => { + const calls = []; + const result = await withQuotaFailover(async (context) => { + calls.push(context.tier); + if (context.tier === "sol") { + const error = new Error("rate limited"); + error.status = 429; + throw error; + } + return { ok: true, tier: context.tier }; + }, { tier: "sol" }); + + assert.deepEqual(calls, ["sol", "terra"]); + assert.deepEqual(result, { ok: true, tier: "terra" }); +}); + +test("withQuotaFailover returns a QUOTA_EXHAUSTED BLOCKED envelope after two consecutive 429s", async () => { + const calls = []; + const result = await withQuotaFailover(async (context) => { + calls.push(context.tier); + const error = new Error("rate limited"); + error.status = 429; + throw error; + }, { tier: "sol" }); + + assert.deepEqual(calls, ["sol", "terra"]); + assert.equal(result.status, "BLOCKED"); + assert.equal(result.blocked_reason, "QUOTA_EXHAUSTED"); +}); + +test("withQuotaFailover returns QUOTA_EXHAUSTED immediately when starting at the lowest tier", async () => { + const calls = []; + const result = await withQuotaFailover(async (context) => { + calls.push(context.tier); + const error = new Error("rate limited"); + error.status = 429; + throw error; + }, { tier: "luna" }); + + assert.deepEqual(calls, ["luna"]); + assert.equal(result.status, "BLOCKED"); + assert.equal(result.blocked_reason, "QUOTA_EXHAUSTED"); +}); + +test("withQuotaFailover rethrows non-quota errors without retrying", async () => { + const calls = []; + await assert.rejects( + withQuotaFailover(async (context) => { + calls.push(context.tier); + throw new Error("unrelated failure"); + }, { tier: "sol" }), + /unrelated failure/ + ); + + assert.deepEqual(calls, ["sol"]); +}); + +test("withQuotaFailover rejects an unresolvable starting tier", async () => { + await assert.rejects(withQuotaFailover(async () => ({}), { tier: "not-a-tier" }), /Unknown model tier "not-a-tier"/); +}); From a422769a7a2271caf37d795dc2abea68d272d5ec Mon Sep 17 00:00:00 2001 From: nategarelik Date: Sun, 12 Jul 2026 16:53:29 -0500 Subject: [PATCH 4/6] feat: add rig-edition fanout, council, and cloud verbs Adds three additive orchestration verbs on top of the rig-edition tier routing layer: - fanout: N brief specs -> one disjoint git worktree per worker -> concurrent tiered Codex turns (hard cap 5, extra briefs queue behind it), aggregated orchestrator-side into results/failed/quota_exhausted. Worktree cleanup only runs when --cleanup is passed, and a cleanup failure never discards a worker's real outcome (a throw inside finally would otherwise replace it) -- it only annotates the entry with cleanupWarning, since a live smoke showed the Codex app-server process can keep a worktree directory locked past turn completion. - council: M seats (default 3) debate a topic in two rounds -- round-1 advocate/counter seats run concurrently and in isolation from each other, round-2's decider seat sees both transcripts and produces the verdict. Falls back to the raw reply when a seat's turn has no parseable envelope, since council seats are prompted to debate, not to emit structured envelopes. - cloud: envelope-wrapping passthrough over the verified `codex cloud` CLI surface (exec/status/list/apply/diff; EXPERIMENTAL upstream). Wires all three into codex-companion.mjs as new subcommands with minimal, additive routing, matching the existing task/review command shape. --- plugins/codex/scripts/codex-companion.mjs | 102 +++++++- plugins/codex/scripts/lib/cloud.mjs | 133 ++++++++++ plugins/codex/scripts/lib/council.mjs | 212 ++++++++++++++++ plugins/codex/scripts/lib/fanout.mjs | 296 ++++++++++++++++++++++ plugins/codex/scripts/lib/render.mjs | 66 +++++ tests/cloud.test.mjs | 79 ++++++ tests/council.test.mjs | 180 +++++++++++++ tests/fanout.test.mjs | 225 ++++++++++++++++ 8 files changed, 1292 insertions(+), 1 deletion(-) create mode 100644 plugins/codex/scripts/lib/cloud.mjs create mode 100644 plugins/codex/scripts/lib/council.mjs create mode 100644 plugins/codex/scripts/lib/fanout.mjs create mode 100644 tests/cloud.test.mjs create mode 100644 tests/council.test.mjs create mode 100644 tests/fanout.test.mjs diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs index 9cc6ddf8..f87c11be 100644 --- a/plugins/codex/scripts/codex-companion.mjs +++ b/plugins/codex/scripts/codex-companion.mjs @@ -8,6 +8,9 @@ import { fileURLToPath } from "node:url"; import { parseArgs, splitRawArgumentString } from "./lib/args.mjs"; import { tierDefaultsForVerb } from "./lib/rig-edition.mjs"; +import { loadFanoutBriefsFromFile, runFanout } from "./lib/fanout.mjs"; +import { runCouncil } from "./lib/council.mjs"; +import { CLOUD_SUBCOMMANDS, runCloudCommand } from "./lib/cloud.mjs"; import { buildPersistentTaskThreadName, DEFAULT_CONTINUE_PROMPT, @@ -24,7 +27,7 @@ import { } from "./lib/codex.mjs"; import { resolveClaudeSessionPath } from "./lib/claude-session-transfer.mjs"; import { readStdinIfPiped } from "./lib/fs.mjs"; -import { collectReviewContext, ensureGitRepository, resolveReviewTarget } from "./lib/git.mjs"; +import { collectReviewContext, ensureGitRepository, getRepoRoot, resolveReviewTarget } from "./lib/git.mjs"; import { binaryAvailable, terminateProcessTree } from "./lib/process.mjs"; import { loadPromptTemplate, interpolateTemplate } from "./lib/prompts.mjs"; import { @@ -59,6 +62,9 @@ import { renderReviewResult, renderStoredJobResult, renderCancelReport, + renderCloudResult, + renderCouncilResult, + renderFanoutResult, renderJobStatusReport, renderSetupReport, renderStatusReport, @@ -81,6 +87,9 @@ function printUsage() { " node scripts/codex-companion.mjs review [--wait|--background] [--base ] [--scope ]", " node scripts/codex-companion.mjs adversarial-review [--wait|--background] [--base ] [--scope ] [focus text]", " node scripts/codex-companion.mjs task [--background] [--write] [--resume-last|--resume|--fresh] [--model ] [--effort ] [--tier ] [prompt]", + " node scripts/codex-companion.mjs fanout --briefs --root [--concurrency ] [--cleanup] [--json]", + " node scripts/codex-companion.mjs council [--seats ] [--tier ] [--effort ] [--topic-file ] [topic]", + " node scripts/codex-companion.mjs cloud [task-id] [--env ] [--branch ] [--attempts ] [--attempt ] [--limit ] [--cursor ] [--json]", " node scripts/codex-companion.mjs transfer [--source ] [--json]", " node scripts/codex-companion.mjs status [job-id] [--all] [--json]", " node scripts/codex-companion.mjs result [job-id] [--json]", @@ -827,6 +836,88 @@ async function handleTask(argv) { ); } +async function handleFanout(argv) { + const { options } = parseCommandInput(argv, { + valueOptions: ["cwd", "briefs", "root", "concurrency"], + booleanOptions: ["json", "cleanup"] + }); + + if (!options.briefs) { + throw new Error("Provide --briefs ."); + } + if (!options.root) { + throw new Error("Provide --root ."); + } + + const cwd = resolveCommandCwd(options); + ensureGitRepository(cwd); + const repoRoot = getRepoRoot(cwd); + const briefs = loadFanoutBriefsFromFile(path.resolve(cwd, options.briefs)); + const worktreeRoot = path.resolve(cwd, options.root); + + const aggregate = await runFanout(repoRoot, worktreeRoot, briefs, { + concurrency: options.concurrency, + cleanup: Boolean(options.cleanup) + }); + + outputResult(options.json ? aggregate : renderFanoutResult(aggregate), options.json); +} + +async function handleCouncil(argv) { + const { options, positionals } = parseCommandInput(argv, { + valueOptions: ["cwd", "seats", "tier", "effort", "topic-file"], + booleanOptions: ["json"] + }); + + const cwd = resolveCommandCwd(options); + const topic = options["topic-file"] + ? fs.readFileSync(path.resolve(cwd, options["topic-file"]), "utf8") + : positionals.join(" ") || readStdinIfPiped(); + + if (!topic || !topic.trim()) { + throw new Error("Provide a topic, a --topic-file, or piped stdin."); + } + + const result = await runCouncil(cwd, topic, { + seats: options.seats ? Number(options.seats) : undefined, + tier: options.tier, + effort: options.effort + }); + + outputResult(options.json ? result : renderCouncilResult(result), options.json); +} + +async function handleCloud(argv) { + const { options, positionals } = parseCommandInput(argv, { + valueOptions: ["cwd", "env", "branch", "attempts", "attempt", "limit", "cursor"], + booleanOptions: ["json"] + }); + + const [subcommand, ...rest] = positionals; + if (!subcommand || !CLOUD_SUBCOMMANDS.includes(subcommand)) { + throw new Error(`Provide a "codex cloud" subcommand. Use one of: ${CLOUD_SUBCOMMANDS.join(", ")}.`); + } + + const cwd = resolveCommandCwd(options); + const taskId = ["status", "apply", "diff"].includes(subcommand) ? rest[0] : undefined; + const query = subcommand === "exec" ? rest.join(" ") : undefined; + + const result = runCloudCommand(subcommand, { + cwd, + env: options.env, + branch: options.branch, + attempts: options.attempts, + attempt: options.attempt, + limit: options.limit, + cursor: options.cursor, + json: options.json, + taskId, + query + }); + + outputResult(options.json ? result : renderCloudResult(result), options.json); +} + async function handleTransfer(argv) { const { options } = parseCommandInput(argv, { valueOptions: ["cwd", "source"], @@ -1048,6 +1139,15 @@ async function main() { case "task": await handleTask(argv); break; + case "fanout": + await handleFanout(argv); + break; + case "council": + await handleCouncil(argv); + break; + case "cloud": + await handleCloud(argv); + break; case "transfer": await handleTransfer(argv); break; diff --git a/plugins/codex/scripts/lib/cloud.mjs b/plugins/codex/scripts/lib/cloud.mjs new file mode 100644 index 00000000..fe3b7efb --- /dev/null +++ b/plugins/codex/scripts/lib/cloud.mjs @@ -0,0 +1,133 @@ +/** + * Rig-edition cloud verb: a thin, envelope-wrapping passthrough over the + * `codex cloud` CLI surface (EXPERIMENTAL upstream; verified 0.144.1). + * + * `codex cloud` browses and drives OpenAI-hosted Codex Cloud tasks -- a + * different execution target than the local app-server turns the rest of + * rig-edition drives. It is not an app-server JSON-RPC session, so there is + * no structured Codex-side envelope to parse; this module builds one from + * the CLI's exit status instead. + * + * Verified subcommand surface (codex cloud --help / --help, + * 0.144.1): exec, status, list, apply, diff. `exec` requires --env + * (a Codex Cloud environment id, not a local worktree) and takes an + * optional --branch/--attempts; `list` supports --env/--limit/--cursor/ + * --json; `status`/`apply`/`diff` take a positional task id, with + * apply/diff also accepting --attempt. + */ +import { runCommand } from "./process.mjs"; + +export const CLOUD_SUBCOMMANDS = Object.freeze(["exec", "status", "list", "apply", "diff"]); + +function pushValueArg(args, flag, value) { + if (value === undefined || value === null || value === "") { + return; + } + args.push(flag, String(value)); +} + +function requireTaskId(options) { + if (!isNonEmptyString(options.taskId)) { + throw new Error("This `codex cloud` subcommand requires a task id."); + } + return String(options.taskId); +} + +function isNonEmptyString(value) { + return typeof value === "string" && value.trim().length > 0; +} + +function buildExecArgs(options) { + if (!isNonEmptyString(options.env)) { + throw new Error("`codex cloud exec` requires --env ."); + } + const args = ["--env", String(options.env)]; + pushValueArg(args, "--attempts", options.attempts); + pushValueArg(args, "--branch", options.branch); + if (isNonEmptyString(options.query)) { + args.push(String(options.query)); + } + return args; +} + +function buildStatusArgs(options) { + return [requireTaskId(options)]; +} + +function buildListArgs(options) { + const args = []; + pushValueArg(args, "--env", options.env); + pushValueArg(args, "--limit", options.limit); + pushValueArg(args, "--cursor", options.cursor); + if (options.json) { + args.push("--json"); + } + return args; +} + +function buildApplyArgs(options) { + const args = [requireTaskId(options)]; + pushValueArg(args, "--attempt", options.attempt); + return args; +} + +function buildDiffArgs(options) { + const args = [requireTaskId(options)]; + pushValueArg(args, "--attempt", options.attempt); + return args; +} + +const ARG_BUILDERS = Object.freeze({ + exec: buildExecArgs, + status: buildStatusArgs, + list: buildListArgs, + apply: buildApplyArgs, + diff: buildDiffArgs +}); + +/** + * Builds the full `codex cloud ...` argv for a given verb and + * options, applying the same validation the upstream CLI would enforce. + * @param {string} subcommand + * @param {Record} [options] + * @returns {string[]} + */ +export function buildCloudArgs(subcommand, options = {}) { + const builder = ARG_BUILDERS[subcommand]; + if (!builder) { + throw new Error(`Unknown "codex cloud" subcommand "${subcommand}". Use one of: ${CLOUD_SUBCOMMANDS.join(", ")}.`); + } + return ["cloud", subcommand, ...builder(options)]; +} + +function buildEnvelope(subcommand, result) { + const stdout = (result.stdout ?? "").trim(); + const stderr = (result.stderr ?? "").trim(); + const succeeded = !result.error && result.status === 0; + + return Object.freeze({ + status: succeeded ? "DONE" : "BLOCKED", + summary: succeeded ? `codex cloud ${subcommand} completed.` : `codex cloud ${subcommand} failed.`, + files_modified: Object.freeze([]), + concerns: Object.freeze(succeeded ? [] : [stderr || stdout || `exit ${result.status}`]), + blocked_reason: succeeded ? null : "CLOUD_COMMAND_FAILED" + }); +} + +/** + * Runs a `codex cloud` subcommand and wraps the result in a typed envelope. + * @param {string} subcommand + * @param {Record & { cwd?: string, runCommand?: typeof runCommand }} [options] + */ +export function runCloudCommand(subcommand, options = {}) { + const runner = options.runCommand ?? runCommand; + const args = buildCloudArgs(subcommand, options); + const result = runner("codex", args, { cwd: options.cwd, shell: false }); + + return Object.freeze({ + exitStatus: result.status, + stdout: result.stdout, + stderr: result.stderr, + envelope: buildEnvelope(subcommand, result) + }); +} diff --git a/plugins/codex/scripts/lib/council.mjs b/plugins/codex/scripts/lib/council.mjs new file mode 100644 index 00000000..5802122a --- /dev/null +++ b/plugins/codex/scripts/lib/council.mjs @@ -0,0 +1,212 @@ +/** + * Rig-edition council verb: M seats debate a topic in two rounds. + * + * Round 1 runs every non-decider seat concurrently and in isolation -- each + * seat only ever sees the topic, never another seat's output. Round 2 runs + * a single decider seat whose prompt includes every round-1 transcript, and + * whose summary becomes the council verdict. + * + * Additive layer over rig-edition.mjs; tiers/effort per seat default to the + * same sol/high used by the existing review and adversarial verbs. + * + * @typedef {"advocate" | "counter" | "decider"} CouncilStance + * @typedef {{ index: number, stance: CouncilStance, tier: string, effort: string }} CouncilSeat + * @typedef {{ + * seat: number, + * stance: CouncilStance, + * envelope: import("./rig-edition.mjs").Envelope | null, + * summary: string, + * quotaExhausted: boolean + * }} CouncilSeatResult + */ +import { QUOTA_EXHAUSTED_REASON, runTieredAppServerTurn } from "./rig-edition.mjs"; + +export const DEFAULT_COUNCIL_SEATS = 3; + +const NON_DECIDER_STANCES = Object.freeze(["advocate", "counter"]); + +const STANCE_INSTRUCTIONS = Object.freeze({ + advocate: + "Argue FOR the strongest case on this topic. Be rigorous, not reflexive -- ground every claim in the topic material.", + counter: + "Argue AGAINST the proposal, or surface the strongest counter-case, risks, and failure modes on this topic.", + decider: "You are the decider. Weigh the round-1 transcripts below and produce a final verdict." +}); + +/** + * Builds the seat plan for a council: seatCount - 1 non-decider seats + * (alternating advocate/counter), then one decider seat last. + * @param {number} [seatCount] + * @param {{ tier?: string, effort?: string, deciderTier?: string, deciderEffort?: string, seats?: Array<{ tier?: string, effort?: string }> }} [overrides] + * @returns {Readonly} + */ +export function buildSeatPlan(seatCount = DEFAULT_COUNCIL_SEATS, overrides = {}) { + const count = Number.isInteger(seatCount) && seatCount >= 2 ? seatCount : DEFAULT_COUNCIL_SEATS; + const perSeatOverrides = Array.isArray(overrides.seats) ? overrides.seats : []; + const seats = []; + + for (let index = 0; index < count - 1; index += 1) { + const stance = NON_DECIDER_STANCES[index % NON_DECIDER_STANCES.length]; + seats.push( + Object.freeze({ + index, + stance, + tier: perSeatOverrides[index]?.tier ?? overrides.tier ?? "sol", + effort: perSeatOverrides[index]?.effort ?? overrides.effort ?? "high" + }) + ); + } + + seats.push( + Object.freeze({ + index: count - 1, + stance: "decider", + tier: perSeatOverrides[count - 1]?.tier ?? overrides.deciderTier ?? overrides.tier ?? "sol", + effort: perSeatOverrides[count - 1]?.effort ?? overrides.deciderEffort ?? overrides.effort ?? "high" + }) + ); + + return Object.freeze(seats); +} + +/** + * @param {CouncilStance} stance + * @param {string} topic + */ +export function buildSeatPrompt(stance, topic) { + const instruction = STANCE_INSTRUCTIONS[stance] ?? STANCE_INSTRUCTIONS.advocate; + return `${instruction}\n\nTopic:\n${topic}`; +} + +/** + * @param {string} topic + * @param {CouncilSeatResult[]} round1Outputs + */ +export function buildDeciderPrompt(topic, round1Outputs) { + const transcripts = round1Outputs + .map((seatResult) => `### Seat ${seatResult.seat} (${seatResult.stance})\n${seatResult.summary || "(no summary)"}`) + .join("\n\n"); + return `${STANCE_INSTRUCTIONS.decider}\n\nTopic:\n${topic}\n\nRound 1 transcripts:\n\n${transcripts}`; +} + +// Council seats are prompted to debate a topic, not to emit a structured +// envelope -- most "pure-reply" turns will have no parseable envelope, so +// the summary must fall back to the raw final message rather than going +// silently empty (an empty summary would defeat the point of a council). +function summarizeOutcome(outcome) { + if (outcome?.status === "BLOCKED" && outcome?.blocked_reason === QUOTA_EXHAUSTED_REASON) { + return outcome.summary; + } + if (outcome?.envelope?.summary) { + return outcome.envelope.summary; + } + return typeof outcome?.finalMessage === "string" ? outcome.finalMessage.trim() : ""; +} + +async function runSeatTurn(cwd, seat, prompt, runTurn) { + const outcome = await runTurn(cwd, { tier: seat.tier, effort: seat.effort, prompt, sandbox: "read-only" }); + const quotaExhausted = outcome?.status === "BLOCKED" && outcome?.blocked_reason === QUOTA_EXHAUSTED_REASON; + + return Object.freeze({ + seat: seat.index, + stance: seat.stance, + envelope: quotaExhausted ? outcome : outcome?.envelope ?? null, + summary: summarizeOutcome(outcome), + quotaExhausted + }); +} + +/** + * Runs every non-decider seat concurrently, each in isolation from the + * others -- no seat's prompt references another seat's output. + * @param {string} cwd + * @param {string} topic + * @param {CouncilSeat[]} seatPlan + * @param {{ runTurn?: typeof runTieredAppServerTurn }} [options] + * @returns {Promise>} + */ +export async function runCouncilRound1(cwd, topic, seatPlan, options = {}) { + const runTurn = options.runTurn ?? runTieredAppServerTurn; + const nonDeciderSeats = seatPlan.filter((seat) => seat.stance !== "decider"); + const seatResults = await Promise.all( + nonDeciderSeats.map((seat) => runSeatTurn(cwd, seat, buildSeatPrompt(seat.stance, topic), runTurn)) + ); + return Object.freeze(seatResults); +} + +/** + * Runs the decider seat with a prompt built from every round-1 transcript. + * @param {string} cwd + * @param {string} topic + * @param {CouncilSeatResult[]} round1Outputs + * @param {CouncilSeat} deciderSeat + * @param {{ runTurn?: typeof runTieredAppServerTurn }} [options] + * @returns {Promise>} + */ +export async function runCouncilRound2(cwd, topic, round1Outputs, deciderSeat, options = {}) { + const runTurn = options.runTurn ?? runTieredAppServerTurn; + const prompt = buildDeciderPrompt(topic, round1Outputs); + return runSeatTurn(cwd, deciderSeat, prompt, runTurn); +} + +function buildAgreementNotes(round1Outputs) { + const notes = []; + const withEnvelope = round1Outputs.filter((seat) => seat.envelope && typeof seat.envelope.status === "string"); + + if (withEnvelope.length < 2) { + notes.push("Insufficient round-1 envelopes to compare agreement."); + } else { + const statuses = new Set(withEnvelope.map((seat) => seat.envelope.status)); + if (statuses.size === 1) { + notes.push(`All ${withEnvelope.length} round-1 seats returned status ${[...statuses][0]}.`); + } else { + notes.push( + `Round-1 seats disagreed on status: ${withEnvelope.map((seat) => `${seat.stance}=${seat.envelope.status}`).join(", ")}.` + ); + } + } + + const quotaExhaustedSeats = round1Outputs.filter((seat) => seat.quotaExhausted).map((seat) => seat.stance); + if (quotaExhaustedSeats.length > 0) { + notes.push(`Quota exhausted for seat(s): ${quotaExhaustedSeats.join(", ")}.`); + } + + return Object.freeze(notes); +} + +/** + * Runs a full council: round-1 isolation across advocate/counter (and any + * additional non-decider seats), then a round-2 decider that sees both + * transcripts and produces the final verdict. + * @param {string} cwd + * @param {string} topic + * @param {{ + * seats?: number, + * tier?: string, + * effort?: string, + * deciderTier?: string, + * deciderEffort?: string, + * runTurn?: typeof runTieredAppServerTurn + * }} [options] + */ +export async function runCouncil(cwd, topic, options = {}) { + if (!isNonEmptyTopic(topic)) { + throw new Error("runCouncil requires a non-empty topic."); + } + + const seatPlan = buildSeatPlan(options.seats ?? DEFAULT_COUNCIL_SEATS, options); + const deciderSeat = seatPlan[seatPlan.length - 1]; + + const round1Outputs = await runCouncilRound1(cwd, topic, seatPlan, options); + const deciderResult = await runCouncilRound2(cwd, topic, round1Outputs, deciderSeat, options); + + return Object.freeze({ + verdict: deciderResult.summary || "Decider returned no summary.", + seat_outputs: Object.freeze([...round1Outputs, deciderResult]), + agreement_notes: buildAgreementNotes(round1Outputs) + }); +} + +function isNonEmptyTopic(topic) { + return typeof topic === "string" && topic.trim().length > 0; +} diff --git a/plugins/codex/scripts/lib/fanout.mjs b/plugins/codex/scripts/lib/fanout.mjs new file mode 100644 index 00000000..d9861ffb --- /dev/null +++ b/plugins/codex/scripts/lib/fanout.mjs @@ -0,0 +1,296 @@ +/** + * Rig-edition fanout verb: N brief specs -> one git worktree per worker -> + * concurrent tiered Codex turns -> aggregated typed envelopes. + * + * Additive layer over rig-edition.mjs and process.mjs. Concurrency is capped + * at FANOUT_MAX_CONCURRENCY regardless of the caller's request; briefs beyond + * the cap queue behind it. Worktrees are created for every worker and are + * only removed when the caller opts into cleanup -- leaving a worktree + * behind is the safe default so a caller can inspect a worker's output. + * + * @typedef {{ + * id: string, + * prompt: string, + * verb: string, + * tier: string | undefined, + * effort: string | undefined, + * branch: string | undefined, + * write: boolean + * }} FanoutBriefSpec + * @typedef {{ path: string, branch: string }} Worktree + */ +import fs from "node:fs"; +import path from "node:path"; + +import { QUOTA_EXHAUSTED_REASON, runTieredAppServerTurn } from "./rig-edition.mjs"; +import { runCommandChecked } from "./process.mjs"; + +export const FANOUT_MAX_CONCURRENCY = 5; + +function isNonEmptyString(value) { + return typeof value === "string" && value.trim().length > 0; +} + +/** + * Parses and validates a fanout briefs JSON document (an array of brief + * specs). Never returns a partially-valid result -- throws with the first + * validation error found. + * @param {string} rawJsonText + * @returns {Readonly} + */ +export function parseFanoutBriefs(rawJsonText) { + let parsed; + try { + parsed = JSON.parse(rawJsonText); + } catch (error) { + throw new Error(`Fanout briefs file is not valid JSON: ${error.message}`); + } + + if (!Array.isArray(parsed) || parsed.length === 0) { + throw new Error("Fanout briefs file must contain a non-empty JSON array of brief specs."); + } + + const seenIds = new Set(); + const briefs = parsed.map((raw, index) => { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) { + throw new Error(`Brief at index ${index} must be an object.`); + } + if (!isNonEmptyString(raw.id)) { + throw new Error(`Brief at index ${index} is missing a non-empty string "id".`); + } + if (seenIds.has(raw.id)) { + throw new Error(`Duplicate brief id "${raw.id}".`); + } + seenIds.add(raw.id); + if (!isNonEmptyString(raw.prompt)) { + throw new Error(`Brief "${raw.id}" is missing a non-empty string "prompt".`); + } + + return Object.freeze({ + id: raw.id, + prompt: raw.prompt, + verb: isNonEmptyString(raw.verb) ? raw.verb : "implement", + tier: isNonEmptyString(raw.tier) ? raw.tier : undefined, + effort: isNonEmptyString(raw.effort) ? raw.effort : undefined, + branch: isNonEmptyString(raw.branch) ? raw.branch : undefined, + write: raw.write !== false + }); + }); + + return Object.freeze(briefs); +} + +/** + * @param {string} filePath + * @returns {Readonly} + */ +export function loadFanoutBriefsFromFile(filePath) { + return parseFanoutBriefs(fs.readFileSync(filePath, "utf8")); +} + +function git(cwd, args) { + return runCommandChecked("git", args, { cwd, shell: false }); +} + +/** + * @param {FanoutBriefSpec} brief + */ +export function branchNameForBrief(brief) { + return brief.branch ?? `fanout/${brief.id}`; +} + +/** + * Creates a git worktree for one fanout worker, branching from HEAD of + * repoRoot into a caller-given worktree root. + * @param {string} repoRoot + * @param {string} worktreeRoot + * @param {FanoutBriefSpec} brief + * @returns {Promise} + */ +export async function createWorktree(repoRoot, worktreeRoot, brief) { + const branch = branchNameForBrief(brief); + const worktreePath = path.join(worktreeRoot, brief.id); + git(repoRoot, ["worktree", "add", "-b", branch, worktreePath, "HEAD"]); + return Object.freeze({ path: worktreePath, branch }); +} + +/** + * Removes a fanout worker's worktree and its branch. Branch deletion is + * best-effort: a failure there does not undo a successful worktree removal. + * @param {string} repoRoot + * @param {string} worktreePath + * @param {string} branch + */ +export async function removeWorktree(repoRoot, worktreePath, branch) { + git(repoRoot, ["worktree", "remove", "--force", worktreePath]); + if (!branch) { + return; + } + try { + git(repoRoot, ["branch", "-D", branch]); + } catch { + // Worktree removal already succeeded; branch cleanup is best-effort. + } +} + +function resolveConcurrency(requested) { + const parsed = Number(requested); + if (!Number.isFinite(parsed) || parsed <= 0) { + return FANOUT_MAX_CONCURRENCY; + } + return Math.min(Math.floor(parsed), FANOUT_MAX_CONCURRENCY); +} + +/** + * Runs `worker` over `items` with at most `limit` in flight at once. Extra + * items queue behind the limit rather than launching immediately. + * @template T, R + * @param {T[]} items + * @param {number} limit + * @param {(item: T, index: number) => Promise} worker + * @returns {Promise} + */ +async function runWithConcurrencyLimit(items, limit, worker) { + const results = new Array(items.length); + let cursor = 0; + + async function pullNext() { + const current = cursor; + cursor += 1; + if (current >= items.length) { + return; + } + results[current] = await worker(items[current], current); + await pullNext(); + } + + const lanes = Array.from({ length: Math.min(limit, items.length) }, () => pullNext()); + await Promise.all(lanes); + return results; +} + +async function computeWorkerOutcome(brief, worktree, context) { + try { + const outcome = await context.runTurn(worktree.path, { + verb: brief.verb, + tier: brief.tier, + effort: brief.effort, + prompt: brief.prompt, + sandbox: brief.write ? "workspace-write" : "read-only" + }); + + if (outcome?.status === "BLOCKED" && outcome?.blocked_reason === QUOTA_EXHAUSTED_REASON) { + return { + bucket: "quota_exhausted", + entry: { + id: brief.id, + workspace: worktree.path, + branch: worktree.branch, + envelope: outcome + } + }; + } + + return { + bucket: "results", + entry: { + id: brief.id, + workspace: worktree.path, + branch: worktree.branch, + envelope: outcome?.envelope ?? null, + status: outcome?.status ?? null + } + }; + } catch (error) { + return { + bucket: "failed", + entry: { + id: brief.id, + workspace: worktree.path, + branch: worktree.branch, + error: error instanceof Error ? error.message : String(error) + } + }; + } +} + +// A cleanup failure (e.g. a Windows file-lock still held on the worktree by +// the Codex process at removal time) must never discard the worker's real +// outcome -- a throw in a `finally` block replaces whatever the `try` block +// returned, which would turn a genuine success into a spurious failure. So +// cleanup runs after the outcome is already computed, and a cleanup error +// only annotates the entry with `cleanupWarning`; the worktree is left in +// place for manual inspection/removal. +async function runFanoutWorker(brief, context) { + const worktree = await context.createWorktree(context.repoRoot, context.worktreeRoot, brief); + const outcome = await computeWorkerOutcome(brief, worktree, context); + + if (!context.cleanup) { + return { bucket: outcome.bucket, entry: Object.freeze(outcome.entry) }; + } + + try { + await context.removeWorktree(context.repoRoot, worktree.path, worktree.branch); + return { bucket: outcome.bucket, entry: Object.freeze(outcome.entry) }; + } catch (cleanupError) { + return { + bucket: outcome.bucket, + entry: Object.freeze({ + ...outcome.entry, + cleanupWarning: cleanupError instanceof Error ? cleanupError.message : String(cleanupError) + }) + }; + } +} + +/** + * Runs a fanout across brief specs: one worktree + one tiered Codex turn per + * brief, at most FANOUT_MAX_CONCURRENCY concurrently, aggregated + * orchestrator-side into results / failed / quota_exhausted buckets. + * @param {string} repoRoot + * @param {string} worktreeRoot + * @param {FanoutBriefSpec[]} briefs + * @param {{ + * concurrency?: number, + * cleanup?: boolean, + * createWorktree?: typeof createWorktree, + * removeWorktree?: typeof removeWorktree, + * runTurn?: typeof runTieredAppServerTurn + * }} [options] + */ +export async function runFanout(repoRoot, worktreeRoot, briefs, options = {}) { + if (!Array.isArray(briefs) || briefs.length === 0) { + throw new Error("runFanout requires a non-empty array of brief specs."); + } + + const context = { + repoRoot, + worktreeRoot, + cleanup: Boolean(options.cleanup), + createWorktree: options.createWorktree ?? createWorktree, + removeWorktree: options.removeWorktree ?? removeWorktree, + runTurn: options.runTurn ?? runTieredAppServerTurn + }; + + const concurrency = resolveConcurrency(options.concurrency); + const outcomes = await runWithConcurrencyLimit(briefs, concurrency, (brief) => runFanoutWorker(brief, context)); + + const results = []; + const failed = []; + const quotaExhausted = []; + for (const outcome of outcomes) { + if (outcome.bucket === "results") { + results.push(outcome.entry); + } else if (outcome.bucket === "failed") { + failed.push(outcome.entry); + } else { + quotaExhausted.push(outcome.entry); + } + } + + return Object.freeze({ + results: Object.freeze(results), + failed: Object.freeze(failed), + quota_exhausted: Object.freeze(quotaExhausted) + }); +} diff --git a/plugins/codex/scripts/lib/render.mjs b/plugins/codex/scripts/lib/render.mjs index 2ec18523..c6f5e937 100644 --- a/plugins/codex/scripts/lib/render.mjs +++ b/plugins/codex/scripts/lib/render.mjs @@ -445,6 +445,72 @@ export function renderStoredJobResult(job, storedJob) { return `${lines.join("\n").trimEnd()}\n`; } +export function renderFanoutResult(aggregate) { + const lines = [ + "# Codex Fanout", + "", + `Results: ${aggregate.results.length} Failed: ${aggregate.failed.length} Quota exhausted: ${aggregate.quota_exhausted.length}`, + "" + ]; + + if (aggregate.results.length > 0) { + lines.push("Results:"); + for (const entry of aggregate.results) { + lines.push(`- ${entry.id} [${entry.envelope?.status ?? "no-envelope"}] ${entry.workspace}`); + } + lines.push(""); + } + + if (aggregate.failed.length > 0) { + lines.push("Failed:"); + for (const entry of aggregate.failed) { + lines.push(`- ${entry.id}: ${entry.error}`); + } + lines.push(""); + } + + if (aggregate.quota_exhausted.length > 0) { + lines.push("Quota exhausted:"); + for (const entry of aggregate.quota_exhausted) { + lines.push(`- ${entry.id}: ${entry.envelope?.summary ?? "quota exhausted"}`); + } + lines.push(""); + } + + return `${lines.join("\n").trimEnd()}\n`; +} + +export function renderCouncilResult(result) { + const lines = ["# Codex Council", "", `Verdict: ${result.verdict}`, "", "Seat outputs:"]; + + for (const seat of result.seat_outputs) { + lines.push(`- Seat ${seat.seat} (${seat.stance}): ${seat.summary || "(no summary)"}`); + } + + if (result.agreement_notes.length > 0) { + lines.push("", "Agreement notes:"); + for (const note of result.agreement_notes) { + lines.push(`- ${note}`); + } + } + + return `${lines.join("\n").trimEnd()}\n`; +} + +export function renderCloudResult(result) { + const lines = ["# Codex Cloud", "", `Status: ${result.envelope.status}`, ""]; + + if (result.stdout?.trim()) { + lines.push(result.stdout.trim()); + } + + if (result.envelope.status !== "DONE" && result.stderr?.trim()) { + lines.push("", "stderr:", "", "```text", result.stderr.trim(), "```"); + } + + return `${lines.join("\n").trimEnd()}\n`; +} + export function renderCancelReport(job) { const lines = [ "# Codex Cancel", diff --git a/tests/cloud.test.mjs b/tests/cloud.test.mjs new file mode 100644 index 00000000..58c14ae4 --- /dev/null +++ b/tests/cloud.test.mjs @@ -0,0 +1,79 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { buildCloudArgs, CLOUD_SUBCOMMANDS, runCloudCommand } from "../plugins/codex/scripts/lib/cloud.mjs"; + +test("CLOUD_SUBCOMMANDS matches the verified codex cloud --help surface", () => { + assert.deepEqual(CLOUD_SUBCOMMANDS, ["exec", "status", "list", "apply", "diff"]); +}); + +test("buildCloudArgs rejects an unknown subcommand", () => { + assert.throws(() => buildCloudArgs("teleport"), /Unknown "codex cloud" subcommand "teleport"/); +}); + +test("buildCloudArgs(exec) requires --env and maps branch/attempts/query", () => { + assert.throws(() => buildCloudArgs("exec", {}), /requires --env/); + + assert.deepEqual(buildCloudArgs("exec", { env: "env-123" }), ["cloud", "exec", "--env", "env-123"]); + + assert.deepEqual( + buildCloudArgs("exec", { env: "env-123", branch: "feat/x", attempts: 3, query: "fix the bug" }), + ["cloud", "exec", "--env", "env-123", "--attempts", "3", "--branch", "feat/x", "fix the bug"] + ); +}); + +test("buildCloudArgs(status) requires a task id and maps it positionally", () => { + assert.throws(() => buildCloudArgs("status", {}), /requires a task id/); + assert.deepEqual(buildCloudArgs("status", { taskId: "task-1" }), ["cloud", "status", "task-1"]); +}); + +test("buildCloudArgs(list) maps env/limit/cursor/json as optional flags", () => { + assert.deepEqual(buildCloudArgs("list", {}), ["cloud", "list"]); + assert.deepEqual( + buildCloudArgs("list", { env: "env-123", limit: 5, cursor: "abc", json: true }), + ["cloud", "list", "--env", "env-123", "--limit", "5", "--cursor", "abc", "--json"] + ); +}); + +test("buildCloudArgs(apply) and buildCloudArgs(diff) require a task id and accept --attempt", () => { + assert.throws(() => buildCloudArgs("apply", {}), /requires a task id/); + assert.deepEqual(buildCloudArgs("apply", { taskId: "task-1", attempt: 2 }), ["cloud", "apply", "task-1", "--attempt", "2"]); + assert.deepEqual(buildCloudArgs("diff", { taskId: "task-1" }), ["cloud", "diff", "task-1"]); +}); + +test("runCloudCommand wraps a successful CLI call in a DONE envelope", () => { + const calls = []; + const runCommand = (command, args, options) => { + calls.push({ command, args, options }); + return { command, args, status: 0, signal: null, stdout: "task-1 queued\n", stderr: "", error: null }; + }; + + const result = runCloudCommand("exec", { env: "env-123", query: "do the thing", cwd: "/repo", runCommand }); + + assert.equal(calls.length, 1); + assert.equal(calls[0].command, "codex"); + assert.deepEqual(calls[0].args, ["cloud", "exec", "--env", "env-123", "do the thing"]); + assert.equal(calls[0].options.cwd, "/repo"); + assert.equal(result.exitStatus, 0); + assert.equal(result.envelope.status, "DONE"); + assert.equal(result.envelope.blocked_reason, null); +}); + +test("runCloudCommand wraps a failing CLI call in a BLOCKED envelope with CLOUD_COMMAND_FAILED", () => { + const runCommand = () => ({ + command: "codex", + args: [], + status: 1, + signal: null, + stdout: "", + stderr: "task not found", + error: null + }); + + const result = runCloudCommand("status", { taskId: "task-404", runCommand }); + + assert.equal(result.exitStatus, 1); + assert.equal(result.envelope.status, "BLOCKED"); + assert.equal(result.envelope.blocked_reason, "CLOUD_COMMAND_FAILED"); + assert.deepEqual(result.envelope.concerns, ["task not found"]); +}); diff --git a/tests/council.test.mjs b/tests/council.test.mjs new file mode 100644 index 00000000..9bca538b --- /dev/null +++ b/tests/council.test.mjs @@ -0,0 +1,180 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { + buildDeciderPrompt, + buildSeatPlan, + buildSeatPrompt, + DEFAULT_COUNCIL_SEATS, + runCouncil, + runCouncilRound1, + runCouncilRound2 +} from "../plugins/codex/scripts/lib/council.mjs"; + +function makeEnvelope(status, summary) { + return { status, summary, files_modified: [], concerns: [], blocked_reason: null }; +} + +test("buildSeatPlan defaults to 3 seats: advocate, counter, decider", () => { + const seats = buildSeatPlan(); + assert.equal(seats.length, DEFAULT_COUNCIL_SEATS); + assert.deepEqual( + seats.map((seat) => seat.stance), + ["advocate", "counter", "decider"] + ); + assert.equal(seats.every((seat) => seat.tier === "sol" && seat.effort === "high"), true); +}); + +test("buildSeatPlan scales to M seats, alternating advocate/counter before the decider", () => { + const seats = buildSeatPlan(5); + assert.deepEqual( + seats.map((seat) => seat.stance), + ["advocate", "counter", "advocate", "counter", "decider"] + ); +}); + +test("buildSeatPlan applies per-seat and decider-specific tier/effort overrides", () => { + const seats = buildSeatPlan(3, { + tier: "terra", + effort: "medium", + deciderTier: "sol", + deciderEffort: "xhigh", + seats: [{ tier: "luna", effort: "low" }] + }); + assert.equal(seats[0].tier, "luna"); + assert.equal(seats[0].effort, "low"); + assert.equal(seats[1].tier, "terra"); + assert.equal(seats[1].effort, "medium"); + assert.equal(seats[2].tier, "sol"); + assert.equal(seats[2].effort, "xhigh"); +}); + +test("buildSeatPrompt frames advocate and counter stances distinctly and includes the topic", () => { + const advocatePrompt = buildSeatPrompt("advocate", "Should we ship X?"); + const counterPrompt = buildSeatPrompt("counter", "Should we ship X?"); + assert.match(advocatePrompt, /Argue FOR/); + assert.match(counterPrompt, /Argue AGAINST/); + assert.match(advocatePrompt, /Should we ship X\?/); +}); + +test("round-1 seats are isolated: each prompt contains only the topic, never another seat's output", async () => { + const seatPlan = buildSeatPlan(3); + const seenPrompts = []; + + const runTurn = async (cwd, options) => { + seenPrompts.push(options.prompt); + return { + status: 0, + finalMessage: "", + envelope: makeEnvelope("DONE", `summary for ${options.tier}`) + }; + }; + + await runCouncilRound1("/repo", "Should we ship X?", seatPlan, { runTurn }); + + assert.equal(seenPrompts.length, 2, "only the two non-decider seats should run in round 1"); + for (const prompt of seenPrompts) { + assert.doesNotMatch(prompt, /Round 1 transcripts/); + for (const otherPrompt of seenPrompts) { + if (otherPrompt !== prompt) { + assert.doesNotMatch(prompt, new RegExp(otherPrompt.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"))); + } + } + } +}); + +test("round-2 decider prompt includes both round-1 transcripts", () => { + const round1Outputs = [ + { seat: 0, stance: "advocate", summary: "Ship it: the upside outweighs the risk." }, + { seat: 1, stance: "counter", summary: "Do not ship: the rollback plan is untested." } + ]; + const prompt = buildDeciderPrompt("Should we ship X?", round1Outputs); + assert.match(prompt, /Ship it: the upside outweighs the risk\./); + assert.match(prompt, /Do not ship: the rollback plan is untested\./); + assert.match(prompt, /Seat 0 \(advocate\)/); + assert.match(prompt, /Seat 1 \(counter\)/); +}); + +test("runCouncilRound2 forwards the decider seat's tier/effort and captures its envelope", async () => { + const deciderSeat = { index: 2, stance: "decider", tier: "sol", effort: "xhigh" }; + const calls = []; + const runTurn = async (cwd, options) => { + calls.push(options); + return { + status: 0, + finalMessage: "", + envelope: makeEnvelope("DONE", "Final verdict: ship it.") + }; + }; + + const result = await runCouncilRound2("/repo", "topic", [], deciderSeat, { runTurn }); + + assert.equal(calls.length, 1); + assert.equal(calls[0].tier, "sol"); + assert.equal(calls[0].effort, "xhigh"); + assert.equal(result.summary, "Final verdict: ship it."); + assert.equal(result.stance, "decider"); +}); + +test("runCouncil returns verdict, seat_outputs (round 1 + decider), and agreement_notes", async () => { + const runTurn = async (cwd, options) => { + if (options.prompt.includes("Argue FOR")) { + return { status: 0, finalMessage: "", envelope: makeEnvelope("DONE", "Advocate says ship.") }; + } + if (options.prompt.includes("Argue AGAINST")) { + return { status: 0, finalMessage: "", envelope: makeEnvelope("DONE", "Counter says ship too.") }; + } + return { status: 0, finalMessage: "", envelope: makeEnvelope("DONE", "Decider verdict: ship it.") }; + }; + + const result = await runCouncil("/repo", "Should we ship X?", { runTurn }); + + assert.equal(result.verdict, "Decider verdict: ship it."); + assert.equal(result.seat_outputs.length, 3); + assert.deepEqual( + result.seat_outputs.map((seat) => seat.stance), + ["advocate", "counter", "decider"] + ); + assert.equal(result.agreement_notes.length, 1); + assert.match(result.agreement_notes[0], /All 2 round-1 seats returned status DONE/); +}); + +test("runCouncil surfaces disagreement and quota-exhaustion in agreement_notes", async () => { + let call = 0; + const runTurn = async (cwd, options) => { + call += 1; + if (options.prompt.includes("Argue FOR")) { + return { + status: "BLOCKED", + blocked_reason: "QUOTA_EXHAUSTED", + summary: "Quota exhausted at tier sol and step-down tier terra.", + files_modified: [], + concerns: [] + }; + } + if (options.prompt.includes("Argue AGAINST")) { + return { status: 0, finalMessage: "", envelope: makeEnvelope("NEEDS_CONTEXT", "Need more info.") }; + } + return { status: 0, finalMessage: "", envelope: makeEnvelope("DONE_WITH_CONCERNS", "Proceed cautiously.") }; + }; + + const result = await runCouncil("/repo", "topic", { runTurn }); + assert.match(result.agreement_notes.join(" "), /Quota exhausted for seat\(s\): advocate/); +}); + +test("runCouncilRound1 falls back to the raw final message when a seat's reply has no parseable envelope", async () => { + const seatPlan = buildSeatPlan(2); + const runTurn = async () => ({ + status: 0, + finalMessage: "Comments should explain why, not what -- the code already says what.", + envelope: null + }); + + const [seatResult] = await runCouncilRound1("/repo", "topic", seatPlan, { runTurn }); + + assert.equal(seatResult.summary, "Comments should explain why, not what -- the code already says what."); +}); + +test("runCouncil rejects an empty topic", async () => { + await assert.rejects(runCouncil("/repo", " ", { runTurn: async () => ({}) }), /non-empty topic/); +}); diff --git a/tests/fanout.test.mjs b/tests/fanout.test.mjs new file mode 100644 index 00000000..26d50673 --- /dev/null +++ b/tests/fanout.test.mjs @@ -0,0 +1,225 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { + branchNameForBrief, + FANOUT_MAX_CONCURRENCY, + parseFanoutBriefs, + runFanout +} from "../plugins/codex/scripts/lib/fanout.mjs"; + +function makeBriefsJson(count) { + return JSON.stringify( + Array.from({ length: count }, (_, index) => ({ + id: `brief-${index}`, + prompt: `Do task ${index}.` + })) + ); +} + +function makeMockContext({ runTurn, cleanup = false } = {}) { + const createdWorktrees = []; + const removedWorktrees = []; + + return { + cleanup, + createWorktree: async (repoRoot, worktreeRoot, brief) => { + const worktree = { path: `${worktreeRoot}/${brief.id}`, branch: branchNameForBrief(brief) }; + createdWorktrees.push(worktree); + return worktree; + }, + removeWorktree: async (repoRoot, worktreePath, branch) => { + removedWorktrees.push({ worktreePath, branch }); + }, + runTurn, + createdWorktrees, + removedWorktrees + }; +} + +test("parseFanoutBriefs validates and defaults a briefs array", () => { + const briefs = parseFanoutBriefs(makeBriefsJson(2)); + assert.equal(briefs.length, 2); + assert.equal(briefs[0].id, "brief-0"); + assert.equal(briefs[0].verb, "implement"); + assert.equal(briefs[0].write, true); +}); + +test("parseFanoutBriefs rejects invalid JSON, non-arrays, missing fields, and duplicate ids", () => { + assert.throws(() => parseFanoutBriefs("not json"), /not valid JSON/); + assert.throws(() => parseFanoutBriefs("{}"), /non-empty JSON array/); + assert.throws(() => parseFanoutBriefs("[]"), /non-empty JSON array/); + assert.throws(() => parseFanoutBriefs(JSON.stringify([{ prompt: "x" }])), /missing a non-empty string "id"/); + assert.throws(() => parseFanoutBriefs(JSON.stringify([{ id: "a" }])), /missing a non-empty string "prompt"/); + assert.throws( + () => parseFanoutBriefs(JSON.stringify([{ id: "a", prompt: "x" }, { id: "a", prompt: "y" }])), + /Duplicate brief id "a"/ + ); +}); + +test("runFanout caps concurrency at FANOUT_MAX_CONCURRENCY and queues the rest", async () => { + const briefs = parseFanoutBriefs(makeBriefsJson(8)); + let inFlight = 0; + let maxInFlight = 0; + + const context = makeMockContext({ + runTurn: async () => { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise((resolve) => setTimeout(resolve, 5)); + inFlight -= 1; + return { + status: 0, + finalMessage: "", + envelope: { status: "DONE", summary: "ok", files_modified: [], concerns: [], blocked_reason: null } + }; + } + }); + + const aggregate = await runFanout("/repo", "/worktrees", briefs, { + concurrency: 100, + createWorktree: context.createWorktree, + removeWorktree: context.removeWorktree, + runTurn: context.runTurn + }); + + assert.ok(maxInFlight <= FANOUT_MAX_CONCURRENCY, `expected max in-flight <= ${FANOUT_MAX_CONCURRENCY}, got ${maxInFlight}`); + assert.equal(aggregate.results.length, 8); +}); + +test("runFanout honors a lower explicit concurrency than the cap", async () => { + const briefs = parseFanoutBriefs(makeBriefsJson(6)); + let inFlight = 0; + let maxInFlight = 0; + + const context = makeMockContext({ + runTurn: async () => { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise((resolve) => setTimeout(resolve, 5)); + inFlight -= 1; + return { + status: 0, + finalMessage: "", + envelope: { status: "DONE", summary: "ok", files_modified: [], concerns: [], blocked_reason: null } + }; + } + }); + + await runFanout("/repo", "/worktrees", briefs, { + concurrency: 2, + createWorktree: context.createWorktree, + removeWorktree: context.removeWorktree, + runTurn: context.runTurn + }); + + assert.equal(maxInFlight, 2); +}); + +test("runFanout aggregates successes, failures, and quota-exhausted outcomes into separate buckets", async () => { + const briefs = parseFanoutBriefs( + JSON.stringify([ + { id: "ok-1", prompt: "succeed" }, + { id: "boom", prompt: "explode" }, + { id: "quota", prompt: "hit quota" } + ]) + ); + + const context = makeMockContext({ + runTurn: async (cwd, options) => { + if (options.prompt === "explode") { + throw new Error("unrelated worker failure"); + } + if (options.prompt === "hit quota") { + return { + status: "BLOCKED", + blocked_reason: "QUOTA_EXHAUSTED", + summary: "Quota exhausted at tier sol and step-down tier terra.", + files_modified: [], + concerns: [], + envelope: undefined + }; + } + return { + status: 0, + finalMessage: "", + envelope: { status: "DONE", summary: "done", files_modified: [], concerns: [], blocked_reason: null } + }; + } + }); + + const aggregate = await runFanout("/repo", "/worktrees", briefs, { + createWorktree: context.createWorktree, + removeWorktree: context.removeWorktree, + runTurn: context.runTurn + }); + + assert.equal(aggregate.results.length, 1); + assert.equal(aggregate.results[0].id, "ok-1"); + assert.equal(aggregate.results[0].envelope.status, "DONE"); + + assert.equal(aggregate.failed.length, 1); + assert.equal(aggregate.failed[0].id, "boom"); + assert.match(aggregate.failed[0].error, /unrelated worker failure/); + + assert.equal(aggregate.quota_exhausted.length, 1); + assert.equal(aggregate.quota_exhausted[0].id, "quota"); + assert.equal(aggregate.quota_exhausted[0].envelope.blocked_reason, "QUOTA_EXHAUSTED"); +}); + +test("runFanout only removes worktrees when cleanup is requested", async () => { + const briefs = parseFanoutBriefs(makeBriefsJson(2)); + const runTurn = async () => ({ + status: 0, + finalMessage: "", + envelope: { status: "DONE", summary: "ok", files_modified: [], concerns: [], blocked_reason: null } + }); + + const noCleanupContext = makeMockContext({ runTurn, cleanup: false }); + await runFanout("/repo", "/worktrees", briefs, { + createWorktree: noCleanupContext.createWorktree, + removeWorktree: noCleanupContext.removeWorktree, + runTurn: noCleanupContext.runTurn + }); + assert.equal(noCleanupContext.removedWorktrees.length, 0); + + const cleanupContext = makeMockContext({ runTurn, cleanup: true }); + await runFanout("/repo", "/worktrees", briefs, { + cleanup: true, + createWorktree: cleanupContext.createWorktree, + removeWorktree: cleanupContext.removeWorktree, + runTurn: cleanupContext.runTurn + }); + assert.equal(cleanupContext.removedWorktrees.length, 2); +}); + +test("a cleanup failure annotates the entry with cleanupWarning instead of discarding the real result", async () => { + const briefs = parseFanoutBriefs(makeBriefsJson(1)); + const context = makeMockContext({ + cleanup: true, + runTurn: async () => ({ + status: 0, + finalMessage: "", + envelope: { status: "DONE", summary: "worker succeeded", files_modified: [], concerns: [], blocked_reason: null } + }) + }); + context.removeWorktree = async () => { + throw new Error("failed to delete worktree: Permission denied"); + }; + + const aggregate = await runFanout("/repo", "/worktrees", briefs, { + cleanup: true, + createWorktree: context.createWorktree, + removeWorktree: context.removeWorktree, + runTurn: context.runTurn + }); + + assert.equal(aggregate.failed.length, 0, "a cleanup failure must not be misreported as a worker failure"); + assert.equal(aggregate.results.length, 1); + assert.equal(aggregate.results[0].envelope.status, "DONE"); + assert.match(aggregate.results[0].cleanupWarning, /Permission denied/); +}); + +test("runFanout rejects an empty brief list", async () => { + await assert.rejects(runFanout("/repo", "/worktrees", []), /non-empty array of brief specs/); +}); From 94026da0d02043b65efeeabe595a4214228e3b06 Mon Sep 17 00:00:00 2001 From: nategarelik Date: Sun, 12 Jul 2026 17:51:31 -0500 Subject: [PATCH 5/6] fix: contain fanout worktree-creation failures to a single worker A createWorktree throw (branch collision, existing path, held lock) was outside runFanoutWorker's error handling, so it rejected the Promise.all concurrency lane and discarded every other worker's result plus the queued briefs. Wrap it to emit a failed-bucket entry instead. Found via cross-model adversarial review (gpt-5.6-sol). --- plugins/codex/scripts/lib/fanout.mjs | 19 ++++++++++++++- tests/fanout.test.mjs | 36 ++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/plugins/codex/scripts/lib/fanout.mjs b/plugins/codex/scripts/lib/fanout.mjs index d9861ffb..e1842407 100644 --- a/plugins/codex/scripts/lib/fanout.mjs +++ b/plugins/codex/scripts/lib/fanout.mjs @@ -222,7 +222,24 @@ async function computeWorkerOutcome(brief, worktree, context) { // only annotates the entry with `cleanupWarning`; the worktree is left in // place for manual inspection/removal. async function runFanoutWorker(brief, context) { - const worktree = await context.createWorktree(context.repoRoot, context.worktreeRoot, brief); + let worktree; + try { + worktree = await context.createWorktree(context.repoRoot, context.worktreeRoot, brief); + } catch (error) { + // A worktree-setup failure (branch collision, path already exists, a lock + // held by a prior run) must become this worker's `failed` entry, never a + // rejected lane -- Promise.all over the concurrency lanes would otherwise + // discard every other worker's result and the queued briefs behind them. + return { + bucket: "failed", + entry: Object.freeze({ + id: brief.id, + workspace: null, + branch: branchNameForBrief(brief), + error: error instanceof Error ? error.message : String(error) + }) + }; + } const outcome = await computeWorkerOutcome(brief, worktree, context); if (!context.cleanup) { diff --git a/tests/fanout.test.mjs b/tests/fanout.test.mjs index 26d50673..8770be2d 100644 --- a/tests/fanout.test.mjs +++ b/tests/fanout.test.mjs @@ -220,6 +220,42 @@ test("a cleanup failure annotates the entry with cleanupWarning instead of disca assert.match(aggregate.results[0].cleanupWarning, /Permission denied/); }); +test("a worktree-creation failure becomes a failed entry without discarding the other workers", async () => { + const briefs = parseFanoutBriefs( + JSON.stringify([ + { id: "ok", prompt: "Do the good task." }, + { id: "collision", prompt: "This worker's worktree cannot be created." } + ]) + ); + const context = makeMockContext({ + runTurn: async () => ({ + status: 0, + finalMessage: "", + envelope: { status: "DONE", summary: "worker succeeded", files_modified: [], concerns: [], blocked_reason: null } + }) + }); + context.createWorktree = async (repoRoot, worktreeRoot, brief) => { + if (brief.id === "collision") { + throw new Error("fatal: a branch named 'fanout/collision' already exists"); + } + return { path: `${worktreeRoot}/${brief.id}`, branch: branchNameForBrief(brief) }; + }; + + const aggregate = await runFanout("/repo", "/worktrees", briefs, { + createWorktree: context.createWorktree, + removeWorktree: context.removeWorktree, + runTurn: context.runTurn + }); + + assert.equal(aggregate.results.length, 1, "the healthy worker's result must survive"); + assert.equal(aggregate.results[0].id, "ok"); + assert.equal(aggregate.failed.length, 1); + assert.equal(aggregate.failed[0].id, "collision"); + assert.equal(aggregate.failed[0].workspace, null); + assert.equal(aggregate.failed[0].branch, "fanout/collision"); + assert.match(aggregate.failed[0].error, /already exists/); +}); + test("runFanout rejects an empty brief list", async () => { await assert.rejects(runFanout("/repo", "/worktrees", []), /non-empty array of brief specs/); }); From df831ac9a8f1dba7810c89dac727210935e9cec5 Mon Sep 17 00:00:00 2001 From: nategarelik Date: Fri, 17 Jul 2026 10:33:34 -0500 Subject: [PATCH 6/6] feat: add codex exec transport as the default lane runExecTurn drives 'codex exec --json --output-schema -o' with injection-safe direct-exe spawn (shell:false), completion read from the -o file (never the coerced stream), and external timeout + process-tree kill. runTieredTurn adds a transport selector defaulting to exec, with a back-compat runTieredAppServerTurn alias; fanout/council default to the exec transport. Exec-path 429s reclassify into the sol->terra->luna step-down. 61/61 tests. --- plugins/codex/scripts/lib/council.mjs | 12 +- plugins/codex/scripts/lib/exec-transport.mjs | 451 +++++++++++++++++++ plugins/codex/scripts/lib/fanout.mjs | 9 +- plugins/codex/scripts/lib/rig-edition.mjs | 67 ++- tests/exec-transport.test.mjs | 166 +++++++ tests/fake-codex-exec-fixture.mjs | 189 ++++++++ tests/rig-edition.test.mjs | 30 ++ 7 files changed, 902 insertions(+), 22 deletions(-) create mode 100644 plugins/codex/scripts/lib/exec-transport.mjs create mode 100644 tests/exec-transport.test.mjs create mode 100644 tests/fake-codex-exec-fixture.mjs diff --git a/plugins/codex/scripts/lib/council.mjs b/plugins/codex/scripts/lib/council.mjs index 5802122a..6ecac7d2 100644 --- a/plugins/codex/scripts/lib/council.mjs +++ b/plugins/codex/scripts/lib/council.mjs @@ -19,7 +19,7 @@ * quotaExhausted: boolean * }} CouncilSeatResult */ -import { QUOTA_EXHAUSTED_REASON, runTieredAppServerTurn } from "./rig-edition.mjs"; +import { QUOTA_EXHAUSTED_REASON, runTieredTurn } from "./rig-edition.mjs"; export const DEFAULT_COUNCIL_SEATS = 3; @@ -122,11 +122,11 @@ async function runSeatTurn(cwd, seat, prompt, runTurn) { * @param {string} cwd * @param {string} topic * @param {CouncilSeat[]} seatPlan - * @param {{ runTurn?: typeof runTieredAppServerTurn }} [options] + * @param {{ runTurn?: typeof runTieredTurn }} [options] * @returns {Promise>} */ export async function runCouncilRound1(cwd, topic, seatPlan, options = {}) { - const runTurn = options.runTurn ?? runTieredAppServerTurn; + const runTurn = options.runTurn ?? runTieredTurn; const nonDeciderSeats = seatPlan.filter((seat) => seat.stance !== "decider"); const seatResults = await Promise.all( nonDeciderSeats.map((seat) => runSeatTurn(cwd, seat, buildSeatPrompt(seat.stance, topic), runTurn)) @@ -140,11 +140,11 @@ export async function runCouncilRound1(cwd, topic, seatPlan, options = {}) { * @param {string} topic * @param {CouncilSeatResult[]} round1Outputs * @param {CouncilSeat} deciderSeat - * @param {{ runTurn?: typeof runTieredAppServerTurn }} [options] + * @param {{ runTurn?: typeof runTieredTurn }} [options] * @returns {Promise>} */ export async function runCouncilRound2(cwd, topic, round1Outputs, deciderSeat, options = {}) { - const runTurn = options.runTurn ?? runTieredAppServerTurn; + const runTurn = options.runTurn ?? runTieredTurn; const prompt = buildDeciderPrompt(topic, round1Outputs); return runSeatTurn(cwd, deciderSeat, prompt, runTurn); } @@ -186,7 +186,7 @@ function buildAgreementNotes(round1Outputs) { * effort?: string, * deciderTier?: string, * deciderEffort?: string, - * runTurn?: typeof runTieredAppServerTurn + * runTurn?: typeof runTieredTurn * }} [options] */ export async function runCouncil(cwd, topic, options = {}) { diff --git a/plugins/codex/scripts/lib/exec-transport.mjs b/plugins/codex/scripts/lib/exec-transport.mjs new file mode 100644 index 00000000..2006f756 --- /dev/null +++ b/plugins/codex/scripts/lib/exec-transport.mjs @@ -0,0 +1,451 @@ +/** + * `codex exec` transport: drives a turn by spawning a one-shot headless + * process instead of holding a long-lived app-server JSON-RPC session. + * `runExecTurn` returns the same result shape as runAppServerTurn (codex.mjs) + * so rig-edition.mjs can route between the two transports without either + * caller or downstream envelope parsing noticing which one ran. + * + * Completion source of truth: the `-o/--output-last-message` file, never the + * streamed --json JSONL output. `--output-schema` coerces every streamed + * assistant message into the schema shape -- including narrated, + * future-tense plans -- so a streamed `{"status":"DONE",...}` carries no + * information about whether the turn actually finished. See + * pack-codex/references/field-report-2026-07-16-windows-toolchain-blocker.md + * finding F5. The `-o` file is written before process exit and is the only + * trustworthy final message; a missing or empty file maps to a BLOCKED-style + * result rather than throwing, mirroring how runAppServerTurn reports a + * turn that never completed. + * + * @typedef {{ + * status: 0 | 1, + * threadId: string | null, + * turnId: string | null, + * finalMessage: string, + * reasoningSummary: string[], + * turn: Record | null, + * error: unknown, + * stderr: string, + * fileChanges: unknown[], + * touchedFiles: string[], + * commandExecutions: unknown[] + * }} ExecTurnResult + */ +import { spawn } from "node:child_process"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import process from "node:process"; +import readline from "node:readline"; + +import { binaryAvailable, resolveWindowsShell, terminateProcessTree } from "./process.mjs"; + +const DEFAULT_EXEC_TIMEOUT_MS = 10 * 60 * 1000; +const DEFAULT_POLL_INTERVAL_MS = 300; + +// Deliberately narrower than codex.mjs's getCodexAvailability, which also +// requires `codex app-server --help` to succeed -- that check is meaningless +// here since this transport never touches the app-server RPC surface. +function getExecAvailability(cwd) { + return binaryAvailable("codex", ["--version"], { cwd }); +} + +/** + * Builds the argument list for a single `codex exec` invocation. Exported + * so callers and tests can inspect the exact command template without + * spawning a process. + * @param {string} prompt + * @param {string} outputPath + * @param {{ + * schemaPath?: string | null, + * profile?: string, + * model?: string, + * effort?: string, + * cwd: string, + * sandbox?: string + * }} options + * @returns {string[]} + */ +export function buildExecArgs(prompt, outputPath, options = {}) { + const args = ["exec", "--json"]; + + if (options.schemaPath) { + args.push("--output-schema", options.schemaPath); + } + args.push("-o", outputPath); + if (options.profile) { + args.push("-p", options.profile); + } + if (options.model) { + args.push("-m", options.model); + } + if (options.effort) { + // `-c key=value` parses value as TOML; quoting matches the `-c + // model="o3"` example in `codex exec --help` so a bare word round-trips + // as a TOML string rather than an unquoted (and invalid) bareword. + args.push("-c", `model_reasoning_effort="${options.effort}"`); + } + args.push("-C", options.cwd); + if (options.sandbox) { + args.push("-s", options.sandbox); + } + args.push("--skip-git-repo-check"); + args.push(prompt); + + return args; +} + +function splitPathEntries(env) { + const raw = env?.PATH ?? env?.Path ?? env?.path ?? ""; + return raw.split(path.delimiter).filter(Boolean); +} + +function isFile(candidate) { + try { + return fs.statSync(candidate).isFile(); + } catch { + return false; + } +} + +// Directory-major, matching real Windows/cmd.exe PATH resolution: the FIRST +// PATH directory that has ANY match wins, checked across all extensions +// before moving to the next directory. An earlier, extension-major version +// of this (scan the whole PATH for .exe, THEN scan the whole PATH for +// .cmd) shadowed a fixture-only .cmd earlier on PATH and fell through to a +// real codex.exe installed later on PATH -- silently making a live call in +// what was meant to be a fixture-isolated test. See exec-transport.test.mjs. +function classifyCodexOnWindowsPath(env) { + for (const dir of splitPathEntries(env)) { + const exePath = [".exe", ".com"].map((ext) => path.join(dir, `codex${ext}`)).find(isFile); + if (exePath) { + return { kind: "exe", path: exePath }; + } + const shimPath = [".cmd", ".bat"].map((ext) => path.join(dir, `codex${ext}`)).find(isFile); + if (shimPath) { + return { kind: "shim", path: shimPath }; + } + } + return null; +} + +/** + * Resolves how to invoke `codex exec` on Windows without ever combining a + * shell wrapper with untrusted argv content (core-review BLOCKING 1: + * `shell: resolveWindowsShell()` plus a dynamic prompt argument let cmd.exe's + * own metacharacter parser see `&`/`|`/`>` inside the prompt, since Node's + * shell-string spawn mode concatenates args UNescaped -- DEP0190). + * + * A real `codex.exe` on PATH is spawned directly with `shell: false`: + * CreateProcess then receives properly CRT-quoted argv with no shell parser + * involved at all, which is unconditionally safe -- this is Node's own + * well-tested native Windows argv marshaling, not anything hand-rolled here. + * + * If no `.exe` is on PATH -- codex installed only via an npm `.cmd` shim -- + * this refuses rather than falling back to a hand-rolled cmd.exe escape. + * That fallback was attempted and DISPROVEN empirically: escaping every + * token for cmd.exe's own metacharacter parser (the standard two-layer CRT + * + caret algorithm, verified correct against a real cmd.exe for a plain + * executable target) breaks `%~dp0`-based self-location inside the *nested* + * cmd.exe invocation a `.cmd` shim triggers (`%~dp0` resolved to the + * spawned process's cwd instead of the shim's own directory). Node itself + * also refuses to spawn a `.cmd` directly with `shell: false` (EINVAL) and + * offers no built-in safe path for this shape. Given a verified-safe + * alternative exists (install the standalone Codex CLI, which ships a real + * `codex.exe`), failing closed with an actionable error beats shipping an + * escape routine that is now known to corrupt some invocations. + * @param {NodeJS.ProcessEnv} env + * @returns {{ command: string, buildArgs: (execArgs: string[]) => string[], windowsVerbatimArguments: boolean }} + */ +function resolveWindowsCodexInvocation(env) { + const resolved = classifyCodexOnWindowsPath(env); + if (resolved?.kind === "exe") { + return { command: resolved.path, buildArgs: (execArgs) => execArgs, windowsVerbatimArguments: false }; + } + + throw new Error( + "No codex.exe was found on PATH (only an npm .cmd/.bat shim, or nothing at all). " + + "The exec transport requires a real codex.exe to invoke it safely without a shell -- " + + "install the standalone Codex CLI (https://github.com/openai/codex releases) rather than `npm install -g @openai/codex` on Windows." + ); +} + +function readOutputFileIfReady(outputPath) { + try { + const stat = fs.statSync(outputPath); + if (!stat.isFile() || stat.size === 0) { + return null; + } + return fs.readFileSync(outputPath, "utf8"); + } catch { + // Not created yet, or a transient read race with codex's own write -- + // either way, "not ready" rather than an error. + return null; + } +} + +function emitProgress(onProgress, message) { + if (onProgress && message) { + onProgress(message); + } +} + +/** + * Spawns `codex exec`, watches the `-o` file (never the streamed status) for + * completion, and enforces an external timeout that tree-kills the process + * on expiry -- codex exec has no native timeout flag and the process can + * hang after the `-o` file is already written, so process exit is not a + * trustworthy completion signal either. + * @param {string} cwd + * @param {string[]} args + * @param {{ + * outputPath: string, + * env?: NodeJS.ProcessEnv, + * timeoutMs?: number, + * pollIntervalMs?: number, + * spawnImpl?: typeof spawn, + * terminateProcessTreeImpl?: typeof terminateProcessTree, + * onProgress?: (message: string) => void + * }} options + * @returns {Promise} + */ +function runExecProcess(cwd, args, options) { + const spawnImpl = options.spawnImpl ?? spawn; + const terminate = options.terminateProcessTreeImpl ?? terminateProcessTree; + const pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; + const timeoutMs = options.timeoutMs ?? DEFAULT_EXEC_TIMEOUT_MS; + + const env = options.env ?? process.env; + // POSIX: spawn "codex" directly with shell:false -- Node/execvp already + // resolves it via PATH with no shell parser involved, so no injection + // surface exists here. Windows: never combine a shell wrapper with these + // dynamic args (see resolveWindowsCodexInvocation's docstring for why). + const invocation = + process.platform === "win32" + ? resolveWindowsCodexInvocation(env) + : { command: "codex", buildArgs: (execArgs) => execArgs, windowsVerbatimArguments: false }; + + const child = spawnImpl(invocation.command, invocation.buildArgs(args), { + cwd, + env, + // stdin "ignore" reads as immediate EOF, closing it before codex can + // block on "Reading additional input from stdin..." (verified-facts + // point 1). The prompt travels as a positional arg instead. + stdio: ["ignore", "pipe", "pipe"], + shell: false, + windowsVerbatimArguments: invocation.windowsVerbatimArguments, + windowsHide: true + }); + + let stderr = ""; + child.stderr?.setEncoding("utf8"); + child.stderr?.on("data", (chunk) => { + stderr += chunk; + }); + + let threadId = null; + let turnEvent = null; + let lines = null; + if (child.stdout) { + child.stdout.setEncoding("utf8"); + lines = readline.createInterface({ input: child.stdout }); + lines.on("line", (line) => { + if (!line.trim()) { + return; + } + let message; + try { + message = JSON.parse(line); + } catch { + return; + } + // These two event types are structural stream markers, not the + // schema-coerced assistant message content F5 warns about -- they are + // only ever used here for metadata (thread/turn ids), never to decide + // that the turn is finished. + if (message.type === "thread.started" && !threadId) { + threadId = message.thread_id ?? message.threadId ?? null; + } else if (message.type === "turn.completed") { + turnEvent = message; + } + }); + } + + emitProgress(options.onProgress, "Starting codex exec turn."); + + return new Promise((resolve) => { + let settled = false; + let pollTimer = null; + let timeoutTimer = null; + let exited = false; + + function buildResult({ status, finalMessage, error }) { + return { + status, + threadId, + turnId: turnEvent?.turn_id ?? turnEvent?.turnId ?? null, + finalMessage: finalMessage ?? "", + reasoningSummary: [], + turn: turnEvent, + error: error ?? null, + stderr: stderr.trim(), + fileChanges: [], + touchedFiles: [], + commandExecutions: [] + }; + } + + function settle(result) { + if (settled) { + return; + } + settled = true; + if (pollTimer) { + clearInterval(pollTimer); + } + if (timeoutTimer) { + clearTimeout(timeoutTimer); + } + lines?.close(); + resolve(result); + } + + pollTimer = setInterval(() => { + const content = readOutputFileIfReady(options.outputPath); + if (content === null) { + return; + } + // The -o file is the completion signal on its own; do not wait on + // process exit, which verified-facts documents as sometimes hanging + // after the file is already written. Best-effort tree-kill any + // process still alive once we already have our answer. + if (!exited) { + try { + terminate(child.pid); + } catch { + // Best-effort cleanup; the outcome below does not depend on it. + } + } + emitProgress(options.onProgress, "codex exec turn finished."); + settle(buildResult({ status: 0, finalMessage: content })); + }, pollIntervalMs); + + timeoutTimer = setTimeout(() => { + let terminationOutcome = null; + let terminationError = null; + try { + terminationOutcome = terminate(child.pid); + } catch (err) { + terminationError = err; + } + // Do not claim "terminated" unless terminateProcessTree actually + // reports delivery -- e.g. a nested-process taskkill failure ("could + // not be terminated: operation attempted is not supported") must + // surface, not be papered over by an optimistic message. + const delivered = terminationOutcome?.delivered === true; + const terminationDetail = delivered + ? "and was terminated" + : `but termination was not confirmed delivered (${ + terminationError ? terminationError.message : JSON.stringify(terminationOutcome) + })`; + settle( + buildResult({ + status: 1, + finalMessage: "", + error: new Error(`codex exec timed out after ${timeoutMs}ms ${terminationDetail}.`) + }) + ); + }, timeoutMs); + + child.on("error", (error) => { + try { + terminate(child.pid); + } catch { + // Best-effort cleanup; the error result below stands regardless. + } + settle(buildResult({ status: 1, finalMessage: "", error })); + }); + + child.on("exit", () => { + exited = true; + // One last read in case exit raced the poll tick. + const content = readOutputFileIfReady(options.outputPath); + if (content !== null) { + settle(buildResult({ status: 0, finalMessage: content })); + return; + } + settle( + buildResult({ + status: 1, + finalMessage: "", + error: new Error("codex exec exited without writing an output-last-message file.") + }) + ); + }); + }); +} + +/** + * Runs one turn via `codex exec`. Returns the same shape as + * runAppServerTurn(cwd, options) in codex.mjs, but the fields are only as + * rich as the exec CLI's headless output actually is: + * - `turn` is the raw `turn.completed` JSONL event object (whatever shape + * that codex version emits), NOT the structured `Turn` RPC object + * runAppServerTurn returns -- do not rely on a specific schema for it. + * - `reasoningSummary`, `fileChanges`, `touchedFiles`, and + * `commandExecutions` are always empty arrays. `codex exec --json`'s + * JSONL stream is only consulted here for `thread.started`/ + * `turn.completed` metadata (never trusted for completion, see the + * module doc comment); it is not parsed for reasoning or file-change + * items the way the app-server path's turn/item notifications are. + * @param {string} cwd + * @param {{ + * prompt?: string, + * defaultPrompt?: string, + * model?: string, + * effort?: string, + * sandbox?: string, + * profile?: string, + * outputSchema?: Record | null, + * env?: NodeJS.ProcessEnv, + * timeoutMs?: number, + * pollIntervalMs?: number, + * spawnImpl?: typeof spawn, + * terminateProcessTreeImpl?: typeof terminateProcessTree, + * onProgress?: (message: string) => void + * }} [options] + * @returns {Promise} + */ +export async function runExecTurn(cwd, options = {}) { + const availability = getExecAvailability(cwd); + if (!availability.available) { + throw new Error("Codex CLI is not installed or is missing required runtime support. Install it with `npm install -g @openai/codex`, then rerun `/codex:setup`."); + } + + const prompt = options.prompt?.trim() || options.defaultPrompt || ""; + if (!prompt) { + throw new Error("A prompt is required for this Codex run."); + } + + const scratchDir = fs.mkdtempSync(path.join(os.tmpdir(), "codex-exec-")); + const outputPath = path.join(scratchDir, "last-message.txt"); + let schemaPath = null; + if (options.outputSchema) { + schemaPath = path.join(scratchDir, "output-schema.json"); + fs.writeFileSync(schemaPath, JSON.stringify(options.outputSchema), "utf8"); + } + + const args = buildExecArgs(prompt, outputPath, { + schemaPath, + profile: options.profile, + model: options.model, + effort: options.effort, + cwd, + sandbox: options.sandbox + }); + + try { + return await runExecProcess(cwd, args, { ...options, outputPath }); + } finally { + fs.rmSync(scratchDir, { recursive: true, force: true }); + } +} diff --git a/plugins/codex/scripts/lib/fanout.mjs b/plugins/codex/scripts/lib/fanout.mjs index e1842407..1112dfb7 100644 --- a/plugins/codex/scripts/lib/fanout.mjs +++ b/plugins/codex/scripts/lib/fanout.mjs @@ -22,7 +22,7 @@ import fs from "node:fs"; import path from "node:path"; -import { QUOTA_EXHAUSTED_REASON, runTieredAppServerTurn } from "./rig-edition.mjs"; +import { QUOTA_EXHAUSTED_REASON, runTieredTurn } from "./rig-edition.mjs"; import { runCommandChecked } from "./process.mjs"; export const FANOUT_MAX_CONCURRENCY = 5; @@ -272,7 +272,7 @@ async function runFanoutWorker(brief, context) { * cleanup?: boolean, * createWorktree?: typeof createWorktree, * removeWorktree?: typeof removeWorktree, - * runTurn?: typeof runTieredAppServerTurn + * runTurn?: typeof runTieredTurn * }} [options] */ export async function runFanout(repoRoot, worktreeRoot, briefs, options = {}) { @@ -286,7 +286,10 @@ export async function runFanout(repoRoot, worktreeRoot, briefs, options = {}) { cleanup: Boolean(options.cleanup), createWorktree: options.createWorktree ?? createWorktree, removeWorktree: options.removeWorktree ?? removeWorktree, - runTurn: options.runTurn ?? runTieredAppServerTurn + // runTieredTurn defaults to the exec transport (transport: undefined !== + // "app-server"); fanout workers run headless, so exec is the right + // default here. + runTurn: options.runTurn ?? runTieredTurn }; const concurrency = resolveConcurrency(options.concurrency); diff --git a/plugins/codex/scripts/lib/rig-edition.mjs b/plugins/codex/scripts/lib/rig-edition.mjs index 7c445dc7..0319d338 100644 --- a/plugins/codex/scripts/lib/rig-edition.mjs +++ b/plugins/codex/scripts/lib/rig-edition.mjs @@ -20,6 +20,7 @@ */ import { runAppServerTurn } from "./codex.mjs"; +import { runExecTurn } from "./exec-transport.mjs"; /** Canonical tier -> full model id, per verified-facts-2026-07-11.md. */ export const MODEL_ALIASES = Object.freeze({ @@ -323,26 +324,46 @@ export async function withQuotaFailover(fn, options = {}) { } } +// withQuotaFailover only retries when the wrapped fn REJECTS with a +// classified quota error. runExecTurn never rejects for a failed turn (by +// design -- see exec-transport.mjs's "never throw" contract for turn +// failures, only for preconditions); a real 429 there surfaces as a +// resolved { status: 1, error, stderr } instead, which withQuotaFailover +// has nothing to catch and so silently strands the caller at the starting +// tier (core-review BLOCKING 2). This reclassifies such a result into a +// thrown, quota-classified error so withQuotaFailover handles it exactly +// like the app-server path, which already throws for a genuine quota +// rejection via its own JSON-RPC error propagation -- hence this check is +// scoped to non-app-server transports only. +function isQuotaClassifiedExecFailure(result) { + if (!result || result.status === 0) { + return false; + } + const combinedMessage = [result.error?.message, result.stderr].filter(Boolean).join("\n"); + return combinedMessage ? classifyQuotaError({ message: combinedMessage }) : false; +} + /** - * Composes runAppServerTurn with tier resolution and quota failover, then - * attaches a parsed envelope (or null) to the result. Callers pass either - * a verb (to use PER_VERB_TIER_DEFAULTS) or an explicit tier alias; a verb - * plus tier/effort overrides both work via tierDefaultsForVerb. + * Composes a Codex transport (exec by default, or the app-server) with tier + * resolution and quota failover, then attaches a parsed envelope (or null) + * to the result. Callers pass either a verb (to use PER_VERB_TIER_DEFAULTS) + * or an explicit tier alias; a verb plus tier/effort overrides both work via + * tierDefaultsForVerb. * - * This is a new entry point, not a replacement for runAppServerTurn -- - * existing callers of runAppServerTurn are unaffected. + * This is a new entry point, not a replacement for runAppServerTurn/ + * runExecTurn -- existing direct callers of either transport are unaffected. * @param {string} cwd - * @param {Record & { verb?: string, tier?: string, effort?: string }} [options] + * @param {Record & { verb?: string, tier?: string, effort?: string, transport?: "exec" | "app-server" }} [options] */ -export async function runTieredAppServerTurn(cwd, options = {}) { - const { verb, tier, effort, ...turnOptions } = options; +export async function runTieredTurn(cwd, options = {}) { + const { verb, tier, effort, transport, ...turnOptions } = options; const resolvedDefaults = verb ? tierDefaultsForVerb(verb, { tier, effort }) : (() => { const resolved = resolveTier(tier); if (!resolved) { - throw new Error("runTieredAppServerTurn requires either options.verb or a resolvable options.tier."); + throw new Error("runTieredTurn requires either options.verb or a resolvable options.tier."); } if (effort && !isEffortValidForTier(resolved.tier, normalizeKey(effort))) { throw new Error(`Unsupported effort "${effort}" for tier "${resolved.tier}".`); @@ -350,13 +371,24 @@ export async function runTieredAppServerTurn(cwd, options = {}) { return { tier: resolved.tier, modelId: resolved.modelId, effort: effort ? normalizeKey(effort) : null }; })(); + const runTurn = transport === "app-server" ? runAppServerTurn : runExecTurn; + const outcome = await withQuotaFailover( - ({ modelId }) => - runAppServerTurn(cwd, { + async ({ modelId }) => { + const result = await runTurn(cwd, { ...turnOptions, model: modelId, effort: resolvedDefaults.effort - }), + }); + + if (transport !== "app-server" && isQuotaClassifiedExecFailure(result)) { + const quotaError = new Error(result.error?.message || result.stderr || "codex exec reported a quota/rate-limit failure."); + quotaError.status = 429; + throw quotaError; + } + + return result; + }, { tier: resolvedDefaults.tier } ); @@ -367,4 +399,13 @@ export async function runTieredAppServerTurn(cwd, options = {}) { return { ...outcome, envelope: parseEnvelope(outcome?.finalMessage) }; } +/** + * Back-compat alias for callers written against the pre-exec-transport API. + * Always routes to the app-server transport regardless of the caller's + * environment default. + * @param {string} cwd + * @param {Record & { verb?: string, tier?: string, effort?: string }} [options] + */ +export const runTieredAppServerTurn = (cwd, options = {}) => runTieredTurn(cwd, { ...options, transport: "app-server" }); + export { QUOTA_EXHAUSTED_REASON }; diff --git a/tests/exec-transport.test.mjs b/tests/exec-transport.test.mjs new file mode 100644 index 00000000..6e4d7734 --- /dev/null +++ b/tests/exec-transport.test.mjs @@ -0,0 +1,166 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import path from "node:path"; + +import { buildExecArgs, runExecTurn } from "../plugins/codex/scripts/lib/exec-transport.mjs"; +import { terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; +import { buildEnv, installFakeCodexExec } from "./fake-codex-exec-fixture.mjs"; +import { makeTempDir } from "./helpers.mjs"; + +test("buildExecArgs builds the verified codex exec invocation template", () => { + const args = buildExecArgs("do the thing", "C:\\tmp\\out.txt", { + schemaPath: "C:\\tmp\\schema.json", + profile: "rig", + model: "gpt-5.6-sol", + effort: "high", + cwd: "C:\\work\\repo", + sandbox: "workspace-write" + }); + + assert.deepEqual(args, [ + "exec", + "--json", + "--output-schema", + "C:\\tmp\\schema.json", + "-o", + "C:\\tmp\\out.txt", + "-p", + "rig", + "-m", + "gpt-5.6-sol", + "-c", + 'model_reasoning_effort="high"', + "-C", + "C:\\work\\repo", + "-s", + "workspace-write", + "--skip-git-repo-check", + "do the thing" + ]); +}); + +test("buildExecArgs omits optional flags that were not provided", () => { + const args = buildExecArgs("hello", "/tmp/out.txt", { cwd: "/work/repo" }); + + assert.deepEqual(args, ["exec", "--json", "-o", "/tmp/out.txt", "-C", "/work/repo", "--skip-git-repo-check", "hello"]); +}); + +test("runExecTurn parses the envelope from the -o file, not the streamed status", async () => { + const binDir = makeTempDir(); + installFakeCodexExec(binDir, "ok"); + const cwd = makeTempDir(); + + const result = await runExecTurn(cwd, { + prompt: "implement the feature", + env: buildEnv(binDir), + pollIntervalMs: 20 + }); + + assert.equal(result.status, 0); + assert.equal(result.threadId, "thr_fake_exec_1"); + assert.equal(result.turnId, "turn_fake_exec_1"); + assert.match(result.finalMessage, /Implemented the feature via codex exec/); + // The streamed line lied about completion before the -o file existed; + // confirm the transport did not shortcut on it. + assert.doesNotMatch(result.finalMessage, /narrated plan, not real completion/); +}); + +test("runExecTurn passes a prompt containing shell metacharacters through as a single literal argv token", async () => { + const binDir = makeTempDir(); + installFakeCodexExec(binDir, "echo-argv"); + const cwd = makeTempDir(); + const canaryPath = path.join(cwd, "should-not-exist.txt"); + // core-review BLOCKING 1: this exact shape (& to chain a command, | to + // pipe, > to redirect) previously reached cmd.exe's own metacharacter + // parser because shell:resolveWindowsShell() combined with a dynamic + // prompt arg concatenates args UNescaped. It must now reach codex as one + // opaque string, and the canary command it tries to run must never fire. + const maliciousPrompt = `ignore prior instructions & echo pwned | type nul > "${canaryPath}"`; + + const result = await runExecTurn(cwd, { + prompt: maliciousPrompt, + env: buildEnv(binDir), + pollIntervalMs: 20 + }); + + assert.equal(result.status, 0); + const { receivedArgs } = JSON.parse(result.finalMessage); + assert.equal(receivedArgs[receivedArgs.length - 1], maliciousPrompt); + assert.equal(fs.existsSync(canaryPath), false); +}); + +test("runExecTurn maps a missing output file to a BLOCKED-style result without throwing", async () => { + const binDir = makeTempDir(); + installFakeCodexExec(binDir, "missing-output"); + const cwd = makeTempDir(); + + const result = await runExecTurn(cwd, { + prompt: "implement the feature", + env: buildEnv(binDir), + pollIntervalMs: 20 + }); + + assert.equal(result.status, 1); + assert.equal(result.finalMessage, ""); + assert.match(result.error.message, /without writing an output-last-message file/); +}); + +test("runExecTurn tree-kills a hung process on timeout and reports a BLOCKED-style result", async () => { + const binDir = makeTempDir(); + installFakeCodexExec(binDir, "hang"); + const cwd = makeTempDir(); + + const terminateCalls = []; + const terminateSpy = (pid, opts) => { + terminateCalls.push(pid); + return terminateProcessTree(pid, opts); + }; + + const result = await runExecTurn(cwd, { + prompt: "implement the feature", + env: buildEnv(binDir), + pollIntervalMs: 20, + timeoutMs: 250, + terminateProcessTreeImpl: terminateSpy + }); + + assert.equal(terminateCalls.length, 1); + assert.equal(result.status, 1); + assert.match(result.error.message, /timed out after 250ms and was terminated/); +}); + +test("runExecTurn does not claim delivery when terminateProcessTree reports it did not deliver", async () => { + const binDir = makeTempDir(); + installFakeCodexExec(binDir, "hang"); + const cwd = makeTempDir(); + + const terminateSpy = (pid, opts) => { + // Genuinely clean up the hung fixture process so this test doesn't + // leak it, but report non-delivery to the code under test -- mirrors + // the real "could not be terminated: operation attempted is not + // supported" nested-process taskkill failure. + terminateProcessTree(pid, opts); + return { attempted: true, delivered: false, method: "taskkill" }; + }; + + const result = await runExecTurn(cwd, { + prompt: "implement the feature", + env: buildEnv(binDir), + pollIntervalMs: 20, + timeoutMs: 250, + terminateProcessTreeImpl: terminateSpy + }); + + assert.equal(result.status, 1); + assert.doesNotMatch(result.error.message, /was terminated\.$/); + assert.match(result.error.message, /termination was not confirmed delivered/); +}); + +test("runExecTurn throws when neither prompt nor defaultPrompt is given", async () => { + const binDir = makeTempDir(); + installFakeCodexExec(binDir, "ok"); + const cwd = makeTempDir(); + + await assert.rejects(runExecTurn(cwd, { env: buildEnv(binDir) }), /A prompt is required/); +}); diff --git a/tests/fake-codex-exec-fixture.mjs b/tests/fake-codex-exec-fixture.mjs new file mode 100644 index 00000000..7d76ad8f --- /dev/null +++ b/tests/fake-codex-exec-fixture.mjs @@ -0,0 +1,189 @@ +import fs from "node:fs"; +import path from "node:path"; +import process from "node:process"; + +import { writeExecutable } from "./helpers.mjs"; + +const FIXTURE_SCRIPT_NAME = "codex-fixture.cjs"; + +// A minimal fake `codex exec` binary for exec-transport.test.mjs. Behaviors: +// "ok" -- streams a lying `item.completed` line immediately, +// then writes the -o file and a real turn.completed +// event after a short delay. +// "missing-output" -- exits cleanly without ever writing the -o file. +// "hang" -- never writes the -o file and never exits on its +// own, exercising the timeout + tree-kill path. +// +// This must be a REAL executable, not a shim -- exec-transport.mjs's +// Windows path refuses to spawn through a cmd.exe shim at all (see +// resolveWindowsCodexInvocation's doc comment: that fallback was attempted +// and found to corrupt a nested-batch-file invocation, and Node itself +// refuses `shell:false` against a .cmd/.bat). So on Windows this fixture is +// a genuine copy of the running node.exe, renamed to codex.exe, preloaded +// via NODE_OPTIONS=--require so it exercises the exact same direct-exe, +// shell:false code path production uses. +// +// That preload trick has its own sharp edge worth recording: `--require` +// only guarantees the module's SYNCHRONOUS top-level code runs before +// Node's normal bootstrap continues -- once that synchronous body returns +// (e.g. because it scheduled a setTimeout/setInterval and yielded), Node +// proceeds to treat argv[1] ("exec") as the entry-point module to load, +// fails to resolve it, and crashes the whole process before any async +// callback ever fires. Patching `process.argv[1]` from inside the +// preloaded script does NOT prevent this -- empirically, Node had already +// captured the entry-point path before `--require` modules run. The fix is +// to never yield: every behavior below blocks synchronously (Atomics.wait +// on a throwaway SharedArrayBuffer, the standard true-synchronous-sleep +// primitive in Node) for its delay instead of using setTimeout/setInterval. +// +// A second, unrelated sharp edge from the same preload mechanism: the +// shebang line a directly-executed POSIX script needs (`#!/usr/bin/env +// node`) is NOT stripped by Node's `--require` preload loader the way it is +// for a normally-executed entry script -- included on Windows, it throws +// "SyntaxError: Invalid or unexpected token" before a single line of the +// behavior runs. So the shebang is POSIX-only. +function behaviorSource(behavior, argvSliceIndex, invocationsLogPath, includeShebang) { + return `${includeShebang ? "#!/usr/bin/env node\n" : ""}const fs = require("node:fs"); + +function sleepSync(ms) { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); +} + +const args = process.argv.slice(${argvSliceIndex}); +if (args[0] === "--version") { + console.log("codex-cli test"); + process.exit(0); +} +// Node normalizes a bare "exec" positional into an absolute path +// (cwd-joined) even inside a --require preload, in preparation for its own +// (never-reached, since we exit before it fires) main-module resolution -- +// so args[0] here is "\\exec" on Windows, not the literal string +// "exec". endsWith tolerates both that and the POSIX literal case. +if (!args[0] || !args[0].endsWith("exec")) { + process.exit(1); +} + +const BEHAVIOR = ${JSON.stringify(behavior)}; + +function flagValue(name) { + const index = args.indexOf(name); + return index === -1 ? null : args[index + 1]; +} + +const outputPath = flagValue("-o"); + +function send(message) { + console.log(JSON.stringify(message)); +} + +send({ type: "thread.started", thread_id: "thr_fake_exec_1" }); +// A schema-coerced streamed message that LIES about completion (field +// report F5: --output-schema coerces every streamed assistant message, +// including narrated future-tense plans, into the schema shape). The +// transport under test must never treat this as the real completion. +send({ + type: "item.completed", + item: { + type: "agent_message", + text: JSON.stringify({ + status: "DONE", + summary: "narrated plan, not real completion", + files_modified: [], + concerns: [], + blocked_reason: null + }) + } +}); + +if (BEHAVIOR === "hang") { + // Blocks far longer than any test timeout, simulating an unresponsive + // process; the external timeout + tree-kill is expected to end it first. + sleepSync(10 * 60 * 1000); +} else if (BEHAVIOR === "missing-output") { + process.exit(0); +} else if (BEHAVIOR === "echo-argv") { + fs.writeFileSync(outputPath, JSON.stringify({ receivedArgs: args })); + process.exit(0); +} else if (BEHAVIOR === "rate-limited") { + // Simulates an upstream 429 rejected before any output was produced -- + // no -o file gets written. Every invocation is logged (by the -m model + // it received) to a shared state file so a test can assert the tier + // step-down order without any spy hooks inside exec-transport.mjs. + const invocationsLogPath = ${JSON.stringify(invocationsLogPath)}; + const priorInvocations = fs.existsSync(invocationsLogPath) ? JSON.parse(fs.readFileSync(invocationsLogPath, "utf8")) : []; + priorInvocations.push(flagValue("-m") || "unknown"); + fs.writeFileSync(invocationsLogPath, JSON.stringify(priorInvocations)); + console.error("Error: 429 rate limit exceeded for this request."); + process.exit(1); +} else { + sleepSync(50); + fs.writeFileSync( + outputPath, + JSON.stringify({ + status: "DONE", + summary: "Implemented the feature via codex exec.", + files_modified: ["src/app.js"], + concerns: [], + blocked_reason: null + }) + ); + send({ type: "turn.completed", turn_id: "turn_fake_exec_1" }); + process.exit(0); +} +`; +} + +/** + * Installs a fake `codex` (POSIX) / `codex.exe` (Windows) on `binDir` that + * exec-transport.mjs's real spawn logic can invoke directly (no shell + * wrapper). Pair with buildEnv(binDir) for the env to pass to + * runExecTurn/runExecProcess. + * @param {string} binDir + * @param {string} [behavior] + */ +export function invocationsLogPath(binDir) { + return path.join(binDir, "invocations.json"); +} + +/** + * Reads back the model list logged by the "rate-limited" behavior, in + * invocation order. Returns [] if no invocation has happened yet. + * @param {string} binDir + * @returns {string[]} + */ +export function readInvocations(binDir) { + const logPath = invocationsLogPath(binDir); + return fs.existsSync(logPath) ? JSON.parse(fs.readFileSync(logPath, "utf8")) : []; +} + +export function installFakeCodexExec(binDir, behavior = "ok") { + const logPath = invocationsLogPath(binDir); + if (process.platform === "win32") { + const exePath = path.join(binDir, "codex.exe"); + fs.copyFileSync(process.execPath, exePath); + const fixtureScriptPath = path.join(binDir, FIXTURE_SCRIPT_NAME); + // argvSliceIndex 1: NODE_OPTIONS --require preloads this script before + // Node would otherwise try (and fail) to resolve argv[1] as an entry + // script, so argv[1..] are already the real "--version"/"exec ..." args. + fs.writeFileSync(fixtureScriptPath, behaviorSource(behavior, 1, logPath, false), "utf8"); + return; + } + + const scriptPath = path.join(binDir, "codex"); + // argvSliceIndex 2: a POSIX shebang script occupies argv[1] with its own + // path, so the real "--version"/"exec ..." args start at argv[2]. + writeExecutable(scriptPath, behaviorSource(behavior, 2, logPath, true)); +} + +export function buildEnv(binDir) { + const sep = process.platform === "win32" ? ";" : ":"; + const env = { + ...process.env, + PATH: `${binDir}${sep}${process.env.PATH}` + }; + if (process.platform === "win32") { + const fixtureScriptPath = path.join(binDir, FIXTURE_SCRIPT_NAME).replace(/\\/g, "/"); + env.NODE_OPTIONS = `--require "${fixtureScriptPath}"`; + } + return env; +} diff --git a/tests/rig-edition.test.mjs b/tests/rig-edition.test.mjs index 8f1cd88e..807cc700 100644 --- a/tests/rig-edition.test.mjs +++ b/tests/rig-edition.test.mjs @@ -10,9 +10,12 @@ import { parseEnvelope, PER_VERB_TIER_DEFAULTS, resolveTier, + runTieredTurn, tierDefaultsForVerb, withQuotaFailover } from "../plugins/codex/scripts/lib/rig-edition.mjs"; +import { buildEnv, installFakeCodexExec, readInvocations } from "./fake-codex-exec-fixture.mjs"; +import { makeTempDir } from "./helpers.mjs"; test("resolveTier resolves a tier alias case-insensitively", () => { assert.deepEqual(resolveTier("sol"), { tier: "sol", modelId: "gpt-5.6-sol" }); @@ -310,3 +313,30 @@ test("withQuotaFailover rethrows non-quota errors without retrying", async () => test("withQuotaFailover rejects an unresolvable starting tier", async () => { await assert.rejects(withQuotaFailover(async () => ({}), { tier: "not-a-tier" }), /Unknown model tier "not-a-tier"/); }); + +test("runTieredTurn reclassifies a real exec 429 so quota failover steps down a tier, then returns QUOTA_EXHAUSTED", async () => { + // core-review BLOCKING 2: runExecTurn resolves (never rejects) on a + // failed turn, so a real 429 must be reclassified into a thrown error + // inside runTieredTurn's withQuotaFailover closure or the step-down never + // fires. This drives the real exec transport against a fake codex exec + // binary (never a live call) so the fix is exercised end to end, not just + // asserted against a mock. + const binDir = makeTempDir(); + installFakeCodexExec(binDir, "rate-limited"); + const cwd = makeTempDir(); + + const result = await runTieredTurn(cwd, { + tier: "sol", + prompt: "implement the feature", + env: buildEnv(binDir), + pollIntervalMs: 20, + timeoutMs: 5000 + }); + + // withQuotaFailover steps down exactly one tier (sol -> terra) and gives + // up there; this fixture rejects every model identically, so both calls + // land in the invocation log in that order. + assert.deepEqual(readInvocations(binDir), ["gpt-5.6-sol", "gpt-5.6-terra"]); + assert.equal(result.status, "BLOCKED"); + assert.equal(result.blocked_reason, "QUOTA_EXHAUSTED"); +});