diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs index 83df468ad..28794d73e 100644 --- a/plugins/codex/scripts/codex-companion.mjs +++ b/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, diff --git a/plugins/codex/scripts/lib/job-control.mjs b/plugins/codex/scripts/lib/job-control.mjs index ad152c157..521f1fd72 100644 --- a/plugins/codex/scripts/lib/job-control.mjs +++ b/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); diff --git a/plugins/codex/scripts/lib/state.mjs b/plugins/codex/scripts/lib/state.mjs index 2da23498f..c1833bcd8 100644 --- a/plugins/codex/scripts/lib/state.mjs +++ b/plugins/codex/scripts/lib/state.mjs @@ -11,20 +11,34 @@ const FALLBACK_STATE_ROOT_DIR = path.join(os.tmpdir(), "codex-companion"); const STATE_FILE_NAME = "state.json"; const JOBS_DIR_NAME = "jobs"; const MAX_JOBS = 50; +const TERMINAL_STATUSES = new Set(["completed", "failed", "cancelled"]); + +export function isTerminalStatus(status) { + return TERMINAL_STATUSES.has(status); +} + +// A job id becomes a filename, so reject anything that could escape the jobs dir. +function isValidJobId(id) { + return typeof id === "string" && id.length > 0 && !id.includes("/") && !id.includes("\\") && id !== "." && id !== ".."; +} + +// Job state is stored one file per job under /jobs/.json, and the +// job list is derived by scanning that directory. There is deliberately NO shared +// index and therefore NO cross-process lock: concurrent `task --background` +// launches (and session cleanup) operate on DIFFERENT files and cannot clobber +// each other, so the "a stale snapshot of a shared array overwrites a sibling +// job" corruption is impossible by construction rather than merely serialized. +// A rename publishes a whole record atomically, so a reader never observes a +// torn/partial file. Only `config` (rarely written, by /setup) lives in +// state.json. See PR openai/codex-plugin-cc#689 for the history behind this. function nowIso() { return new Date().toISOString(); } -function defaultState() { - return { - version: STATE_VERSION, - config: { - stopReviewGate: false - }, - jobs: [] - }; -} +// --------------------------------------------------------------------------- +// Paths +// --------------------------------------------------------------------------- export function resolveStateDir(cwd) { const workspaceRoot = resolveWorkspaceRoot(cwd); @@ -55,70 +69,75 @@ export function ensureStateDir(cwd) { fs.mkdirSync(resolveJobsDir(cwd), { recursive: true }); } -export function loadState(cwd) { - const stateFile = resolveStateFile(cwd); - if (!fs.existsSync(stateFile)) { - return defaultState(); - } +export function resolveJobLogFile(cwd, jobId) { + ensureStateDir(cwd); + return path.join(resolveJobsDir(cwd), `${jobId}.log`); +} - try { - const parsed = JSON.parse(fs.readFileSync(stateFile, "utf8")); - return { - ...defaultState(), - ...parsed, - config: { - ...defaultState().config, - ...(parsed.config ?? {}) - }, - jobs: Array.isArray(parsed.jobs) ? parsed.jobs : [] - }; - } catch { - return defaultState(); - } +export function resolveJobFile(cwd, jobId) { + ensureStateDir(cwd); + return path.join(resolveJobsDir(cwd), `${jobId}.json`); } -function pruneJobs(jobs) { - return [...jobs] - .sort((left, right) => String(right.updatedAt ?? "").localeCompare(String(left.updatedAt ?? ""))) - .slice(0, MAX_JOBS); +// Cancellation is expressed as a SEPARATE, immutable marker file, never by mutating +// the worker-owned record. `.cancelled` is created atomically (O_EXCL) and never +// overwritten, so a cancellation can never be lost to a racing record write. Readers +// overlay it (a record with a live marker reads as cancelled) and the worker honors +// it. This is the compare-and-swap primitive the mutable-record design lacked. +export function resolveJobCancelFile(cwd, jobId) { + ensureStateDir(cwd); + return path.join(resolveJobsDir(cwd), `${jobId}.cancelled`); } -function removeFileIfExists(filePath) { - if (filePath && fs.existsSync(filePath)) { - fs.unlinkSync(filePath); - } +// A per-session "ended" marker (also immutable, atomic-create). It closes the window +// where a task is enqueued AFTER session cleanup's one-shot directory scan: enqueue +// refuses, and a worker started in that window aborts, because both consult it. +// Hash (not sanitize) the session id into the marker filename, so distinct ids like +// "a/b" and "a-b" cannot collide onto the same marker. +function sessionEndedBasename(sessionId) { + const hash = createHash("sha256").update(String(sessionId)).digest("hex").slice(0, 32); + return `session-${hash}.ended`; } -export function saveState(cwd, state) { - const previousJobs = loadState(cwd).jobs; +export function resolveSessionEndedFile(cwd, sessionId) { ensureStateDir(cwd); - const nextJobs = pruneJobs(state.jobs ?? []); - const nextState = { - version: STATE_VERSION, - config: { - ...defaultState().config, - ...(state.config ?? {}) - }, - jobs: nextJobs - }; + return path.join(resolveJobsDir(cwd), sessionEndedBasename(sessionId)); +} - const retainedIds = new Set(nextJobs.map((job) => job.id)); - for (const job of previousJobs) { - if (retainedIds.has(job.id)) { - continue; - } - removeJobFile(resolveJobFile(cwd, job.id)); - removeFileIfExists(job.logFile); +// Create `file` atomically iff absent (wx). Returns true if we created it, false if it +// already existed. Any other error propagates. Used for the immutable markers. +function createMarkerFile(file, payload) { + try { + fs.writeFileSync(file, `${JSON.stringify({ ...payload, at: nowIso() }, null, 2)}\n`, { encoding: "utf8", flag: "wx" }); + return true; + } catch (err) { + if (err.code === "EEXIST") return false; // already marked; idempotent + throw err; } +} + +export function markJobCancelled(cwd, jobId, reason) { + if (!isValidJobId(jobId)) return false; + return createMarkerFile(resolveJobCancelFile(cwd, jobId), { reason: reason ?? "Cancelled." }); +} - fs.writeFileSync(resolveStateFile(cwd), `${JSON.stringify(nextState, null, 2)}\n`, "utf8"); - return nextState; +export function isJobCancelled(cwd, jobId) { + if (!isValidJobId(jobId)) return false; + return fs.existsSync(resolveJobCancelFile(cwd, jobId)); } -export function updateState(cwd, mutate) { - const state = loadState(cwd); - mutate(state); - return saveState(cwd, state); +export function markSessionEnded(cwd, sessionId) { + if (!sessionId) return false; + return createMarkerFile(resolveSessionEndedFile(cwd, sessionId), { sessionId: String(sessionId) }); +} + +export function isSessionEnded(cwd, sessionId) { + if (!sessionId) return false; + try { + return fs.existsSync(resolveSessionEndedFile(cwd, sessionId)); + } catch { + return false; + } } export function generateJobId(prefix = "job") { @@ -126,66 +145,405 @@ export function generateJobId(prefix = "job") { return `${prefix}-${Date.now().toString(36)}-${random}`; } -export function upsertJob(cwd, jobPatch) { - return updateState(cwd, (state) => { - const timestamp = nowIso(); - const existingIndex = state.jobs.findIndex((job) => job.id === jobPatch.id); - if (existingIndex === -1) { - state.jobs.unshift({ - createdAt: timestamp, - updatedAt: timestamp, - ...jobPatch - }); - return; +// --------------------------------------------------------------------------- +// Atomic writes +// --------------------------------------------------------------------------- + +function uniqueTmp(file) { + return `${file}.tmp.${process.pid}.${Math.random().toString(36).slice(2, 8)}`; +} + +// Publish `value` at `file` atomically. The temp lives in the same directory (so +// rename is a same-filesystem atomic replace) and its name carries `.tmp.` so it +// is never mistaken for a job file by listJobs. +function atomicWriteJson(file, value) { + const tmp = uniqueTmp(file); + try { + fs.writeFileSync(tmp, `${JSON.stringify(value, null, 2)}\n`, "utf8"); + fs.renameSync(tmp, file); + } catch (err) { + try { fs.unlinkSync(tmp); } catch {} // never leave a partial temp behind + throw err; + } +} + +// Atomically create `file` iff it does not already exist, and return true; return false +// if it exists. Uses temp + hardlink (link fails with EEXIST if the target exists) so the +// published file is never torn. This is the single-writer CLAIM primitive: the first +// caller to create a job record owns it; a second caller (a duplicate id, a double worker +// launch, a racing migrator) gets false and must not proceed as owner. +function createJsonExclusive(file, value) { + const tmp = uniqueTmp(file); + try { + fs.writeFileSync(tmp, `${JSON.stringify(value, null, 2)}\n`, "utf8"); + try { + fs.linkSync(tmp, file); + return true; + } catch (err) { + if (err.code === "EEXIST") return false; + throw err; } - state.jobs[existingIndex] = { - ...state.jobs[existingIndex], - ...jobPatch, - updatedAt: timestamp - }; + } finally { + try { fs.unlinkSync(tmp); } catch {} + } +} + +// Claim job 's record for the first time. Returns true on success, false if the +// record already exists (another enqueuer/worker owns it -- caller must not proceed). +export function claimJobRecord(cwd, jobId, payload) { + ensureStateDir(cwd); + const now = nowIso(); + return createJsonExclusive(resolveJobFile(cwd, jobId), { + ...payload, + id: jobId, + createdAt: payload.createdAt ?? now, + updatedAt: now }); } -export function listJobs(cwd) { - return loadState(cwd).jobs; +// --------------------------------------------------------------------------- +// Liveness (used only to keep pruning from evicting a live job) +// --------------------------------------------------------------------------- + +function pidAlive(pid) { + if (!Number.isInteger(pid) || pid <= 0) return false; + try { + process.kill(pid, 0); + return true; + } catch (err) { + return err.code === "EPERM"; // exists but not ours (still alive); ESRCH => dead + } } -export function setConfig(cwd, key, value) { - return updateState(cwd, (state) => { - state.config = { - ...state.config, - [key]: value - }; - }); +// --------------------------------------------------------------------------- +// Legacy migration: older installs kept a jobs[] index array in state.json, with some +// fields (startedAt, completedAt, summary, threadId, ...) living only in that index. +// Materialize each index entry as its per-job file, then rewrite state.json config-only. +// - No per-job file yet: create it exclusively (claim). +// - A per-job file exists AND has no live worker (isEvictable: terminal, or dead pid): +// fold the index-only fields IN (payload wins on conflicts), so a finished job keeps +// its summary/threadId/duration. This is safe precisely because no worker can be +// writing that record. +// - A per-job file exists for a LIVE/booting job (queued pid-less, or running with a +// live pid): leave it untouched -- a worker owns it, and legacy index metadata for an +// active job is stale/minimal anyway. This is what keeps migration from racing a +// worker (it now runs in the session hook too). +// Idempotent; after the first run state.json has no jobs array and this no-ops. +// --------------------------------------------------------------------------- + +function migrateLegacyState(cwd) { + const stateFile = resolveStateFile(cwd); + let parsed; + try { + parsed = JSON.parse(fs.readFileSync(stateFile, "utf8")); + } catch { + return; // no state.json (or unreadable) -> nothing to migrate + } + if (!parsed || !Array.isArray(parsed.jobs) || parsed.jobs.length === 0) { + return; + } + ensureStateDir(cwd); + for (const job of parsed.jobs) { + if (!job || !isValidJobId(job.id)) continue; + // `readJobRecord` is the RAW on-disk payload (NOT the cancel/session overlay), so + // isEvictable() classifies the record by its true status: a marked-but-still-running + // record still reads `running` here and is correctly treated as owned, not terminal. + // The evictable set (raw terminal, or raw running with an ESRCH-dead pid) has no + // process that can still write it -- there is no reclaimer of a dead-pid job in this + // design -- so folding index metadata in cannot lose a live/booting worker's write. + const existing = readJobRecord(cwd, job.id); + if (existing == null) { + createJsonExclusive(resolveJobFile(cwd, job.id), job); + } else if (isEvictable(existing)) { + atomicWriteJson(resolveJobFile(cwd, job.id), { ...job, ...existing }); // add index-only fields; payload wins + } + // else: a live/booting worker owns the record -> leave it alone + } + // Rewrite state.json config-only to drop the legacy `jobs` array. Re-read the CURRENT + // state fresh rather than reusing the snapshot parsed at the top, so a config committed + // by a concurrent setConfig while we migrated per-job files is preserved. If the re-read + // FAILS, do NOT rewrite from the stale snapshot (that could revert a concurrent config + // update or a rewrite another migrator already did) -- migration is idempotent and + // re-runs next time. (A residual TOCTOU between this re-read and the rename remains; it + // is the accepted no-CAS class and is much smaller than the pre-read window.) + let current; + try { + current = JSON.parse(fs.readFileSync(stateFile, "utf8")); + } catch { + return; // couldn't re-read -> leave state.json as-is; retry on the next migration + } + if (!Array.isArray(current.jobs) || current.jobs.length === 0) { + return; // another migrator already dropped the jobs array -> nothing to do + } + try { + atomicWriteJson(stateFile, { + version: STATE_VERSION, + config: { stopReviewGate: false, ...(current.config ?? {}) } + }); + } catch { + // A concurrent migrator may have rewritten it between our re-read and here; harmless. + } +} + +// --------------------------------------------------------------------------- +// Config (state.json holds config only) +// --------------------------------------------------------------------------- + +function readConfig(cwd) { + try { + const parsed = JSON.parse(fs.readFileSync(resolveStateFile(cwd), "utf8")); + return { stopReviewGate: false, ...(parsed?.config ?? {}) }; + } catch { + return { stopReviewGate: false }; + } } export function getConfig(cwd) { - return loadState(cwd).config; + migrateLegacyState(cwd); + return readConfig(cwd); } -export function writeJobFile(cwd, jobId, payload) { +export function setConfig(cwd, key, value) { ensureStateDir(cwd); - const jobFile = resolveJobFile(cwd, jobId); - fs.writeFileSync(jobFile, `${JSON.stringify(payload, null, 2)}\n`, "utf8"); - return jobFile; + migrateLegacyState(cwd); + const config = { ...readConfig(cwd), [key]: value }; + atomicWriteJson(resolveStateFile(cwd), { version: STATE_VERSION, config }); + return config; } -export function readJobFile(jobFile) { - return JSON.parse(fs.readFileSync(jobFile, "utf8")); +// --------------------------------------------------------------------------- +// Job records +// --------------------------------------------------------------------------- + +function isJobFileName(name) { + return name.endsWith(".json") && !name.includes(".tmp."); } -function removeJobFile(jobFile) { - if (fs.existsSync(jobFile)) { - fs.unlinkSync(jobFile); +function readJobRecord(cwd, jobId) { + try { + return JSON.parse(fs.readFileSync(resolveJobFile(cwd, jobId), "utf8")); + } catch { + return null; } } -export function resolveJobLogFile(cwd, jobId) { - ensureStateDir(cwd); - return path.join(resolveJobsDir(cwd), `${jobId}.log`); +// The RAW (non-overlaid) pid currently on the record, or null. A canceller reads this +// AFTER creating the cancel marker (never before), so the marker/pid handshake holds: +// if the worker had already published its pid, we see it here and kill it; if not, the +// worker will see our marker on its post-pid re-check and self-abort. +export function readJobPid(cwd, jobId) { + const pid = readJobRecord(cwd, jobId)?.pid; + return Number.isInteger(pid) && pid > 0 ? pid : null; } -export function resolveJobFile(cwd, jobId) { +// Read job 's cancel marker: its parsed content ({reason, at}), `{}` if the marker +// exists but is unreadable/non-object (existence is authoritative), or null if there is +// no marker. A marker whose content parses to a non-object (e.g. literal `null`) must NOT +// be mistaken for "absent" -- normalize it to `{}`. +function readCancelMarker(cwd, jobId) { + let parsed; + try { + parsed = JSON.parse(fs.readFileSync(resolveJobCancelFile(cwd, jobId), "utf8")); + } catch (err) { + return err.code === "ENOENT" ? null : {}; // absent vs present-but-unreadable + } + return parsed && typeof parsed === "object" ? parsed : {}; +} + +// Overlay the immutable markers onto a record so a caller observes a cancellation +// atomically with the record and can never see a cancelled job as live: +// - a `.cancelled` marker forces cancelled, authoritatively (overriding even a +// raced completion, and applying even if the raw record already says cancelled so its +// stripping/metadata are always enforced); +// - a `session-.ended` marker forces cancelled for that session's still-LIVE +// (queued/running) jobs -- a genuinely finished job keeps its terminal outcome. +// The completion payload (result/rendered/summary) is dropped so a cancelled job never +// exposes any model-generated output. `cancelMeta` MUST be read AFTER the record (see +// readAllJobs) so the marker/record observation is coherent: cancel/cleanup always +// publish the marker BEFORE the worker's terminal write, so any record carrying that +// write is observed together with its marker. Side-effect free. +function overlayJob(job, cancelMeta, sessionEnded) { + if (!job || typeof job.id !== "string") return job; + const marked = cancelMeta != null; + if (!marked && !sessionEnded) return job; + const meta = cancelMeta ?? {}; + const at = typeof meta.at === "string" ? meta.at : null; + const { result, rendered, summary, ...rest } = job; + return { + ...rest, + status: "cancelled", + phase: "cancelled", + pid: null, + errorMessage: rest.errorMessage ?? (typeof meta.reason === "string" ? meta.reason : rest.errorMessage), + cancelledAt: rest.cancelledAt ?? at ?? undefined, + completedAt: rest.completedAt ?? at ?? undefined + }; +} + +// Read every RAW job record. `readdirSync` failing with ENOENT means the dir does not +// exist yet -> genuinely empty; ANY OTHER error (EACCES/EIO/...) is a real, load-bearing +// scan failure and MUST propagate rather than masquerade as "no jobs" (session cleanup +// relies on this to not silently skip live workers). A record file that vanished or is +// mid-write (parse error) is skipped -- that is a per-file transient, not a scan failure. +function scanRawJobs(cwd) { + const dir = resolveJobsDir(cwd); + let names; + try { + names = fs.readdirSync(dir); + } catch (err) { + if (err.code === "ENOENT") return []; + throw err; + } + const jobs = []; + for (const name of names) { + if (!isJobFileName(name)) continue; + let raw; + try { + raw = fs.readFileSync(path.join(dir, name), "utf8"); + } catch (err) { + if (err.code === "ENOENT") continue; // vanished between readdir and read -> skip + throw err; // EACCES/EIO on a record we can SEE is a real scan failure -> propagate + } + try { + jobs.push(JSON.parse(raw)); + } catch { + // torn/unparseable record -> skip (a transient, not a scan failure) + } + } + return jobs; +} + +// The marker-overlaid view -- what every consumer (status/cancel/result/session cleanup) +// should see. Each record's markers are read AFTER the record itself (not from a stale +// directory snapshot), so a marker published between the directory listing and the record +// read is still observed. +function readAllJobs(cwd) { + return scanRawJobs(cwd).map((job) => { + if (!job || typeof job.id !== "string") return job; + const cancelMeta = readCancelMarker(cwd, job.id); + const sessionEnded = + cancelMeta == null && job.sessionId != null && + !TERMINAL_STATUSES.has(job.status) && isSessionEnded(cwd, job.sessionId); + return overlayJob(job, cancelMeta, sessionEnded); + }); +} + +// A record is reclaimable ONLY if it is terminal, or a running job whose owner pid is +// provably dead. A pid-less non-terminal job (queued, no pid published yet) is NEVER +// age-evicted: a worker could still be booting -- even paused for a long time by machine +// sleep, load, or a debugger -- and prune's re-check + unlink is a non-atomic +// check-then-act, so age-evicting such a record races the worker's `running` publish and +// (worse) can strand its cancel tombstone. Leaving an abandoned queued record until it +// turns terminal or its worker publishes+dies is the safe choice (a rare, bounded leak). +function isEvictable(job) { + if (typeof job.id !== "string") return false; + if (TERMINAL_STATUSES.has(job.status)) return true; + if (Number.isInteger(job.pid) && job.pid > 0) return !pidAlive(job.pid); + return false; // pid-less non-terminal -> a worker may still be booting; never age-evict +} + +// Remove a job's record, its cancel marker, and both its recorded and conventional +// logs. Returns true if the record was removed (or was already gone). +export function deleteJobFiles(cwd, job) { + let gone = false; + try { + fs.unlinkSync(resolveJobFile(cwd, job.id)); + gone = true; + } catch (err) { + gone = err.code === "ENOENT"; // already removed by another actor -> counts + } + if (!gone) return false; // couldn't remove (e.g. EACCES) + try { fs.unlinkSync(resolveJobCancelFile(cwd, job.id)); } catch {} + if (typeof job.logFile === "string") { try { fs.unlinkSync(job.logFile); } catch {} } + try { fs.unlinkSync(resolveJobLogFile(cwd, job.id)); } catch {} + return true; +} + +// Keep the newest MAX_JOBS records; evict the oldest reclaimable ones so a live +// queued/running job is never made undiscoverable (which would break +// status/cancel/session cleanup). Eviction is decided on the RAW record, NOT the cancel +// overlay: a job that merely carries a cancel marker but whose raw record is still +// queued/running is NOT evictable while a worker could still be booting (a pid-less +// queued record only ages out after QUEUED_GRACE_MS; a booting worker boots in seconds +// and honors the marker long before then). Deleting the record+marker of such a job +// would let the booting worker re-create it unmarked -- the resurrection this avoids. +// Session-ended markers are intentionally NOT GC'd here: a safe generation-aware sweep +// is out of scope, and one tiny empty file per session is a negligible, race-free leak. +function pruneJobs(cwd) { + const jobs = scanRawJobs(cwd); // RAW records + if (jobs.length <= MAX_JOBS) return; + // Protect the newest MAX_JOBS: a record is a candidate for eviction ONLY if it falls + // OUTSIDE that window. Otherwise, when the cap is exceeded entirely by non-evictable + // records (e.g. 50 queued pid-less jobs) and one more job just completed, we would + // evict that fresh completion -- the only evictable record -- and lose its output. + // The cap is soft: a non-evictable (live/booting) record beyond the window is kept + // rather than a recent one. Every evictable record outside the window is removed. + const byNewest = [...jobs].sort((a, b) => String(b.updatedAt ?? "").localeCompare(String(a.updatedAt ?? ""))); + const candidates = byNewest.slice(MAX_JOBS).filter(isEvictable); // everything beyond the newest MAX_JOBS + for (const job of candidates) { + const current = readJobRecord(cwd, job.id); // RAW re-check + if (current && !isEvictable(current)) continue; // became live/pending since scan -> spare + deleteJobFiles(cwd, job); + } +} + +// Merge `patch` into job 's record and publish it atomically. This is the +// single write path for both a full record and an incremental patch; merging +// (rather than overwriting) means a field written by one call site is never lost +// by a later call that omits it (e.g. a `summary` added after the payload write). +// +// The record has a SINGLE writer -- the worker owns queued->running->terminal (the +// enqueue's pre-spawn queued write happens-before the worker exists). Cancellation +// does NOT write here; it uses the immutable `.cancelled` marker instead. So there +// is no concurrent read-modify-write on this file and a plain additive merge is safe: +// no cross-process lost update is possible. Returns the published record. +function mergeJobRecord(cwd, jobId, patch) { ensureStateDir(cwd); - return path.join(resolveJobsDir(cwd), `${jobId}.json`); + const existing = readJobRecord(cwd, jobId) ?? {}; + const now = nowIso(); + const record = { + ...existing, + ...patch, + id: jobId, + createdAt: existing.createdAt ?? patch.createdAt ?? now, + updatedAt: now + }; + atomicWriteJson(resolveJobFile(cwd, jobId), record); + // Prune is best-effort GC that runs AFTER the record above is already committed. It must + // never throw back into the caller: a prune-scan failure (e.g. EACCES/EIO) must not be + // mistaken for a failure of this already-published write (which, in runTrackedJob, would + // overwrite a committed `completed` with `failed`). + try { + pruneJobs(cwd); + } catch { + // GC failure is non-fatal; the cap is soft and the next write retries prune. + } + return record; +} + +export function upsertJob(cwd, patch) { + return mergeJobRecord(cwd, patch.id, patch); +} + +export function writeJobFile(cwd, jobId, payload) { + mergeJobRecord(cwd, jobId, payload); + return resolveJobFile(cwd, jobId); +} + +export function readJobFile(jobFile) { + return JSON.parse(fs.readFileSync(jobFile, "utf8")); +} + +export function listJobs(cwd) { + migrateLegacyState(cwd); + return readAllJobs(cwd); +} + +// The RAW records (no marker overlay), but still migrated from any legacy state.json +// jobs[] index so a fresh install and an upgraded one look the same. Session cleanup +// needs this: having just written the session-ended marker, the overlaid view would +// hide the very jobs it must kill. +export function readAllJobsRaw(cwd) { + migrateLegacyState(cwd); + return scanRawJobs(cwd); } diff --git a/plugins/codex/scripts/lib/tracked-jobs.mjs b/plugins/codex/scripts/lib/tracked-jobs.mjs index 902869012..b13de40f8 100644 --- a/plugins/codex/scripts/lib/tracked-jobs.mjs +++ b/plugins/codex/scripts/lib/tracked-jobs.mjs @@ -1,7 +1,7 @@ import fs from "node:fs"; import process from "node:process"; -import { readJobFile, resolveJobFile, resolveJobLogFile, upsertJob, writeJobFile } from "./state.mjs"; +import { isJobCancelled, isSessionEnded, readJobFile, resolveJobFile, resolveJobLogFile, upsertJob } from "./state.mjs"; export const SESSION_ID_ENV = "CODEX_COMPANION_SESSION_ID"; @@ -99,18 +99,11 @@ export function createJobProgressUpdater(workspaceRoot, jobId) { return; } + // upsertJob merges the patch into the single per-job record atomically, so the + // former re-read + full-snapshot writeJobFile is not only redundant now but + // unsafe: passing a stale snapshot could resurrect a concurrently cancelled or + // completed job back to "running". upsertJob(workspaceRoot, patch); - - const jobFile = resolveJobFile(workspaceRoot, jobId); - if (!fs.existsSync(jobFile)) { - return; - } - - const storedJob = readJobFile(jobFile); - writeJobFile(workspaceRoot, jobId, { - ...storedJob, - ...patch - }); }; } @@ -132,11 +125,13 @@ export function createProgressReporter({ stderr = false, logFile = null, onEvent } function readStoredJobOrNull(workspaceRoot, jobId) { - const jobFile = resolveJobFile(workspaceRoot, jobId); - if (!fs.existsSync(jobFile)) { + // Guarded read: the per-job file can be pruned or session-cleaned between an + // 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); } export async function runTrackedJob(job, runner, options = {}) { @@ -148,41 +143,50 @@ export async function runTrackedJob(job, runner, options = {}) { pid: process.pid, logFile: options.logFile ?? job.logFile ?? null }; - writeJobFile(job.workspaceRoot, job.id, runningRecord); + // Cancel handshake (the linearizable half). We PUBLISH our pid first, THEN re-check + // the cancel marker. Cancellation does the mirror: create the marker first, THEN read + // the pid. Because each side writes its flag before reading the other's, at least one + // side always observes the other -- so either the canceller finds our pid and kills + // us, or we find the marker here and abort. No orphan worker, no lost cancellation. upsertJob(job.workspaceRoot, runningRecord); + if (isJobCancelled(job.workspaceRoot, job.id) || isSessionEnded(job.workspaceRoot, job.sessionId)) { + appendLogLine(options.logFile ?? job.logFile ?? null, "Job cancelled before start; not starting."); + upsertJob(job.workspaceRoot, { id: job.id, status: "cancelled", phase: "cancelled", pid: null, completedAt: nowIso() }); + return { aborted: true, status: "cancelled" }; + } try { const execution = await runner(); const completionStatus = execution.exitStatus === 0 ? "completed" : "failed"; const completedAt = nowIso(); - writeJobFile(job.workspaceRoot, job.id, { + // Append the final-output log FIRST, then publish the terminal record LAST, so the + // terminal write is the worker's last filesystem touch of this job. Otherwise session + // cleanup, seeing the terminal record, could delete the log and have this append + // recreate an orphan log afterward. The append is best-effort (its own try/catch) so a + // logging failure never falls into the lifecycle catch and writes a `failed` record. + try { + appendLogBlock(options.logFile ?? job.logFile ?? null, "Final output", execution.rendered); + } catch { + // logging is best-effort; must not affect the published lifecycle state + } + upsertJob(job.workspaceRoot, { ...runningRecord, status: completionStatus, threadId: execution.threadId ?? null, turnId: execution.turnId ?? null, + summary: execution.summary, pid: null, phase: completionStatus === "completed" ? "done" : "failed", completedAt, result: execution.payload, rendered: execution.rendered }); - upsertJob(job.workspaceRoot, { - id: job.id, - status: completionStatus, - threadId: execution.threadId ?? null, - turnId: execution.turnId ?? null, - summary: execution.summary, - phase: completionStatus === "completed" ? "done" : "failed", - pid: null, - completedAt - }); - appendLogBlock(options.logFile ?? job.logFile ?? null, "Final output", execution.rendered); return execution; } catch (error) { 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, { + upsertJob(job.workspaceRoot, { ...existing, status: "failed", phase: "failed", @@ -191,14 +195,6 @@ export async function runTrackedJob(job, runner, options = {}) { completedAt, logFile: options.logFile ?? job.logFile ?? existing.logFile ?? null }); - upsertJob(job.workspaceRoot, { - id: job.id, - status: "failed", - phase: "failed", - pid: null, - errorMessage, - completedAt - }); throw error; } } diff --git a/plugins/codex/scripts/session-lifecycle-hook.mjs b/plugins/codex/scripts/session-lifecycle-hook.mjs index 778571e6c..42a0cf0d8 100644 --- a/plugins/codex/scripts/session-lifecycle-hook.mjs +++ b/plugins/codex/scripts/session-lifecycle-hook.mjs @@ -1,7 +1,9 @@ #!/usr/bin/env node +import { spawnSync } from "node:child_process"; import fs from "node:fs"; import process from "node:process"; +import { fileURLToPath } from "node:url"; import { terminateProcessTree } from "./lib/process.mjs"; import { BROKER_ENDPOINT_ENV } from "./lib/app-server.mjs"; @@ -13,7 +15,13 @@ import { sendBrokerShutdown, teardownBrokerSession } from "./lib/broker-lifecycle.mjs"; -import { loadState, resolveStateFile, saveState } from "./lib/state.mjs"; +import { + deleteJobFiles, + markJobCancelled, + markSessionEnded, + readAllJobsRaw, + readJobPid +} from "./lib/state.mjs"; import { TRANSCRIPT_PATH_ENV } from "./lib/claude-session-transfer.mjs"; import { resolveWorkspaceRoot } from "./lib/workspace.mjs"; @@ -39,39 +47,144 @@ function appendEnvVar(name, value) { fs.appendFileSync(process.env.CLAUDE_ENV_FILE, `export ${name}=${shellEscape(value)}\n`, "utf8"); } -function cleanupSessionJobs(cwd, sessionId) { +function pidAlive(pid) { + try { + process.kill(pid, 0); + return true; + } catch (err) { + return err.code === "EPERM"; + } +} + +// A pid that is still RUNNABLE (can execute code), as opposed to a zombie/defunct that +// `kill(pid, 0)` still reports as existing but which can never resurrect a record. Used +// only to decide whether to DELETE a job's record. It must FAIL SAFE: any uncertainty +// returns true (keep the record+tombstone), because wrongly classifying a live worker as +// non-runnable would delete a tombstone the worker could still overwrite. Only a +// conclusive `ps` result reporting a zombie state returns false. +function pidRunnable(pid) { + if (!pidAlive(pid)) return false; // ESRCH => definitely gone + const out = spawnSync("ps", ["-o", "state=", "-p", String(pid)], { encoding: "utf8" }); + // Treat launch failure / non-zero exit / no output / read error as UNCERTAIN -> runnable. + if (out.error || out.status !== 0) return true; + const state = (out.stdout ?? "").trim(); + if (state === "") return true; // ambiguous (some ps print nothing transiently) -> keep + return state[0].toUpperCase() !== "Z"; // conclusive Z/Z+ zombie -> not runnable +} + +// Block (bounded) until every pid has exited, or capMs elapses. Workers install no +// SIGTERM handler so they normally die within a few ms; the cap prevents a hang. +function waitForExit(pids, capMs) { + const deadline = Date.now() + capMs; + let remaining = pids.slice(); + while (remaining.length > 0 && Date.now() < deadline) { + remaining = remaining.filter(pidAlive); + if (remaining.length > 0) { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50); + } + } +} + +export function cleanupSessionJobs(cwd, sessionId) { if (!cwd || !sessionId) { return; } const workspaceRoot = resolveWorkspaceRoot(cwd); - const stateFile = resolveStateFile(workspaceRoot); - if (!fs.existsSync(stateFile)) { - return; + let sessionMarkerFailed = false; + const jobMarkerFailed = new Set(); + + // 1. Publish the session-ended marker FIRST -- before ANY existence check or scan. + // A `!existsSync(state) return` optimization here would be a correctness hole: when + // SessionEnd races the very FIRST background launch in a workspace, both paths can be + // absent at that check, so cleanup would return without writing the marker and the + // launcher's worker would then run after the session ended. Writing the marker first + // (it ensures the state dir) closes that race and makes the scan below race-free: if a + // worker slips past its own marker checks (reads them before this marker exists), it + // must have published its pid before this marker -- hence before the scan -- so the + // scan sees its record and kills it. Either the worker honors the marker, or we find + // and kill it; it also refuses any later enqueue for this session. Load-bearing: if it + // cannot be written (ENOSPC/EACCES), surface the failure loudly rather than silently + // proceeding as if the session were cleanly closed; the per-job kill below still runs. + try { + markSessionEnded(workspaceRoot, sessionId); + } catch (err) { + sessionMarkerFailed = true; + process.stderr.write(`codex: failed to publish session-ended marker for ${sessionId}: ${err instanceof Error ? err.message : String(err)}\n`); } - const state = loadState(workspaceRoot); - const removedJobs = state.jobs.filter((job) => job.sessionId === sessionId); - if (removedJobs.length === 0) { + // 2. Scan the RAW records (not the overlaid view -- the marker we just wrote would + // otherwise hide this session's active jobs from us). A scan failure here (EACCES/EIO; + // an absent dir returns [] rather than throwing) is load-bearing: we may be leaving a + // live worker unscanned/unkilled, so fail LOUD rather than silently returning success. + const isActive = (job) => job.status === "queued" || job.status === "running"; + let jobs; + try { + jobs = readAllJobsRaw(workspaceRoot).filter((job) => typeof job.id === "string" && job.sessionId === sessionId); + } catch (err) { + process.exitCode = 1; + process.stderr.write(`codex: session cleanup for ${sessionId} could not scan jobs: ${err instanceof Error ? err.message : String(err)}\n`); return; } + const active = jobs.filter(isActive); - for (const job of removedJobs) { - const stillRunning = job.status === "queued" || job.status === "running"; - if (!stillRunning) { - continue; - } + // 3. Publish a per-job cancel marker for each active job BEFORE reading its pid, so a + // worker that publishes its pid concurrently sees the marker on its post-pid re-check + // and self-aborts (the pid-less startup window is safe: no pid to kill, marker stands). + for (const job of active) { + try { markJobCancelled(workspaceRoot, job.id, "Session ended."); } catch { jobMarkerFailed.add(job.id); } + } + + // 4. Only NOW read each active job's pid (raw, AFTER its marker exists) and terminate + // it, escalating to SIGKILL. Reading the pid after the marker is what makes the + // handshake hold: a worker that publishes its pid after this read still sees the marker + // on its own re-check and self-aborts. + const pidByJob = new Map(active.map((job) => [job.id, readJobPid(workspaceRoot, job.id)])); + const pids = [...pidByJob.values()].filter((pid) => pid != null); + for (const pid of pids) { try { - terminateProcessTree(job.pid ?? Number.NaN); + terminateProcessTree(pid); } catch { // Ignore teardown failures during session shutdown. } } + waitForExit(pids, 2000); + const survivors = pids.filter(pidAlive); + for (const pid of survivors) { + try { process.kill(pid, "SIGKILL"); } catch {} + } + if (survivors.length > 0) { + waitForExit(survivors, 500); + } - saveState(workspaceRoot, { - ...state, - jobs: state.jobs.filter((job) => job.sessionId !== sessionId) - }); + // 5. Delete records, EXCEPT those we cannot prove are worker-free: an active job for + // which we found NO pid (a worker may still be booting), or one whose pid is STILL + // alive after SIGKILL (an unkillable/stuck worker). Those are left as record + marker + // (overlay => cancelled; prune GCs once the raw record ages out or turns terminal); + // the booting worker honors the marker and aborts. Everything else -- a job we killed + // and confirmed gone, or an already-finished job -- is safe to remove; deleteJobFiles + // removes the record, its cancel marker, and its logs together. + const stillRunnable = new Set(pids.filter(pidRunnable)); // excludes zombies (dead, unreaped) + const keepMarker = new Set( + active + .filter((job) => pidByJob.get(job.id) == null || stillRunnable.has(pidByJob.get(job.id))) + .map((job) => job.id) + ); + for (const job of jobs) { + if (keepMarker.has(job.id)) continue; + deleteJobFiles(workspaceRoot, job); + } + + // 6. Fail loud if a LOAD-BEARING marker could not be written. Best-effort scanning and + // killing above still ran, but without the session-ended marker (or the cancel marker of + // a job we KEPT because a worker may still be booting) we cannot claim the session was + // cleanly closed: a task racing enqueue, or that kept booting worker, could still run. + // A non-zero exit surfaces that to the hook runner rather than reporting success. + const keptWithFailedMarker = [...keepMarker].some((id) => jobMarkerFailed.has(id)); + if (sessionMarkerFailed || keptWithFailedMarker) { + process.exitCode = 1; + process.stderr.write(`codex: session cleanup for ${sessionId} could not durably publish a required marker\n`); + } } function handleSessionStart(input) { @@ -127,7 +240,20 @@ async function main() { } } -main().catch((error) => { - process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); - process.exit(1); -}); +// Only run the hook when executed directly (node session-lifecycle-hook.mjs ); +// importing the module (e.g. from tests) must not read stdin or run a lifecycle event. +function isDirectRun() { + if (!process.argv[1]) return false; + try { + return fs.realpathSync(process.argv[1]) === fs.realpathSync(fileURLToPath(import.meta.url)); + } catch { + return false; + } +} + +if (isDirectRun()) { + main().catch((error) => { + process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); + process.exit(1); + }); +} diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 8f276835b..997d1ae0f 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -8,7 +8,23 @@ import { fileURLToPath } from "node:url"; import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs"; import { initGitRepo, makeTempDir, run } from "./helpers.mjs"; import { loadBrokerSession, saveBrokerSession } from "../plugins/codex/scripts/lib/broker-lifecycle.mjs"; -import { resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; +import { listJobs, resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs"; + +// Jobs are stored one file per job under /jobs/.json (no shared +// index array). Read them back newest-first, the way the CLI lists them. +function jobsFromStateDir(stateDir) { + const jobsDir = path.join(stateDir, "jobs"); + let names; + try { + names = fs.readdirSync(jobsDir); + } catch { + return []; + } + return names + .filter((name) => name.endsWith(".json") && !name.includes(".tmp.")) + .map((name) => JSON.parse(fs.readFileSync(path.join(jobsDir, name), "utf8"))) + .sort((a, b) => String(b.updatedAt ?? "").localeCompare(String(a.updatedAt ?? ""))); +} const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); const PLUGIN_ROOT = path.join(ROOT, "plugins", "codex"); @@ -471,7 +487,7 @@ test("review logs reasoning summaries and review output to the job log", () => { assert.equal(result.status, 0, result.stderr); const stateDir = resolveStateDir(repo); - const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const state = { jobs: jobsFromStateDir(stateDir) }; const log = fs.readFileSync(state.jobs[0].logFile, "utf8"); assert.match(log, /Reasoning summary/); assert.match(log, /Reviewed the changed files and checked the likely regression paths/); @@ -800,7 +816,7 @@ test("task logs reasoning summaries and assistant messages to the job log", () = assert.equal(result.status, 0, result.stderr); const stateDir = resolveStateDir(repo); - const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const state = { jobs: jobsFromStateDir(stateDir) }; const log = fs.readFileSync(state.jobs[0].logFile, "utf8"); assert.match(log, /Reasoning summary/); assert.match(log, /Inspected the prompt, gathered evidence, and checked the highest-risk paths first/); @@ -824,7 +840,7 @@ test("task logs subagent reasoning and messages with a subagent prefix", () => { assert.equal(result.status, 0, result.stderr); const stateDir = resolveStateDir(repo); - const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const state = { jobs: jobsFromStateDir(stateDir) }; const log = fs.readFileSync(state.jobs[0].logFile, "utf8"); assert.match(log, /Starting subagent design-challenger via collaboration tool: wait\./); assert.match(log, /Subagent design-challenger reasoning:/); @@ -1428,6 +1444,7 @@ test("result without a job id prefers the latest finished job from the current C id: "review-current", status: "completed", title: "Codex Review", + sessionId: "sess-current", threadId: "thr_current", result: { codex: { @@ -1448,6 +1465,7 @@ test("result without a job id prefers the latest finished job from the current C id: "review-other", status: "completed", title: "Codex Review", + sessionId: "sess-other", threadId: "thr_other", result: { codex: { @@ -1565,22 +1583,10 @@ test("cancel stops an active background job and marks it cancelled", async (t) = }); const logFile = path.join(jobsDir, "task-live.log"); - const jobFile = path.join(jobsDir, "task-live.json"); fs.writeFileSync(logFile, "[2026-03-18T15:30:00.000Z] Starting Codex Task.\n", "utf8"); - fs.writeFileSync( - jobFile, - JSON.stringify( - { - id: "task-live", - status: "running", - title: "Codex Task", - logFile - }, - null, - 2 - ), - "utf8" - ); + // The task-live.json record (with its pid) is materialized from the legacy state.json + // jobs[] index by migration; do NOT pre-write a bare file, which create-exclusive + // migration would skip, leaving the record without its pid. fs.writeFileSync( path.join(stateDir, "state.json"), `${JSON.stringify( @@ -1624,13 +1630,13 @@ test("cancel stops an active background job and marks it cancelled", async (t) = } }); - const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); - const cancelled = state.jobs.find((job) => job.id === "task-live"); + // Cancellation is recorded via an immutable marker + read overlay (not a racy record + // write), so the observable state (what listJobs returns) is cancelled and the marker + // file exists. + const cancelled = listJobs(workspace).find((job) => job.id === "task-live"); assert.equal(cancelled.status, "cancelled"); assert.equal(cancelled.pid, null); - - const stored = JSON.parse(fs.readFileSync(jobFile, "utf8")); - assert.equal(stored.status, "cancelled"); + assert.equal(fs.existsSync(path.join(jobsDir, "task-live.cancelled")), true, "cancel marker published"); assert.match(fs.readFileSync(logFile, "utf8"), /Cancelled by user/); }); @@ -1685,7 +1691,7 @@ test("cancel without a job id ignores active jobs from other Claude sessions", ( assert.equal(cancel.status, 1); assert.match(cancel.stderr, /No active Codex jobs to cancel for this session\./); - const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const state = { jobs: jobsFromStateDir(stateDir) }; assert.equal(state.jobs[0].status, "running"); }); @@ -1733,8 +1739,7 @@ test("cancel with a job id can still target an active job from another Claude se assert.equal(cancel.status, 0, cancel.stderr); assert.equal(JSON.parse(cancel.stdout).jobId, "task-other"); - const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); - assert.equal(state.jobs[0].status, "cancelled"); + assert.equal(listJobs(workspace).find((job) => job.id === "task-other").status, "cancelled"); }); test("cancel sends turn interrupt to the shared app-server before killing a brokered task", async () => { @@ -1760,7 +1765,7 @@ test("cancel sends turn interrupt to the shared app-server before killing a brok const stateDir = resolveStateDir(repo); const runningJob = await waitFor(() => { - const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const state = { jobs: jobsFromStateDir(stateDir) }; const job = state.jobs.find((candidate) => candidate.id === jobId); if (job?.status === "running" && job.threadId && job.turnId) { return job; @@ -1821,8 +1826,9 @@ test("session end fully cleans up jobs for the ending session", async (t) => { fs.writeFileSync(completedLog, "completed\n", "utf8"); fs.writeFileSync(runningLog, "running\n", "utf8"); fs.writeFileSync(otherSessionLog, "other\n", "utf8"); - fs.writeFileSync(completedJobFile, JSON.stringify({ id: "review-completed" }, null, 2), "utf8"); - fs.writeFileSync(otherJobFile, JSON.stringify({ id: "review-other" }, null, 2), "utf8"); + // The per-job .json records are materialized from the legacy state.json jobs[] index + // below by migration; we deliberately do NOT pre-write bare {id} files, because + // migration is now create-exclusive and would skip (not enrich) an existing file. const sleeper = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { cwd: repo, @@ -1830,7 +1836,6 @@ test("session end fully cleans up jobs for the ending session", async (t) => { stdio: "ignore" }); sleeper.unref(); - fs.writeFileSync(runningJobFile, JSON.stringify({ id: "review-running" }, null, 2), "utf8"); t.after(() => { try { @@ -1903,8 +1908,14 @@ test("session end fully cleans up jobs for the ending session", async (t) => { assert.equal(result.status, 0, result.stderr); assert.equal(fs.existsSync(otherSessionLog), true); assert.equal(fs.existsSync(otherJobFile), true); + // The ending session's jobs (and their logs) are removed; the other session's job is + // untouched. Session end also drops an immutable `session-.ended` marker that + // refuses/aborts a task raced in after this cleanup scan. + const remaining = fs.readdirSync(path.dirname(otherJobFile)).sort(); + const sessionMarkers = remaining.filter((n) => n.startsWith("session-") && n.endsWith(".ended")); + assert.equal(sessionMarkers.length, 1, "one session-ended marker was written"); assert.deepEqual( - fs.readdirSync(path.dirname(otherJobFile)).sort(), + remaining.filter((n) => !n.startsWith("session-")), [path.basename(otherJobFile), path.basename(otherSessionLog)].sort() ); @@ -1917,7 +1928,7 @@ test("session end fully cleans up jobs for the ending session", async (t) => { } }); - const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")); + const state = { jobs: jobsFromStateDir(stateDir) }; assert.deepEqual(state.jobs.map((job) => job.id), ["review-other"]); const otherJob = state.jobs[0]; assert.equal(otherJob.logFile, otherSessionLog); diff --git a/tests/session-cleanup.test.mjs b/tests/session-cleanup.test.mjs new file mode 100644 index 000000000..0582c24a0 --- /dev/null +++ b/tests/session-cleanup.test.mjs @@ -0,0 +1,54 @@ +import fs from "node:fs"; +import test from "node:test"; +import assert from "node:assert/strict"; + +import { makeTempDir } from "./helpers.mjs"; +import { isJobCancelled, listJobs, resolveJobFile, upsertJob } from "../plugins/codex/scripts/lib/state.mjs"; +import { resolveWorkspaceRoot } from "../plugins/codex/scripts/lib/workspace.mjs"; +import { cleanupSessionJobs } from "../plugins/codex/scripts/session-lifecycle-hook.mjs"; + +// A pid that is (practically) guaranteed not to exist, so it reads as a dead worker. +const DEAD_PID = 2 ** 31 - 1; + +test("cleanup marks a queued pid-less job cancelled and keeps it (a booting worker cannot resurrect it)", () => { + const cwd = makeTempDir(); + const workspace = resolveWorkspaceRoot(cwd); + // Enqueue wrote the queued record but the worker has not published its pid yet. + upsertJob(workspace, { id: "q", status: "queued", phase: "queued", pid: null, sessionId: "s1" }); + + cleanupSessionJobs(cwd, "s1"); + + assert.equal(isJobCancelled(workspace, "q"), true, "an immutable cancel marker was published"); + const job = listJobs(workspace).find((entry) => entry.id === "q"); + assert.ok(job, "record must NOT be deleted while a worker could still be booting"); + assert.equal(job.status, "cancelled", "the marker overlays the record as cancelled"); + + // A worker that already read the queued record then publishes running: the record + // says running, but the immutable marker overlays every read back to cancelled. + upsertJob(workspace, { id: "q", status: "running", pid: 12345 }); + const after = listJobs(workspace).find((entry) => entry.id === "q"); + assert.equal(after.status, "cancelled", "no resurrection to running: the marker wins on every read"); + assert.equal(after.pid, null, "an overlaid-cancelled job exposes no live pid"); +}); + +test("cleanup deletes a running job whose worker is provably gone", () => { + const cwd = makeTempDir(); + const workspace = resolveWorkspaceRoot(cwd); + upsertJob(workspace, { id: "r", status: "running", phase: "starting", pid: DEAD_PID, sessionId: "s1" }); + + cleanupSessionJobs(cwd, "s1"); + + assert.equal(fs.existsSync(resolveJobFile(workspace, "r")), false, "a confirmed-dead worker's record is removed"); +}); + +test("cleanup deletes an already-finished job and ignores other sessions", () => { + const cwd = makeTempDir(); + const workspace = resolveWorkspaceRoot(cwd); + upsertJob(workspace, { id: "done", status: "completed", pid: null, sessionId: "s1" }); + upsertJob(workspace, { id: "other", status: "running", pid: DEAD_PID, sessionId: "s2" }); + + cleanupSessionJobs(cwd, "s1"); + + assert.equal(fs.existsSync(resolveJobFile(workspace, "done")), false, "finished job of this session is removed"); + assert.ok(listJobs(workspace).find((entry) => entry.id === "other"), "another session's job is untouched"); +}); diff --git a/tests/state.test.mjs b/tests/state.test.mjs index 0f8f57cea..90c09a141 100644 --- a/tests/state.test.mjs +++ b/tests/state.test.mjs @@ -4,8 +4,28 @@ 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 } from "./helpers.mjs"; -import { resolveJobFile, resolveJobLogFile, resolveStateDir, resolveStateFile, saveState } from "../plugins/codex/scripts/lib/state.mjs"; +import { + ensureStateDir, + isJobCancelled, + isSessionEnded, + listJobs, + markJobCancelled, + markSessionEnded, + readJobPid, + resolveJobFile, + resolveJobLogFile, + resolveStateDir, + resolveStateFile, + upsertJob, + writeJobFile +} from "../plugins/codex/scripts/lib/state.mjs"; + +const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); +const STATE_URL = pathToFileURL(path.join(ROOT, "plugins", "codex", "scripts", "lib", "state.mjs")).href; test("resolveStateDir uses a temp-backed per-workspace directory", () => { const workspace = makeTempDir(); @@ -40,66 +60,264 @@ test("resolveStateDir uses CLAUDE_PLUGIN_DATA when it is provided", () => { } }); -test("saveState prunes dropped job artifacts when indexed jobs exceed the cap", () => { - const workspace = makeTempDir(); - const stateFile = resolveStateFile(workspace); - fs.mkdirSync(path.dirname(stateFile), { recursive: true }); - - const jobs = Array.from({ length: 51 }, (_, index) => { - const jobId = `job-${index}`; - const updatedAt = new Date(Date.UTC(2026, 0, 1, 0, index, 0)).toISOString(); - const logFile = resolveJobLogFile(workspace, jobId); - const jobFile = resolveJobFile(workspace, jobId); - fs.writeFileSync(logFile, `log ${jobId}\n`, "utf8"); - fs.writeFileSync(jobFile, JSON.stringify({ id: jobId, status: "completed" }, null, 2), "utf8"); - return { - id: jobId, - status: "completed", - logFile, - updatedAt, - createdAt: updatedAt - }; - }); +test("upsertJob merges a patch into a job's record without dropping fields", () => { + const workspace = makeTempDir(); + writeJobFile(workspace, "j1", { id: "j1", status: "running", sessionId: "s1", pid: 123 }); + upsertJob(workspace, { id: "j1", status: "completed", summary: "done" }); + const job = listJobs(workspace).find((entry) => entry.id === "j1"); + assert.equal(job.status, "completed"); + assert.equal(job.summary, "done"); + assert.equal(job.sessionId, "s1"); // field from the earlier write is preserved + assert.ok(job.createdAt); + assert.ok(job.updatedAt); +}); + +test("concurrent upserts of different jobs never lose a record (no lock)", async () => { + const workspace = makeTempDir(); + const jobCount = 16; + const worker = + `import { upsertJob } from ${JSON.stringify(STATE_URL)};\n` + + "const [cwd, id] = process.argv.slice(1);\n" + + "upsertJob(cwd, { id, status: \"queued\", jobClass: \"task\", summary: id });\n"; + + await Promise.all( + Array.from({ length: jobCount }, (_, index) => + new Promise((resolve, reject) => { + const child = spawn( + process.execPath, + ["--input-type=module", "-e", worker, workspace, `job-${index}`], + { stdio: "ignore" } + ); + child.on("exit", (code) => + code === 0 ? resolve() : reject(new Error(`worker ${index} exited ${code}`)) + ); + }) + ) + ); + + const ids = new Set(listJobs(workspace).map((job) => job.id)); + const missing = Array.from({ length: jobCount }, (_, index) => `job-${index}`).filter( + (id) => !ids.has(id) + ); + assert.deepEqual(missing, [], "every concurrently launched job must be present"); +}); + +test("prune evicts oldest terminal jobs but keeps live and non-terminal ones over the cap", () => { + const workspace = makeTempDir(); + ensureStateDir(workspace); + + // 50 old terminal jobs (completed), staggered updatedAt so ordering is defined. + for (let i = 0; i < 50; i += 1) { + const id = `done-${String(i).padStart(2, "0")}`; + const ts = new Date(Date.UTC(2026, 0, 1, 0, i, 0)).toISOString(); + fs.writeFileSync(resolveJobFile(workspace, id), JSON.stringify({ id, status: "completed", updatedAt: ts, createdAt: ts })); + fs.writeFileSync(resolveJobLogFile(workspace, id), `log ${id}\n`); + } + // A running job owned by THIS (alive) process, with the oldest timestamp of all. + const liveTs = new Date(Date.UTC(2025, 0, 1)).toISOString(); fs.writeFileSync( - stateFile, - `${JSON.stringify( - { - version: 1, - config: { stopReviewGate: false }, - jobs - }, - null, - 2 - )}\n`, - "utf8" + resolveJobFile(workspace, "live"), + JSON.stringify({ id: "live", status: "running", pid: process.pid, updatedAt: liveTs, createdAt: liveTs }) ); - saveState(workspace, { + // 52 files now (> MAX_JOBS=50). One more write triggers prune (overflow 2). + upsertJob(workspace, { id: "trigger", status: "completed" }); + + const ids = new Set(listJobs(workspace).map((job) => job.id)); + assert.equal(ids.has("live"), true, "a live running job must never be pruned, even as the oldest"); + // Only records OUTSIDE the newest MAX_JOBS window are eviction candidates. With 52 + // records the two oldest are `done-00` and `live`; `done-00` (terminal) is evicted, + // `live` (non-evictable) is kept -> the cap stays soft at 51. `done-01` is inside the + // protected newest-50 window, so it survives. + assert.equal(ids.has("done-00"), false, "the oldest terminal job (outside the window) is evicted"); + assert.equal(fs.existsSync(resolveJobLogFile(workspace, "done-00")), false, "evicted job's log is removed too"); + assert.equal(ids.has("done-01"), true, "a job inside the newest-MAX_JOBS window is protected"); + assert.equal(ids.has("done-02"), true, "newer terminal jobs are kept"); +}); + +test("prune never evicts a just-completed job when the cap is full of non-evictable jobs", () => { + const workspace = makeTempDir(); + ensureStateDir(workspace); + // MAX_JOBS (50) queued, pid-less jobs -> all non-evictable. + for (let i = 0; i < 50; i += 1) { + const id = `q-${String(i).padStart(2, "0")}`; + const ts = new Date(Date.UTC(2026, 0, 1, 0, i, 0)).toISOString(); + fs.writeFileSync(resolveJobFile(workspace, id), JSON.stringify({ id, status: "queued", pid: null, updatedAt: ts, createdAt: ts })); + } + // A 51st job completes with full output; its write triggers prune. + upsertJob(workspace, { id: "fresh", status: "completed", result: { codex: { stdout: "important" } } }); + + const job = listJobs(workspace).find((entry) => entry.id === "fresh"); + assert.ok(job, "the just-completed job is the ONLY evictable record but must not be pruned"); + assert.equal(job.status, "completed"); + assert.equal(fs.existsSync(resolveJobFile(workspace, "fresh")), true); +}); + +test("legacy state.json jobs[] array migrates to per-job files on read", () => { + const workspace = makeTempDir(); + ensureStateDir(workspace); + const legacy = { version: 1, - config: { stopReviewGate: false }, - jobs - }); + config: { stopReviewGate: true }, + jobs: [ + { id: "old-a", status: "completed", sessionId: "s", updatedAt: "2026-01-01T00:00:00.000Z" }, + { id: "old-b", status: "failed", sessionId: "s", updatedAt: "2026-01-01T00:01:00.000Z" } + ] + }; + fs.writeFileSync(resolveStateFile(workspace), JSON.stringify(legacy, null, 2)); + + const ids = new Set(listJobs(workspace).map((job) => job.id)); + assert.equal(ids.has("old-a"), true); + assert.equal(ids.has("old-b"), true); + assert.equal(fs.existsSync(resolveJobFile(workspace, "old-a")), true, "legacy entry materialized as a per-job file"); + + // state.json is rewritten config-only (no jobs array left to re-migrate/resurrect). + const rewritten = JSON.parse(fs.readFileSync(resolveStateFile(workspace), "utf8")); + assert.equal(Array.isArray(rewritten.jobs), false); + assert.equal(rewritten.config.stopReviewGate, true); +}); + +test("a no-status patch (e.g. a progress update) leaves status untouched", () => { + const workspace = makeTempDir(); + upsertJob(workspace, { id: "j", status: "running", pid: 42 }); + upsertJob(workspace, { id: "j", phase: "thinking" }); + + const job = listJobs(workspace).find((entry) => entry.id === "j"); + assert.equal(job.status, "running"); + assert.equal(job.phase, "thinking"); +}); + +test("a cancel marker overlays as cancelled and cannot be resurrected by a later running write", () => { + const workspace = makeTempDir(); + upsertJob(workspace, { id: "j", status: "queued", pid: null }); + assert.equal(markJobCancelled(workspace, "j", "test"), true, "marker created"); + + // A racing worker publishes running AFTER the marker (the record itself says running): + upsertJob(workspace, { id: "j", status: "running", pid: 99, phase: "starting" }); + + const job = listJobs(workspace).find((entry) => entry.id === "j"); + assert.equal(job.status, "cancelled", "the immutable marker overlays the record"); + assert.equal(job.pid, null, "an overlaid-cancelled job exposes no live pid"); + assert.equal(isJobCancelled(workspace, "j"), true); +}); + +test("the cancel marker is immutable: a second mark is a no-op, not an error", () => { + const workspace = makeTempDir(); + upsertJob(workspace, { id: "j", status: "running", pid: 1 }); + assert.equal(markJobCancelled(workspace, "j", "first"), true); + assert.equal(markJobCancelled(workspace, "j", "second"), false, "already marked -> false, no throw"); +}); + +test("a completion that lands after a cancel marker still reads as cancelled", () => { + const workspace = makeTempDir(); + upsertJob(workspace, { id: "j", status: "running", pid: 5 }); + markJobCancelled(workspace, "j", "cancelled by user"); + upsertJob(workspace, { id: "j", status: "completed", summary: "done" }); + + const job = listJobs(workspace).find((entry) => entry.id === "j"); + assert.equal(job.status, "cancelled", "cancellation wins over a later completion via the overlay"); +}); + +test("normal forward transitions still advance (queued -> running -> completed)", () => { + const workspace = makeTempDir(); + upsertJob(workspace, { id: "j", status: "queued", pid: null }); + upsertJob(workspace, { id: "j", status: "running", pid: 7 }); + const done = upsertJob(workspace, { id: "j", status: "completed" }); + assert.equal(done.status, "completed"); +}); + +test("a no-status progress patch advances phase without touching status", () => { + const workspace = makeTempDir(); + upsertJob(workspace, { id: "j", status: "running", pid: 7, phase: "starting" }); + const after = upsertJob(workspace, { id: "j", phase: "thinking" }); + assert.equal(after.status, "running"); + assert.equal(after.phase, "thinking", "phase update must not be dropped"); +}); - const prunedJobFile = resolveJobFile(workspace, "job-0"); - const prunedLogFile = resolveJobLogFile(workspace, "job-0"); - const retainedJobFile = resolveJobFile(workspace, "job-50"); - const retainedLogFile = resolveJobLogFile(workspace, "job-50"); - const jobsDir = path.dirname(prunedJobFile); +test("session-ended marker is create-once and observable", () => { + const workspace = makeTempDir(); + assert.equal(isSessionEnded(workspace, "s1"), false); + assert.equal(markSessionEnded(workspace, "s1"), true, "first mark creates it"); + assert.equal(markSessionEnded(workspace, "s1"), false, "second mark is a no-op"); + assert.equal(isSessionEnded(workspace, "s1"), true); + assert.equal(isSessionEnded(workspace, "s2"), false, "unrelated session is unaffected"); +}); - assert.equal(fs.existsSync(retainedJobFile), true); - assert.equal(fs.existsSync(retainedLogFile), true); +test("readJobPid returns the raw record pid (not overlaid) so a canceller can kill after marking", () => { + const workspace = makeTempDir(); + upsertJob(workspace, { id: "j", status: "running", pid: 4321 }); + assert.equal(readJobPid(workspace, "j"), 4321); + // After marking, listJobs overlays pid:null, but the RAW pid is still readable to kill. + markJobCancelled(workspace, "j", "test"); + assert.equal(readJobPid(workspace, "j"), 4321, "raw pid survives the overlay"); + assert.equal(listJobs(workspace).find((e) => e.id === "j").pid, null, "overlay hides the pid from readers"); +}); - const savedState = JSON.parse(fs.readFileSync(stateFile, "utf8")); - assert.equal(savedState.jobs.length, 50); - assert.deepEqual( - savedState.jobs.map((job) => job.id), - Array.from({ length: 50 }, (_, index) => `job-${50 - index}`) +test("legacy migration folds index-only metadata into an existing FINISHED per-job file", () => { + const workspace = makeTempDir(); + // A finished per-job payload already exists but lacks the index-only fields. + writeJobFile(workspace, "review-done", { id: "review-done", status: "completed", title: "Review" }); + fs.writeFileSync( + resolveStateFile(workspace), + JSON.stringify({ + version: 1, + config: { stopReviewGate: false }, + jobs: [ + { + id: "review-done", + status: "completed", + title: "Review", + summary: "found 2 issues", + threadId: "thr_abc", + startedAt: "2026-01-01T00:00:00.000Z", + completedAt: "2026-01-01T00:01:00.000Z" + } + ] + }, null, 2) ); - assert.deepEqual( - fs.readdirSync(jobsDir).sort(), - Array.from({ length: 50 }, (_, index) => `job-${index + 1}`) - .flatMap((jobId) => [`${jobId}.json`, `${jobId}.log`]) - .sort() + + const job = listJobs(workspace).find((entry) => entry.id === "review-done"); + assert.equal(job.status, "completed"); + assert.equal(job.summary, "found 2 issues", "index-only summary is merged in, not lost"); + assert.equal(job.threadId, "thr_abc", "index-only threadId is merged in"); + assert.equal(job.startedAt, "2026-01-01T00:00:00.000Z"); + assert.equal(job.completedAt, "2026-01-01T00:01:00.000Z"); +}); + +test("legacy migration does NOT overwrite a LIVE (running, live-pid) per-job file", () => { + const workspace = makeTempDir(); + // A running job owned by THIS (alive) process; its worker is the sole writer. + writeJobFile(workspace, "live", { id: "live", status: "running", pid: process.pid, phase: "thinking" }); + fs.writeFileSync( + resolveStateFile(workspace), + JSON.stringify({ + version: 1, + config: {}, + jobs: [{ id: "live", status: "queued", phase: "queued", summary: "stale index" }] + }, null, 2) ); + + const job = listJobs(workspace).find((entry) => entry.id === "live"); + assert.equal(job.status, "running", "a live worker's record is never reverted by a stale index"); + assert.equal(job.phase, "thinking"); +}); + +test("cancel overlay strips result/rendered and surfaces the marker's reason + timestamp", () => { + const workspace = makeTempDir(); + // A job that finished (has result/rendered) and is THEN cancel-marked. + upsertJob(workspace, { + id: "j", + status: "completed", + result: { codex: { stdout: "secret output" } }, + rendered: "full rendered output" + }); + assert.equal(markJobCancelled(workspace, "j", "Cancelled by user."), true); + + const job = listJobs(workspace).find((entry) => entry.id === "j"); + assert.equal(job.status, "cancelled"); + assert.equal(job.result, undefined, "result payload is stripped from a cancelled job"); + assert.equal(job.rendered, undefined, "rendered output is stripped from a cancelled job"); + assert.equal(job.errorMessage, "Cancelled by user.", "marker reason surfaces as errorMessage"); + assert.ok(job.cancelledAt, "marker timestamp surfaces as cancelledAt"); });