diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs index 83df468ad..47bee18a5 100644 --- a/plugins/codex/scripts/codex-companion.mjs +++ b/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; } diff --git a/plugins/codex/scripts/lib/background-dispatch.mjs b/plugins/codex/scripts/lib/background-dispatch.mjs new file mode 100644 index 000000000..de39cae5a --- /dev/null +++ b/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 + }; +} diff --git a/plugins/codex/scripts/lib/tracked-jobs.mjs b/plugins/codex/scripts/lib/tracked-jobs.mjs index 902869012..8207ae8d6 100644 --- a/plugins/codex/scripts/lib/tracked-jobs.mjs +++ b/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", diff --git a/tests/background-dispatch.test.mjs b/tests/background-dispatch.test.mjs new file mode 100644 index 000000000..b4335a19a --- /dev/null +++ b/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/); +}); diff --git a/tests/tracked-jobs.test.mjs b/tests/tracked-jobs.test.mjs new file mode 100644 index 000000000..d6ceb776a --- /dev/null +++ b/tests/tracked-jobs.test.mjs @@ -0,0 +1,53 @@ +import fs from "node:fs"; +import path from "node:path"; +import test from "node:test"; +import assert from "node:assert/strict"; + +import { makeTempDir } from "./helpers.mjs"; +import { listJobs, readJobFile, resolveJobFile, upsertJob, writeJobFile } from "../plugins/codex/scripts/lib/state.mjs"; +import { runTrackedJob } from "../plugins/codex/scripts/lib/tracked-jobs.mjs"; + +test("runTrackedJob handles an initial running-state write failure before invoking the runner", async () => { + const workspace = makeTempDir(); + const job = { + id: "task-initial-write-failure", + workspaceRoot: workspace, + title: "Codex Task", + status: "queued", + phase: "queued", + pid: null, + logFile: path.join(workspace, "task.log") + }; + fs.writeFileSync(job.logFile, "queued\n", "utf8"); + writeJobFile(workspace, job.id, job); + upsertJob(workspace, job); + + let writeCalls = 0; + let runnerCalls = 0; + await assert.rejects( + runTrackedJob( + job, + async () => { + runnerCalls += 1; + return { exitStatus: 0 }; + }, + { + logFile: job.logFile, + writeJobFileImpl(cwd, jobId, payload) { + writeCalls += 1; + if (writeCalls === 1) { + throw new Error("initial running write failed"); + } + return writeJobFile(cwd, jobId, payload); + } + } + ), + /initial running write failed/ + ); + + assert.equal(runnerCalls, 0); + const stored = readJobFile(resolveJobFile(workspace, job.id)); + assert.equal(stored.status, "failed"); + assert.equal(stored.errorMessage, "initial running write failed"); + assert.equal(listJobs(workspace)[0].status, "failed"); +}); diff --git a/tsconfig.app-server.json b/tsconfig.app-server.json index 3f8c11f4a..47de8ba6e 100644 --- a/tsconfig.app-server.json +++ b/tsconfig.app-server.json @@ -14,6 +14,7 @@ }, "include": [ "plugins/codex/scripts/lib/app-server.mjs", + "plugins/codex/scripts/lib/background-dispatch.mjs", "plugins/codex/scripts/lib/codex.mjs", "plugins/codex/scripts/lib/fs.mjs", "plugins/codex/scripts/lib/process.mjs",