fix(task): publish jobs before worker spawn

plugins/codex/scripts/codex-companion.mjs

@@ -22,6 +22,7 @@ import {
     runAppServerTurn
   } from "./lib/codex.mjs";
 import { resolveClaudeSessionPath } from "./lib/claude-session-transfer.mjs";
+import { dispatchBackgroundJob } from "./lib/background-dispatch.mjs";
 import { readStdinIfPiped } from "./lib/fs.mjs";
 import { collectReviewContext, ensureGitRepository, resolveReviewTarget } from "./lib/git.mjs";
 import { binaryAvailable, terminateProcessTree } from "./lib/process.mjs";
@@ -677,36 +678,21 @@ function spawnDetachedTaskWorker(cwd, jobId) {
     stdio: "ignore",
     windowsHide: true
   });
-  child.unref();
   return child;
 }
 
-function enqueueBackgroundTask(cwd, job, request) {
+async function enqueueBackgroundTask(cwd, job, request) {
   const { logFile } = createTrackedProgress(job);
   appendLogLine(logFile, "Queued for background execution.");
 
-  const child = spawnDetachedTaskWorker(cwd, job.id);
-  const queuedRecord = {
-    ...job,
-    status: "queued",
-    phase: "queued",
-    pid: child.pid ?? null,
+  const payload = await dispatchBackgroundJob({
+    job,
+    request,
     logFile,
-    request
-  };
-  writeJobFile(job.workspaceRoot, job.id, queuedRecord);
-  upsertJob(job.workspaceRoot, queuedRecord);
+    spawnWorker: () => spawnDetachedTaskWorker(cwd, job.id)
+  });
 
-  return {
-    payload: {
-      jobId: job.id,
-      status: "queued",
-      title: job.title,
-      summary: job.summary,
-      logFile
-    },
-    logFile
-  };
+  return { payload, logFile };
 }
 
 async function handleReviewCommand(argv, config) {
@@ -799,7 +785,7 @@ async function handleTask(argv) {
       resumeLast,
       jobId: job.id
     });
-    const { payload } = enqueueBackgroundTask(cwd, job, request);
+    const { payload } = await enqueueBackgroundTask(cwd, job, request);
     outputCommandResult(payload, renderQueuedTaskLaunch(payload), options.json);
     return;
   }

plugins/codex/scripts/lib/background-dispatch.mjs

@@ -0,0 +1,91 @@
+import { readJobFile, resolveJobFile, upsertJob, writeJobFile } from "./state.mjs";
+import { nowIso } from "./tracked-jobs.mjs";
+
+function errorMessage(error) {
+  return error instanceof Error ? error.message : String(error);
+}
+
+function persistSpawnFailure(workspaceRoot, jobId, error) {
+  const message = errorMessage(error);
+  const completedAt = nowIso();
+  let existing = { id: jobId };
+  try {
+    existing = readJobFile(resolveJobFile(workspaceRoot, jobId));
+  } catch {
+    // Preserve the launch error even if the prepublished record cannot be reread.
+  }
+  const failedRecord = {
+    ...existing,
+    id: jobId,
+    status: "failed",
+    phase: "failed",
+    pid: null,
+    errorMessage: message,
+    completedAt
+  };
+  writeJobFile(workspaceRoot, jobId, failedRecord);
+  upsertJob(workspaceRoot, {
+    id: jobId,
+    status: "failed",
+    phase: "failed",
+    pid: null,
+    errorMessage: message,
+    completedAt
+  });
+}
+
+function awaitSpawn(child) {
+  if (!child || typeof child.once !== "function") {
+    return Promise.reject(new Error("Detached task worker did not return a child process."));
+  }
+
+  return new Promise((resolve, reject) => {
+    const onSpawn = () => {
+      child.off?.("error", onError);
+      resolve(child);
+    };
+    const onError = (error) => {
+      child.off?.("spawn", onSpawn);
+      reject(error);
+    };
+    child.once("spawn", onSpawn);
+    child.once("error", onError);
+  });
+}
+
+export async function dispatchBackgroundJob({ job, request, logFile, spawnWorker }) {
+  const queuedRecord = {
+    ...job,
+    status: "queued",
+    phase: "queued",
+    pid: null,
+    logFile,
+    request
+  };
+
+  // The worker's first operation is to read this record, so publish it before
+  // process creation. The worker owns the later running transition and PID.
+  writeJobFile(job.workspaceRoot, job.id, queuedRecord);
+  upsertJob(job.workspaceRoot, queuedRecord);
+
+  let child;
+  try {
+    child = spawnWorker();
+    await awaitSpawn(child);
+    if (!Number.isFinite(child.pid)) {
+      throw new Error("Detached task worker spawned without a process ID.");
+    }
+    child.unref?.();
+  } catch (error) {
+    persistSpawnFailure(job.workspaceRoot, job.id, error);
+    throw error;
+  }
+
+  return {
+    jobId: job.id,
+    status: "queued",
+    title: job.title,
+    summary: job.summary,
+    logFile
+  };
+}

plugins/codex/scripts/lib/tracked-jobs.mjs

@@ -140,6 +140,8 @@ function readStoredJobOrNull(workspaceRoot, jobId) {
 }
 
 export async function runTrackedJob(job, runner, options = {}) {
+  const writeJobFileImpl = options.writeJobFileImpl ?? writeJobFile;
+  const upsertJobImpl = options.upsertJobImpl ?? upsertJob;
   const runningRecord = {
     ...job,
     status: "running",
@@ -148,14 +150,13 @@ export async function runTrackedJob(job, runner, options = {}) {
     pid: process.pid,
     logFile: options.logFile ?? job.logFile ?? null
   };
-  writeJobFile(job.workspaceRoot, job.id, runningRecord);
-  upsertJob(job.workspaceRoot, runningRecord);
-
   try {
+    writeJobFileImpl(job.workspaceRoot, job.id, runningRecord);
+    upsertJobImpl(job.workspaceRoot, runningRecord);
     const execution = await runner();
     const completionStatus = execution.exitStatus === 0 ? "completed" : "failed";
     const completedAt = nowIso();
-    writeJobFile(job.workspaceRoot, job.id, {
+    writeJobFileImpl(job.workspaceRoot, job.id, {
       ...runningRecord,
       status: completionStatus,
       threadId: execution.threadId ?? null,
@@ -166,7 +167,7 @@ export async function runTrackedJob(job, runner, options = {}) {
       result: execution.payload,
       rendered: execution.rendered
     });
-    upsertJob(job.workspaceRoot, {
+    upsertJobImpl(job.workspaceRoot, {
       id: job.id,
       status: completionStatus,
       threadId: execution.threadId ?? null,
@@ -182,7 +183,7 @@ export async function runTrackedJob(job, runner, options = {}) {
     const errorMessage = error instanceof Error ? error.message : String(error);
     const existing = readStoredJobOrNull(job.workspaceRoot, job.id) ?? runningRecord;
     const completedAt = nowIso();
-    writeJobFile(job.workspaceRoot, job.id, {
+    writeJobFileImpl(job.workspaceRoot, job.id, {
       ...existing,
       status: "failed",
       phase: "failed",
@@ -191,7 +192,7 @@ export async function runTrackedJob(job, runner, options = {}) {
       completedAt,
       logFile: options.logFile ?? job.logFile ?? existing.logFile ?? null
     });
-    upsertJob(job.workspaceRoot, {
+    upsertJobImpl(job.workspaceRoot, {
       id: job.id,
       status: "failed",
       phase: "failed",

tests/background-dispatch.test.mjs

@@ -0,0 +1,159 @@
+import { EventEmitter } from "node:events";
+import fs from "node:fs";
+import path from "node:path";
+import test from "node:test";
+import assert from "node:assert/strict";
+
+import { dispatchBackgroundJob } from "../plugins/codex/scripts/lib/background-dispatch.mjs";
+import { makeTempDir } from "./helpers.mjs";
+import {
+  listJobs,
+  readJobFile,
+  resolveJobFile,
+  upsertJob,
+  writeJobFile
+} from "../plugins/codex/scripts/lib/state.mjs";
+
+function createJob(workspaceRoot) {
+  return {
+    id: "task-ordering-test",
+    kind: "task",
+    kindLabel: "rescue",
+    title: "Codex Task",
+    workspaceRoot,
+    jobClass: "task",
+    summary: "ordering test",
+    write: false,
+    createdAt: "2026-09-02T00:00:00.000Z"
+  };
+}
+
+function fakeChild(schedule, pid = 4242) {
+  const child = new EventEmitter();
+  child.pid = pid;
+  child.unrefCalled = false;
+  child.unref = () => {
+    child.unrefCalled = true;
+  };
+  process.nextTick(() => schedule(child));
+  return child;
+}
+
+test("dispatchBackgroundJob publishes the complete request before spawning", async () => {
+  const workspace = makeTempDir();
+  const job = createJob(workspace);
+  const request = { cwd: workspace, prompt: "inspect the race", write: false };
+  const logFile = path.join(workspace, "task.log");
+  fs.writeFileSync(logFile, "queued\n", "utf8");
+  let observedAtSpawn = null;
+  let child = null;
+
+  const payload = await dispatchBackgroundJob({
+    job,
+    request,
+    logFile,
+    spawnWorker() {
+      observedAtSpawn = readJobFile(resolveJobFile(workspace, job.id));
+      child = fakeChild((value) => value.emit("spawn"));
+      return child;
+    }
+  });
+
+  assert.equal(observedAtSpawn.status, "queued");
+  assert.equal(observedAtSpawn.pid, null);
+  assert.deepEqual(observedAtSpawn.request, request);
+  assert.equal(observedAtSpawn.logFile, logFile);
+  assert.equal(payload.jobId, job.id);
+  assert.equal(payload.status, "queued");
+  assert.equal(child.unrefCalled, true);
+  assert.equal(listJobs(workspace)[0].status, "queued");
+});
+
+test("dispatchBackgroundJob never regresses a worker-authored running record", async () => {
+  const workspace = makeTempDir();
+  const job = createJob(workspace);
+  const request = { cwd: workspace, prompt: "run immediately", write: false };
+  const logFile = path.join(workspace, "task.log");
+  fs.writeFileSync(logFile, "queued\n", "utf8");
+
+  await dispatchBackgroundJob({
+    job,
+    request,
+    logFile,
+    spawnWorker() {
+      const queued = readJobFile(resolveJobFile(workspace, job.id));
+      const running = {
+        ...queued,
+        status: "running",
+        phase: "starting",
+        pid: 4242,
+        startedAt: "2026-09-02T00:00:01.000Z"
+      };
+      writeJobFile(workspace, job.id, running);
+      upsertJob(workspace, running);
+      return fakeChild((value) => value.emit("spawn"));
+    }
+  });
+
+  const stored = readJobFile(resolveJobFile(workspace, job.id));
+  assert.equal(stored.status, "running");
+  assert.equal(stored.pid, 4242);
+  assert.equal(listJobs(workspace)[0].status, "running");
+  assert.equal(listJobs(workspace)[0].pid, 4242);
+});
+
+test("dispatchBackgroundJob persists a terminal failure when spawn fails", async () => {
+  const workspace = makeTempDir();
+  const job = createJob(workspace);
+  const request = { cwd: workspace, prompt: "cannot spawn", write: false };
+  const logFile = path.join(workspace, "task.log");
+  fs.writeFileSync(logFile, "queued\n", "utf8");
+
+  await assert.rejects(
+    dispatchBackgroundJob({
+      job,
+      request,
+      logFile,
+      spawnWorker() {
+        return fakeChild((value) => value.emit("error", new Error("spawn denied")));
+      }
+    }),
+    /spawn denied/
+  );
+
+  const stored = readJobFile(resolveJobFile(workspace, job.id));
+  const indexed = listJobs(workspace)[0];
+  for (const record of [stored, indexed]) {
+    assert.equal(record.status, "failed");
+    assert.equal(record.phase, "failed");
+    assert.equal(record.pid, null);
+    assert.equal(record.errorMessage, "spawn denied");
+    assert.ok(record.completedAt);
+  }
+});
+
+
+test("dispatchBackgroundJob rejects a spawn acknowledgement without a process ID", async () => {
+  const workspace = makeTempDir();
+  const job = createJob(workspace);
+  const request = { cwd: workspace, prompt: "missing pid", write: false };
+  const logFile = path.join(workspace, "task.log");
+  fs.writeFileSync(logFile, "queued\n", "utf8");
+
+  await assert.rejects(
+    dispatchBackgroundJob({
+      job,
+      request,
+      logFile,
+      spawnWorker() {
+        return fakeChild((value) => value.emit("spawn"), Number.NaN);
+      }
+    }),
+    /spawned without a process ID/
+  );
+
+  const stored = readJobFile(resolveJobFile(workspace, job.id));
+  assert.equal(stored.status, "failed");
+  assert.equal(stored.pid, null);
+  assert.match(stored.errorMessage, /without a process ID/);
+});