diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs index 83df468a..f8fd1949 100644 --- a/plugins/codex/scripts/codex-companion.mjs +++ b/plugins/codex/scripts/codex-companion.mjs @@ -49,6 +49,8 @@ import { createJobRecord, createProgressReporter, nowIso, + reapDeadJobs, + registerWorkerCrashGuard, runTrackedJob, SESSION_ID_ENV } from "./lib/tracked-jobs.mjs"; @@ -336,7 +338,7 @@ async function waitForSingleJobSnapshot(cwd, reference, options = {}) { async function resolveLatestTrackedTaskThread(cwd, options = {}) { const workspaceRoot = resolveWorkspaceRoot(cwd); const sessionId = getCurrentClaudeSessionId(); - const jobs = sortJobsNewestFirst(listJobs(workspaceRoot)).filter((job) => job.id !== options.excludeJobId); + const jobs = sortJobsNewestFirst(reapDeadJobs(workspaceRoot, listJobs(workspaceRoot))).filter((job) => job.id !== options.excludeJobId); const visibleJobs = filterJobsForCurrentClaudeSession(jobs); const activeTask = visibleJobs.find((job) => job.jobClass === "task" && (job.status === "queued" || job.status === "running")); if (activeTask) { @@ -865,6 +867,7 @@ async function handleTaskWorker(argv) { logFile: storedJob.logFile ?? null } ); + registerWorkerCrashGuard(workspaceRoot, options["job-id"], logFile); await runTrackedJob( { ...storedJob, @@ -934,7 +937,7 @@ function handleTaskResumeCandidate(argv) { const cwd = resolveCommandCwd(options); const workspaceRoot = resolveCommandWorkspace(options); const sessionId = getCurrentClaudeSessionId(); - const jobs = filterJobsForCurrentClaudeSession(sortJobsNewestFirst(listJobs(workspaceRoot))); + const jobs = filterJobsForCurrentClaudeSession(sortJobsNewestFirst(reapDeadJobs(workspaceRoot, listJobs(workspaceRoot)))); const candidate = findLatestResumableTaskJob(jobs); const payload = { diff --git a/plugins/codex/scripts/lib/job-control.mjs b/plugins/codex/scripts/lib/job-control.mjs index ad152c15..68ab7db1 100644 --- a/plugins/codex/scripts/lib/job-control.mjs +++ b/plugins/codex/scripts/lib/job-control.mjs @@ -2,7 +2,7 @@ import fs from "node:fs"; import { getSessionRuntimeStatus } from "./codex.mjs"; import { getConfig, listJobs, readJobFile, resolveJobFile } from "./state.mjs"; -import { SESSION_ID_ENV } from "./tracked-jobs.mjs"; +import { reapDeadJobs, SESSION_ID_ENV } from "./tracked-jobs.mjs"; import { resolveWorkspaceRoot } from "./workspace.mjs"; export const DEFAULT_MAX_STATUS_JOBS = 8; @@ -213,7 +213,7 @@ function matchJobReference(jobs, reference, predicate = () => true) { export function buildStatusSnapshot(cwd, options = {}) { const workspaceRoot = resolveWorkspaceRoot(cwd); const config = getConfig(workspaceRoot); - const jobs = sortJobsNewestFirst(filterJobsForCurrentSession(listJobs(workspaceRoot), options)); + const jobs = sortJobsNewestFirst(filterJobsForCurrentSession(reapDeadJobs(workspaceRoot, listJobs(workspaceRoot)), options)); const maxJobs = options.maxJobs ?? DEFAULT_MAX_STATUS_JOBS; const maxProgressLines = options.maxProgressLines ?? DEFAULT_MAX_PROGRESS_LINES; @@ -241,7 +241,7 @@ export function buildStatusSnapshot(cwd, options = {}) { export function buildSingleJobSnapshot(cwd, reference, options = {}) { const workspaceRoot = resolveWorkspaceRoot(cwd); - const jobs = sortJobsNewestFirst(listJobs(workspaceRoot)); + const jobs = sortJobsNewestFirst(reapDeadJobs(workspaceRoot, listJobs(workspaceRoot))); const selected = matchJobReference(jobs, reference); if (!selected) { throw new Error(`No job found for "${reference}". Run /codex:status to inspect known jobs.`); @@ -255,7 +255,7 @@ export function buildSingleJobSnapshot(cwd, reference, options = {}) { export function resolveResultJob(cwd, reference) { const workspaceRoot = resolveWorkspaceRoot(cwd); - const jobs = sortJobsNewestFirst(reference ? listJobs(workspaceRoot) : filterJobsForCurrentSession(listJobs(workspaceRoot))); + const jobs = sortJobsNewestFirst(reference ? reapDeadJobs(workspaceRoot, listJobs(workspaceRoot)) : filterJobsForCurrentSession(reapDeadJobs(workspaceRoot, listJobs(workspaceRoot)))); const selected = matchJobReference( jobs, reference, @@ -280,7 +280,7 @@ export function resolveResultJob(cwd, reference) { export function resolveCancelableJob(cwd, reference, options = {}) { const workspaceRoot = resolveWorkspaceRoot(cwd); - const jobs = sortJobsNewestFirst(listJobs(workspaceRoot)); + const jobs = sortJobsNewestFirst(reapDeadJobs(workspaceRoot, listJobs(workspaceRoot))); const activeJobs = jobs.filter((job) => job.status === "queued" || job.status === "running"); if (reference) { diff --git a/plugins/codex/scripts/lib/tracked-jobs.mjs b/plugins/codex/scripts/lib/tracked-jobs.mjs index 90286901..41cdaf78 100644 --- a/plugins/codex/scripts/lib/tracked-jobs.mjs +++ b/plugins/codex/scripts/lib/tracked-jobs.mjs @@ -202,3 +202,93 @@ export async function runTrackedJob(job, runner, options = {}) { throw error; } } + +function isPidAlive(pid) { + if (!Number.isInteger(pid) || pid <= 0) { + return null; + } + try { + process.kill(pid, 0); + return true; + } catch (error) { + // ESRCH: no such process. EPERM: exists but not ours — treat as alive. + return error?.code === "ESRCH" ? false : true; + } +} + +function markJobDead(workspaceRoot, jobSummary, errorMessage) { + const jobFile = resolveJobFile(workspaceRoot, jobSummary.id); + const stored = fs.existsSync(jobFile) ? readJobFile(jobFile) : null; + const base = stored ?? jobSummary; + if (base.status !== "running" && base.status !== "queued") { + // The job finished between the caller's read and now — keep the real result. + return base; + } + const completedAt = nowIso(); + const record = { + ...base, + status: "failed", + phase: "failed", + errorMessage, + pid: null, + completedAt, + // Keep updatedAt current so the reaped job sorts newest-first in the same + // read that recorded it — otherwise a stale updatedAt can page it out of + // the first /codex:status report. + updatedAt: completedAt + }; + writeJobFile(workspaceRoot, jobSummary.id, record); + upsertJob(workspaceRoot, { + id: jobSummary.id, + status: "failed", + phase: "failed", + pid: null, + errorMessage, + completedAt + }); + appendLogLine(base.logFile ?? null, `Marked failed: ${errorMessage}`); + return record; +} + +// A worker that dies without throwing (SIGKILL, OOM, native crash) never +// reaches runTrackedJob's catch, so its job stays "running" forever. Rewrite +// any active job whose recorded pid is no longer alive as failed. +export function reapDeadJobs(workspaceRoot, jobs) { + return jobs.map((job) => { + if (job.status !== "running" && job.status !== "queued") { + return job; + } + if (isPidAlive(job.pid) === false) { + return markJobDead( + workspaceRoot, + job, + `worker process (pid ${job.pid}) died without recording a result — ` + + `likely a crash; resume with task --resume-last` + ); + } + return job; + }); +} + +// 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 the reader-side reapDeadJobs must cover +// it regardless, and /codex:cancel delivers SIGTERM as its teardown signal after +// writing the job "cancelled" — catching it here would race that terminal state +// back to "failed". reapDeadJobs handles every signal death and never rewrites a +// job that already reached a terminal status. +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}`); + markJobDead(workspaceRoot, { id: jobId, status: "running", 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/plugins/codex/scripts/stop-review-gate-hook.mjs b/plugins/codex/scripts/stop-review-gate-hook.mjs index 2346bdcf..55756b0f 100644 --- a/plugins/codex/scripts/stop-review-gate-hook.mjs +++ b/plugins/codex/scripts/stop-review-gate-hook.mjs @@ -10,7 +10,7 @@ import { getCodexAvailability } from "./lib/codex.mjs"; import { loadPromptTemplate, interpolateTemplate } from "./lib/prompts.mjs"; import { getConfig, listJobs } from "./lib/state.mjs"; import { sortJobsNewestFirst } from "./lib/job-control.mjs"; -import { SESSION_ID_ENV } from "./lib/tracked-jobs.mjs"; +import { reapDeadJobs, SESSION_ID_ENV } from "./lib/tracked-jobs.mjs"; import { resolveWorkspaceRoot } from "./lib/workspace.mjs"; const STOP_REVIEW_TIMEOUT_MS = 15 * 60 * 1000; @@ -145,7 +145,7 @@ function main() { const workspaceRoot = resolveWorkspaceRoot(cwd); const config = getConfig(workspaceRoot); - const jobs = sortJobsNewestFirst(filterJobsForCurrentSession(listJobs(workspaceRoot), input)); + const jobs = sortJobsNewestFirst(filterJobsForCurrentSession(reapDeadJobs(workspaceRoot, listJobs(workspaceRoot)), input)); const runningJob = jobs.find((job) => job.status === "queued" || job.status === "running"); const runningTaskNote = runningJob ? `Codex task ${runningJob.id} is still running. Check /codex:status and use /codex:cancel ${runningJob.id} if you want to stop it before ending the session.` diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 8f276835..6a03725f 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -570,6 +570,48 @@ test("task-resume-candidate returns the latest rescue thread from the current se assert.equal(payload.candidate.threadId, "thr_current"); }); +test("task-resume-candidate reaps a crashed running task so it becomes resumable", () => { + const workspace = makeTempDir(); + const stateDir = resolveStateDir(workspace); + const jobsDir = path.join(stateDir, "jobs"); + fs.mkdirSync(jobsDir, { recursive: true }); + + // A pid that has already exited: process.kill(pid, 0) will throw ESRCH. + const deadPid = run(process.execPath, ["-e", ""]).pid; + const crashedJob = { + id: "task-crashed", + status: "running", + phase: "delegating", + title: "Codex Task", + jobClass: "task", + sessionId: "sess-current", + threadId: "thr_crashed", + summary: "Investigate the crash", + pid: deadPid, + updatedAt: "2026-03-24T20:00:00.000Z" + }; + fs.writeFileSync(path.join(jobsDir, "task-crashed.json"), `${JSON.stringify(crashedJob, null, 2)}\n`, "utf8"); + fs.writeFileSync( + path.join(stateDir, "state.json"), + `${JSON.stringify({ version: 1, config: { stopReviewGate: false }, jobs: [crashedJob] }, null, 2)}\n`, + "utf8" + ); + + const result = run("node", [SCRIPT, "task-resume-candidate", "--json"], { + cwd: workspace, + env: { ...process.env, CODEX_COMPANION_SESSION_ID: "sess-current" } + }); + + assert.equal(result.status, 0, result.stderr); + const payload = JSON.parse(result.stdout); + // Without the reaper the job would still read as "running" and be skipped, + // leaving the resume probe with no candidate. + assert.equal(payload.available, true); + assert.equal(payload.candidate.id, "task-crashed"); + assert.equal(payload.candidate.status, "failed"); + assert.equal(payload.candidate.threadId, "thr_crashed"); +}); + test("task --resume-last does not resume a task from another Claude session", () => { const repo = makeTempDir(); const binDir = makeTempDir(); diff --git a/tests/tracked-jobs.test.mjs b/tests/tracked-jobs.test.mjs new file mode 100644 index 00000000..2d8f0c76 --- /dev/null +++ b/tests/tracked-jobs.test.mjs @@ -0,0 +1,150 @@ +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 { reapDeadJobs } from "../plugins/codex/scripts/lib/tracked-jobs.mjs"; +import { listJobs, 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); +} + +function spawnDeadPid() { + const result = run(process.execPath, ["-e", ""]); + assert.equal(result.status, 0); + return result.pid; +} + +test("reapDeadJobs marks a running job with a dead pid as failed", () => { + const workspace = makeTempDir(); + seedJob(workspace, { id: "job-dead", status: "running", phase: "delegating", pid: spawnDeadPid(), logFile: null }); + + const reaped = reapDeadJobs(workspace, listJobs(workspace)); + + assert.equal(reaped.length, 1); + assert.equal(reaped[0].status, "failed"); + assert.equal(reaped[0].pid, null); + assert.match(reaped[0].errorMessage, /died without recording a result/); + + const stored = readJobFile(resolveJobFile(workspace, "job-dead")); + assert.equal(stored.status, "failed"); + assert.equal(stored.pid, null); + assert.equal(listJobs(workspace).find((job) => job.id === "job-dead").status, "failed"); +}); + +test("reapDeadJobs refreshes updatedAt so the reaped job sorts newest-first", () => { + const workspace = makeTempDir(); + const stale = "2000-01-01T00:00:00.000Z"; + seedJob(workspace, { id: "job-stale", status: "running", phase: "delegating", pid: spawnDeadPid(), updatedAt: stale, logFile: null }); + + const reaped = reapDeadJobs(workspace, listJobs(workspace)); + + assert.notEqual(reaped[0].updatedAt, stale); + assert.equal(reaped[0].updatedAt, reaped[0].completedAt); + assert.equal(readJobFile(resolveJobFile(workspace, "job-stale")).updatedAt, reaped[0].updatedAt); +}); + +test("reapDeadJobs leaves a running job with a live pid untouched", () => { + const workspace = makeTempDir(); + seedJob(workspace, { id: "job-live", status: "running", phase: "delegating", pid: process.pid, logFile: null }); + + const reaped = reapDeadJobs(workspace, listJobs(workspace)); + + assert.equal(reaped[0].status, "running"); + assert.equal(readJobFile(resolveJobFile(workspace, "job-live")).status, "running"); +}); + +test("reapDeadJobs leaves jobs without a recorded pid untouched", () => { + const workspace = makeTempDir(); + seedJob(workspace, { id: "job-no-pid", status: "queued", phase: "queued", pid: null, logFile: null }); + + const reaped = reapDeadJobs(workspace, listJobs(workspace)); + + assert.equal(reaped[0].status, "queued"); +}); + +test("reapDeadJobs keeps the stored result when the job finished between read and probe", () => { + const workspace = makeTempDir(); + const deadPid = spawnDeadPid(); + writeJobFile(workspace, "job-raced", { id: "job-raced", status: "completed", phase: "completed", result: "done" }); + upsertJob(workspace, { id: "job-raced", status: "running", phase: "delegating", pid: deadPid, logFile: null }); + + const reaped = reapDeadJobs(workspace, [{ id: "job-raced", status: "running", pid: deadPid, logFile: null }]); + + assert.equal(reaped[0].status, "completed"); + assert.equal(reaped[0].result, "done"); + assert.equal(readJobFile(resolveJobFile(workspace, "job-raced")).status, "completed"); +}); + +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: null, 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."); +});