diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs index 4cd16b7b..8ea9da19 100644 --- a/plugins/codex/scripts/codex-companion.mjs +++ b/plugins/codex/scripts/codex-companion.mjs @@ -50,6 +50,7 @@ import { createJobRecord, createProgressReporter, nowIso, + registerWorkerCrashGuard, runTrackedJob, SESSION_ID_ENV } from "./lib/tracked-jobs.mjs"; @@ -1004,6 +1005,7 @@ async function handleTaskWorker(argv) { logFile: storedJob.logFile ?? null } ); + registerWorkerCrashGuard(workspaceRoot, options["job-id"], logFile); await runTrackedJob( { ...storedJob, diff --git a/plugins/codex/scripts/lib/codex.mjs b/plugins/codex/scripts/lib/codex.mjs index 4f5a95f6..1eeaa1fb 100644 --- a/plugins/codex/scripts/lib/codex.mjs +++ b/plugins/codex/scripts/lib/codex.mjs @@ -1206,7 +1206,12 @@ function resolveInterruptTimeoutMs(timeoutMs) { return DEFAULT_INTERRUPT_TIMEOUT_MS; } -export async function interruptAppServerTurn(cwd, { threadId, turnId, timeoutMs } = {}) { +/** + * @param {string} cwd + * @param {{ threadId?: string, turnId?: string, timeoutMs?: number }} [options] + */ +export async function interruptAppServerTurn(cwd, options = {}) { + const { threadId, turnId, timeoutMs } = options; if (!threadId || !turnId) { return { attempted: false, diff --git a/plugins/codex/scripts/lib/state.mjs b/plugins/codex/scripts/lib/state.mjs index 542804cc..76e79b55 100644 --- a/plugins/codex/scripts/lib/state.mjs +++ b/plugins/codex/scripts/lib/state.mjs @@ -152,7 +152,19 @@ function reconcileRunningJobs(cwd, state) { const completedAt = nowIso(); const staleJobs = []; const jobs = state.jobs.map((job) => { - if (job.status !== "running" || !Number.isInteger(job.pid) || job.pid <= 0 || isProcessAlive(job.pid)) { + // "queued" also needs reconciling: enqueueBackgroundTask records the + // detached worker's pid at enqueue time, before that worker has run far + // enough to flip the record to "running" via runTrackedJob. A worker that + // dies in that window (crash, immediate OOM kill) leaves the job stuck + // "queued" forever with an already-dead pid — invisible to this check if + // it only looked at "running" — permanently blocking --resume-last and + // any other gate that treats queued/running as active. + if ( + (job.status !== "running" && job.status !== "queued") || + !Number.isInteger(job.pid) || + job.pid <= 0 || + isProcessAlive(job.pid) + ) { return job; } diff --git a/plugins/codex/scripts/lib/tracked-jobs.mjs b/plugins/codex/scripts/lib/tracked-jobs.mjs index 2239fc18..b23c060c 100644 --- a/plugins/codex/scripts/lib/tracked-jobs.mjs +++ b/plugins/codex/scripts/lib/tracked-jobs.mjs @@ -252,3 +252,55 @@ export async function runTrackedJob(job, runner, options = {}) { throw error; } } + +function markWorkerJobDead(workspaceRoot, jobId, logFile, errorMessage) { + const stored = readStoredJobOrNull(workspaceRoot, jobId); + if (stored && stored.status !== "running" && stored.status !== "queued") { + // Already terminal (e.g. /codex:cancel wrote "cancelled" and delivered the + // SIGTERM this guard is reacting to) — don't race that state back to failed. + return; + } + const base = stored ?? { id: jobId, status: "running", logFile }; + const completedAt = nowIso(); + writeJobFile(workspaceRoot, jobId, { + ...base, + status: "failed", + phase: "failed", + errorMessage, + pid: null, + completedAt + }); + upsertJob(workspaceRoot, { + id: jobId, + status: "failed", + phase: "failed", + pid: null, + errorMessage, + completedAt + }); + appendLogLine(logFile ?? base.logFile ?? null, `Marked failed: ${errorMessage}`); +} + +// Guards only against in-process crashes (uncaughtException / unhandledRejection) +// where a precise error is available and no other command is writing the job. +// Signal-based deaths (SIGTERM/SIGINT/SIGHUP/SIGKILL) are intentionally NOT +// caught here: SIGKILL is uncatchable, so a reader-side liveness check (see +// state.mjs's reconcileRunningJobs, which now also covers "queued") must cover +// process death regardless, and /codex:cancel delivers SIGTERM as its teardown +// signal after already writing the job "cancelled" — catching it here would +// race that terminal state back to "failed". markWorkerJobDead never rewrites +// a job that already reached a terminal status, so a same-tick cancel wins. +export function registerWorkerCrashGuard(workspaceRoot, jobId, logFile = null) { + const mark = (label) => (reason) => { + try { + const detail = reason instanceof Error ? (reason.stack ?? reason.message) : String(reason ?? ""); + appendLogLine(logFile, `Worker ${label}: ${detail}`); + markWorkerJobDead(workspaceRoot, jobId, logFile, `worker ${label}: ${detail.split("\n")[0]}`); + } catch { + // Never let the guard itself throw during teardown. + } + process.exit(1); + }; + process.on("uncaughtException", mark("uncaughtException")); + process.on("unhandledRejection", mark("unhandledRejection")); +} diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 9f8e7465..cdd8dec3 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -623,6 +623,113 @@ test("task --resume-last resumes the latest persisted task thread", () => { assert.equal(fakeState.lastThreadResume.sandbox, null); }); +test("task --resume-last is not permanently blocked by a job stuck 'running' with a dead worker pid (upstream #392)", () => { + // resolveLatestTrackedTaskThread() reads jobs via listJobs(), which already + // reconciles a "running" job with a dead pid to "failed" (state.mjs's + // reconcileRunningJobs, exercised directly by "status and resume candidates + // mark a running job with a dead pid as failed" above). This test proves + // that reconciliation actually unblocks the --resume-last gate itself, not + // just /status and task-resume-candidate — the exact throw upstream #392 + // reports ("Task is still running. Use /codex:status before continuing + // it.") never fires for a worker that has actually exited. + const repo = makeTempDir(); + const binDir = makeTempDir(); + const statePath = path.join(binDir, "fake-codex-state.json"); + 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 env = { ...buildEnv(binDir), CODEX_COMPANION_SESSION_ID: "sess-dead-worker" }; + + const firstRun = run("node", [SCRIPT, "task", "initial task"], { cwd: repo, env }); + assert.equal(firstRun.status, 0, firstRun.stderr); + + const stateDir = resolveStateDir(repo); + const stateFile = path.join(stateDir, "state.json"); + const state = JSON.parse(fs.readFileSync(stateFile, "utf8")); + const job = state.jobs.find((entry) => entry.jobClass === "task"); + assert.ok(job, "expected the completed task job to be recorded"); + + const exitedWorker = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + const deadPid = exitedWorker.pid; + const exitedForReal = new Promise((resolve, reject) => { + exitedWorker.once("error", reject); + exitedWorker.once("exit", resolve); + }); + + // Simulate the worker crashing mid-run: still "running", pid now dead, no + // completedAt/result was ever written (the exact state a SIGKILL or an + // uncaught exception leaves behind, since neither reaches runTrackedJob's + // catch or a normal completion write). + const stuckJob = { ...job, status: "running", phase: "running", pid: deadPid }; + const jobFile = path.join(stateDir, "jobs", `${job.id}.json`); + writeJobFile(repo, job.id, { ...JSON.parse(fs.readFileSync(jobFile, "utf8")), status: "running", phase: "running", pid: deadPid }); + fs.writeFileSync(stateFile, `${JSON.stringify({ ...state, jobs: [stuckJob] }, null, 2)}\n`, "utf8"); + + return exitedForReal.then(() => { + const resume = run("node", [SCRIPT, "task", "--resume-last", "follow up"], { cwd: repo, env }); + + assert.equal(resume.status, 0, resume.stderr); + assert.doesNotMatch(resume.stderr, /is still running/i); + assert.equal(resume.stdout, "Resumed the prior run.\nFollow-up prompt accepted.\n"); + const fakeState = JSON.parse(fs.readFileSync(statePath, "utf8")); + assert.equal(fakeState.lastThreadResume.sandbox, null); + }); +}); + +test("task --resume-last is not permanently blocked by a job stuck 'queued' with a dead worker pid (upstream #425 gap)", () => { + // reconcileRunningJobs (state.mjs) used to only reconcile job.status === + // "running". A detached worker that dies BEFORE runTrackedJob's first + // write — the window between enqueueBackgroundTask recording + // status:"queued" with the spawned child's pid, and that worker calling + // runTrackedJob to flip it to "running" — was never reconciled: it stayed + // "queued" forever, with a pid that's already dead, invisible to the + // reconciliation check and permanently blocking --resume-last. This is the + // detached-worker-dies-early half of what upstream #425's reapDeadJobs + // (which reaps any queued/running job with a dead pid, not just running) + // covers and #392's throw-site fix does not. Fixed by also reconciling + // "queued" jobs, not just "running" ones. + 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 env = { ...buildEnv(binDir), CODEX_COMPANION_SESSION_ID: "sess-dead-queued-worker" }; + + const firstRun = run("node", [SCRIPT, "task", "initial task"], { cwd: repo, env }); + assert.equal(firstRun.status, 0, firstRun.stderr); + + const stateDir = resolveStateDir(repo); + const stateFile = path.join(stateDir, "state.json"); + const state = JSON.parse(fs.readFileSync(stateFile, "utf8")); + const job = state.jobs.find((entry) => entry.jobClass === "task"); + assert.ok(job, "expected the completed task job to be recorded"); + + const exitedWorker = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + const deadPid = exitedWorker.pid; + const exitedForReal = new Promise((resolve, reject) => { + exitedWorker.once("error", reject); + exitedWorker.once("exit", resolve); + }); + + const stuckJob = { ...job, status: "queued", phase: "queued", pid: deadPid }; + const jobFile = path.join(stateDir, "jobs", `${job.id}.json`); + writeJobFile(repo, job.id, { ...JSON.parse(fs.readFileSync(jobFile, "utf8")), status: "queued", phase: "queued", pid: deadPid }); + fs.writeFileSync(stateFile, `${JSON.stringify({ ...state, jobs: [stuckJob] }, null, 2)}\n`, "utf8"); + + return exitedForReal.then(() => { + const resume = run("node", [SCRIPT, "task", "--resume-last", "follow up"], { cwd: repo, env }); + + assert.equal(resume.status, 0, resume.stderr); + assert.doesNotMatch(resume.stderr, /is still running/i); + }); +}); + test("task-resume-candidate uses an explicit workspace cwd from an unrelated invocation directory", () => { const workspace = makeTempDir(); const invocationDir = makeTempDir(); diff --git a/tests/tracked-jobs.test.mjs b/tests/tracked-jobs.test.mjs new file mode 100644 index 00000000..b48ce26c --- /dev/null +++ b/tests/tracked-jobs.test.mjs @@ -0,0 +1,84 @@ +import fs from "node:fs"; +import path from "node:path"; +import test from "node:test"; +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { fileURLToPath, pathToFileURL } from "node:url"; + +import { makeTempDir, run } from "./helpers.mjs"; +import { readJobFile, resolveJobFile, upsertJob, writeJobFile } from "../plugins/codex/scripts/lib/state.mjs"; + +const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); +const TRACKED_JOBS_URL = pathToFileURL( + path.join(ROOT, "plugins", "codex", "scripts", "lib", "tracked-jobs.mjs") +).href; + +function seedJob(workspace, job) { + writeJobFile(workspace, job.id, job); + upsertJob(workspace, job); +} + +test("registerWorkerCrashGuard marks the job failed when the worker dies on an unhandled rejection", () => { + const workspace = makeTempDir(); + seedJob(workspace, { id: "job-crash", status: "running", phase: "delegating", pid: process.pid, logFile: null }); + + const workerFile = path.join(makeTempDir(), "crashing-worker.mjs"); + fs.writeFileSync( + workerFile, + [ + `import { registerWorkerCrashGuard } from ${JSON.stringify(TRACKED_JOBS_URL)};`, + "registerWorkerCrashGuard(process.argv[2], process.argv[3], null);", + 'Promise.reject(new Error("boom"));', + "setTimeout(() => {}, 5000);", + "" + ].join("\n"), + "utf8" + ); + + const result = run(process.execPath, [workerFile, workspace, "job-crash"]); + + assert.equal(result.status, 1); + const stored = readJobFile(resolveJobFile(workspace, "job-crash")); + assert.equal(stored.status, "failed"); + assert.match(stored.errorMessage, /unhandledRejection/); + assert.match(stored.errorMessage, /boom/); +}); + +test("registerWorkerCrashGuard does not rewrite a cancelled job when the worker is SIGTERMed", async () => { + const workspace = makeTempDir(); + // Simulate handleCancel having already written the terminal state before the + // worker processes the teardown SIGTERM it delivered. + seedJob(workspace, { id: "job-cancelled", status: "cancelled", phase: "cancelled", pid: null, errorMessage: "Cancelled by user." }); + + const workerFile = path.join(makeTempDir(), "long-worker.mjs"); + fs.writeFileSync( + workerFile, + [ + `import { registerWorkerCrashGuard } from ${JSON.stringify(TRACKED_JOBS_URL)};`, + "registerWorkerCrashGuard(process.argv[2], process.argv[3], null);", + 'process.stdout.write("ready\\n");', + "setInterval(() => {}, 1000);", + "" + ].join("\n"), + "utf8" + ); + + const child = spawn(process.execPath, [workerFile, workspace, "job-cancelled"], { stdio: ["ignore", "pipe", "ignore"] }); + await new Promise((resolve, reject) => { + child.stdout.on("data", (chunk) => { + if (chunk.toString().includes("ready")) { + resolve(); + } + }); + child.on("error", reject); + }); + + const exited = new Promise((resolve) => child.on("exit", (code, signal) => resolve({ code, signal }))); + child.kill("SIGTERM"); + const { signal } = await exited; + + assert.equal(signal, "SIGTERM"); + const stored = readJobFile(resolveJobFile(workspace, "job-cancelled")); + assert.equal(stored.status, "cancelled"); + assert.equal(stored.errorMessage, "Cancelled by user."); +});