fix: reap detached workers that die before reaching 'running', and mark in-process crashes failed immediately

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,

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,

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;
     }
 

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"));
+}

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 <id> 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();

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.");
+});