Fix job records lost on concurrent background task launches

plugins/codex/scripts/codex-companion.mjs

@@ -27,11 +27,15 @@ import { collectReviewContext, ensureGitRepository, resolveReviewTarget } from "
 import { binaryAvailable, terminateProcessTree } from "./lib/process.mjs";
 import { loadPromptTemplate, interpolateTemplate } from "./lib/prompts.mjs";
 import {
+  claimJobRecord,
   generateJobId,
   getConfig,
+  isJobCancelled,
+  isSessionEnded,
   listJobs,
+  markJobCancelled,
+  readJobPid,
   setConfig,
-  upsertJob,
   writeJobFile
 } from "./lib/state.mjs";
 import {
@@ -304,13 +308,16 @@ function filterJobsForCurrentClaudeSession(jobs) {
 }
 
 function findLatestResumableTaskJob(jobs) {
+  // Only a COMPLETED or FAILED task is resumable. A CANCELLED task must NOT be -- the user
+  // explicitly stopped it, and (because the overlay keeps threadId while marking the job
+  // cancelled) treating "any non-active" as resumable would let a freshly cancelled task
+  // shadow an older genuinely-finished one and resume the wrong Codex thread.
   return (
     jobs.find(
       (job) =>
         job.jobClass === "task" &&
         job.threadId &&
-        job.status !== "queued" &&
-        job.status !== "running"
+        (job.status === "completed" || job.status === "failed")
     ) ?? null
   );
 }
@@ -682,20 +689,52 @@ function spawnDetachedTaskWorker(cwd, jobId) {
 }
 
 function enqueueBackgroundTask(cwd, job, request) {
+  // Refuse to enqueue into a session that has already ended: otherwise a task launched
+  // after session cleanup's one-shot scan would be neither tombstoned nor terminated.
+  if (isSessionEnded(job.workspaceRoot, job.sessionId)) {
+    throw new Error("This session has ended; cannot launch a new background task.");
+  }
+
   const { logFile } = createTrackedProgress(job);
   appendLogLine(logFile, "Queued for background execution.");
 
-  const child = spawnDetachedTaskWorker(cwd, job.id);
+  // CLAIM the queued record BEFORE spawning the worker, with an atomic create (never an
+  // overwrite). This both makes the record findable by a fast worker AND enforces
+  // single-writer ownership: if the id already exists (an astronomically unlikely
+  // duplicate from generateJobId, or a double launch), we refuse rather than spawn a
+  // second worker onto the same record.
   const queuedRecord = {
     ...job,
     status: "queued",
     phase: "queued",
-    pid: child.pid ?? null,
+    pid: null,
     logFile,
     request
   };
-  writeJobFile(job.workspaceRoot, job.id, queuedRecord);
-  upsertJob(job.workspaceRoot, queuedRecord);
+  if (!claimJobRecord(job.workspaceRoot, job.id, queuedRecord)) {
+    throw new Error(`Job id ${job.id} already exists; refusing to launch a duplicate worker.`);
+  }
+
+  try {
+    spawnDetachedTaskWorker(cwd, job.id);
+  } catch (error) {
+    // Spawn failed synchronously: mark the record failed so it does not linger as
+    // a pending job.
+    writeJobFile(job.workspaceRoot, job.id, {
+      ...queuedRecord,
+      status: "failed",
+      phase: "failed",
+      errorMessage: error instanceof Error ? error.message : String(error)
+    });
+    throw error;
+  }
+
+  // The worker owns the record from here: it publishes `running` with its own pid as
+  // its first act (see runTrackedJob). The parent deliberately does NOT patch the pid
+  // in, so the per-job record has a SINGLE writer -- no concurrent read-modify-write,
+  // hence no lost update. A cancel/session-end arriving before the worker publishes
+  // `running` is honored by the immutable cancel/session-ended markers, which the
+  // worker consults at startup and again right after it publishes its pid.
 
   return {
     payload: {
@@ -851,6 +890,24 @@ async function handleTaskWorker(argv) {
     throw new Error(`No stored job found for ${options["job-id"]}.`);
   }
 
+  // Honor a cancellation or a session end that landed during our startup window
+  // (before we could publish a pid): the immutable cancel marker and the session-ended
+  // marker are the durable, race-free signals, so consult them (not just the record,
+  // which a stale reader could have missed). We are the single writer, so record a
+  // cancelled terminal state before returning (keeps the raw record truthful; the marker
+  // overlay would show cancelled regardless). runTrackedJob re-checks the markers again
+  // after it publishes our pid, closing the rest of the window.
+  if (storedJob.status === "cancelled" || isJobCancelled(workspaceRoot, options["job-id"]) || isSessionEnded(workspaceRoot, storedJob.sessionId)) {
+    writeJobFile(workspaceRoot, options["job-id"], {
+      ...storedJob,
+      status: "cancelled",
+      phase: "cancelled",
+      pid: null,
+      completedAt: nowIso()
+    });
+    return;
+  }
+
   const request = storedJob.request;
   if (!request || typeof request !== "object") {
     throw new Error(`Stored job ${options["job-id"]} is missing its task request payload.`);
@@ -915,8 +972,12 @@ function handleResult(argv) {
 
   const cwd = resolveCommandCwd(options);
   const reference = positionals[0] ?? "";
-  const { workspaceRoot, job } = resolveResultJob(cwd, reference);
-  const storedJob = readStoredJob(workspaceRoot, job.id);
+  const { job } = resolveResultJob(cwd, reference);
+  // Render from the OVERLAID record (resolveResultJob returns it via listJobs), NOT a raw
+  // readStoredJob: a cancel marker strips result/rendered, so a job that completed and was
+  // then cancel-marked must not leak its output here. The overlay is the single source of
+  // truth for what a cancelled job exposes.
+  const storedJob = job;
   const payload = {
     job,
     storedJob
@@ -983,7 +1044,31 @@ async function handleCancel(argv) {
     );
   }
 
-  terminateProcessTree(job.pid ?? Number.NaN);
+  // Publish the immutable cancel marker BEFORE terminating, so a worker that publishes
+  // its pid concurrently is guaranteed to see the marker on its post-pid re-check and
+  // self-abort (the mirror of runTrackedJob's handshake). The marker is the RACE-SAFE
+  // authority: readers overlay it, so even if a worker writes `running`/`completed`
+  // after this, the job still reads as cancelled and can never be resurrected.
+  //
+  // Ordering is load-bearing: create the marker FIRST, THEN read the pid to kill. The
+  // worker does the mirror (publish pid, then read marker), so at least one side always
+  // observes the other -- we kill the worker, or it self-aborts. Reading a stale pid
+  // from before the marker would let a worker slip through both checks.
+  markJobCancelled(workspaceRoot, job.id, "Cancelled by user.");
+  // Read the pid FRESH, after the marker. No fallback to the pre-marker snapshot pid: a
+  // fresh null means the worker either never published or already exited, and killing a
+  // stale pid risks signalling an unrelated process that reused it. If it's null, the
+  // worker will honor the marker on its own post-pid re-check.
+  const killPid = readJobPid(workspaceRoot, job.id);
+  // Best-effort: a termination failure must NOT abort the command and leave the job
+  // marked-cancelled-but-live with no way to retry. resolveCancelableJob works off the RAW
+  // record, so re-running cancel still finds a marked job whose worker is raw-queued/running
+  // and re-attempts the kill.
+  try {
+    terminateProcessTree(killPid ?? Number.NaN);
+  } catch (error) {
+    appendLogLine(job.logFile, `Termination failed: ${error instanceof Error ? error.message : String(error)}`);
+  }
   appendLogLine(job.logFile, "Cancelled by user.");
 
   const completedAt = nowIso();
@@ -996,19 +1081,10 @@ async function handleCancel(argv) {
     errorMessage: "Cancelled by user."
   };
 
-  writeJobFile(workspaceRoot, job.id, {
-    ...existing,
-    ...nextJob,
-    cancelledAt: completedAt
-  });
-  upsertJob(workspaceRoot, {
-    id: job.id,
-    status: "cancelled",
-    phase: "cancelled",
-    pid: null,
-    errorMessage: "Cancelled by user.",
-    completedAt
-  });
+  // We deliberately do NOT write the record here. The marker is the authority: every
+  // read overlays it to cancelled, and the worker (the record's single writer) records
+  // the terminal cancelled state when it observes the marker. A cancel-command record
+  // write would reintroduce a second writer racing the worker.
 
   const payload = {
     jobId: job.id,

plugins/codex/scripts/lib/job-control.mjs

@@ -1,13 +1,24 @@
 import fs from "node:fs";
 
 import { getSessionRuntimeStatus } from "./codex.mjs";
-import { getConfig, listJobs, readJobFile, resolveJobFile } from "./state.mjs";
+import { getConfig, isJobCancelled, isSessionEnded, listJobs, readAllJobsRaw, readJobFile, resolveJobFile } from "./state.mjs";
 import { SESSION_ID_ENV } from "./tracked-jobs.mjs";
 import { resolveWorkspaceRoot } from "./workspace.mjs";
 
 export const DEFAULT_MAX_STATUS_JOBS = 8;
 export const DEFAULT_MAX_PROGRESS_LINES = 4;
 
+// True while pid exists and could still execute code (EPERM = alive but not ours).
+function isPidAlive(pid) {
+  if (!Number.isInteger(pid) || pid <= 0) return false;
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch (err) {
+    return err.code === "EPERM";
+  }
+}
+
 export function sortJobsNewestFirst(jobs) {
   return [...jobs].sort((left, right) => String(right.updatedAt ?? "").localeCompare(String(left.updatedAt ?? "")));
 }
@@ -181,11 +192,13 @@ export function enrichJob(job, options = {}) {
 }
 
 export function readStoredJob(workspaceRoot, jobId) {
-  const jobFile = resolveJobFile(workspaceRoot, jobId);
-  if (!fs.existsSync(jobFile)) {
+  // Guarded read: the per-job file can be pruned or session-cleaned between the
+  // existence check and the read, so treat any read/parse failure as "absent".
+  try {
+    return readJobFile(resolveJobFile(workspaceRoot, jobId));
+  } catch {
     return null;
   }
-  return readJobFile(jobFile);
 }
 
 function matchJobReference(jobs, reference, predicate = () => true) {
@@ -280,8 +293,24 @@ export function resolveResultJob(cwd, reference) {
 
 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");
+  // Work off RAW records (so a failed-kill retry can still see the true running state),
+  // but classify cancellability precisely so a job that was already cancelled cannot
+  // shadow a genuinely active one:
+  //  - unmarked queued/running  -> cancellable (normal case);
+  //  - MARKED + running + LIVE pid -> cancellable (a previous kill threw; retry it);
+  //  - MARKED + running with a dead pid (killed, died before writing a terminal record),
+  //    or MARKED + queued/pid-less -> NOT cancellable (the marker is already authoritative
+  //    and the worker must honor it; re-selecting it would shadow other active jobs).
+  const jobs = sortJobsNewestFirst(readAllJobsRaw(workspaceRoot));
+  const activeJobs = jobs.filter((job) => {
+    if (job.status !== "queued" && job.status !== "running") return false;
+    // A job is logically cancelled by EITHER its own cancel marker OR its session's
+    // ended marker (both are authoritative in the read overlay). Only a marked job that
+    // is still running with a live pid stays targetable (to retry a kill that threw).
+    const marked = isJobCancelled(workspaceRoot, job.id) || isSessionEnded(workspaceRoot, job.sessionId);
+    if (!marked) return true;
+    return job.status === "running" && isPidAlive(job.pid);
+  });
 
   if (reference) {
     const selected = matchJobReference(activeJobs, reference);