diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs index 0361247e..4cd16b7b 100644 --- a/plugins/codex/scripts/codex-companion.mjs +++ b/plugins/codex/scripts/codex-companion.mjs @@ -35,6 +35,8 @@ import { import { buildSingleJobSnapshot, buildStatusSnapshot, + CANCELLATION_INTERRUPT_REQUIRED_MESSAGE, + isOrphanedTurn, readStoredJob, resolveCancelableJob, resolveResultJob, @@ -1135,6 +1137,44 @@ async function handleCancel(argv) { return cancelledJob; }; + const isOrphan = isOrphanedTurn(job); + + // An orphaned job has no local pid to fall back on — the remote interrupt is + // the ONLY mechanism that can actually stop it, so cancellation must not be + // persisted until that interrupt is confirmed. Do this BEFORE the optimistic + // persistCancellation() below: interruptAppServerTurn is bounded (see + // DEFAULT_INTERRUPT_TIMEOUT_MS), but persisting "cancelled" first and rolling + // back only after the await returns leaves a false-cancelled state on disk if + // the process is killed or crashes while still awaiting it. + if (isOrphan) { + const interrupt = await interruptAppServerTurn(cwd, { threadId, turnId }); + if (interrupt.attempted) { + appendLogLine( + job.logFile, + interrupt.interrupted + ? `Requested Codex turn interrupt for ${turnId} on ${threadId}.` + : `Codex turn interrupt failed${interrupt.detail ? `: ${interrupt.detail}` : "."}` + ); + } + if (!interrupt.interrupted) { + const detail = interrupt.detail ? `: ${interrupt.detail}` : "."; + const message = `${CANCELLATION_INTERRUPT_REQUIRED_MESSAGE}${detail}`; + appendLogLine(job.logFile, message); + throw new Error(message); + } + const nextJob = persistCancellation(null); + appendLogLine(job.logFile, "Cancelled by user."); + const payload = { + jobId: job.id, + status: "cancelled", + title: job.title, + turnInterruptAttempted: interrupt.attempted, + turnInterrupted: interrupt.interrupted + }; + outputCommandResult(payload, renderCancelReport(nextJob), options.json); + return; + } + persistCancellation(job.pid ?? null); appendLogLine(job.logFile, "Cancelled by user."); diff --git a/plugins/codex/scripts/lib/codex.mjs b/plugins/codex/scripts/lib/codex.mjs index 2020490c..4f5a95f6 100644 --- a/plugins/codex/scripts/lib/codex.mjs +++ b/plugins/codex/scripts/lib/codex.mjs @@ -813,7 +813,8 @@ async function captureTurn(client, threadId, startRequest, options = {}) { if (state.threadId && state.turnId && options.cwd) { await interruptAppServerTurn(options.cwd, { threadId: state.threadId, - turnId: state.turnId + turnId: state.turnId, + timeoutMs: TURN_INTERRUPT_TIMEOUT_MS }).catch(() => {}); } throw error; @@ -1186,7 +1187,26 @@ export async function getCodexAuthStatus(cwd, options = {}) { } } -export async function interruptAppServerTurn(cwd, { threadId, turnId }) { +// A control-plane RPC (send-interrupt-and-acknowledge), not agent work — it +// must not hang the way a turn can. Callers that gate a state transition on +// the result (e.g. /codex:cancel finalizing an orphaned job) need a bounded +// wait so a stuck broker/transport fails fast instead of leaving the caller +// stalled indefinitely on this await. CODEX_INTERRUPT_TIMEOUT_MS lets tests +// shrink this the same way CODEX_TURN_TIMEOUT_MS overrides the turn budget. +const DEFAULT_INTERRUPT_TIMEOUT_MS = 15000; + +function resolveInterruptTimeoutMs(timeoutMs) { + if (Number.isFinite(timeoutMs) && timeoutMs > 0) { + return timeoutMs; + } + const fromEnv = Number(process.env.CODEX_INTERRUPT_TIMEOUT_MS); + if (Number.isFinite(fromEnv) && fromEnv > 0) { + return fromEnv; + } + return DEFAULT_INTERRUPT_TIMEOUT_MS; +} + +export async function interruptAppServerTurn(cwd, { threadId, turnId, timeoutMs } = {}) { if (!threadId || !turnId) { return { attempted: false, @@ -1206,6 +1226,8 @@ export async function interruptAppServerTurn(cwd, { threadId, turnId }) { }; } + const resolvedTimeoutMs = resolveInterruptTimeoutMs(timeoutMs); + let client = null; try { client = await CodexAppServerClient.connect(cwd, { @@ -1213,10 +1235,10 @@ export async function interruptAppServerTurn(cwd, { threadId, turnId }) { allowBusyStaleBroker: true }); await client.request("turn/interrupt", { threadId, turnId }, { - timeoutMs: TURN_INTERRUPT_TIMEOUT_MS, + timeoutMs: resolvedTimeoutMs, // A wedged peer that accepted the connection but goes silent must not // linger past the timeout: force-close so callers aren't blocked - // beyond TURN_INTERRUPT_TIMEOUT_MS regardless of transport. + // beyond resolvedTimeoutMs regardless of transport. onTimeout: () => client?.close().catch(() => {}) }); return { diff --git a/plugins/codex/scripts/lib/job-control.mjs b/plugins/codex/scripts/lib/job-control.mjs index 791f4506..c2fc04ef 100644 --- a/plugins/codex/scripts/lib/job-control.mjs +++ b/plugins/codex/scripts/lib/job-control.mjs @@ -2,7 +2,15 @@ import fs from "node:fs"; import { getSessionRuntimeStatus } from "./codex.mjs"; import { isProcessAlive } from "./process.mjs"; -import { getConfig, listJobs, readJobFile, resolveJobFile, upsertJob, writeJobFile } from "./state.mjs"; +import { + getConfig, + listJobs, + readJobFile, + resolveJobFile, + UNREPORTED_PROCESS_EXIT_MESSAGE, + upsertJob, + writeJobFile +} from "./state.mjs"; import { SESSION_ID_ENV } from "./tracked-jobs.mjs"; import { resolveWorkspaceRoot } from "./workspace.mjs"; @@ -10,6 +18,8 @@ export const DEFAULT_MAX_STATUS_JOBS = 8; export const DEFAULT_MAX_PROGRESS_LINES = 4; export const CANCELLATION_TERMINATION_FAILED_MESSAGE = "Cancellation requested but process termination failed; retry /codex:cancel."; +export const CANCELLATION_INTERRUPT_REQUIRED_MESSAGE = + "Cancellation requested but the remote Codex turn interrupt failed; retry /codex:cancel."; export function sortJobsNewestFirst(jobs) { return [...jobs].sort((left, right) => String(right.updatedAt ?? "").localeCompare(String(left.updatedAt ?? ""))); @@ -323,10 +333,21 @@ export function resolveResultJob(cwd, reference) { throw new Error("No finished Codex jobs found for this repository yet."); } +export function isOrphanedTurn(job) { + return ( + job.status === "failed" && + job.errorMessage === UNREPORTED_PROCESS_EXIT_MESSAGE && + Boolean(job.threadId) && + Boolean(job.turnId) + ); +} + export function resolveCancelableJob(cwd, reference, options = {}) { const workspaceRoot = resolveWorkspaceRoot(cwd); const jobs = sortJobsNewestFirst(listJobs(workspaceRoot)); - const activeJobs = jobs.filter((job) => job.status === "queued" || job.status === "running"); + const activeJobs = jobs.filter( + (job) => job.status === "queued" || job.status === "running" || isOrphanedTurn(job) + ); if (reference) { const selected = matchJobReference(activeJobs, reference); diff --git a/plugins/codex/scripts/lib/state.mjs b/plugins/codex/scripts/lib/state.mjs index 471ff628..542804cc 100644 --- a/plugins/codex/scripts/lib/state.mjs +++ b/plugins/codex/scripts/lib/state.mjs @@ -12,7 +12,7 @@ const FALLBACK_STATE_ROOT_DIR = path.join(os.tmpdir(), "codex-companion"); const STATE_FILE_NAME = "state.json"; const JOBS_DIR_NAME = "jobs"; const MAX_JOBS = 50; -const UNREPORTED_PROCESS_EXIT_MESSAGE = "Process exited without reporting."; +export const UNREPORTED_PROCESS_EXIT_MESSAGE = "Process exited without reporting."; function nowIso() { return new Date().toISOString(); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index ac6dd4c9..cd1ab6ec 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -927,6 +927,14 @@ rl.on("line", (line) => { } case "turn/interrupt": { + if (BEHAVIOR === "stalled-interrupt") { + // Never respond — simulates a wedged/hung broker or app-server during + // turn/interrupt. interruptAppServerTurn() must bound this request + // (DEFAULT_INTERRUPT_TIMEOUT_MS) instead of hanging forever, so a + // caller gating a state transition on the result (e.g. /codex:cancel + // finalizing an orphaned job) is never left stuck. + break; + } state.lastInterrupt = { threadId: message.params.threadId, turnId: message.params.turnId @@ -1007,6 +1015,9 @@ export function buildEnv(binDir) { CLAUDE_PLUGIN_DATA: getTestPluginDataDir(), // Production keeps an idle broker warm for 15 minutes. Tests only need a // brief reuse window and should not leave dozens of detached helpers. - CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS: "2000" + CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS: "2000", + // Production bounds a stuck turn/interrupt request at 15s. Tests exercising + // a hung interrupt (BEHAVIOR "stalled-interrupt") should not wait that long. + CODEX_INTERRUPT_TIMEOUT_MS: "1000" }; } diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 517c4b37..9f8e7465 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -2262,6 +2262,179 @@ test("cancel stops an active background job and marks it cancelled", async (t) = assert.match(fs.readFileSync(logFile, "utf8"), /Cancelled by user/); }); +function writeOrphanedJobFixture(workspace, { threadId = "thr_orphaned", turnId = "turn_orphaned" } = {}) { + const stateDir = resolveStateDir(workspace); + const jobsDir = path.join(stateDir, "jobs"); + fs.mkdirSync(jobsDir, { recursive: true }); + + const logFile = path.join(jobsDir, "task-orphaned.log"); + const jobFile = path.join(jobsDir, "task-orphaned.json"); + fs.writeFileSync(logFile, "[2026-03-18T15:30:00.000Z] Starting Codex Task.\n", "utf8"); + fs.writeFileSync( + jobFile, + JSON.stringify( + { + id: "task-orphaned", + status: "running", + title: "Codex Task", + threadId, + turnId, + logFile + }, + null, + 2 + ), + "utf8" + ); + fs.writeFileSync( + path.join(stateDir, "state.json"), + `${JSON.stringify( + { + version: 1, + config: { stopReviewGate: false }, + jobs: [ + { + id: "task-orphaned", + status: "running", + title: "Codex Task", + jobClass: "task", + summary: "Investigate flaky test", + threadId, + turnId, + // A pid that is guaranteed to be dead simulates the SIGKILLed foreground process. + pid: 999999, + logFile, + createdAt: "2026-03-18T15:30:00.000Z", + startedAt: "2026-03-18T15:30:01.000Z", + updatedAt: "2026-03-18T15:30:02.000Z" + } + ] + }, + null, + 2 + )}\n`, + "utf8" + ); + + return { stateDir, jobsDir, logFile, jobFile }; +} + +test("cancel reaches a job orphaned by a SIGKILLed companion process once the remote turn interrupt succeeds (closes #42)", () => { + const workspace = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir); + // buildEnv() pins process.env.CLAUDE_PLUGIN_DATA for this worker (see its doc + // comment) — resolve it before writeOrphanedJobFixture computes resolveStateDir, + // so the in-process read and the spawned companion's state dir agree. + const env = buildEnv(binDir); + const { stateDir, logFile, jobFile } = writeOrphanedJobFixture(workspace); + + // Reconciliation (triggered by /codex:status) flips the dead-pid job to "failed" + // before cancel ever runs, mirroring the real orphan sequence from the issue. + const statusResult = run("node", [SCRIPT, "status", "--json"], { cwd: workspace, env }); + assert.equal(statusResult.status, 0, statusResult.stderr); + const reconciled = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")).jobs.find( + (job) => job.id === "task-orphaned" + ); + assert.equal(reconciled.status, "failed"); + assert.equal(reconciled.errorMessage, "Process exited without reporting."); + + const cancelResult = run("node", [SCRIPT, "cancel", "task-orphaned", "--json"], { + cwd: workspace, + env + }); + + assert.equal(cancelResult.status, 0, cancelResult.stderr); + const payload = JSON.parse(cancelResult.stdout); + assert.equal(payload.status, "cancelled"); + assert.equal(payload.turnInterrupted, true); + + const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const cancelled = state.jobs.find((job) => job.id === "task-orphaned"); + assert.equal(cancelled.status, "cancelled"); + + const stored = JSON.parse(fs.readFileSync(jobFile, "utf8")); + assert.equal(stored.status, "cancelled"); + assert.match(fs.readFileSync(logFile, "utf8"), /Cancelled by user/); +}); + +test("cancel does not finalize an orphaned job as cancelled when the remote turn interrupt fails, and stays retryable", () => { + const workspace = makeTempDir(); + const { stateDir, logFile, jobFile } = writeOrphanedJobFixture(workspace); + + // No fake codex on PATH: interruptAppServerTurn cannot reach a real broker/thread, + // so the interrupt attempt fails (or is never attempted) — this must NOT be + // treated as a successful cancellation, since nothing actually stopped the + // orphaned remote turn. + const statusResult = run("node", [SCRIPT, "status", "--json"], { cwd: workspace }); + assert.equal(statusResult.status, 0, statusResult.stderr); + const reconciled = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")).jobs.find( + (job) => job.id === "task-orphaned" + ); + assert.equal(reconciled.status, "failed"); + + const cancelResult = run("node", [SCRIPT, "cancel", "task-orphaned", "--json"], { + cwd: workspace + }); + + assert.notEqual(cancelResult.status, 0); + assert.match(cancelResult.stderr, /remote Codex turn interrupt failed/i); + + // The job must remain in its original orphaned-failed state so a retried + // /codex:cancel can still select and re-attempt it via isOrphanedTurn. + const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const stillOrphaned = state.jobs.find((job) => job.id === "task-orphaned"); + assert.equal(stillOrphaned.status, "failed"); + assert.equal(stillOrphaned.errorMessage, "Process exited without reporting."); + assert.equal(stillOrphaned.pid, null); + + const stored = JSON.parse(fs.readFileSync(jobFile, "utf8")); + assert.equal(stored.status, "failed"); + assert.equal(stored.errorMessage, "Process exited without reporting."); + + const retryResult = run("node", [SCRIPT, "cancel", "task-orphaned", "--json"], { + cwd: workspace + }); + assert.notEqual(retryResult.status, 0); + assert.match(retryResult.stderr, /remote Codex turn interrupt failed/i); +}); + +test("cancel on an orphaned job does not persist cancelled while the remote interrupt is still hung, and stays retryable once it times out", () => { + const workspace = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "stalled-interrupt"); + const env = buildEnv(binDir); + const { stateDir, logFile, jobFile } = writeOrphanedJobFixture(workspace); + + const statusResult = run("node", [SCRIPT, "status", "--json"], { cwd: workspace, env }); + assert.equal(statusResult.status, 0, statusResult.stderr); + const reconciled = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")).jobs.find( + (job) => job.id === "task-orphaned" + ); + assert.equal(reconciled.status, "failed"); + + // The fake broker never answers turn/interrupt. Without a bounded timeout on + // that request, /codex:cancel would hang here; with one, it must return + // (non-zero) once the timeout fires rather than persisting "cancelled" before + // the interrupt is confirmed. + const cancelResult = run("node", [SCRIPT, "cancel", "task-orphaned", "--json"], { + cwd: workspace, + env + }); + + assert.notEqual(cancelResult.status, 0); + assert.match(cancelResult.stderr, /remote Codex turn interrupt failed/i); + + const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const stillOrphaned = state.jobs.find((job) => job.id === "task-orphaned"); + assert.equal(stillOrphaned.status, "failed"); + assert.equal(stillOrphaned.errorMessage, "Process exited without reporting."); + + const stored = JSON.parse(fs.readFileSync(jobFile, "utf8")); + assert.equal(stored.status, "failed"); + assert.match(fs.readFileSync(logFile, "utf8"), /remote Codex turn interrupt failed/i); +}); + test("failed cancellation restores a live job so cancellation can be retried", () => { const workspace = makeTempDir(); const job = {