Skip to content

fix: persist background jobs before worker spawn #733

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fscfede-beep wants to merge 7 commits into
base: main
Choose a base branch
from
Open

fix: persist background jobs before worker spawn #733

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
ce6563b
fix: persist background jobs before worker spawn
Sep 4, 2026
f8452a1
fix: prevent cancelled workers from reclaiming jobs
Sep 4, 2026
2f35b69
fix: make startup cancellation durable
Sep 4, 2026
5e7cfb3
fix: harden background worker teardown
Sep 4, 2026
d7a435d
fix: preserve cancellation over late spawn errors
Sep 4, 2026
0d8cb00
fix: preserve terminal state on spawn failure races
Sep 5, 2026
d5fda68
fix: preserve removal after startup cancellation
fscfede-beep Sep 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 23 additions & 8 deletions plugins/codex/scripts/codex-companion.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import {
generateJobId,
getConfig,
listJobs,
markJobCancellationRequested,
setConfig,
upsertJob,
writeJobFile
Expand All @@ -48,6 +49,7 @@ import {
createJobProgressUpdater,
createJobRecord,
createProgressReporter,
failTrackedJobLaunch,
nowIso,
runTrackedJob,
SESSION_ID_ENV
Expand Down Expand Up @@ -668,7 +670,7 @@ async function runForegroundCommand(job, runner, options = {}) {
return execution;
}

function spawnDetachedTaskWorker(cwd, jobId) {
function spawnDetachedTaskWorker(cwd, jobId, onError) {
const scriptPath = path.join(ROOT_DIR, "scripts", "codex-companion.mjs");
const child = spawn(process.execPath, [scriptPath, "task-worker", "--cwd", cwd, "--job-id", jobId], {
cwd,
Expand All @@ -677,25 +679,32 @@ function spawnDetachedTaskWorker(cwd, jobId) {
stdio: "ignore",
windowsHide: true
});
child.unref();
child.once("error", onError);
child.once("spawn", () => child.unref());
return child;
}

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,
pid: null,
Comment thread
fscfede-beep marked this conversation as resolved.
Comment thread
fscfede-beep marked this conversation as resolved.
logFile,
request
};
writeJobFile(job.workspaceRoot, job.id, queuedRecord);
upsertJob(job.workspaceRoot, queuedRecord);
const failSpawn = (error) => failTrackedJobLaunch(queuedRecord, error);
try {
spawnDetachedTaskWorker(cwd, job.id, failSpawn);
} catch (error) {
failSpawn(error);
throw error;
}

return {
payload: {
Expand Down Expand Up @@ -850,6 +859,10 @@ async function handleTaskWorker(argv) {
if (!storedJob) {
throw new Error(`No stored job found for ${options["job-id"]}.`);
}
if (storedJob.status !== "queued") {
Comment thread
fscfede-beep marked this conversation as resolved.
appendLogLine(storedJob.logFile ?? null, `Background worker skipped ${storedJob.status} job.`);
return;
}

const request = storedJob.request;
if (!request || typeof request !== "object") {
Expand Down Expand Up @@ -969,9 +982,11 @@ async function handleCancel(argv) {
const cwd = resolveCommandCwd(options);
const reference = positionals[0] ?? "";
const { workspaceRoot, job } = resolveCancelableJob(cwd, reference, { env: process.env });
markJobCancellationRequested(workspaceRoot, job.id);

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve SessionEnd removal in the cancel handler

When /codex:cancel pauses in interruptAppServerTurn and SessionEnd runs in that window, cleanup publishes .removed and deletes the job, but this handler subsequently writes and indexes cancelled unconditionally, recreating the removed artifacts. A late worker will not repair this because it reads the recreated cancelled record and returns before runTrackedJob. The fresh evidence after the line-232 fix is that removal revalidation was added only to the worker's startup-cancellation branch; handleCancel still needs to recheck removal after its cancellation writes and delete the per-job file and indexed state when SessionEnd won.

Useful? React with 👍 / 👎.

const existing = readStoredJob(workspaceRoot, job.id) ?? {};
const threadId = existing.threadId ?? job.threadId ?? null;
const turnId = existing.turnId ?? job.turnId ?? null;
const latestJob = { ...job, ...existing };
const threadId = latestJob.threadId ?? null;
const turnId = latestJob.turnId ?? null;

const interrupt = await interruptAppServerTurn(cwd, { threadId, turnId });
if (interrupt.attempted) {
Expand All @@ -983,8 +998,8 @@ async function handleCancel(argv) {
);
}

terminateProcessTree(job.pid ?? Number.NaN);
appendLogLine(job.logFile, "Cancelled by user.");
terminateProcessTree(latestJob.pid ?? Number.NaN);
appendLogLine(latestJob.logFile ?? job.logFile, "Cancelled by user.");

const completedAt = nowIso();
const nextJob = {
Expand Down
37 changes: 37 additions & 0 deletions plugins/codex/scripts/lib/state.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,7 @@ export function saveState(cwd, state) {
continue;
}
removeJobFile(resolveJobFile(cwd, job.id));
removeFileIfExists(resolveJobCancellationFile(cwd, job.id));
removeFileIfExists(job.logFile);
}

Expand Down Expand Up @@ -189,3 +190,39 @@ export function resolveJobFile(cwd, jobId) {
ensureStateDir(cwd);
return path.join(resolveJobsDir(cwd), `${jobId}.json`);
}

export function resolveJobCancellationFile(cwd, jobId) {
ensureStateDir(cwd);
return path.join(resolveJobsDir(cwd), `${jobId}.cancelled`);
}

export function markJobCancellationRequested(cwd, jobId) {
const marker = resolveJobCancellationFile(cwd, jobId);
fs.writeFileSync(marker, "cancel requested\n", "utf8");
return marker;
}

export function isJobCancellationRequested(cwd, jobId) {
return fs.existsSync(resolveJobCancellationFile(cwd, jobId));
}

export function resolveJobRemovalFile(cwd, jobId) {
ensureStateDir(cwd);
return path.join(resolveJobsDir(cwd), `${jobId}.removed`);
}

export function markJobRemovalRequested(cwd, jobId) {
const marker = resolveJobRemovalFile(cwd, jobId);
fs.writeFileSync(marker, "removed\n", "utf8");
return marker;
}

export function isJobRemovalRequested(cwd, jobId) {
return fs.existsSync(resolveJobRemovalFile(cwd, jobId));
}

export function removeJobFromState(cwd, jobId) {
return updateState(cwd, (state) => {
state.jobs = state.jobs.filter((job) => job.id !== jobId);
});
}
119 changes: 115 additions & 4 deletions plugins/codex/scripts/lib/tracked-jobs.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import fs from "node:fs";
import process from "node:process";

import { readJobFile, resolveJobFile, resolveJobLogFile, upsertJob, writeJobFile } from "./state.mjs";
import { isJobCancellationRequested, isJobRemovalRequested, readJobFile, removeJobFromState, resolveJobFile, resolveJobLogFile, upsertJob, writeJobFile } from "./state.mjs";

export const SESSION_ID_ENV = "CODEX_COMPANION_SESSION_ID";

Expand Down Expand Up @@ -73,6 +73,9 @@ export function createJobProgressUpdater(workspaceRoot, jobId) {
let lastTurnId = null;

return (event) => {
if (isJobRemovalRequested(workspaceRoot, jobId)) {
return;
}
const normalized = normalizeProgressEvent(event);
const patch = { id: jobId };
let changed = false;
Expand Down Expand Up @@ -139,7 +142,62 @@ function readStoredJobOrNull(workspaceRoot, jobId) {
return readJobFile(jobFile);
}

function removedExecution(job) {
return {
exitStatus: 0,
payload: { jobId: job.id, status: "removed" },
rendered: "",
summary: "Session ended.",
threadId: null,
turnId: null
};
}

export function failTrackedJobLaunch(job, error) {
if (isJobRemovalRequested(job.workspaceRoot, job.id)) {
return { ...job, status: "removed", phase: "removed", pid: null };
}
const currentJob = readStoredJobOrNull(job.workspaceRoot, job.id) ?? job;
if (isJobCancellationRequested(job.workspaceRoot, job.id) || currentJob.status !== "queued") {

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Serialize spawn-failure handling with stop markers

When an emitted spawn error overlaps /codex:cancel or SessionEnd, this marker/status check and the subsequent failed-record writes are not atomic. If the callback passes this guard just before the other process publishes its marker and terminalizes/removes the job, it can resume afterward and overwrite cancelled with failed or recreate a job that session cleanup removed. The fresh evidence is that failTrackedJobLaunch performs its only cancellation/removal checks before the writes at lines 167-168; coordinate the transition or revalidate without allowing a check-to-write race.

Useful? React with 👍 / 👎.

@fscfede-beep fscfede-beep Sep 5, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in 0d8cb00. I reproduced both check→write races deterministically by triggering cancellation/removal from error.toString() after the initial guard but before the failed-record write. RED on d7a435d: both cases ended as failed. The fix now revalidates durable terminal markers after persisting failed: removal deletes job file/state; cancellation rewrites canonical cancelled state, so terminal authority wins even if it arrives in that window. Validation on Windows: tracked-jobs + commands + state 20/20 PASS; targeted runtime background|cancel|SessionEnd 6 relevant cases PASS with one known Windows taskkill failure. I reproduced that exact taskkill ... operation not permitted failure on untouched upstream/main (db52e28), confirming it is baseline and unrelated. node --check and git diff --check clean. @codex review

return currentJob;
}
const completedAt = nowIso();
const errorMessage = `Background worker failed to start: ${error instanceof Error ? error.message : String(error)}`;
const failedRecord = { ...currentJob, status: "failed", phase: "failed", pid: null, completedAt, errorMessage };
writeJobFile(job.workspaceRoot, job.id, failedRecord);
upsertJob(job.workspaceRoot, failedRecord);

if (isJobRemovalRequested(job.workspaceRoot, job.id)) {
const jobFile = resolveJobFile(job.workspaceRoot, job.id);
if (fs.existsSync(jobFile)) fs.unlinkSync(jobFile);
removeJobFromState(job.workspaceRoot, job.id);
return { ...currentJob, status: "removed", phase: "removed", pid: null };
}

if (isJobCancellationRequested(job.workspaceRoot, job.id)) {
const cancelledAt = nowIso();
const cancelledRecord = {
...currentJob,
status: "cancelled",
phase: "cancelled",
pid: null,
completedAt: cancelledAt,
cancelledAt,
errorMessage: "Cancelled by user."
};
writeJobFile(job.workspaceRoot, job.id, cancelledRecord);
upsertJob(job.workspaceRoot, cancelledRecord);
return cancelledRecord;
}

appendLogLine(currentJob.logFile ?? null, errorMessage);
return failedRecord;
}

export async function runTrackedJob(job, runner, options = {}) {
if (isJobRemovalRequested(job.workspaceRoot, job.id)) {
return removedExecution(job);
}
const runningRecord = {
...job,
status: "running",
Expand All @@ -148,11 +206,60 @@ 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 {
writeJobFile(job.workspaceRoot, job.id, runningRecord);
if (isJobRemovalRequested(job.workspaceRoot, job.id)) {
const jobFile = resolveJobFile(job.workspaceRoot, job.id);
if (fs.existsSync(jobFile)) fs.unlinkSync(jobFile);
return removedExecution(job);
}
upsertJob(job.workspaceRoot, runningRecord);
if (isJobRemovalRequested(job.workspaceRoot, job.id)) {
removeJobFromState(job.workspaceRoot, job.id);
return removedExecution(job);
}
if (isJobCancellationRequested(job.workspaceRoot, job.id)) {
const completedAt = nowIso();
const cancelledRecord = {
...runningRecord,
status: "cancelled",
phase: "cancelled",
pid: null,
completedAt,
cancelledAt: completedAt,
errorMessage: "Cancelled by user."
};
writeJobFile(job.workspaceRoot, job.id, cancelledRecord);

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Recheck removal after persisting startup cancellation

When /codex:cancel and SessionEnd overlap during the queued-worker handoff, the worker can pass the removal check at line 217, observe the cancellation marker, and then have session cleanup publish .removed and delete the job before this write. If process termination is delayed or fails, this branch recreates the cancelled job file and index after cleanup and never revalidates the removal marker, unlike failTrackedJobLaunch; recheck removal after these writes and delete the artifacts/state so session removal remains authoritative.

Useful? React with 👍 / 👎.

@fscfede-beep fscfede-beep Sep 5, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in d5fda68. The startup-cancellation branch now rechecks the durable removal marker after persisting cancelled; if SessionEnd won in that window, it deletes the per-job file, removes the indexed state, and returns removedExecution instead of recreating terminalized work. Added a RED structural regression on 0d8cb00 requiring removal revalidation after the cancellation writes; GREEN on this commit, plus existing behavioral coverage. Fresh validation: tracked-jobs + commands + state 21/21 PASS. Targeted runtime background|cancel|SessionEnd has 6 relevant cases PASS and one Windows taskkill failure; I reproduced that exact failure again on untouched upstream/main (db52e28), confirming baseline. node --check/git diff --check clean. @codex review

upsertJob(job.workspaceRoot, {
id: job.id,
status: "cancelled",
phase: "cancelled",
pid: null,
completedAt,
cancelledAt: completedAt,
errorMessage: "Cancelled by user."
});
if (isJobRemovalRequested(job.workspaceRoot, job.id)) {
const jobFile = resolveJobFile(job.workspaceRoot, job.id);
if (fs.existsSync(jobFile)) fs.unlinkSync(jobFile);
removeJobFromState(job.workspaceRoot, job.id);
return removedExecution(job);
}
appendLogLine(options.logFile ?? job.logFile ?? null, "Cancelled before task execution.");
return {
exitStatus: 0,
payload: { jobId: job.id, status: "cancelled" },
rendered: "Cancelled by user.\n",
summary: "Cancelled by user.",
threadId: null,
turnId: null
};
}
const execution = await runner();
if (isJobRemovalRequested(job.workspaceRoot, job.id)) {

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Recheck removal after persisting completion

When cleanupSessionJobs starts just after this check, the worker can write the completed job file with pid: null, after which SessionEnd reads the still-running index entry, cannot terminate the worker using that file, and deletes the job before the worker's subsequent upsertJob. The upsert then recreates the session's job index without its result file, so cleanup is undone and /codex:result cannot retrieve the advertised completed result. Revalidate the removal marker after the completion file/index writes or serialize terminalization with removal.

Useful? React with 👍 / 👎.

removeJobFromState(job.workspaceRoot, job.id);
return removedExecution(job);
}
const completionStatus = execution.exitStatus === 0 ? "completed" : "failed";
const completedAt = nowIso();
writeJobFile(job.workspaceRoot, job.id, {
Expand All @@ -179,6 +286,10 @@ export async function runTrackedJob(job, runner, options = {}) {
appendLogBlock(options.logFile ?? job.logFile ?? null, "Final output", execution.rendered);
return execution;
} catch (error) {
if (isJobRemovalRequested(job.workspaceRoot, job.id)) {
removeJobFromState(job.workspaceRoot, job.id);
return removedExecution(job);
}
const errorMessage = error instanceof Error ? error.message : String(error);
const existing = readStoredJobOrNull(job.workspaceRoot, job.id) ?? runningRecord;
const completedAt = nowIso();
Expand Down
10 changes: 8 additions & 2 deletions plugins/codex/scripts/session-lifecycle-hook.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ import {
sendBrokerShutdown,
teardownBrokerSession
} from "./lib/broker-lifecycle.mjs";
import { loadState, resolveStateFile, saveState } from "./lib/state.mjs";
import { loadState, markJobRemovalRequested, readJobFile, resolveJobFile, resolveStateFile, saveState } from "./lib/state.mjs";
import { TRANSCRIPT_PATH_ENV } from "./lib/claude-session-transfer.mjs";
import { resolveWorkspaceRoot } from "./lib/workspace.mjs";

Expand Down Expand Up @@ -61,8 +61,14 @@ function cleanupSessionJobs(cwd, sessionId) {
if (!stillRunning) {
continue;
}
markJobRemovalRequested(workspaceRoot, job.id);
let effectiveJob = job;
try {
terminateProcessTree(job.pid ?? Number.NaN);
const jobFile = resolveJobFile(workspaceRoot, job.id);
if (fs.existsSync(jobFile)) effectiveJob = { ...job, ...readJobFile(jobFile) };
} catch {}
try {
terminateProcessTree(effectiveJob.pid ?? Number.NaN);
} catch {
// Ignore teardown failures during session shutdown.
}
Expand Down
31 changes: 31 additions & 0 deletions tests/commands.test.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,6 +210,37 @@ test("hooks keep session-end cleanup and stop gating enabled", () => {
assert.match(source, /session-lifecycle-hook\.mjs/);
});

test("background task persists its bootstrap record before spawning the worker", () => {
const source = read("scripts/codex-companion.mjs");
const start = source.indexOf("function enqueueBackgroundTask");
const end = source.indexOf("\nasync function handleReviewCommand", start);
const body = source.slice(start, end);
assert.notEqual(start, -1);
assert.notEqual(end, -1);
assert.ok(body.indexOf("writeJobFile") < body.indexOf("spawnDetachedTaskWorker"));
assert.ok(body.indexOf("upsertJob") < body.indexOf("spawnDetachedTaskWorker"));

const tracked = read("scripts/lib/tracked-jobs.mjs");
const trackedStart = tracked.indexOf("export async function runTrackedJob");
const trackedBody = tracked.slice(trackedStart);
assert.ok(trackedBody.indexOf("try {") < trackedBody.indexOf("writeJobFile"));
assert.ok(trackedBody.indexOf("try {") < trackedBody.indexOf("upsertJob"));
});

test("cancel publishes its durable request marker before awaiting the app-server", () => {
const source = read("scripts/codex-companion.mjs");
const start = source.indexOf("async function handleCancel");
const end = source.indexOf("\nasync function main", start);
const body = source.slice(start, end);
const markerWrite = body.indexOf("markJobCancellationRequested");
const interrupt = body.indexOf("await interruptAppServerTurn");
const reread = body.indexOf("readStoredJob");
const terminate = body.indexOf("terminateProcessTree(latestJob.pid");
assert.ok(markerWrite >= 0 && markerWrite < reread);
assert.ok(markerWrite < interrupt);
assert.ok(reread < terminate);
});

test("setup command can offer Codex install and still points users to codex login", () => {
const setup = read("commands/setup.md");
const readme = fs.readFileSync(path.join(ROOT, "README.md"), "utf8");
Expand Down
Loading