From 8f8c3ae32a417b84091de12bf0535b995d1e6634 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Thu, 10 Sep 2026 14:43:51 +0000 Subject: [PATCH 1/4] feat: retry transient Codex capacity failures --- src/billing-events.ts | 6 + src/billing-run.ts | 70 ++++++++++- src/billing-usage.ts | 6 +- src/capacity-retry.ts | 37 ++++++ tests/billing-usage.test.ts | 3 +- tests/capacity-retry.test.ts | 221 +++++++++++++++++++++++++++++++++++ 6 files changed, 335 insertions(+), 8 deletions(-) create mode 100644 src/capacity-retry.ts create mode 100644 tests/capacity-retry.test.ts diff --git a/src/billing-events.ts b/src/billing-events.ts index b6e223b..e1c07d6 100644 --- a/src/billing-events.ts +++ b/src/billing-events.ts @@ -1,4 +1,5 @@ import { StringDecoder } from "node:string_decoder"; +import { CODEX_CAPACITY_MESSAGE } from "./capacity-retry.js"; import type { AgentName } from "./types.js"; export type BillingFailureReason = "subscription-quota" | "subscription-model-unsupported"; @@ -55,6 +56,7 @@ export class BillingEventCollector { nativeSessionId: string | undefined; hasWork = false; failed = false; + capacityFailure = false; private pending = ""; private pendingBytes = 0; private skipping = false; @@ -106,6 +108,7 @@ export class BillingEventCollector { private consumeCodex(event: RecordValue): void { if (event.type === "turn.completed") { this.failed = false; + this.capacityFailure = false; this.failureReason = undefined; } if (event.type === "thread.started") { @@ -115,6 +118,9 @@ export class BillingEventCollector { const item = object(event.item); if (typeof item.type === "string" && item.type !== "error") this.hasWork = true; } + if (event.type === "turn.failed") { + this.capacityFailure = object(event.error).message === CODEX_CAPACITY_MESSAGE; + } if (event.type === "error" || event.type === "turn.failed") { this.failureReason ??= codexFailure(event.error) ?? codexFailure(event.message) ?? codexFailure(object(event.error).message) ?? codexFailure(event); diff --git a/src/billing-run.ts b/src/billing-run.ts index 980bdd4..96dc3d1 100644 --- a/src/billing-run.ts +++ b/src/billing-run.ts @@ -3,6 +3,8 @@ import { BillingEventCollector, type BillingFailureReason } from "./billing-even import { aggregateBillingUsage, type BillingUsageReport } from "./billing-usage.js"; import type { AgentName, BillingMode, BuildOptions, Env } from "./types.js"; import type { UsageSummary } from "./usage.js"; +import { CAPACITY_CONTINUATION, MAX_EXECUTION_ATTEMPTS, cancellationCode, capacityRetryDelay, + resolveCapacityRetries, waitForCapacityRetry, type CapacityRetryEvent } from "./capacity-retry.js"; export interface BillingExecutionResult { code: number; @@ -31,6 +33,10 @@ export interface BillingRunOptions { reportUsage?: (trace: string, route: BillingRoute) => Promise; onTransition?: (event: { type: "billing_transition"; from: BillingRoute; to: BillingRoute; reason: BillingFailureReason }) => void; now?: () => number; + random?: () => number; + sleep?: (delayMs: number, signal?: AbortSignal) => Promise; + signal?: AbortSignal; + onCapacityRetry?: (event: CapacityRetryEvent) => void; } export interface BillingRunResult { @@ -43,9 +49,15 @@ export interface BillingRunResult { const terminated = new Set([124, 130, 137, 143]); const continuation = "Continue from where you left off. Your previous turn was interrupted by a subscription usage limit. Preserve completed work and do not repeat completed commands."; -/** One billing transition, sharing the invocation's deadline and native transcript. */ +/** Capacity retries and one billing transition share a deadline and native transcript. */ export async function runWithBilling(input: BillingRunOptions): Promise { const now = input.now ?? Date.now; + const capacityLimit = input.agent === "codex" ? resolveCapacityRetries(input.env) : 0; + const sleep = input.sleep ?? waitForCapacityRetry; + let capacityRetries = 0; + let billingTransitioned = false; + let hasWork = false; + let retryReason: "model-capacity" | undefined; const deadline = input.timeoutSeconds === undefined ? undefined : now() + input.timeoutSeconds * 1000; let options = input.options; let attempt = prepareBillingAttempt(input.agent, options, input.env, input.mode); @@ -55,7 +67,12 @@ export async function runWithBilling(input: BillingRunOptions): Promise events.write(chunk) }); events.end(); nativeSessionId = events.nativeSessionId ?? nativeSessionId; + hasWork ||= events.hasWork; if (input.reportUsage) { const trace = result.usageTrace || result.stdout || result.finalMessageTrace || ""; - reports.push({ route: attempt.route, reason, usage: await input.reportUsage(trace, attempt.route) }); + reports.push({ route: attempt.route, reason, ...(retryReason ? { retryReason } : {}), + usage: await input.reportUsage(trace, attempt.route) }); + } + reason = undefined; + retryReason = undefined; + const cancelledAfterExecution = cancellationCode(input.signal); + if (cancelledAfterExecution !== undefined) { + result = { ...result, code: cancelledAfterExecution }; + break; } if (terminated.has(result.code)) break; if (events.failed && result.code === 0) result = { ...result, code: 1 }; + if (events.capacityFailure && !events.failureReason) { + if (capacityRetries >= capacityLimit) { + error = `model capacity retries exhausted (${capacityRetries} retries)`; + break; + } + const resumeId = nativeSessionId ?? options.sessionId; + if (hasWork && !resumeId) { + error = "capacity retry cannot safely resume partial work: native session ID unavailable"; + break; + } + const remainingMs = deadline === undefined ? Infinity : deadline - now(); + if (remainingMs <= 0) { + result = { ...result, code: 124 }; + break; + } + capacityRetries++; + const delayMs = Math.min(capacityRetryDelay(capacityRetries, input.random ?? Math.random), remainingMs); + input.onCapacityRetry?.({ type: "capacity_retry", retry: capacityRetries, delayMs, reason: "model-capacity" }); + try { + await sleep(delayMs, input.signal); + } catch (failure) { + const code = cancellationCode(input.signal); + if (code === undefined) throw failure; + result = { ...result, code }; + break; + } + retryReason = "model-capacity"; + if (resumeId) { + options = { ...options, prompt: CAPACITY_CONTINUATION, promptFile: undefined, + sessionMode: "resume", sessionId: resumeId }; + } + continue; + } if (!events.failureReason) break; // Explicit subscription-only policy and exhausted paid routes are terminal. - if (input.mode !== "auto" || attempt.route !== "subscription" || index !== 0) { + if (input.mode !== "auto" || attempt.route !== "subscription" || billingTransitioned) { result = { ...result, code: 78 }; error = `billing unavailable: ${events.failureReason}; no further billing fallback`; break; @@ -83,7 +142,7 @@ export async function runWithBilling(input: BillingRunOptions): Promise 2) { - throw new Error("Billing usage requires one or two attempts"); + if (attempts.length < 1 || attempts.length > MAX_EXECUTION_ATTEMPTS) { + throw new Error(`Billing usage requires one to ${MAX_EXECUTION_ATTEMPTS} attempts`); } const summaries = attempts.map((attempt) => attempt.usage); const last = summaries[summaries.length - 1]; diff --git a/src/capacity-retry.ts b/src/capacity-retry.ts new file mode 100644 index 0000000..4b75d00 --- /dev/null +++ b/src/capacity-retry.ts @@ -0,0 +1,37 @@ +import { setTimeout } from "node:timers/promises"; +import type { Env } from "./types.js"; + +export const MAX_CAPACITY_RETRIES = 3; +export const MAX_EXECUTION_ATTEMPTS = MAX_CAPACITY_RETRIES + 2; +export const CODEX_CAPACITY_MESSAGE = "Selected model is at capacity. Please try a different model."; +export const CAPACITY_CONTINUATION = "Continue from where you left off. Your previous turn was interrupted by temporary model capacity exhaustion. Preserve completed work and do not repeat completed commands."; + +export interface CapacityRetryEvent { + type: "capacity_retry"; + retry: number; + delayMs: number; + reason: "model-capacity"; +} + +export function resolveCapacityRetries(env: Env): number { + const value = env.HEADLESS_CAPACITY_RETRIES; + if (value === undefined) return MAX_CAPACITY_RETRIES; + if (!/^[0-3]$/.test(value)) { + throw new Error("HEADLESS_CAPACITY_RETRIES must be 0, 1, 2, or 3"); + } + return Number(value); +} + +export function capacityRetryDelay(retry: number, random: () => number): number { + return Math.round(30_000 * 2 ** (retry - 1) * (0.8 + 0.4 * random())); +} + +export async function waitForCapacityRetry(delayMs: number, signal?: AbortSignal): Promise { + await setTimeout(delayMs, undefined, { signal }); +} + +export function cancellationCode(signal?: AbortSignal): number | undefined { + if (!signal?.aborted) return undefined; + return Number.isInteger(signal.reason) && signal.reason > 0 && signal.reason <= 255 + ? signal.reason as number : 130; +} diff --git a/tests/billing-usage.test.ts b/tests/billing-usage.test.ts index 2d615f4..b63ba36 100644 --- a/tests/billing-usage.test.ts +++ b/tests/billing-usage.test.ts @@ -49,7 +49,8 @@ test("partial cost components stay null; enforce bounded attempts", () => { assert.equal(aggregateBillingUsage([attempt, attempt]).cost?.input, null); assert.equal(aggregateBillingUsage([attempt, attempt]).cost?.total, 20); assert.throws(() => aggregateBillingUsage([])); - assert.throws(() => aggregateBillingUsage([attempt, attempt, attempt])); + assert.equal(aggregateBillingUsage(Array(5).fill(attempt)).totalTokens, 85); + assert.throws(() => aggregateBillingUsage(Array(6).fill(attempt))); }); const claudeResult = { diff --git a/tests/capacity-retry.test.ts b/tests/capacity-retry.test.ts new file mode 100644 index 0000000..d804ac0 --- /dev/null +++ b/tests/capacity-retry.test.ts @@ -0,0 +1,221 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { BillingEventCollector } from "../src/billing-events.ts"; +import { runWithBilling, type BillingRunOptions } from "../src/billing-run.ts"; +import { resolveCapacityRetries, waitForCapacityRetry } from "../src/capacity-retry.ts"; + +const message = "Selected model is at capacity. Please try a different model."; +const record = (value: unknown) => `${JSON.stringify(value)}\n`; +const capacity = record({ type: "turn.failed", error: { message } }); +const thread = record({ type: "thread.started", thread_id: "thread-123" }); +const work = record({ type: "item.completed", item: { type: "command_execution" } }); +const quota = record({ type: "turn.failed", error: { type: "usage_limit_reached" } }); +const base: Pick = { + agent: "codex", mode: "subscription", env: {}, + options: { prompt: "original", model: "gpt-test", reasoningEffort: "high" }, +}; + +function collect(trace: string) { + const events = new BillingEventCollector("codex"); + for (const chunk of [trace.slice(0, 11), trace.slice(11)]) events.write(chunk); + events.end(); + return events; +} + +test("capacity classification requires the exact native terminal failure", () => { + assert.equal(collect(capacity).capacityFailure, true); + for (const trace of [ + record({ type: "error", message }), + record({ type: "turn.failed", error: { message: `${message} extra` } }), + record({ type: "item.completed", item: { type: "agent_message", text: capacity } }), + record({ type: "item.completed", item: { type: "command_execution", aggregated_output: capacity } }), + capacity + record({ type: "turn.completed" }), + capacity + record({ type: "turn.failed", error: { message: "other failure" } }), + ]) assert.equal(collect(trace).capacityFailure, false, trace); + const claude = new BillingEventCollector("claude"); + claude.write(capacity); + assert.equal(claude.capacityFailure, false); +}); + +test("retry configuration is bounded and defaults to three", () => { + assert.equal(resolveCapacityRetries({}), 3); + for (let count = 0; count <= 3; count++) { + assert.equal(resolveCapacityRetries({ HEADLESS_CAPACITY_RETRIES: String(count) }), count); + } + for (const value of ["", "4", "-1", "1.5", "03", " 1", "yes"]) { + assert.throws(() => resolveCapacityRetries({ HEADLESS_CAPACITY_RETRIES: value }), /HEADLESS_CAPACITY_RETRIES/); + } +}); + +test("capacity retry preserves routing and resumes with remaining deadline", async () => { + let now = 0; + const attempts: Parameters[0][] = []; + const transitions: unknown[] = []; + const outcome = await runWithBilling({ ...base, timeoutSeconds: 100, + now: () => now, random: () => 0.5, + sleep: async (delay) => { assert.equal(delay, 30_000); now += delay; }, + onCapacityRetry: (event) => transitions.push(event), + execute: async (attempt) => { + attempts.push(attempt); + if (attempts.length === 1) { + attempt.observe(thread + work + capacity); + now += 10_000; + return { code: 1, stdout: "first" }; + } + return { code: 0, stdout: "done" }; + }, + }); + assert.equal(outcome.result.code, 0); + assert.equal(attempts.length, 2); + assert.equal(attempts[1].timeoutSeconds, 60); + assert.equal(attempts[1].route, attempts[0].route); + assert.deepEqual(attempts[1].env, attempts[0].env); + assert.equal(attempts[1].options.model, base.options.model); + assert.equal(attempts[1].options.reasoningEffort, "high"); + assert.equal(attempts[1].options.sessionId, "thread-123"); + assert.equal(attempts[1].options.sessionMode, "resume"); + assert.match(attempts[1].options.prompt, /capacity/); + assert.equal(attempts[1].options.promptFile, undefined); + assert.deepEqual(transitions, [{ type: "capacity_retry", retry: 1, delayMs: 30_000, reason: "model-capacity" }]); +}); + +test("three capacity retries exhaust with bounded exponential jitter", async () => { + let calls = 0; + const delays: number[] = []; + const outcome = await runWithBilling({ ...base, random: () => 0, + sleep: async (delay) => { delays.push(delay); }, + execute: async ({ observe }) => { calls++; observe(capacity); return { code: 1, stdout: "failed" }; }, + }); + assert.equal(calls, 4); + assert.deepEqual(delays, [24_000, 48_000, 96_000]); + assert.equal(outcome.result.code, 1); + assert.match(outcome.error ?? "", /capacity.*exhausted/); +}); + +for (const retries of [0, 1, 2]) { + test(`configured capacity retry limit ${retries}`, async () => { + let calls = 0; + await runWithBilling({ ...base, env: { HEADLESS_CAPACITY_RETRIES: String(retries) }, + sleep: async () => {}, + execute: async ({ observe }) => { calls++; observe(capacity); return { code: 1, stdout: "" }; }, + }); + assert.equal(calls, retries + 1); + }); +} + +test("no-work retry can replay prompt, but partial work without a session cannot", async () => { + for (const trace of [capacity, work + capacity]) { + let calls = 0; + const outcome = await runWithBilling({ ...base, sleep: async () => {}, + execute: async ({ observe, options }) => { + calls++; + assert.equal(options.prompt, "original"); + if (calls === 1) { observe(trace); return { code: 1, stdout: "retained" }; } + return { code: 0, stdout: "done" }; + }, + }); + assert.equal(calls, trace === capacity ? 2 : 1); + if (trace !== capacity) { + assert.equal(outcome.result.stdout, "retained"); + assert.match(outcome.error ?? "", /cannot safely resume/); + } + } +}); + +test("successful native recovery and ordinary failures never trigger capacity retry", async () => { + for (const trace of [record({ type: "error", message }) + record({ type: "turn.completed" }), + record({ type: "turn.failed", error: { message: "bad input" } }), ""]) { + let calls = 0; + await runWithBilling({ ...base, sleep: async () => { assert.fail("unexpected wait"); }, + execute: async ({ observe }) => { calls++; observe(trace); return { code: 1, stdout: "" }; }, + }); + assert.equal(calls, 1); + } +}); + +test("remaining deadline bounds waiting and prevents another launch", async () => { + let now = 0; + let calls = 0; + const outcome = await runWithBilling({ ...base, timeoutSeconds: 5, now: () => now, + sleep: async (delay) => { assert.equal(delay, 5_000); now += delay; }, + execute: async ({ observe }) => { calls++; observe(capacity); return { code: 1, stdout: "retained" }; }, + }); + assert.equal(calls, 1); + assert.equal(outcome.result.code, 124); +}); + +for (const code of [124, 130, 137, 143]) { + test(`terminated execution ${code} never retries capacity`, async () => { + const outcome = await runWithBilling({ ...base, + sleep: async () => { assert.fail("unexpected sleep"); }, + execute: async ({ observe }) => { observe(capacity); return { code, stdout: "" }; }, + }); + assert.equal(outcome.result.code, code); + }); +} + +test("cancellation during backoff preserves the signal status and transcript", async () => { + const controller = new AbortController(); + let calls = 0; + const outcome = await runWithBilling({ ...base, signal: controller.signal, + sleep: async () => { controller.abort(143); throw new Error("aborted"); }, + execute: async ({ observe }) => { calls++; observe(capacity); return { code: 1, stdout: "retained" }; }, + }); + assert.equal(calls, 1); + assert.equal(outcome.result.code, 143); + assert.equal(outcome.result.stdout, "retained"); +}); + +test("already cancelled invocations do not launch", async () => { + const outcome = await runWithBilling({ ...base, signal: AbortSignal.abort(), + execute: async () => { assert.fail("unexpected launch"); }, + }); + assert.equal(outcome.result.code, 130); +}); + +test("default backoff timer can be interrupted", async () => { + await waitForCapacityRetry(1); + const controller = new AbortController(); + const waiting = waitForCapacityRetry(120_000, controller.signal); + controller.abort(); + await assert.rejects(waiting, { name: "AbortError" }); +}); + +test("billing transition and capacity budgets are independent and usage is counted once", async () => { + let calls = 0; + let transitions = 0; + const routes: string[] = []; + const outcome = await runWithBilling({ ...base, mode: "auto", + env: { CODEX_ACCESS_TOKEN: "subscription", OPENAI_API_KEY: "test" }, sleep: async () => {}, + onTransition: () => { transitions++; }, + execute: async ({ observe, route }) => { + calls++; routes.push(route); + observe(thread + (calls === 2 ? quota : capacity)); + return { code: 1, stdout: String(calls) }; + }, + reportUsage: async (trace) => ({ agent: "codex", inputTokens: Number(trace), cacheReadTokens: 0, + cacheWriteTokens: 0, outputTokens: 0, reasoningOutputTokens: 0, totalTokens: Number(trace), + usageStatus: "reported", cost: null, costBasis: null, pricingSource: null, pricingStatus: "missing" }), + }); + assert.equal(calls, 5); + assert.equal(transitions, 1); + assert.deepEqual(routes, ["subscription", "subscription", "openai-api", "openai-api", "openai-api"]); + assert.equal(outcome.usage?.inputTokens, 15); + assert.equal(outcome.usage?.billing.attempts.length, 5); + assert.equal(outcome.usage?.billing.attempts[1].retryReason, "model-capacity"); + assert.equal(outcome.usage?.billing.attempts[2].reason, "subscription-quota"); + assert.equal(outcome.usage?.billing.attempts[3].reason, undefined); +}); + +test("subscription-only quota remains terminal after a capacity retry", async () => { + let calls = 0; + const outcome = await runWithBilling({ ...base, env: { OPENAI_API_KEY: "unused" }, sleep: async () => {}, + execute: async ({ observe, route }) => { + calls++; assert.equal(route, "subscription"); + observe(thread + (calls === 1 ? capacity : quota)); + return { code: 1, stdout: "" }; + }, + }); + assert.equal(calls, 2); + assert.equal(outcome.result.code, 78); +}); From 66c70d09815690a2cbf6992165b8eeab2bb2b332 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Thu, 10 Sep 2026 14:46:05 +0000 Subject: [PATCH 2/4] feat: integrate capacity retries across execution backends --- src/billing-run.ts | 5 +- src/cli.ts | 40 +++++++- tests/capacity-cli.test.ts | 190 +++++++++++++++++++++++++++++++++++ tests/capacity-retry.test.ts | 13 ++- tests/modal.test.ts | 49 +++++++++ 5 files changed, 290 insertions(+), 7 deletions(-) create mode 100644 tests/capacity-cli.test.ts diff --git a/src/billing-run.ts b/src/billing-run.ts index 96dc3d1..c211db7 100644 --- a/src/billing-run.ts +++ b/src/billing-run.ts @@ -8,6 +8,7 @@ import { CAPACITY_CONTINUATION, MAX_EXECUTION_ATTEMPTS, cancellationCode, capaci export interface BillingExecutionResult { code: number; + terminationSignal?: NodeJS.Signals; stdout: string; usageTrace?: string; finalMessageTrace?: string; @@ -46,7 +47,7 @@ export interface BillingRunResult { error?: string; } -const terminated = new Set([124, 130, 137, 143]); +const terminated = new Set([124, 129, 130, 131, 137, 143, 149]); const continuation = "Continue from where you left off. Your previous turn was interrupted by a subscription usage limit. Preserve completed work and do not repeat completed commands."; /** Capacity retries and one billing transition share a deadline and native transcript. */ @@ -95,7 +96,7 @@ export async function runWithBilling(input: BillingRunOptions): Promise= capacityLimit) { diff --git a/src/cli.ts b/src/cli.ts index 84934c3..7f023d5 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -17,7 +17,7 @@ import { spawn, spawnSync } from "node:child_process"; import { runAcpClient, runAcpStdioAgent } from "./acp.js"; import { randomUUID } from "node:crypto"; import { dirname, join } from "node:path"; -import { tmpdir } from "node:os"; +import { constants as osConstants, tmpdir } from "node:os"; import { fileURLToPath } from "node:url"; import { PiCompletionObserver } from "./pi-completion.js"; @@ -40,6 +40,7 @@ import { import { prepareAntigravityUsageCapture, type AntigravityUsageCapture } from "./antigravity-usage.js"; import { BillingError, prepareBillingAttempt, prepareBillingPreview, resolveBillingMode } from "./billing.js"; import { runWithBilling, type BillingExecutionAttempt, type BillingRunResult } from "./billing-run.js"; +import { resolveCapacityRetries, waitForCapacityRetry } from "./capacity-retry.js"; import { buildDockerBillingVolumeInitCommand, removeDockerBillingVolume } from "./docker-billing.js"; import { checkAgents, checkDocker, commandExists, commandForAgent, renderAgentChecks, renderDockerCheck } from "./check.js"; import { @@ -245,6 +246,8 @@ interface CliDeps { stderrIsTTY?: boolean; stdout?: (text: string) => void; stderr?: (text: string) => void; + capacityRetrySleep?: (delayMs: number, signal?: AbortSignal) => Promise; + capacityRetryRandom?: () => number; } class CliError extends Error { @@ -1122,6 +1125,7 @@ function selectDefaultAgent(env: Env, preferredAgent: AgentName | undefined): Ag interface ExecuteResult { code: number; + terminationSignal?: NodeJS.Signals; stdout: string; finalMessageTrace?: string; usageTrace?: string; @@ -1791,6 +1795,7 @@ async function executeCommand( let settled = false; let timeout: NodeJS.Timeout | undefined; let termination: { code: number } | undefined; + let parentTerminationSignal: NodeJS.Signals | undefined; let forceKill: NodeJS.Timeout | undefined; let forceFinish: NodeJS.Timeout | undefined; let forceDrain: NodeJS.Timeout | undefined; @@ -1903,6 +1908,7 @@ async function executeCommand( result.usageTrace = readRelevantTrace() || undefined; result.stdoutReceived = stdoutReceived; result.stdoutEndsWithNewline = stdoutEndsWithNewline; + result.terminationSignal ??= parentTerminationSignal; resolve(result); }; const asyncMessageWorker = env.HEADLESS_ASYNC_MESSAGE_WORKER === "1"; @@ -1966,6 +1972,7 @@ async function executeCommand( child.stderr?.destroy(); finish({ code: signal ? 1 : (code ?? 1), + ...(signal ? { terminationSignal: signal } : {}), stdout: capturedStdout, stdoutReceived, stdoutEndsWithNewline, @@ -1983,6 +1990,7 @@ async function executeCommand( }; for (const signal of parentExitSignals()) { const handler = () => { + parentTerminationSignal = signal; const inheritedListeners = options.inheritedSignalListeners?.get(signal); const hasExternalListener = inheritedListeners ? process.listeners(signal).some((listener) => inheritedListeners.has(listener)) @@ -2122,7 +2130,8 @@ async function executeCommand( signalChildTree("SIGKILL"); } if (signal) { - finish({ code: termination?.code ?? 1, stdout: capturedStdout, stdoutReceived, stdoutEndsWithNewline }); + finish({ code: termination?.code ?? 1, terminationSignal: signal, + stdout: capturedStdout, stdoutReceived, stdoutEndsWithNewline }); return; } finish({ code: termination?.code ?? code ?? 1, stdout: capturedStdout, stdoutReceived, stdoutEndsWithNewline }); @@ -4412,6 +4421,7 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise 0) billingEnv[entry.slice(0, split)] = entry.slice(split + 1); } + const capacityRetries = resolveCapacityRetries(billingEnv); const billingPreviewOptions = { model: configuredDefaults.model, profile, prompt: composedPrompt, workDir: cwd }; let sessionAlias = parsed.sessionAlias; if (parsed.runId && parsed.role && coordination === "session" && !parsed.sessionAlias) { @@ -4420,8 +4430,9 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise 0; + if (parsed.docker && !parsed.printCommand && !sessionAlias && (needsRetryHome || (billingMode === "auto" && + prepareBillingAttempt(parsed.agent, billingPreviewOptions, billingEnv, billingMode).route === "subscription"))) { if (process.platform === "win32") temporaryBillingVolume = `headless-billing-${randomUUID()}`; else temporaryBillingRoot = mkdtempSync(join(tmpdir(), "headless-billing-")); } @@ -4586,10 +4597,25 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise Promise) => { + const retryController = new AbortController(); billingResult = await runWithBilling({ agent: parsed.agent!, mode: billingMode, env: billingEnv, options: nativeOptions, timeoutSeconds: parsed.modal ? modalTimeoutSeconds : commandTimeoutSeconds, execute, + signal: retryController.signal, + random: deps.capacityRetryRandom, + sleep: async (delayMs, signal) => { + const handlers = parentExitSignals().map((signalName) => { + const handler = () => retryController.abort(128 + osConstants.signals[signalName]); + process.on(signalName, handler); + return [signalName, handler] as const; + }); + try { + await (deps.capacityRetrySleep ?? waitForCapacityRetry)(delayMs, signal); + } finally { + for (const [signalName, handler] of handlers) process.off(signalName, handler); + } + }, reportUsage: parsed.usage && (parsed.agent === "claude" || parsed.agent === "codex") ? async (trace, route) => { const context = usageContext(parsed.agent!, configuredDefaults, env, effectiveProfile); if (route === "bedrock") context.provider = "amazon-bedrock"; @@ -4601,6 +4627,12 @@ export async function runCli(argv: string[], deps: CliDeps = {}): Promise { + displayStderr(`headless: model capacity retry ${event.retry} in ${(event.delayMs / 1000).toFixed(1)}s (${event.reason})\n`); + const line = `${JSON.stringify(event)}\n`; + commandStdoutLog?.(line); + if (stdoutHandling !== "capture") commandStdout(line); + }, }); if (billingResult.error) displayStderr(`headless: ${billingResult.error}\n`); return billingResult.result; diff --git a/tests/capacity-cli.test.ts b/tests/capacity-cli.test.ts new file mode 100644 index 0000000..bc2e270 --- /dev/null +++ b/tests/capacity-cli.test.ts @@ -0,0 +1,190 @@ +import assert from "node:assert/strict"; +import { existsSync, mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { runCli } from "../src/cli.ts"; + +function fixture(binary = "codex") { + const home = mkdtempSync(join(tmpdir(), "capacity-cli-")); + mkdirSync(join(home, ".codex")); + writeFileSync(join(home, ".codex/auth.json"), JSON.stringify({ tokens: { access_token: "subscription" } })); + writeFileSync(join(home, binary), `#!/usr/bin/env node +const fs = require('node:fs'); +const callsFile = process.env.HOME + '/calls'; +const previous = fs.existsSync(callsFile) ? fs.readFileSync(callsFile, 'utf8').trim().split('\\n').length : 0; +fs.appendFileSync(callsFile, JSON.stringify({args: process.argv.slice(2), paid: !!process.env.CODEX_API_KEY}) + '\\n'); +console.log(JSON.stringify({type:'thread.started', thread_id:'12345678-1234-1234-1234-123456789abc'})); +if (previous === 0) { + console.log(JSON.stringify({type:'turn.failed', error:{message:'Selected model is at capacity. Please try a different model.'}})); + process.exitCode = 1; +} else { + console.log(JSON.stringify({type:'item.completed', item:{type:'agent_message', text:'finished'}})); + console.log(JSON.stringify({type:'turn.completed', usage:{input_tokens:12, output_tokens:4, cached_input_tokens:0}})); +} +`, { mode: 0o755 }); + return { + home, + env: { HOME: home, PATH: `${home}:${process.env.PATH}` }, + calls: (): Array<{ args: string[]; paid: boolean }> => existsSync(join(home, "calls")) + ? readFileSync(join(home, "calls"), "utf8").trim().split("\n").map((line) => JSON.parse(line)) : [], + cleanup: () => rmSync(home, { recursive: true, force: true }), + }; +} + +test("Codex capacity retry resumes and emits a structured retry without changing billing", async () => { + const run = fixture(); + try { + const output: string[] = []; + const errors: string[] = []; + const delays: number[] = []; + const code = await runCli(["codex", "--prompt", "task", "--json"], { + env: run.env, stdout: (text) => output.push(text), stderr: (text) => errors.push(text), + capacityRetryRandom: () => 0.5, + capacityRetrySleep: async (delay) => { delays.push(delay); }, + }); + assert.equal(code, 0, errors.join("")); + assert.deepEqual(delays, [30000]); + assert.equal(run.calls().length, 2); + assert.ok(run.calls()[1].args.includes("resume")); + assert.ok(run.calls().every((call) => !call.paid)); + const retry = output.join("").trim().split("\n").map((line) => JSON.parse(line)) + .find((event) => event.type === "capacity_retry"); + assert.deepEqual(retry, { type: "capacity_retry", retry: 1, delayMs: 30000, reason: "model-capacity" }); + assert.match(errors.join(""), /capacity.*retry/i); + } finally { run.cleanup(); } +}); + +test("disabled capacity retries return the native failure after one execution", async () => { + const run = fixture(); + try { + const code = await runCli(["codex", "--prompt", "task", "--json"], { + env: { ...run.env, HEADLESS_CAPACITY_RETRIES: "0" }, stdout: () => {}, stderr: () => {}, + capacityRetrySleep: async () => { assert.fail("retry disabled"); }, + }); + assert.equal(code, 1); + assert.equal(run.calls().length, 1); + } finally { run.cleanup(); } +}); + +test("invalid capacity configuration fails before agent launch", async () => { + const run = fixture(); + try { + const errors: string[] = []; + const code = await runCli(["codex", "--prompt", "task"], { + env: { ...run.env, HEADLESS_CAPACITY_RETRIES: "4" }, stdout: () => {}, stderr: (text) => errors.push(text), + }); + assert.equal(code, 2); + assert.match(errors.join(""), /HEADLESS_CAPACITY_RETRIES/); + assert.equal(run.calls().length, 0); + } finally { run.cleanup(); } +}); + +test("SIGTERM during capacity backoff stops promptly without another launch or leaked handler", async () => { + const run = fixture(); + const listeners = process.listeners("SIGTERM"); + try { + const code = await runCli(["codex", "--prompt", "task", "--json"], { + env: run.env, stdout: () => {}, stderr: () => {}, + capacityRetrySleep: async (_delay, signal) => { + process.emit("SIGTERM"); + assert.equal(signal?.aborted, true); + assert.equal(signal?.reason, 143); + signal?.throwIfAborted(); + }, + }); + assert.equal(code, 143); + assert.equal(run.calls().length, 1); + assert.deepEqual(process.listeners("SIGTERM"), listeners); + } finally { run.cleanup(); } +}); + +for (const billing of ["subscription", "api", "native"] as const) { + test(`unnamed Docker ${billing} capacity retries preserve one home and clean it after success`, async () => { + const run = fixture("docker"); + let sessionRoot: string | undefined; + try { + const errors: string[] = []; + if (billing === "native") writeFileSync(join(run.home, ".codex/custom.config.toml"), 'model_provider = "custom"\n'); + const route = billing === "native" ? ["--profile", "custom"] : ["--billing", billing]; + const code = await runCli(["codex", "--docker", ...route, "--prompt", "task", "--json"], { + env: { ...run.env, OPENAI_API_KEY: "test-api" }, stdout: () => {}, stderr: (text) => errors.push(text), + capacityRetrySleep: async () => {}, + }); + assert.equal(code, 0, errors.join("")); + const calls = run.calls(); + assert.equal(calls.length, 2); + const mounts = calls.map((call) => call.args.find((arg) => arg.includes("headless-billing-") && arg.includes("/codex/home"))); + assert.ok(mounts[0]); + assert.equal(mounts[0], mounts[1]); + sessionRoot = mounts[0].match(/(?:src=|source=)?([^,:]*headless-billing-[^/]+)/)?.[1]; + assert.ok(sessionRoot); + assert.equal(existsSync(sessionRoot), false); + assert.ok(calls[1].args.includes("resume")); + } finally { + run.cleanup(); + if (sessionRoot) rmSync(sessionRoot, { recursive: true, force: true }); + } + }); +} + +test("disabled capacity retries do not create an anonymous subscription Docker home", async () => { + const run = fixture("docker"); + try { + const code = await runCli(["codex", "--docker", "--billing", "subscription", "--prompt", "task", "--json"], { + env: { ...run.env, HEADLESS_CAPACITY_RETRIES: "0" }, stdout: () => {}, stderr: () => {}, + }); + assert.equal(code, 1); + assert.equal(run.calls().length, 1); + assert.ok(run.calls()[0].args.every((arg) => !arg.includes("headless-billing-"))); + } finally { run.cleanup(); } +}); + +test("a signal-terminated Codex process does not retry its preceding capacity failure", { skip: process.platform === "win32" }, async () => { + const run = fixture(); + try { + const binary = join(run.home, "codex"); + writeFileSync(binary, readFileSync(binary, "utf8").replace("process.exitCode = 1;", + "process.stdout.write('', () => process.kill(process.pid, 'SIGTERM'));")); + let delays = 0; + const code = await runCli(["codex", "--prompt", "task", "--json"], { + env: run.env, stdout: () => {}, stderr: () => {}, + capacityRetrySleep: async () => { delays++; }, + }); + assert.notEqual(code, 0); + assert.equal(run.calls().length, 1); + assert.equal(delays, 0); + } finally { run.cleanup(); } +}); + +test("embedded parent cancellation stays terminal when Codex catches SIGTERM and exits one", { skip: process.platform === "win32" }, async () => { + const run = fixture(); + const inheritedListener = () => {}; + process.on("SIGTERM", inheritedListener); + try { + const binary = join(run.home, "codex"); + writeFileSync(binary, readFileSync(binary, "utf8") + .replace("const fs = require('node:fs');", "const fs = require('node:fs');\nprocess.on('SIGTERM', () => process.exit(1));") + .replace("process.exitCode = 1;", "setInterval(() => {}, 1000);")); + let signalled = false; + let delays = 0; + const code = await runCli(["codex", "--prompt", "task", "--json", "--timeout", "5"], { + env: run.env, stderr: () => {}, + stdout: (text) => { + if (!signalled && text.includes('"turn.failed"')) { + signalled = true; + process.emit("SIGTERM"); + } + }, + capacityRetrySleep: async () => { delays++; }, + }); + assert.equal(signalled, true); + assert.equal(code, 1); + assert.equal(run.calls().length, 1); + assert.equal(delays, 0); + assert.ok(process.listeners("SIGTERM").includes(inheritedListener)); + } finally { + process.off("SIGTERM", inheritedListener); + run.cleanup(); + } +}); diff --git a/tests/capacity-retry.test.ts b/tests/capacity-retry.test.ts index d804ac0..1b1f9a4 100644 --- a/tests/capacity-retry.test.ts +++ b/tests/capacity-retry.test.ts @@ -144,7 +144,7 @@ test("remaining deadline bounds waiting and prevents another launch", async () = assert.equal(outcome.result.code, 124); }); -for (const code of [124, 130, 137, 143]) { +for (const code of [124, 129, 130, 131, 137, 143, 149]) { test(`terminated execution ${code} never retries capacity`, async () => { const outcome = await runWithBilling({ ...base, sleep: async () => { assert.fail("unexpected sleep"); }, @@ -219,3 +219,14 @@ test("subscription-only quota remains terminal after a capacity retry", async () assert.equal(calls, 2); assert.equal(outcome.result.code, 78); }); + +test("native termination signal after capacity failure never retries", async () => { + const outcome = await runWithBilling({ ...base, + sleep: async () => { assert.fail("terminated execution must not retry"); }, + execute: async ({ observe }) => { + observe(capacity); + return { code: 1, stdout: "retained", terminationSignal: "SIGTERM" }; + }, + }); + assert.equal(outcome.result.code, 1); +}); diff --git a/tests/modal.test.ts b/tests/modal.test.ts index 8a10dbb..73828d7 100644 --- a/tests/modal.test.ts +++ b/tests/modal.test.ts @@ -27,6 +27,8 @@ import { } from "../src/modal.ts"; import { quoteCommand } from "../src/shell.ts"; import { PiCompletionObserver } from "../src/pi-completion.ts"; +import { buildAgentCommand } from "../src/agents.ts"; +import { runWithBilling } from "../src/billing-run.ts"; test("default Modal image is immutable", () => { assert.equal( @@ -953,6 +955,53 @@ test("Modal billing retries share sandbox, observe captured output and mask inhe } finally { rmSync(dir, { recursive: true, force: true }); } }); +test("Modal capacity failure resumes in the same sandbox without repeating bootstrap", async () => { + const dir = mkdtempSync(join(tmpdir(), "headless-modal-capacity-")); + try { + const work = join(dir, "work"), remote = join(dir, "remote"); + mkdirSync(work); + mkdirSync(remote); + initGitWorkdir(work); + const nativeId = "12345678-1234-1234-1234-123456789abc"; + const trace = [ + { type: "thread.started", thread_id: nativeId }, + { type: "turn.failed", error: { message: "Selected model is at capacity. Please try a different model." } }, + ].map((event) => `${JSON.stringify(event)}\n`); + const sandbox = new FakeSandbox(remote, { agentStdoutChunks: trace }); + const client = new FakeModalClient(sandbox); + const env = { HOME: join(dir, "home"), OPENAI_API_KEY: "test-api" }; + const delays: number[] = []; + const result = await executeModalAgent({ + agent: "codex", appName: "test", command: buildAgentCommand("codex", { prompt: "task" }, env), + cpu: 1, env, image: DEFAULT_MODAL_IMAGE, includeGit: false, memoryMiB: 1024, + modalEnv: [], modalSecrets: [], stdout: () => {}, stderr: () => {}, + stdoutHandling: "capture", timeoutSeconds: 180, workDir: work, clientFactory: async () => client, + invoke: async (execute) => (await runWithBilling({ + agent: "codex", mode: "api", env, options: { prompt: "task" }, timeoutSeconds: 180, + random: () => 0.5, + sleep: async (delay) => { + delays.push(delay); + sandbox.options.agentStdoutChunks = [ + JSON.stringify({ type: "item.completed", item: { type: "agent_message", text: "finished" } }) + "\n", + JSON.stringify({ type: "turn.completed", usage: {} }) + "\n", + ]; + }, + execute: (attempt) => execute(buildAgentCommand("codex", attempt.options, attempt.env), + attempt.env, attempt.timeoutSeconds!, attempt.observe), + })).result, + }); + assert.equal(result.code, 0); + assert.deepEqual(delays, [30000]); + const executions = sandbox.commands.filter((command) => command[0] === "sh"); + assert.equal(executions.length, 2); + assert.match(executions[0][2], /headless-host-home/); + assert.doesNotMatch(executions[1][2], /headless-host-home/); + assert.ok(executions[1].includes("resume") && executions[1].includes(nativeId)); + assert.equal(sandbox.terminated, true); + assert.equal(client.closed, true); + } finally { rmSync(dir, { recursive: true, force: true }); } +}); + test("explicit undefined command credentials cannot be reintroduced by forwarding", () => { assert.equal(collectModalEnv({ OPENAI_API_KEY: "parent" }, { OPENAI_API_KEY: undefined }, ["OPENAI_API_KEY=explicit"]).OPENAI_API_KEY, undefined); }); From 4d0e35f932c5db3dbf6503f1c383f5be2ec4e7f0 Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Thu, 10 Sep 2026 14:46:09 +0000 Subject: [PATCH 3/4] docs: document automatic capacity retries --- CHANGELOG.md | 1 + README.md | 29 ++++++++++++++++++++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2645a1d..a698c98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## TBD +- Added automatic bounded retries for native Codex model-capacity failures in local, Docker, and Modal runs, preserving sessions, billing policy, cancellation, deadlines, and per-attempt usage. Set `HEADLESS_CAPACITY_RETRIES=0` to disable. - Fixed run coordination to recover stale run and node locks after owner crashes, retain async ownership for the full detached process tree, and handle Windows process probing and state replacement safely (#28). ## 0.6.1 - 2026-08-12 diff --git a/README.md b/README.md index 2e19cb5..a4fcf3a 100644 --- a/README.md +++ b/README.md @@ -157,7 +157,7 @@ errors, tool output, interruptions, and timeouts do not trigger fallback. Paid provider limits still apply; Headless does not impose a local dollar cap. The policy applies locally, in Docker, and in Modal. Docker keeps an anonymous -private home across both attempts, removes it after native success, and reports +private home across attempts, removes it after native success, and reports its retained path on failure for recovery. `--session` homes remain durable. On Windows, anonymous Docker runs share a Docker-managed volume across attempts; success removes it and failure reports its name for recovery. Named durable @@ -172,6 +172,33 @@ estimates with reported charges. Subscription cost estimates are API list-price comparisons, not subscription charges. Auth changes affect child environments only; Headless never replaces shared login files. +### Temporary Codex capacity failures + +Noninteractive Codex runs automatically retry the native terminal error +`Selected model is at capacity. Please try a different model.` up to three times. +Delays are 30, 60, and 120 seconds, each with ±20% jitter. Waiting counts against +the original command timeout, and cancellation stops the wait immediately. + +Set `HEADLESS_CAPACITY_RETRIES=0` to disable these retries, or choose `1`, `2`, or +`3` to set their limit. Other values are rejected before launching the agent. +Capacity retries keep the same model, profile, permissions, and billing route. +They resume the native session with a continuation prompt; without a session, +Headless retries the original prompt only if no work has been observed. + +Retries apply to local, Docker, and Modal runs. Unnamed Docker runs keep a private +native home across attempts, including subscription-only and API billing; success +removes it and failure reports its location for recovery. Interactive/tmux launches +remain under the native CLI's control. Assistant text, tool output, generic errors, +and recoverable error notices do not trigger capacity retries. + +Each wait emits a stderr diagnostic and a `capacity_retry` event in streamed +JSON/log output, with `retry`, `delayMs`, and `reason: "model-capacity"`. +`--usage` includes every execution once in `billing.attempts`, with +`retryReason: "model-capacity"` on executions caused by capacity retries. The +three capacity retries and the existing single billing fallback have independent +budgets, allowing at most five executions per invocation. Exhausted capacity +retries preserve the last failure status and transcript. + ### Empty Pi completions Pi can finish artifact-producing work with an empty final assistant message. From 1b74091cdb90a56fc9e7983e050b315565d4dd8c Mon Sep 17 00:00:00 2001 From: RobertTLange Date: Thu, 10 Sep 2026 14:54:45 +0000 Subject: [PATCH 4/4] fix: preserve Codex cancellation without a timeout --- src/cli.ts | 1 + tests/capacity-cli.test.ts | 76 +++++++++++++++++++++----------------- 2 files changed, 44 insertions(+), 33 deletions(-) diff --git a/src/cli.ts b/src/cli.ts index 7f023d5..8f0dd5e 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -1918,6 +1918,7 @@ async function executeCommand( ); const handlesParentSignals = asyncMessageWorker || ownsChildProcessGroup + || agent === "codex" || options.cleanupBeforeParentSignalExit !== undefined; waitForAsyncMessageOwnership(env, command); let childEnv = commandEnv(env, command); diff --git a/tests/capacity-cli.test.ts b/tests/capacity-cli.test.ts index bc2e270..843c2f5 100644 --- a/tests/capacity-cli.test.ts +++ b/tests/capacity-cli.test.ts @@ -13,7 +13,7 @@ function fixture(binary = "codex") { const fs = require('node:fs'); const callsFile = process.env.HOME + '/calls'; const previous = fs.existsSync(callsFile) ? fs.readFileSync(callsFile, 'utf8').trim().split('\\n').length : 0; -fs.appendFileSync(callsFile, JSON.stringify({args: process.argv.slice(2), paid: !!process.env.CODEX_API_KEY}) + '\\n'); +fs.appendFileSync(callsFile, JSON.stringify({args: process.argv.slice(2), paid: !!process.env.CODEX_API_KEY, pid: process.pid}) + '\\n'); console.log(JSON.stringify({type:'thread.started', thread_id:'12345678-1234-1234-1234-123456789abc'})); if (previous === 0) { console.log(JSON.stringify({type:'turn.failed', error:{message:'Selected model is at capacity. Please try a different model.'}})); @@ -26,7 +26,7 @@ if (previous === 0) { return { home, env: { HOME: home, PATH: `${home}:${process.env.PATH}` }, - calls: (): Array<{ args: string[]; paid: boolean }> => existsSync(join(home, "calls")) + calls: (): Array<{ args: string[]; paid: boolean; pid: number }> => existsSync(join(home, "calls")) ? readFileSync(join(home, "calls"), "utf8").trim().split("\n").map((line) => JSON.parse(line)) : [], cleanup: () => rmSync(home, { recursive: true, force: true }), }; @@ -157,34 +157,44 @@ test("a signal-terminated Codex process does not retry its preceding capacity fa } finally { run.cleanup(); } }); -test("embedded parent cancellation stays terminal when Codex catches SIGTERM and exits one", { skip: process.platform === "win32" }, async () => { - const run = fixture(); - const inheritedListener = () => {}; - process.on("SIGTERM", inheritedListener); - try { - const binary = join(run.home, "codex"); - writeFileSync(binary, readFileSync(binary, "utf8") - .replace("const fs = require('node:fs');", "const fs = require('node:fs');\nprocess.on('SIGTERM', () => process.exit(1));") - .replace("process.exitCode = 1;", "setInterval(() => {}, 1000);")); - let signalled = false; - let delays = 0; - const code = await runCli(["codex", "--prompt", "task", "--json", "--timeout", "5"], { - env: run.env, stderr: () => {}, - stdout: (text) => { - if (!signalled && text.includes('"turn.failed"')) { - signalled = true; - process.emit("SIGTERM"); - } - }, - capacityRetrySleep: async () => { delays++; }, - }); - assert.equal(signalled, true); - assert.equal(code, 1); - assert.equal(run.calls().length, 1); - assert.equal(delays, 0); - assert.ok(process.listeners("SIGTERM").includes(inheritedListener)); - } finally { - process.off("SIGTERM", inheritedListener); - run.cleanup(); - } -}); +for (const timeoutArgs of [[], ["--timeout", "5"]]) { + const timeoutLabel = timeoutArgs.length ? "with" : "without"; + test(`embedded parent cancellation stays terminal when Codex catches SIGTERM ${timeoutLabel} a timeout`, { skip: process.platform === "win32" }, async () => { + const run = fixture(); + const listeners = process.listeners("SIGTERM"); + let inheritedSignals = 0; + const inheritedListener = () => { + inheritedSignals++; + if (!timeoutArgs.length) process.kill(run.calls()[0].pid, "SIGTERM"); + }; + process.on("SIGTERM", inheritedListener); + try { + const binary = join(run.home, "codex"); + writeFileSync(binary, readFileSync(binary, "utf8") + .replace("const fs = require('node:fs');", "const fs = require('node:fs');\nprocess.on('SIGTERM', () => process.exit(1));") + .replace("process.exitCode = 1;", "setInterval(() => {}, 1000);")); + let signalled = false; + let delays = 0; + const code = await runCli(["codex", "--prompt", "task", "--json", ...timeoutArgs], { + env: run.env, stderr: () => {}, + stdout: (text) => { + if (!signalled && text.includes('"turn.failed"')) { + signalled = true; + process.emit("SIGTERM"); + } + }, + capacityRetrySleep: async () => { delays++; }, + }); + assert.equal(signalled, true); + assert.equal(inheritedSignals, 1); + assert.equal(code, 1); + assert.equal(run.calls().length, 1); + assert.equal(delays, 0); + assert.ok(process.listeners("SIGTERM").includes(inheritedListener)); + } finally { + process.off("SIGTERM", inheritedListener); + run.cleanup(); + assert.deepEqual(process.listeners("SIGTERM"), listeners); + } + }); +}