diff --git a/README.md b/README.md index 937a3037..27880c55 100644 --- a/README.md +++ b/README.md @@ -283,6 +283,20 @@ Your configuration will be picked up based on: - project-level overrides in `.codex/config.toml` - project-level overrides only load when the [project is trusted](https://developers.openai.com/codex/config-advanced#project-config-files-codexconfigtoml) +#### Stalled-turn salvage + +If Codex stops emitting events on a still-open connection without ever confirming a turn finished, the plugin would otherwise wait forever. To avoid a wedged job, a review or task is salvaged after an inter-event idle window and reported as incomplete (a non-success exit), returning whatever output was captured. + +This window is an inter-event gap (the time since the last event for the turn), not a total-duration limit, so a long but active turn keeps running as long as Codex keeps streaming events. It defaults to 15 minutes and can be tuned with an environment variable: + +```bash +# Salvage a silent turn after 10 minutes of no events (value in milliseconds). +export CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS=600000 + +# Disable the watchdog entirely (a stalled turn will wait indefinitely). +export CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS=0 +``` + Check out the Codex docs for more [configuration options](https://developers.openai.com/codex/config-reference). ### Moving The Work Over To Codex diff --git a/plugins/codex/scripts/lib/codex.mjs b/plugins/codex/scripts/lib/codex.mjs index fead00cc..9002d606 100644 --- a/plugins/codex/scripts/lib/codex.mjs +++ b/plugins/codex/scripts/lib/codex.mjs @@ -24,6 +24,8 @@ * pendingCollaborations: Set, * activeSubagentTurns: Set, * completionTimer: ReturnType | null, + * idleTimer: ReturnType | null, + * timedOut: boolean, * lastAgentMessage: string, * reviewText: string, * reasoningSummary: string[], @@ -51,6 +53,36 @@ const DEFAULT_CONTINUE_PROMPT = const EXTERNAL_AGENT_IMPORT_COMPLETED = "externalAgentConfig/import/completed"; const EXTERNAL_AGENT_IMPORT_TIMEOUT_MS = 2 * 60 * 1000; +export const TURN_IDLE_TIMEOUT_ENV = "CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS"; +export const DEFAULT_TURN_IDLE_TIMEOUT_MS = 900000; +// Node's setTimeout caps delays at 2^31-1 ms; larger values overflow to 1 ms (near-immediate fire). +const MAX_TURN_IDLE_TIMEOUT_MS = 2147483647; + +/** + * Resolve the inter-event idle window (ms) after which a silent turn is salvaged. + * This is an inter-event gap, NOT a total-duration ceiling: it resets on every + * notification that belongs to the turn. The whole trimmed value must be a decimal + * integer of milliseconds; anything else (blank, negative, decimal, or unit-suffixed + * like "15m" / "0ms") falls back to the default rather than parseInt's lenient numeric + * prefix. An explicit 0 is preserved and disables the watchdog; values above setTimeout's + * 2^31-1 ms ceiling are clamped to it. Avoid `Number(value) || DEFAULT` here - it would + * turn 0 into the default. + * @param {NodeJS.ProcessEnv} [env] + * @returns {number} + */ +export function resolveTurnIdleMs(env = process.env) { + const raw = env?.[TURN_IDLE_TIMEOUT_ENV]; + const trimmed = raw == null ? "" : String(raw).trim(); + if (!/^\d+$/.test(trimmed)) { + return DEFAULT_TURN_IDLE_TIMEOUT_MS; + } + const parsed = Number.parseInt(trimmed, 10); + if (!Number.isFinite(parsed)) { + return DEFAULT_TURN_IDLE_TIMEOUT_MS; + } + return Math.min(parsed, MAX_TURN_IDLE_TIMEOUT_MS); +} + function cleanCodexStderr(stderr) { return stderr .split(/\r?\n/) @@ -325,6 +357,8 @@ function createTurnCaptureState(threadId, options = {}) { pendingCollaborations: new Set(), activeSubagentTurns: new Set(), completionTimer: null, + idleTimer: null, + timedOut: false, lastAgentMessage: "", reviewText: "", reasoningSummary: [], @@ -343,12 +377,43 @@ function clearCompletionTimer(state) { } } +function clearIdleTimer(state) { + if (state.idleTimer) { + clearTimeout(state.idleTimer); + state.idleTimer = null; + } +} + +/** + * (Re)arm the idle watchdog. Fires purely on silence: if no turn-belonging + * notification arrives within `idleMs`, the turn is salvaged as timed out. + * It deliberately does NOT consult pendingCollaborations/activeSubagentTurns - + * a subagent whose own terminal event was also dropped would never drain and + * would re-create the hang. Active subagents are protected by the reset surface + * (their streaming notifications re-arm this timer) instead. `idleMs <= 0` disables it. + */ +function armIdleTimer(state, idleMs) { + clearIdleTimer(state); + if (!(idleMs > 0)) { + return; + } + state.idleTimer = setTimeout(() => { + state.idleTimer = null; + if (state.completed) { + return; + } + completeTurn(state, null, { inferred: true, timedOut: true }); + }, Math.min(idleMs, MAX_TURN_IDLE_TIMEOUT_MS)); + state.idleTimer.unref?.(); +} + function completeTurn(state, turn = null, options = {}) { if (state.completed) { return; } clearCompletionTimer(state); + clearIdleTimer(state); state.completed = true; if (turn) { @@ -357,13 +422,27 @@ function completeTurn(state, turn = null, options = {}) { state.turnId = turn.id; } } else if (!state.finalTurn) { - state.finalTurn = { - id: state.turnId ?? "inferred-turn", - status: "completed" - }; + if (options.timedOut) { + // Salvaged without a confirmed completion: report a non-success status so + // buildResultStatus returns 1 and the job is recorded as failed, never a + // false "completed". The normal drained-after-final_answer inferred path + // (no timedOut) keeps the "completed" status. + state.timedOut = true; + state.finalTurn = { + id: state.turnId ?? "inferred-turn", + status: "incomplete" + }; + } else { + state.finalTurn = { + id: state.turnId ?? "inferred-turn", + status: "completed" + }; + } } - if (options.inferred) { + if (options.timedOut) { + emitProgress(state.onProgress, "Turn salvaged: Codex stopped emitting events before confirming completion; returning the captured output.", "finalizing"); + } else if (options.inferred) { emitProgress(state.onProgress, "Turn completion inferred after the main thread finished and subagent work drained.", "finalizing"); } @@ -559,6 +638,14 @@ function applyTurnNotification(state, message) { async function captureTurn(client, threadId, startRequest, options = {}) { const state = createTurnCaptureState(threadId, options); const previousHandler = client.notificationHandler; + const idleMs = options.turnIdleMs ?? resolveTurnIdleMs(options.env ?? process.env); + emitProgress( + state.onProgress, + idleMs > 0 + ? `Turn idle watchdog armed (${idleMs}ms inter-event window).` + : "Turn idle watchdog disabled.", + "starting" + ); client.setNotificationHandler((message) => { if (!state.turnId) { @@ -568,6 +655,12 @@ async function captureTurn(client, threadId, startRequest, options = {}) { if (message.method === "thread/started" || message.method === "thread/name/updated") { applyTurnNotification(state, message); + // thread/started / thread/name/updated register a (subagent) thread for THIS turn - + // that is live turn activity, so reset the watchdog here too. This branch returns + // before the belongsToTurn reset below, so the reset has to be repeated here. + if (!state.completed) { + armIdleTimer(state, idleMs); + } return; } @@ -579,6 +672,12 @@ async function captureTurn(client, threadId, startRequest, options = {}) { } applyTurnNotification(state, message); + // Reset the idle watchdog on every turn-belonging notification (including active + // subagent streaming). Do NOT reset on the forwarded !belongsToTurn branch above, + // or an unrelated broker session would keep a dead turn alive. + if (!state.completed) { + armIdleTimer(state, idleMs); + } }); try { @@ -589,8 +688,18 @@ async function captureTurn(client, threadId, startRequest, options = {}) { state.threadTurnIds.set(state.threadId, state.turnId); } for (const message of state.bufferedNotifications) { - if (belongsToTurn(state, message)) { + if (message.method === "thread/started" || message.method === "thread/name/updated") { + // Mirror the live handler: these register turn threads (e.g. subagents) and carry no + // turnId, so belongsToTurn would reject them. Apply and reset the watchdog here too. + applyTurnNotification(state, message); + if (!state.completed) { + armIdleTimer(state, idleMs); + } + } else if (belongsToTurn(state, message)) { applyTurnNotification(state, message); + if (!state.completed) { + armIdleTimer(state, idleMs); + } } else { if (previousHandler) { previousHandler(message); @@ -603,9 +712,31 @@ async function captureTurn(client, threadId, startRequest, options = {}) { completeTurn(state, response.turn); } - return await state.completion; + // Arm the watchdog only once the turn is live (after startRequest resolved); arming + // before it would race a still-pending RPC that rejects on disconnect. + if (!state.completed) { + armIdleTimer(state, idleMs); + } + + // Layer A: race normal completion against transport death. exitPromise resolves within + // ms when the pipe/socket dies but is never linked to state.completion, so without this + // race a dropped terminal event on a dead connection would hang until the idle window. + const outcome = await Promise.race([ + state.completion.then((value) => ({ type: "completion", value })), + client.exitPromise.then(() => ({ type: "exit" })) + ]); + + if (outcome.type === "exit") { + if (!state.completed) { + completeTurn(state, null, { inferred: true, timedOut: true }); + } + return state; + } + + return outcome.value; } finally { clearCompletionTimer(state); + clearIdleTimer(state); client.setNotificationHandler(previousHandler ?? null); } } @@ -755,6 +886,28 @@ function buildResultStatus(turnState) { return turnState.finalTurn?.status === "completed" ? 0 : 1; } +const SALVAGE_CAPTURED_NOTE = + "Note: Codex stopped emitting events before confirming completion; returning the captured-so-far output, which may be incomplete."; +const SALVAGE_EMPTY_NOTE = + "No verdict was salvageable - Codex went silent before producing output; re-run or inspect the thread."; + +/** + * Prefix a one-line salvage note onto a timed-out turn's body so the surface never + * renders a bare "failed". Two cases: captured-then-silence keeps the partial body, + * empty-then-silence replaces an empty body with an explicit no-verdict message. + * Non-timed-out bodies pass through untouched. + */ +function annotateSalvagedBody(body, timedOut) { + if (!timedOut) { + return body; + } + const text = typeof body === "string" ? body.trim() : ""; + if (text) { + return `${SALVAGE_CAPTURED_NOTE}\n\n${body}`; + } + return SALVAGE_EMPTY_NOTE; +} + const BUILTIN_PROVIDER_LABELS = new Map([ ["openai", "OpenAI"], ["ollama", "Ollama"], @@ -1030,6 +1183,8 @@ export async function runAppServerReview(cwd, options = {}) { }), { onProgress: options.onProgress, + turnIdleMs: options.turnIdleMs, + env: options.env, onResponse(response, state) { if (response.reviewThreadId) { state.threadIds.add(response.reviewThreadId); @@ -1046,9 +1201,10 @@ export async function runAppServerReview(cwd, options = {}) { threadId: turnState.threadId, sourceThreadId, turnId: turnState.turnId, - reviewText: turnState.reviewText, + reviewText: annotateSalvagedBody(turnState.reviewText, turnState.timedOut), reasoningSummary: turnState.reasoningSummary, turn: turnState.finalTurn, + timedOut: turnState.timedOut, error: turnState.error, stderr: cleanCodexStderr(client.stderr) }; @@ -1140,16 +1296,21 @@ export async function runAppServerTurn(cwd, options = {}) { effort: options.effort ?? null, outputSchema: options.outputSchema ?? null }), - { onProgress: options.onProgress } + { + onProgress: options.onProgress, + turnIdleMs: options.turnIdleMs, + env: options.env + } ); return { status: buildResultStatus(turnState), threadId, turnId: turnState.turnId, - finalMessage: turnState.lastAgentMessage, + finalMessage: annotateSalvagedBody(turnState.lastAgentMessage, turnState.timedOut), reasoningSummary: turnState.reasoningSummary, turn: turnState.finalTurn, + timedOut: turnState.timedOut, error: turnState.error, stderr: cleanCodexStderr(client.stderr), fileChanges: turnState.fileChanges, diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index f83c96a0..752788f7 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -452,12 +452,151 @@ rl.on("line", (line) => { prompt }; saveState(state); - send({ id: message.id, result: { turn: buildTurn(turnId) } }); + if (BEHAVIOR === "subagent-buffered-start") { + const subThread = nextThread(state, thread.cwd, true); + const subThreadRecord = ensureThread(state, subThread.id); + subThreadRecord.name = "design-challenger"; + saveState(state); + // Emit the subagent thread/started BEFORE the turn/start response: the client buffers it + // (state.turnId is only set once the response's await continuation runs), so it is handled + // through the buffered-replay path rather than the live handler. + send({ method: "thread/started", params: { thread: { ...buildThread(subThreadRecord), name: "design-challenger", agentNickname: "design-challenger" } } }); + send({ id: message.id, result: { turn: buildTurn(turnId) } }); + send({ method: "turn/started", params: { threadId: thread.id, turn: buildTurn(turnId) } }); + send({ + method: "item/completed", + params: { + threadId: subThread.id, + turnId, + item: { type: "agentMessage", id: "msg_sub_" + turnId, text: "Subagent analysis note.", phase: "analysis" } + } + }); + send({ + method: "item/completed", + params: { + threadId: thread.id, + turnId, + item: { type: "agentMessage", id: "msg_" + turnId, text: taskPayload(prompt, false), phase: "final_answer" } + } + }); + send({ method: "turn/completed", params: { threadId: thread.id, turn: buildTurn(turnId, "completed") } }); + break; + } + + send({ id: message.id, result: { turn: buildTurn(turnId) } }); const payload = message.params.outputSchema && message.params.outputSchema.properties && message.params.outputSchema.properties.verdict ? structuredReviewPayload(prompt) : taskPayload(prompt, thread.name && thread.name.startsWith("Codex Companion Task") && prompt.includes("Continue from the current thread state")); + if (BEHAVIOR === "silent-after-progress") { + send({ method: "turn/started", params: { threadId: thread.id, turn: buildTurn(turnId) } }); + send({ + method: "item/completed", + params: { + threadId: thread.id, + turnId, + item: { type: "agentMessage", id: "msg_" + turnId, text: "Looked into the failing retry path before going quiet.", phase: "analysis" } + } + }); + // Intentionally drop BOTH the final_answer agentMessage and the turn/completed event. + // The connection stays open and silent, reproducing the stalled-turn hang. + break; + } + + if (BEHAVIOR === "transport-death-mid-turn") { + send({ method: "turn/started", params: { threadId: thread.id, turn: buildTurn(turnId) } }); + // Let the turn/start response + turn/started flush and register the live turn, THEN drop the + // transport before any agentMessage. This is connection death on an already-live turn (the + // exitPromise-race scenario), not a pre-response rejection. + setTimeout(() => process.exit(0), 150); + break; + } + + if (BEHAVIOR === "subagent-streaming-spaced") { + const subThread = nextThread(state, thread.cwd, true); + const subThreadRecord = ensureThread(state, subThread.id); + subThreadRecord.name = "design-challenger"; + saveState(state); + const subTurnId = nextTurnId(state); + send({ method: "turn/started", params: { threadId: thread.id, turn: buildTurn(turnId) } }); + send({ method: "thread/started", params: { thread: { ...buildThread(subThreadRecord), name: "design-challenger", agentNickname: "design-challenger" } } }); + send({ + method: "item/started", + params: { + threadId: thread.id, + turnId, + item: { + type: "collabAgentToolCall", + id: "collab_" + turnId, + tool: "wait", + status: "inProgress", + senderThreadId: thread.id, + receiverThreadIds: [subThread.id], + prompt: "Stream design feedback", + model: null, + reasoningEffort: null, + agentsStates: { [subThread.id]: { status: "inProgress", message: "Working" } } + } + } + }); + send({ method: "turn/started", params: { threadId: subThread.id, turn: buildTurn(subTurnId) } }); + let streamed = 0; + const STREAM_STEPS = 7; + const STREAM_GAP_MS = 200; + const streamNext = () => { + if (streamed < STREAM_STEPS) { + send({ + method: "item/completed", + params: { + threadId: subThread.id, + turnId: subTurnId, + item: { + type: "reasoning", + id: "reasoning_" + subTurnId + "_" + streamed, + summary: [{ text: "Subagent streaming step " + streamed }], + content: [] + } + } + }); + streamed++; + setTimeout(streamNext, STREAM_GAP_MS); + return; + } + send({ method: "turn/completed", params: { threadId: subThread.id, turn: buildTurn(subTurnId, "completed") } }); + send({ + method: "item/completed", + params: { + threadId: thread.id, + turnId, + item: { + type: "collabAgentToolCall", + id: "collab_" + turnId, + tool: "wait", + status: "completed", + senderThreadId: thread.id, + receiverThreadIds: [subThread.id], + prompt: "Stream design feedback", + model: null, + reasoningEffort: null, + agentsStates: { [subThread.id]: { status: "completed", message: "Finished" } } + } + } + }); + send({ + method: "item/completed", + params: { + threadId: thread.id, + turnId, + item: { type: "agentMessage", id: "msg_" + turnId, text: payload, phase: "final_answer" } + } + }); + send({ method: "turn/completed", params: { threadId: thread.id, turn: buildTurn(turnId, "completed") } }); + }; + setTimeout(streamNext, STREAM_GAP_MS); + break; + } + if ( BEHAVIOR === "with-subagent" || BEHAVIOR === "with-late-subagent-message" || diff --git a/tests/helpers.mjs b/tests/helpers.mjs index 945ae0e7..2c534dfc 100644 --- a/tests/helpers.mjs +++ b/tests/helpers.mjs @@ -18,6 +18,7 @@ export function run(command, args, options = {}) { env: options.env, encoding: "utf8", input: options.input, + timeout: options.timeout, shell: process.platform === "win32" && !path.isAbsolute(command), windowsHide: true }); diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 8f276835..ed131be0 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -9,6 +9,8 @@ import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs"; import { initGitRepo, makeTempDir, run } from "./helpers.mjs"; import { loadBrokerSession, saveBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; +import * as codexRuntime from "../plugins/codex/scripts/lib/codex.mjs"; +import { createBrokerEndpoint } from "../plugins/codex/scripts/lib/broker-endpoint.mjs"; const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); const PLUGIN_ROOT = path.join(ROOT, "plugins", "codex"); @@ -2257,3 +2259,259 @@ test("setup and status honor --cwd when reading shared session runtime", () => { assert.equal(payload.sessionRuntime.mode, "shared"); assert.equal(payload.sessionRuntime.endpoint, "unix:/tmp/fake-broker.sock"); }); + +test("resolveTurnIdleMs parses the idle window strictly with 0 meaning disabled", () => { + const { resolveTurnIdleMs, DEFAULT_TURN_IDLE_TIMEOUT_MS } = codexRuntime; + assert.equal(typeof resolveTurnIdleMs, "function"); + assert.equal(DEFAULT_TURN_IDLE_TIMEOUT_MS, 900000); + const ENV = "CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS"; + assert.equal(resolveTurnIdleMs({}), DEFAULT_TURN_IDLE_TIMEOUT_MS); + assert.equal(resolveTurnIdleMs({ [ENV]: "" }), DEFAULT_TURN_IDLE_TIMEOUT_MS); + assert.equal(resolveTurnIdleMs({ [ENV]: " " }), DEFAULT_TURN_IDLE_TIMEOUT_MS); + assert.equal(resolveTurnIdleMs({ [ENV]: "not-a-number" }), DEFAULT_TURN_IDLE_TIMEOUT_MS); + assert.equal(resolveTurnIdleMs({ [ENV]: "-5" }), DEFAULT_TURN_IDLE_TIMEOUT_MS); + // 0 is an explicit, valid opt-out and must survive (not collapse to the default). + assert.equal(resolveTurnIdleMs({ [ENV]: "0" }), 0); + assert.equal(resolveTurnIdleMs({ [ENV]: "1500" }), 1500); + // Unit-suffixed or decimal values must NOT parse to their numeric prefix - otherwise "15m" + // would become a 15ms watchdog and "0ms" would silently disable it; they fall back to default. + assert.equal(resolveTurnIdleMs({ [ENV]: "15m" }), DEFAULT_TURN_IDLE_TIMEOUT_MS); + assert.equal(resolveTurnIdleMs({ [ENV]: "0ms" }), DEFAULT_TURN_IDLE_TIMEOUT_MS); + assert.equal(resolveTurnIdleMs({ [ENV]: "900000ms" }), DEFAULT_TURN_IDLE_TIMEOUT_MS); + assert.equal(resolveTurnIdleMs({ [ENV]: "1.5" }), DEFAULT_TURN_IDLE_TIMEOUT_MS); + // Node's setTimeout overflows delays above 2^31-1 ms (firing in ~1 ms); clamp to that max + // so a very large window does not collapse into near-immediate salvage. + assert.equal(resolveTurnIdleMs({ [ENV]: "2147483647" }), 2147483647); + assert.equal(resolveTurnIdleMs({ [ENV]: "2147483648" }), 2147483647); + assert.equal(resolveTurnIdleMs({ [ENV]: "2592000000" }), 2147483647); +}); + +test("task salvages a stalled turn when Codex goes idle without terminal events", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "silent-after-progress"); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + + const result = run("node", [SCRIPT, "task", "investigate the silent hang"], { + cwd: repo, + env: { ...buildEnv(binDir), CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS: "1500" }, + timeout: 15000 + }); + + // A salvaged turn is reported as a non-success (incomplete -> exit 1), never a false completion. + assert.equal(result.status, 1, result.stderr); + assert.match(result.stdout, /Note: Codex stopped emitting events before confirming completion/); + assert.match(result.stdout, /Looked into the failing retry path before going quiet\./); + + const state = JSON.parse(fs.readFileSync(path.join(resolveStateDir(repo), "state.json"), "utf8")); + const job = state.jobs.find((candidate) => candidate.jobClass === "task"); + assert.ok(job, "expected a tracked task job"); + // The core bug: a stalled turn must not leave the job stuck "running". + assert.equal(job.status, "failed"); + assert.equal(job.pid, null); +}); + +test("a healthy task is neither reaped nor pinned by the idle watchdog", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + + const start = Date.now(); + const result = run("node", [SCRIPT, "task", "do a quick task"], { + cwd: repo, + env: { ...buildEnv(binDir), CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS: "600000" }, + timeout: 15000 + }); + const elapsed = Date.now() - start; + + assert.equal(result.status, 0, result.stderr); + assert.equal(result.stdout, "Handled the requested task.\nTask prompt accepted.\n"); + assert.doesNotMatch(result.stdout, /Note: Codex stopped emitting events/); + // An un-unref'd / uncleared 600000ms timer would pin the short-lived CLI until the spawn timeout. + assert.ok(elapsed < 15000, `expected a prompt exit, took ${elapsed}ms`); +}); + +test("a turn streaming subagent activity under the idle window is not reaped", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "subagent-streaming-spaced"); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + + const result = run("node", [SCRIPT, "task", "challenge the design with streaming feedback"], { + cwd: repo, + env: { ...buildEnv(binDir), CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS: "1200" }, + timeout: 20000 + }); + + // Each subagent item (~200ms apart) resets the 1200ms window, so the turn completes normally + // even though its total span (~1.6s) exceeds the window. A broken reset would still fire at + // 1200ms (< the ~1.6s completion) and salvage early, so the reset stays load-bearing while the + // 6x gap-to-window margin keeps the test robust against CI scheduling jitter. + assert.equal(result.status, 0, result.stderr); + assert.equal(result.stdout, "Handled the requested task.\nTask prompt accepted.\n"); + assert.doesNotMatch(result.stdout, /Note: Codex stopped emitting events/); +}); + +test("task salvages immediately when the Codex transport dies mid-turn", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "transport-death-mid-turn"); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + + const start = Date.now(); + const result = run("node", [SCRIPT, "task", "investigate the crash"], { + cwd: repo, + env: { + ...buildEnv(binDir), + // A large idle window proves the salvage comes from the exitPromise race, not the watchdog. + CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS: "600000", + // Force the direct transport. A brokered app-server death is not observable as a client + // disconnect (the broker process stays up), so Layer A's exitPromise race is exercised on + // the direct transport, where killing the app-server is the client's own transport death. + // Pointing at a non-existent broker makes withAppServer fall back to a directly-spawned server. + CODEX_COMPANION_APP_SERVER_ENDPOINT: createBrokerEndpoint(path.join(binDir, "no-such-broker-dir")) + }, + timeout: 20000 + }); + const elapsed = Date.now() - start; + + assert.equal(result.status, 1, result.stderr); + assert.match(result.stdout, /No verdict was salvageable - Codex went silent before producing output/); + assert.ok(elapsed < 15000, `expected a fast transport-death salvage, took ${elapsed}ms`); +}); + +test("a brokered task salvages a stalled turn when terminal events are dropped", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "silent-after-progress"); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + fs.writeFileSync(path.join(repo, "README.md"), "hello again\n"); + + const env = { ...buildEnv(binDir), CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS: "1500" }; + + // review/start is unaffected by the silent-after-progress behavior, so it completes and + // establishes the shared broker (mirrors the existing brokered-subagent test). + const review = run("node", [SCRIPT, "review"], { cwd: repo, env, timeout: 20000 }); + assert.equal(review.status, 0, review.stderr); + + if (!loadBrokerSession(repo)) { + return; + } + + const result = run("node", [SCRIPT, "task", "investigate the silent hang"], { + cwd: repo, + env, + timeout: 20000 + }); + + assert.equal(result.status, 1, result.stderr); + assert.match(result.stdout, /Note: Codex stopped emitting events before confirming completion/); + + run("node", [SESSION_HOOK, "SessionEnd"], { + cwd: repo, + env, + input: JSON.stringify({ hook_event_name: "SessionEnd", cwd: repo }) + }); +}); + +test("the idle watchdog opt-out (0) leaves a stalled turn running in the background", async (t) => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "silent-after-progress"); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + + const env = { + ...buildEnv(binDir), + CODEX_COMPANION_TURN_IDLE_TIMEOUT_MS: "0", + CODEX_COMPANION_SESSION_ID: "sess-idle-disabled" + }; + + t.after(() => { + run("node", [SESSION_HOOK, "SessionEnd"], { + cwd: repo, + env, + input: JSON.stringify({ + hook_event_name: "SessionEnd", + session_id: "sess-idle-disabled", + cwd: repo + }) + }); + }); + + const launched = run("node", [SCRIPT, "task", "--background", "--json", "investigate the silent hang"], { + cwd: repo, + env, + timeout: 15000 + }); + assert.equal(launched.status, 0, launched.stderr); + const launchPayload = JSON.parse(launched.stdout); + assert.match(launchPayload.jobId, /^task-/); + + // Wait for the detached worker to actually begin the (stalling) turn. + await waitFor( + () => { + const snap = run("node", [SCRIPT, "status", launchPayload.jobId, "--json"], { cwd: repo, env }); + if (snap.status !== 0) { + return null; + } + return JSON.parse(snap.stdout).job.status === "running" ? true : null; + }, + { timeoutMs: 15000, intervalMs: 300 } + ); + + // With the watchdog disabled, the stalled turn is never salvaged: the job stays running + // across a wait window instead of flipping to failed. + const waited = run( + "node", + [SCRIPT, "status", launchPayload.jobId, "--wait", "--timeout-ms", "3000", "--json"], + { cwd: repo, env, timeout: 15000 } + ); + assert.equal(waited.status, 0, waited.stderr); + const waitedPayload = JSON.parse(waited.stdout); + assert.equal(waitedPayload.job.status, "running"); + assert.equal(waitedPayload.waitTimedOut, true); +}); + +test("a buffered subagent thread-start is applied during replay so its label is registered", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "subagent-buffered-start"); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + + const result = run("node", [SCRIPT, "task", "challenge with a buffered subagent start"], { + cwd: repo, + env: buildEnv(binDir), + timeout: 20000 + }); + + assert.equal(result.status, 0, result.stderr); + // thread/started for the subagent arrives before the turn/start response, so it is buffered and + // handled through the replay path. That path must apply thread/started (like the live handler) so + // the subagent thread is registered with its name; otherwise the message logs under the raw id. + const stateDir = resolveStateDir(repo); + const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const log = fs.readFileSync(state.jobs[0].logFile, "utf8"); + assert.match(log, /Subagent design-challenger/); +});