diff --git a/docs/superpowers/plans/2026-08-19-fail-closed-codex-gate.md b/docs/superpowers/plans/2026-08-19-fail-closed-codex-gate.md new file mode 100644 index 000000000..ac823cc2c --- /dev/null +++ b/docs/superpowers/plans/2026-08-19-fail-closed-codex-gate.md @@ -0,0 +1,269 @@ +# Fail-Closed Codex Gate Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make detached Codex jobs self-heal after worker loss and make the Claude Stop review gate deterministically fail closed without duplicate reviews of the same turn. + +**Architecture:** Persist queued work before spawning, reconcile active jobs against their worker PID on every control-plane read, and refuse late writes after a terminal transition. Use a deterministic full-hash Stop job ID and immutable claims to single-flight an exact raw Claude turn, while retaining the existing foreground timeout. + +**Tech Stack:** Node.js ESM, built-in `node:test`, filesystem JSON state, Claude Code hooks. + +## Global Constraints + +- No new runtime dependencies or daemon. +- The only job transitions are `queued -> running -> completed|failed|cancelled`. +- Enabled review gates fail closed for infrastructure and persistence errors. +- The raw, untrimmed last assistant message is never stored in a gate cache key; an empty message is not cached. +- All production changes follow RED-GREEN TDD. + +--- + +### Task 1: Self-healing tracked jobs + +**Files:** +- Modify: `tests/process.test.mjs` +- Modify: `tests/runtime.test.mjs` +- Modify: `plugins/codex/scripts/lib/process.mjs` +- Modify: `plugins/codex/scripts/lib/tracked-jobs.mjs` +- Modify: `plugins/codex/scripts/lib/job-control.mjs` +- Modify: `plugins/codex/scripts/lib/render.mjs` + +**Interfaces:** +- Produces: `isProcessAlive(pid, options?) -> boolean` +- Produces: `reconcileTrackedJobs(workspaceRoot, options?) -> Job[]` +- Consumes: existing `listJobs`, `writeJobFile`, and `upsertJob` persistence functions. + +- [ ] **Step 1: Write failing process-liveness tests** + +Add tests proving a successful signal probe and `EPERM` mean alive while `ESRCH` means dead: + +```js +assert.equal(isProcessAlive(123, { killImpl() {} }), true); +assert.equal(isProcessAlive(123, { killImpl() { throw Object.assign(new Error("gone"), { code: "ESRCH" }); } }), false); +assert.equal(isProcessAlive(123, { killImpl() { throw Object.assign(new Error("denied"), { code: "EPERM" }); } }), true); +``` + +- [ ] **Step 2: Verify the process test is RED** + +Run: `node --test --test-name-pattern="isProcessAlive" tests/process.test.mjs` + +Expected: FAIL because `isProcessAlive` is not exported. + +- [ ] **Step 3: Implement the process probe** + +Add `isProcessAlive` to `process.mjs` using `process.kill(pid, 0)`, returning false only for non-finite PIDs and `ESRCH`, and treating `EPERM` as alive. + +- [ ] **Step 4: Write failing reconciliation tests** + +Add runtime tests with persisted jobs proving: + +```js +assert.equal(payload.job.status, "failed"); +assert.equal(payload.waitTimedOut, false); +assert.match(payload.job.errorMessage, /worker exited/i); +``` + +and an old queued job without a PID fails with `/did not start within 5 seconds/i`. Update the existing active-timeout fixture to use `pid: process.pid` so it continues to represent a live worker. + +- [ ] **Step 5: Verify the reconciliation tests are RED** + +Run: `node --test --test-name-pattern="dead worker|startup grace|still active" tests/runtime.test.mjs` + +Expected: dead and unstarted jobs remain active, so the new assertions fail. + +- [ ] **Step 6: Implement reconciliation and rendering** + +In `tracked-jobs.mjs`, reconcile each active job: + +```js +if (job.status === "queued" && !Number.isFinite(job.pid) && ageMs >= 5000) { + return failTrackedJob(workspaceRoot, job, "Background worker did not start within 5 seconds."); +} +if (Number.isFinite(job.pid) && !isProcessAlive(job.pid)) { + return failTrackedJob(workspaceRoot, job, "Background worker exited before completing the job."); +} +``` + +Use reconciled jobs in all job-control read paths. Render `Error: ${job.errorMessage}` in failed-job details. + +- [ ] **Step 7: Verify Task 1 GREEN** + +Run: `node --test tests/process.test.mjs tests/runtime.test.mjs` + +Expected: PASS. + +### Task 2: Race-free launch and terminal-state protection + +**Files:** +- Modify: `tests/runtime.test.mjs` +- Modify: `plugins/codex/scripts/codex-companion.mjs` +- Modify: `plugins/codex/scripts/lib/tracked-jobs.mjs` + +**Interfaces:** +- Consumes: `reconcileTrackedJobs` from Task 1. +- Produces: queued job publication before detached worker spawn. +- Produces: an immutable per-job terminal fence, created with exclusive filesystem creation, whose first writer wins. + +- [ ] **Step 1: Write failing terminal-state tests** + +Persist a terminal fence and assert a late worker cannot run or overwrite its first terminal outcome. Remove a running job during a deferred runner after SessionEnd/cancellation fences it and assert progress/finalization does not recreate a visible job. Add a deterministic reconciliation-versus-worker interleaving test. + +- [ ] **Step 2: Verify the terminal-state tests are RED** + +Run: `node --test --test-name-pattern="terminal job|removed job" tests/runtime.test.mjs` + +Expected: FAIL because current finalization overwrites or recreates the job. + +- [ ] **Step 3: Publish before spawn** + +Change enqueue ordering to: + +```js +writeJobFile(job.workspaceRoot, job.id, queuedRecord); +upsertJob(job.workspaceRoot, queuedRecord); +spawnDetachedTaskWorker(cwd, job.id); +``` + +The worker sets its own PID when it enters `runTrackedJob`; queued jobs receive the five-second startup grace from Task 1. + +- [ ] **Step 4: Protect terminal transitions** + +Workers first claim `jobs/.started.json` with `openSync(..., "wx")`; reconciliation, SessionEnd, and startup compete there before a running publication. After publishing `running`, only the exclusive `jobs/.admission.json` winner may call the runner. Later terminal outcomes compete on `jobs/.terminal.json`. Empty/corrupt claims fail as failed, progress never upserts the state index, and effective reads merge mutable job-file fields. SessionEnd writes a dominant empty `jobs/.removed` marker before cleanup so a late mutable artifact is never visible. + +- [ ] **Step 5: Verify Task 2 GREEN** + +Run: `node --test --test-name-pattern="background|terminal job|removed job|cancel|SessionEnd" tests/runtime.test.mjs` + +Expected: PASS. + +### Task 3: Fail-closed, idempotent Stop review + +**Files:** +- Modify: `tests/runtime.test.mjs` +- Modify: `plugins/codex/scripts/lib/tracked-jobs.mjs` +- Modify: `plugins/codex/scripts/codex-companion.mjs` +- Modify: `plugins/codex/scripts/stop-review-gate-hook.mjs` + +**Interfaces:** +- Produces: `CODEX_COMPANION_GATE_KEY` metadata on Stop-review jobs. +- Produces: deterministic gate-key hashing of session ID and last assistant message. +- Consumes: stored `result.rawOutput` for exact-turn reuse. + +- [ ] **Step 1: Write failing gate tests** + +Change the unavailable-Codex test to require: + +```js +assert.equal(JSON.parse(result.stdout).decision, "block"); +assert.match(JSON.parse(result.stdout).reason, /not set up/i); +``` + +Run the Stop hook twice with the same session and non-empty `last_assistant_message`; assert both decisions match and the fake Codex `nextTurnId` does not increase on the second call. + +- [ ] **Step 2: Verify the gate tests are RED** + +Run: `node --test --test-name-pattern="unavailable|same Claude response" tests/runtime.test.mjs` + +Expected: unavailable Codex produces no decision and the second Stop starts another turn. + +- [ ] **Step 3: Propagate and reuse the gate key** + +Hash the raw, untrimmed non-empty message without retaining its content: + +```js +createHash("sha256").update(`${sessionId}\0${lastAssistantMessage}`).digest("hex") +``` + +Pass it through `CODEX_COMPANION_GATE_KEY`, use deterministic `gate-` as the tracked-job ID, and before availability checks find the matching current-session Stop job. The immutable startup claim makes concurrent matching launches single-flight. Reparse completed output; block with the existing job ID for active, failed, or cancelled matches. With no key, run fresh without caching. + +- [ ] **Step 4: Make setup failures fail closed** + +When `buildSetupNote` returns a message for an enabled gate, emit: + +```js +emitDecision({ decision: "block", reason: setupNote }); +``` + +- [ ] **Step 5: Verify Task 3 GREEN** + +Run: `node --test --test-name-pattern="stop hook" tests/runtime.test.mjs` + +Expected: PASS. + +### Task 4: Atomic, strict JSON persistence + +**Files:** +- Modify: `tests/state.test.mjs` +- Modify: `plugins/codex/scripts/lib/state.mjs` +- Modify: `plugins/codex/scripts/stop-review-gate-hook.mjs` + +**Interfaces:** +- Produces: same-directory temporary write plus `renameSync` for state and job JSON. +- Produces: bounded per-workspace serialization for state read-modify-write mutations. +- Produces: explicit parse errors from invalid state JSON. + +- [ ] **Step 1: Write failing persistence tests** + +Write invalid `state.json` and assert `loadState` throws an error containing the state path. Run the enabled Stop hook against invalid state and assert it emits `decision: block` with a persistence error. + +- [ ] **Step 2: Verify persistence tests are RED** + +Run: `node --test --test-name-pattern="invalid state|corrupt state" tests/state.test.mjs tests/runtime.test.mjs` + +Expected: `loadState` silently returns defaults and the hook emits no block decision. + +- [ ] **Step 3: Implement atomic strict persistence** + +Write JSON through a unique sibling temporary file and `fs.renameSync`. Serialize state read-modify-write mutations with an exclusive per-workspace lock, recover dead owners, and fail clearly after five seconds of contention. On parse failure throw `Failed to read Codex Companion state at : ` instead of returning defaults. Catch top-level Stop-hook errors and emit a block decision. + +- [ ] **Step 4: Verify Task 4 GREEN** + +Run: `node --test tests/state.test.mjs tests/runtime.test.mjs` + +Expected: PASS. + +### Task 5: Full verification and delivery + +**Files:** +- Modify: `plugins/codex/CHANGELOG.md` + +**Interfaces:** +- Consumes: all behavior from Tasks 1-4. +- Produces: release note and verified branch. + +- [ ] **Step 1: Add a concise changelog entry** + +Document dead-worker reconciliation, race-free background launch, exact-turn Stop reuse, and fail-closed setup/persistence failures under the current unreleased section. + +- [ ] **Step 2: Run complete verification** + +Run: + +```bash +npm test +npm run build +npm run check-version +``` + +Expected: all tests pass, TypeScript exits 0, and version metadata is consistent. + +- [ ] **Step 3: Review the final diff** + +Run: `git diff --check && git diff --stat origin/main...HEAD && git status --short --branch` + +Expected: no whitespace errors and only planned files changed. + +- [ ] **Step 4: Commit** + +```bash +git add docs/superpowers plugins/codex tests +git commit -m "fix: harden Codex stop gate supervision" +``` + +- [ ] **Step 5: Request independent review, fix blocking findings, and re-run verification** + +Dispatch a read-only verifier against `origin/main...HEAD`. Critical and Important findings must be fixed before publishing. + +- [ ] **Step 6: Publish and integrate through repository policy** + +Fetch `origin`, reconcile with its current default branch, push `codex/gate-supervision`, and use the required PR/review path. Record the final main-branch SHA or the external blocker if upstream permissions prevent merge. diff --git a/docs/superpowers/specs/2026-08-19-fail-closed-codex-gate-design.md b/docs/superpowers/specs/2026-08-19-fail-closed-codex-gate-design.md new file mode 100644 index 000000000..b643e79e6 --- /dev/null +++ b/docs/superpowers/specs/2026-08-19-fail-closed-codex-gate-design.md @@ -0,0 +1,63 @@ +# Fail-Closed Codex Gate Design + +## Goal + +Make Codex Companion jobs self-heal after worker loss and make the Claude Stop review gate block with an actionable error instead of silently allowing, looping, or launching the same review twice. + +## Confirmed failure modes + +- A detached worker is spawned before its queued job is persisted. The worker can start first, fail to find the job, and leave the parent to publish a job that will remain queued forever. +- Status and wait paths trust `queued` and `running` JSON without checking whether the worker PID still exists. +- A worker killed outside the normal exception path never reaches `runTrackedJob` finalization, so its job remains active forever. +- A late worker completion can recreate or overwrite a job already cancelled or removed by SessionEnd. +- When the review gate is enabled but Codex is unavailable, the Stop hook only writes a note and allows the session to end. +- Repeating Stop for the same Claude response always starts a fresh Codex review. +- State files are overwritten in place, so an interrupted write can be parsed as an empty default state with the gate disabled. + +## Required behavior + +### Job lifecycle + +- The only transitions are `queued -> running -> completed|failed|cancelled`. +- The queued record and request exist before the detached worker is spawned. +- A queued job without a worker PID receives a five-second startup grace period. After that it becomes `failed` with `Background worker did not start within 5 seconds.` +- A running or queued job with a dead PID becomes `failed` with `Background worker exited before completing the job.` +- Reconciliation runs before status, wait, result, cancel, task resume selection, and Stop-hook decisions. +- Terminal state is claimed by an immutable per-job `jobs/.terminal.json` fence, created with exclusive filesystem creation. The first terminal writer wins; terminal fences contain only status and completion time, never request, prompt, or log content. +- Worker startup is separately claimed by `jobs/.started.json`: its first writer is either a running PID/start time or a terminal outcome. The winner publishes `running`, then exclusively claims `jobs/.admission.json` before calling the runner; a duplicate or terminal winner cannot execute. +- A terminal or removed job cannot be overwritten by a late worker. Corrupt or empty fences fail closed as `failed`; all control-plane reads use the fence over mutable job/index JSON. +- SessionEnd writes an immutable empty `jobs/.removed` marker before cleanup. Removal overrides every terminal or running claim, including a completion that won before SessionEnd, and may remain as a tiny orphan fence. +- Failed job status output includes the stored error message. + +### Stop gate + +- An enabled gate is fail-closed for unavailable Codex, task failure, timeout, missing output, invalid output, and corrupt state. +- Every block explains the failure and how to retry. Active jobs also show the exact status and cancel commands. +- The gate key is a SHA-256 hash of the Claude session ID and the raw, untrimmed last assistant message. No message content is stored in the key; an empty message has no key and is never cached. +- A Stop review uses deterministic `gate-` job ID plus the immutable startup claim, making concurrent same-turn invocations single-flight. A completed matching review is reused; an active matching review blocks with its existing job ID instead of starting another review. +- A different last assistant message gets a new gate key and a fresh review. +- Matching cached jobs are checked before Codex availability, so a prior same-turn decision is still reusable when Codex later becomes unavailable. + +### Persistence + +- `state.json` and per-job JSON files are written to a same-directory temporary file and atomically renamed. +- State read-modify-write mutations use a short per-workspace filesystem lock. Dead owners are recovered; contention or invalid lock metadata fails with a clear error after five seconds instead of hanging. +- Invalid persisted JSON is an explicit error. It must not silently reset `stopReviewGate` to `false`. +- Removed job IDs keep a zero-byte tombstone so an arbitrarily late worker cannot reuse them. This is the deliberate correctness tradeoff for avoiding a daemon, lease, heartbeat, or attempt-token protocol. +- No new daemon, dependency, heartbeat file, or long-lived lock is introduced. PID liveness covers the observed worker-loss failure; the existing 15-minute Stop timeout covers a live but non-returning gate review. + +## Verification + +- A dead running worker becomes `failed` during `status --wait` and returns without timing out. +- A queued job missing a PID past startup grace becomes `failed`. +- Existing live-job timeout behavior remains unchanged when the PID is alive. +- Background task enqueue and completion remain green. +- Cancelling or removing a job prevents late finalization from resurrecting it. +- An unavailable Codex emits `decision: block` when the gate is enabled. +- Running the Stop hook twice with the same session and last response starts one Codex turn and returns the same decision. +- Invalid state blocks the Stop hook with a clear persistence error. +- The complete Node test suite and TypeScript build pass. + +## Scope boundary + +This change does not add retries, a supervisor daemon, configurable heartbeat intervals, or arbitrary background-job runtime limits. Add those only after evidence of a live worker hanging while its process remains healthy. diff --git a/plugins/codex/CHANGELOG.md b/plugins/codex/CHANGELOG.md index d647561bb..8937d6ee1 100644 --- a/plugins/codex/CHANGELOG.md +++ b/plugins/codex/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## Unreleased + +- Hardened background supervision: queue records are published before spawn, and queued or running jobs with missing/dead workers are reconciled to failure. +- Added immutable startup, admission, terminal, and removal claims so late workers cannot resurrect jobs or execute duplicates. +- Stop-gate reviews now single-flight per exact Claude turn, reuse the same-turn result, and fail closed for unavailable or corrupt persistence. +- Made mutable state and job JSON writes atomic, and serialized state mutations behind a bounded crash-recovering workspace lock. + ## 1.0.0 - Initial version of the Codex plugin for Claude Code diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs index 83df468ad..5eeef7762 100644 --- a/plugins/codex/scripts/codex-companion.mjs +++ b/plugins/codex/scripts/codex-companion.mjs @@ -29,7 +29,6 @@ import { loadPromptTemplate, interpolateTemplate } from "./lib/prompts.mjs"; import { generateJobId, getConfig, - listJobs, setConfig, upsertJob, writeJobFile @@ -48,9 +47,12 @@ import { createJobProgressUpdater, createJobRecord, createProgressReporter, + GATE_KEY_ENV, nowIso, + reconcileTrackedJobs, runTrackedJob, - SESSION_ID_ENV + SESSION_ID_ENV, + terminalizeTrackedJob } from "./lib/tracked-jobs.mjs"; import { resolveWorkspaceRoot } from "./lib/workspace.mjs"; import { @@ -336,7 +338,7 @@ async function waitForSingleJobSnapshot(cwd, reference, options = {}) { async function resolveLatestTrackedTaskThread(cwd, options = {}) { const workspaceRoot = resolveWorkspaceRoot(cwd); const sessionId = getCurrentClaudeSessionId(); - const jobs = sortJobsNewestFirst(listJobs(workspaceRoot)).filter((job) => job.id !== options.excludeJobId); + const jobs = sortJobsNewestFirst(reconcileTrackedJobs(workspaceRoot)).filter((job) => job.id !== options.excludeJobId); const visibleJobs = filterJobsForCurrentClaudeSession(jobs); const activeTask = visibleJobs.find((job) => job.jobClass === "task" && (job.status === "queued" || job.status === "running")); if (activeTask) { @@ -564,16 +566,17 @@ function getJobKindLabel(kind, jobClass) { return jobClass === "review" ? "review" : "rescue"; } -function createCompanionJob({ prefix, kind, title, workspaceRoot, jobClass, summary, write = false }) { +function createCompanionJob({ prefix, kind, title, workspaceRoot, jobClass, summary, write = false, id, gateKey }) { return createJobRecord({ - id: generateJobId(prefix), + id: id ?? generateJobId(prefix), kind, kindLabel: getJobKindLabel(kind, jobClass), title, workspaceRoot, jobClass, summary, - write + write, + ...(gateKey ? { gateKey } : {}) }); } @@ -589,7 +592,7 @@ function createTrackedProgress(job, options = {}) { }; } -function buildTaskJob(workspaceRoot, taskMetadata, write) { +function buildTaskJob(workspaceRoot, taskMetadata, write, gateKey = null) { return createCompanionJob({ prefix: "task", kind: "task", @@ -597,7 +600,8 @@ function buildTaskJob(workspaceRoot, taskMetadata, write) { workspaceRoot, jobClass: "task", summary: taskMetadata.summary, - write + write, + ...(gateKey ? { id: `gate-${gateKey}`, gateKey } : {}) }); } @@ -661,6 +665,13 @@ async function runForegroundCommand(job, runner, options = {}) { stderr: !options.json }); const execution = await runTrackedJob(job, () => runner(progress), { logFile }); + if (!Number.isFinite(execution?.exitStatus)) { + if (job.gateKey) { + outputResult({ jobId: job.id, status: execution?.status ?? "failed", gateDuplicate: true }, true); + return execution; + } + throw new Error(`Codex job ${job.id} is ${execution?.status ?? "removed"}; no new run was started.`); + } outputResult(options.json ? execution.payload : execution.rendered, options.json); if (execution.exitStatus !== 0) { process.exitCode = execution.exitStatus; @@ -685,17 +696,17 @@ 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, logFile, request }; writeJobFile(job.workspaceRoot, job.id, queuedRecord); upsertJob(job.workspaceRoot, queuedRecord); + spawnDetachedTaskWorker(cwd, job.id); return { payload: { @@ -804,7 +815,10 @@ async function handleTask(argv) { return; } - const job = buildTaskJob(workspaceRoot, taskMetadata, write); + const gateKey = !resumeLast && taskMetadata.title === "Codex Stop Gate Review" && /^[a-f0-9]{64}$/.test(process.env[GATE_KEY_ENV] ?? "") + ? process.env[GATE_KEY_ENV] + : null; + const job = buildTaskJob(workspaceRoot, taskMetadata, write, gateKey); await runForegroundCommand( job, (progress) => @@ -934,7 +948,7 @@ function handleTaskResumeCandidate(argv) { const cwd = resolveCommandCwd(options); const workspaceRoot = resolveCommandWorkspace(options); const sessionId = getCurrentClaudeSessionId(); - const jobs = filterJobsForCurrentClaudeSession(sortJobsNewestFirst(listJobs(workspaceRoot))); + const jobs = filterJobsForCurrentClaudeSession(sortJobsNewestFirst(reconcileTrackedJobs(workspaceRoot))); const candidate = findLatestResumableTaskJob(jobs); const payload = { @@ -973,19 +987,6 @@ async function handleCancel(argv) { const threadId = existing.threadId ?? job.threadId ?? null; const turnId = existing.turnId ?? job.turnId ?? null; - const interrupt = await interruptAppServerTurn(cwd, { threadId, turnId }); - if (interrupt.attempted) { - appendLogLine( - job.logFile, - interrupt.interrupted - ? `Requested Codex turn interrupt for ${turnId} on ${threadId}.` - : `Codex turn interrupt failed${interrupt.detail ? `: ${interrupt.detail}` : "."}` - ); - } - - terminateProcessTree(job.pid ?? Number.NaN); - appendLogLine(job.logFile, "Cancelled by user."); - const completedAt = nowIso(); const nextJob = { ...job, @@ -996,19 +997,37 @@ async function handleCancel(argv) { errorMessage: "Cancelled by user." }; - writeJobFile(workspaceRoot, job.id, { + const terminal = terminalizeTrackedJob(workspaceRoot, { ...existing, + ...nextJob + }, { ...nextJob, cancelledAt: completedAt }); - upsertJob(workspaceRoot, { - id: job.id, - status: "cancelled", - phase: "cancelled", - pid: null, - errorMessage: "Cancelled by user.", - completedAt - }); + if (!terminal.claimed) { + const firstOutcome = terminal.job ?? job; + const payload = { + jobId: job.id, + status: firstOutcome.status, + title: job.title, + turnInterruptAttempted: false, + turnInterrupted: false + }; + outputCommandResult(payload, renderCancelReport(firstOutcome), options.json); + return; + } + + const interrupt = await interruptAppServerTurn(cwd, { threadId, turnId }); + if (interrupt.attempted) { + appendLogLine( + job.logFile, + interrupt.interrupted + ? `Requested Codex turn interrupt for ${turnId} on ${threadId}.` + : `Codex turn interrupt failed${interrupt.detail ? `: ${interrupt.detail}` : "."}` + ); + } + terminateProcessTree(job.pid ?? Number.NaN); + appendLogLine(job.logFile, "Cancelled by user."); 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..89b264ce6 100644 --- a/plugins/codex/scripts/lib/job-control.mjs +++ b/plugins/codex/scripts/lib/job-control.mjs @@ -1,8 +1,8 @@ import fs from "node:fs"; import { getSessionRuntimeStatus } from "./codex.mjs"; -import { getConfig, listJobs, readJobFile, resolveJobFile } from "./state.mjs"; -import { SESSION_ID_ENV } from "./tracked-jobs.mjs"; +import { getConfig } from "./state.mjs"; +import { readEffectiveStoredJob, reconcileTrackedJobs, SESSION_ID_ENV } from "./tracked-jobs.mjs"; import { resolveWorkspaceRoot } from "./workspace.mjs"; export const DEFAULT_MAX_STATUS_JOBS = 8; @@ -181,11 +181,7 @@ export function enrichJob(job, options = {}) { } export function readStoredJob(workspaceRoot, jobId) { - const jobFile = resolveJobFile(workspaceRoot, jobId); - if (!fs.existsSync(jobFile)) { - return null; - } - return readJobFile(jobFile); + return readEffectiveStoredJob(workspaceRoot, jobId); } function matchJobReference(jobs, reference, predicate = () => true) { @@ -213,7 +209,7 @@ function matchJobReference(jobs, reference, predicate = () => true) { export function buildStatusSnapshot(cwd, options = {}) { const workspaceRoot = resolveWorkspaceRoot(cwd); const config = getConfig(workspaceRoot); - const jobs = sortJobsNewestFirst(filterJobsForCurrentSession(listJobs(workspaceRoot), options)); + const jobs = sortJobsNewestFirst(filterJobsForCurrentSession(reconcileTrackedJobs(workspaceRoot), options)); const maxJobs = options.maxJobs ?? DEFAULT_MAX_STATUS_JOBS; const maxProgressLines = options.maxProgressLines ?? DEFAULT_MAX_PROGRESS_LINES; @@ -241,7 +237,7 @@ export function buildStatusSnapshot(cwd, options = {}) { export function buildSingleJobSnapshot(cwd, reference, options = {}) { const workspaceRoot = resolveWorkspaceRoot(cwd); - const jobs = sortJobsNewestFirst(listJobs(workspaceRoot)); + const jobs = sortJobsNewestFirst(reconcileTrackedJobs(workspaceRoot)); const selected = matchJobReference(jobs, reference); if (!selected) { throw new Error(`No job found for "${reference}". Run /codex:status to inspect known jobs.`); @@ -255,7 +251,8 @@ export function buildSingleJobSnapshot(cwd, reference, options = {}) { export function resolveResultJob(cwd, reference) { const workspaceRoot = resolveWorkspaceRoot(cwd); - const jobs = sortJobsNewestFirst(reference ? listJobs(workspaceRoot) : filterJobsForCurrentSession(listJobs(workspaceRoot))); + const reconciledJobs = reconcileTrackedJobs(workspaceRoot); + const jobs = sortJobsNewestFirst(reference ? reconciledJobs : filterJobsForCurrentSession(reconciledJobs)); const selected = matchJobReference( jobs, reference, @@ -280,7 +277,7 @@ export function resolveResultJob(cwd, reference) { export function resolveCancelableJob(cwd, reference, options = {}) { const workspaceRoot = resolveWorkspaceRoot(cwd); - const jobs = sortJobsNewestFirst(listJobs(workspaceRoot)); + const jobs = sortJobsNewestFirst(reconcileTrackedJobs(workspaceRoot)); const activeJobs = jobs.filter((job) => job.status === "queued" || job.status === "running"); if (reference) { diff --git a/plugins/codex/scripts/lib/process.mjs b/plugins/codex/scripts/lib/process.mjs index dd8fc3751..5918b3674 100644 --- a/plugins/codex/scripts/lib/process.mjs +++ b/plugins/codex/scripts/lib/process.mjs @@ -50,12 +50,26 @@ export function binaryAvailable(command, versionArgs = ["--version"], options = return { available: true, detail: result.stdout.trim() || result.stderr.trim() || "ok" }; } +export function isProcessAlive(pid, options = {}) { + if (!Number.isSafeInteger(pid) || pid <= 0) { + return false; + } + + const killImpl = options.killImpl ?? process.kill.bind(process); + try { + killImpl(pid, 0); + return true; + } catch (error) { + return error?.code !== "ESRCH"; + } +} + function looksLikeMissingProcessMessage(text) { return /not found|no running instance|cannot find|does not exist|no such process/i.test(text); } export function terminateProcessTree(pid, options = {}) { - if (!Number.isFinite(pid)) { + if (!Number.isSafeInteger(pid) || pid <= 0) { return { attempted: false, delivered: false, method: null }; } @@ -101,19 +115,15 @@ export function terminateProcessTree(pid, options = {}) { killImpl(-pid, "SIGTERM"); return { attempted: true, delivered: true, method: "process-group" }; } catch (error) { - if (error?.code !== "ESRCH") { - try { - killImpl(pid, "SIGTERM"); - return { attempted: true, delivered: true, method: "process" }; - } catch (innerError) { - if (innerError?.code === "ESRCH") { - return { attempted: true, delivered: false, method: "process" }; - } - throw innerError; + try { + killImpl(pid, "SIGTERM"); + return { attempted: true, delivered: true, method: "process" }; + } catch (innerError) { + if (innerError?.code === "ESRCH") { + return { attempted: true, delivered: false, method: "process" }; } + throw innerError; } - - return { attempted: true, delivered: false, method: "process-group" }; } } diff --git a/plugins/codex/scripts/lib/render.mjs b/plugins/codex/scripts/lib/render.mjs index 2ec185236..e7768484a 100644 --- a/plugins/codex/scripts/lib/render.mjs +++ b/plugins/codex/scripts/lib/render.mjs @@ -126,6 +126,9 @@ function pushJobDetails(lines, job, options = {}) { if (job.summary) { lines.push(` Summary: ${job.summary}`); } + if (job.status === "failed" && job.errorMessage) { + lines.push(` Error: ${job.errorMessage}`); + } if (job.phase) { lines.push(` Phase: ${job.phase}`); } @@ -446,10 +449,11 @@ export function renderStoredJobResult(job, storedJob) { } export function renderCancelReport(job) { + const outcome = job.status === "cancelled" ? `Cancelled ${job.id}.` : `${job.id} already ${job.status}.`; const lines = [ "# Codex Cancel", "", - `Cancelled ${job.id}.`, + outcome, "" ]; diff --git a/plugins/codex/scripts/lib/state.mjs b/plugins/codex/scripts/lib/state.mjs index 2da23498f..a3471e448 100644 --- a/plugins/codex/scripts/lib/state.mjs +++ b/plugins/codex/scripts/lib/state.mjs @@ -1,8 +1,9 @@ -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import { isProcessAlive } from "./process.mjs"; import { resolveWorkspaceRoot } from "./workspace.mjs"; const STATE_VERSION = 1; @@ -11,6 +12,11 @@ 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 STATE_LOCK_FILE_NAME = ".state.lock"; +const STATE_LOCK_WAIT_MS = 5000; +const STATE_LOCK_RETRY_MS = 20; +const JOB_STATUSES = new Set(["queued", "running", "completed", "failed", "cancelled"]); +const SAFE_JOB_ID = /^(?!.*\.\.)[A-Za-z0-9][A-Za-z0-9._-]*$/; function nowIso() { return new Date().toISOString(); @@ -55,32 +61,163 @@ export function ensureStateDir(cwd) { fs.mkdirSync(resolveJobsDir(cwd), { recursive: true }); } -export function loadState(cwd) { - const stateFile = resolveStateFile(cwd); - if (!fs.existsSync(stateFile)) { - return defaultState(); +function isObject(value) { + return value && typeof value === "object" && !Array.isArray(value); +} + +function validateState(parsed) { + if (!isObject(parsed) || parsed.version !== STATE_VERSION || !isObject(parsed.config) || typeof parsed.config.stopReviewGate !== "boolean" || !Array.isArray(parsed.jobs)) { + throw new Error("invalid state schema"); + } + for (const job of parsed.jobs) { + if (!isObject(job) || typeof job.id !== "string" || !SAFE_JOB_ID.test(job.id) || !JOB_STATUSES.has(job.status)) { + throw new Error("invalid job schema"); + } } +} +export function loadState(cwd) { + const stateFile = path.resolve(resolveStateFile(cwd)); 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(); + validateState(parsed); + return parsed; + } catch (error) { + if (error?.code === "ENOENT") { + return defaultState(); + } + const detail = error instanceof Error ? error.message : String(error); + throw new Error(`Failed to read Codex Companion state at ${stateFile}: ${detail}`, { cause: error }); + } +} + +function pauseForStateLock() { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, STATE_LOCK_RETRY_MS); +} + +function readStateLockOwner(lockFile) { + let owner; + try { + owner = JSON.parse(fs.readFileSync(lockFile, "utf8")); + } catch (error) { + if (error?.code === "ENOENT") { + throw error; + } + throw new Error(`Codex Companion state lock at ${path.resolve(lockFile)} has an invalid owner.`, { cause: error }); + } + if (!isObject(owner) || !Number.isSafeInteger(owner.pid) || owner.pid <= 0 || typeof owner.token !== "string" || !owner.token || typeof owner.createdAt !== "string") { + throw new Error(`Codex Companion state lock at ${path.resolve(lockFile)} has an invalid owner.`); + } + return owner; +} + +function tryCreateStateLock(filePath, payload) { + const temporaryFile = `${filePath}.${process.pid}.${payload.token}.tmp`; + try { + fs.writeFileSync(temporaryFile, `${JSON.stringify(payload)}\n`, "utf8"); + fs.linkSync(temporaryFile, filePath); + return true; + } catch (error) { + if (error?.code === "EEXIST") { + return false; + } + throw error; + } finally { + removeFileIfExists(temporaryFile); + } +} + +function releaseStateLock(lockFile, token) { + try { + if (readStateLockOwner(lockFile).token === token) { + fs.unlinkSync(lockFile); + } + } catch (error) { + if (error?.code !== "ENOENT") { + throw error; + } + } +} + +function reapDeadStateLock(lockFile, owner) { + const reapFile = `${lockFile}.reap`; + const token = randomUUID(); + if (!tryCreateStateLock(reapFile, { pid: process.pid, token, createdAt: nowIso() })) { + return false; + } + try { + const current = readStateLockOwner(lockFile); + if (current.token === owner.token && current.pid === owner.pid && !isProcessAlive(current.pid)) { + fs.unlinkSync(lockFile); + return true; + } + } catch (error) { + if (error?.code !== "ENOENT") { + throw error; + } + } finally { + releaseStateLock(reapFile, token); + } + return false; +} + +function withStateLock(cwd, action) { + ensureStateDir(cwd); + const lockFile = path.join(resolveStateDir(cwd), STATE_LOCK_FILE_NAME); + const deadline = Date.now() + STATE_LOCK_WAIT_MS; + const token = randomUUID(); + while (true) { + if (tryCreateStateLock(lockFile, { pid: process.pid, token, createdAt: nowIso() })) { + try { + return action(); + } finally { + releaseStateLock(lockFile, token); + } + } + try { + const owner = readStateLockOwner(lockFile); + if (!isProcessAlive(owner.pid)) { + if (reapDeadStateLock(lockFile, owner)) { + continue; + } + } + } catch (error) { + if (error?.code === "ENOENT") { + continue; + } + if (Date.now() >= deadline) { + throw error; + } + } + if (Date.now() >= deadline) { + throw new Error(`Timed out waiting for Codex Companion state lock at ${path.resolve(lockFile)}.`); + } + pauseForStateLock(); + } +} + +function writeAtomicJson(filePath, value) { + const temporaryFile = `${filePath}.${process.pid}.${randomUUID()}.tmp`; + const content = `${JSON.stringify(value, null, 2)}\n`; + try { + fs.writeFileSync(temporaryFile, content, "utf8"); + fs.renameSync(temporaryFile, filePath); + } catch (error) { + try { + fs.unlinkSync(temporaryFile); + } catch { + // The temporary file may not have been created or may already have been renamed. + } + throw error; } } function pruneJobs(jobs) { - return [...jobs] - .sort((left, right) => String(right.updatedAt ?? "").localeCompare(String(left.updatedAt ?? ""))) - .slice(0, MAX_JOBS); + const sorted = [...jobs].sort((left, right) => String(right.updatedAt ?? "").localeCompare(String(left.updatedAt ?? ""))); + const active = sorted.filter((job) => job.status === "queued" || job.status === "running"); + const terminal = sorted.filter((job) => job.status !== "queued" && job.status !== "running"); + return [...active, ...terminal.slice(0, Math.max(0, MAX_JOBS - active.length))] + .sort((left, right) => String(right.updatedAt ?? "").localeCompare(String(left.updatedAt ?? ""))); } function removeFileIfExists(filePath) { @@ -89,36 +226,79 @@ function removeFileIfExists(filePath) { } } -export function saveState(cwd, state) { - const previousJobs = loadState(cwd).jobs; - ensureStateDir(cwd); - const nextJobs = pruneJobs(state.jobs ?? []); - const nextState = { +function resolveJobSidecarFile(cwd, jobId, suffix) { + return path.join(resolveJobsDir(cwd), `${jobId}${suffix}`); +} + +function markJobRemoved(cwd, jobId) { + try { + fs.closeSync(fs.openSync(resolveJobSidecarFile(cwd, jobId, ".removed"), "wx")); + } catch (error) { + if (error?.code !== "EEXIST") { + throw error; + } + } +} + +function removeJobSidecars(cwd, jobId) { + for (const suffix of [".started.json", ".admission.json", ".terminal.json"]) { + removeFileIfExists(resolveJobSidecarFile(cwd, jobId, suffix)); + } +} + +function saveStateLocked(cwd, state) { + const candidate = { version: STATE_VERSION, config: { ...defaultState().config, ...(state.config ?? {}) }, + jobs: state.jobs ?? [] + }; + validateState(candidate); + const previousJobs = loadState(cwd).jobs; + ensureStateDir(cwd); + const requestedJobs = candidate.jobs; + const nextJobs = pruneJobs(requestedJobs.filter((job) => !fs.existsSync(resolveJobSidecarFile(cwd, job.id, ".removed")))); + const nextState = { + version: STATE_VERSION, + config: candidate.config, jobs: nextJobs }; const retainedIds = new Set(nextJobs.map((job) => job.id)); - for (const job of previousJobs) { + const knownJobs = new Map(previousJobs.map((job) => [job.id, job])); + for (const job of requestedJobs) { + knownJobs.set(job.id, { ...knownJobs.get(job.id), ...job }); + } + for (const job of knownJobs.values()) { if (retainedIds.has(job.id)) { continue; } + markJobRemoved(cwd, job.id); removeJobFile(resolveJobFile(cwd, job.id)); - removeFileIfExists(job.logFile); + removeFileIfExists(resolveJobLogFile(cwd, job.id)); + removeJobSidecars(cwd, job.id); } - fs.writeFileSync(resolveStateFile(cwd), `${JSON.stringify(nextState, null, 2)}\n`, "utf8"); + writeAtomicJson(resolveStateFile(cwd), nextState); return nextState; } +export function saveState(cwd, state) { + return withStateLock(cwd, () => saveStateLocked(cwd, state)); +} + export function updateState(cwd, mutate) { - const state = loadState(cwd); - mutate(state); - return saveState(cwd, state); + return withStateLock(cwd, () => { + const state = loadState(cwd); + mutate(state); + return saveStateLocked(cwd, state); + }); +} + +export function isJobRemovedLocked(cwd, jobId) { + return withStateLock(cwd, () => fs.existsSync(resolveJobSidecarFile(cwd, jobId, ".removed"))); } export function generateJobId(prefix = "job") { @@ -166,7 +346,7 @@ export function getConfig(cwd) { export function writeJobFile(cwd, jobId, payload) { ensureStateDir(cwd); const jobFile = resolveJobFile(cwd, jobId); - fs.writeFileSync(jobFile, `${JSON.stringify(payload, null, 2)}\n`, "utf8"); + writeAtomicJson(jobFile, payload); return jobFile; } diff --git a/plugins/codex/scripts/lib/tracked-jobs.mjs b/plugins/codex/scripts/lib/tracked-jobs.mjs index 902869012..2c4c24a05 100644 --- a/plugins/codex/scripts/lib/tracked-jobs.mjs +++ b/plugins/codex/scripts/lib/tracked-jobs.mjs @@ -1,9 +1,11 @@ import fs from "node:fs"; import process from "node:process"; -import { readJobFile, resolveJobFile, resolveJobLogFile, upsertJob, writeJobFile } from "./state.mjs"; +import { isProcessAlive } from "./process.mjs"; +import { isJobRemovedLocked, listJobs, readJobFile, resolveJobFile, resolveJobLogFile, updateState, upsertJob, writeJobFile } from "./state.mjs"; export const SESSION_ID_ENV = "CODEX_COMPANION_SESSION_ID"; +export const GATE_KEY_ENV = "CODEX_COMPANION_GATE_KEY"; export function nowIso() { return new Date().toISOString(); @@ -50,8 +52,16 @@ export function appendLogBlock(logFile, title, body) { export function createJobLogFile(workspaceRoot, jobId, title) { const logFile = resolveJobLogFile(workspaceRoot, jobId); - fs.writeFileSync(logFile, "", "utf8"); - if (title) { + let created = false; + try { + fs.closeSync(fs.openSync(logFile, "wx")); + created = true; + } catch (error) { + if (error?.code !== "EEXIST") { + throw error; + } + } + if (created && title) { appendLogLine(logFile, `Starting ${title}.`); } return logFile; @@ -99,10 +109,8 @@ export function createJobProgressUpdater(workspaceRoot, jobId) { return; } - upsertJob(workspaceRoot, patch); - const jobFile = resolveJobFile(workspaceRoot, jobId); - if (!fs.existsSync(jobFile)) { + if (isJobRemoved(workspaceRoot, jobId) || readTerminalFence(workspaceRoot, jobId) || !fs.existsSync(jobFile)) { return; } @@ -139,23 +147,369 @@ function readStoredJobOrNull(workspaceRoot, jobId) { return readJobFile(jobFile); } +const TERMINAL_STATUSES = new Set(["completed", "failed", "cancelled"]); + +function resolveTerminalFenceFile(workspaceRoot, jobId) { + return resolveJobFile(workspaceRoot, jobId).replace(/\.json$/, ".terminal.json"); +} + +function resolveInitialClaimFile(workspaceRoot, jobId) { + return resolveJobFile(workspaceRoot, jobId).replace(/\.json$/, ".started.json"); +} + +function resolveRemovedFenceFile(workspaceRoot, jobId) { + return resolveJobFile(workspaceRoot, jobId).replace(/\.json$/, ".removed"); +} + +function resolveAdmissionFile(workspaceRoot, jobId) { + return resolveJobFile(workspaceRoot, jobId).replace(/\.json$/, ".admission.json"); +} + +function isJobRemoved(workspaceRoot, jobId) { + return fs.existsSync(resolveRemovedFenceFile(workspaceRoot, jobId)); +} + +export function markTrackedJobRemoved(workspaceRoot, jobId) { + claimFile(resolveRemovedFenceFile(workspaceRoot, jobId), ""); +} + +function terminalPhase(status) { + return status === "completed" ? "done" : status; +} + +export function readTerminalFence(workspaceRoot, jobId) { + const fenceFile = resolveTerminalFenceFile(workspaceRoot, jobId); + if (!fs.existsSync(fenceFile)) { + return null; + } + + try { + const parsed = JSON.parse(fs.readFileSync(fenceFile, "utf8")); + if (!parsed || typeof parsed !== "object" || !TERMINAL_STATUSES.has(parsed.status)) { + throw new Error("invalid terminal status"); + } + return { + status: parsed.status, + completedAt: typeof parsed.completedAt === "string" ? parsed.completedAt : null, + removed: parsed.removed === true + }; + } catch { + return { status: "failed", completedAt: null, corrupt: true }; + } +} + +function readInitialClaim(workspaceRoot, jobId) { + const claimFile = resolveInitialClaimFile(workspaceRoot, jobId); + if (!fs.existsSync(claimFile)) { + return null; + } + + try { + const parsed = JSON.parse(fs.readFileSync(claimFile, "utf8")); + if (!parsed || typeof parsed !== "object") { + throw new Error("invalid initial claim"); + } + if (parsed.status === "running" && Number.isFinite(parsed.pid) && typeof parsed.startedAt === "string") { + return { status: "running", pid: parsed.pid, startedAt: parsed.startedAt }; + } + if (TERMINAL_STATUSES.has(parsed.status)) { + return { + status: parsed.status, + completedAt: typeof parsed.completedAt === "string" ? parsed.completedAt : null, + removed: parsed.removed === true + }; + } + throw new Error("invalid initial claim status"); + } catch { + return { status: "failed", completedAt: null, corrupt: true }; + } +} + +function readAdmissionClaim(workspaceRoot, jobId) { + const admissionFile = resolveAdmissionFile(workspaceRoot, jobId); + if (!fs.existsSync(admissionFile)) { + return null; + } + try { + const parsed = JSON.parse(fs.readFileSync(admissionFile, "utf8")); + if (parsed?.status === "admitted") { + return { status: "admitted" }; + } + if (TERMINAL_STATUSES.has(parsed?.status)) { + return { status: parsed.status, completedAt: typeof parsed.completedAt === "string" ? parsed.completedAt : null }; + } + throw new Error("invalid admission claim"); + } catch { + return { status: "failed", completedAt: null, corrupt: true }; + } +} + +function applyTerminalFence(job, fence) { + if (!fence) { + return job; + } + return { + ...job, + status: fence.status, + phase: terminalPhase(fence.status), + pid: null, + completedAt: fence.completedAt ?? job.completedAt ?? null, + ...(fence.corrupt ? { errorMessage: "Terminal job fence is corrupt." } : {}) + }; +} + +function removedLifecycleRecord(job) { + return { ...job, status: "cancelled", phase: "cancelled", pid: null, removed: true }; +} + +function missingTrackedJobRecord(job) { + return { ...job, status: "failed", phase: "failed", pid: null, errorMessage: "Tracked job record is missing." }; +} + +function applyInitialClaim(job, claim) { + if (!claim) { + return job; + } + if (claim.status === "running") { + return { ...job, status: "running", phase: job.phase === "queued" ? "starting" : job.phase, pid: claim.pid, startedAt: claim.startedAt }; + } + return applyTerminalFence(job, claim); +} + +function claimFile(file, payload) { + try { + const descriptor = fs.openSync(file, "wx"); + try { + fs.writeFileSync(descriptor, `${JSON.stringify(payload)}\n`, "utf8"); + } finally { + fs.closeSync(descriptor); + } + return true; + } catch (error) { + if (error?.code === "EEXIST") { + return false; + } + throw error; + } +} + +function claimTerminalFence(workspaceRoot, jobId, status, completedAt, removed = false) { + const fenceFile = resolveTerminalFenceFile(workspaceRoot, jobId); + if (claimFile(fenceFile, { status, completedAt, ...(removed ? { removed: true } : {}) })) { + return { fence: { status, completedAt, removed }, claimed: true }; + } + return { fence: readTerminalFence(workspaceRoot, jobId), claimed: false }; +} + +export function readEffectiveStoredJob(workspaceRoot, jobId) { + if (isJobRemoved(workspaceRoot, jobId)) { + return null; + } + const storedJob = readStoredJobOrNull(workspaceRoot, jobId); + if (!storedJob) { + return null; + } + const initial = readInitialClaim(workspaceRoot, jobId); + const admission = initial?.status === "running" ? readAdmissionClaim(workspaceRoot, jobId) : null; + if (admission && admission.status !== "admitted") { + return applyTerminalFence(applyInitialClaim(storedJob, initial), admission); + } + const fence = initial?.status === "running" ? readTerminalFence(workspaceRoot, jobId) : null; + return applyTerminalFence(applyInitialClaim(storedJob, initial), fence); +} + +export function terminalizeTrackedJob(workspaceRoot, job, terminal) { + if (isJobRemoved(workspaceRoot, job.id)) { + return { job: removedLifecycleRecord(job), claimed: false }; + } + const completedAt = terminal.completedAt ?? nowIso(); + let initial = readInitialClaim(workspaceRoot, job.id); + if (!initial) { + const claimed = claimFile(resolveInitialClaimFile(workspaceRoot, job.id), { + status: terminal.status, + completedAt + }); + const winner = claimed ? { status: terminal.status, completedAt } : readInitialClaim(workspaceRoot, job.id); + const storedJob = readStoredJobOrNull(workspaceRoot, job.id); + if (!claimed && winner?.status !== "running") { + return { job: applyTerminalFence(storedJob ?? job, winner), claimed }; + } + if (claimed) { + const effectiveJob = applyTerminalFence({ ...(storedJob ?? job), ...terminal }, winner); + writeJobFile(workspaceRoot, job.id, effectiveJob); + upsertJob(workspaceRoot, effectiveJob); + return { job: effectiveJob, claimed }; + } + initial = winner; + } + if (initial.status !== "running") { + return { job: applyTerminalFence(readStoredJobOrNull(workspaceRoot, job.id) ?? job, initial), claimed: false }; + } + const admission = readAdmissionClaim(workspaceRoot, job.id); + if (!admission) { + const claimed = claimFile(resolveAdmissionFile(workspaceRoot, job.id), { status: terminal.status, completedAt }); + const winner = claimed ? { status: terminal.status, completedAt } : readAdmissionClaim(workspaceRoot, job.id); + if (winner?.status !== "admitted") { + return { job: applyTerminalFence(readStoredJobOrNull(workspaceRoot, job.id) ?? job, winner), claimed }; + } + } else if (admission.status !== "admitted") { + return { job: applyTerminalFence(readStoredJobOrNull(workspaceRoot, job.id) ?? job, admission), claimed: false }; + } + const { fence, claimed } = claimTerminalFence(workspaceRoot, job.id, terminal.status, completedAt, terminal.removed); + const storedJob = readStoredJobOrNull(workspaceRoot, job.id); + const effectiveJob = applyTerminalFence({ ...(storedJob ?? job), ...terminal }, fence); + + if (!claimed) { + return { job: applyTerminalFence(storedJob ?? job, fence), claimed }; + } + + writeJobFile(workspaceRoot, job.id, effectiveJob); + upsertJob(workspaceRoot, effectiveJob); + return { job: effectiveJob, claimed }; +} + +function failTrackedJob(workspaceRoot, job, errorMessage) { + return terminalizeTrackedJob(workspaceRoot, job, { + status: "failed", + phase: "failed", + errorMessage, + pid: null, + completedAt: nowIso() + }).job; +} + +export function reconcileTrackedJobs(workspaceRoot, options = {}) { + const now = options.now ?? Date.now(); + + return listJobs(workspaceRoot).flatMap((job) => { + if (isJobRemoved(workspaceRoot, job.id)) { + return []; + } + const storedJob = readStoredJobOrNull(workspaceRoot, job.id); + if (storedJob) { + job = { ...job, ...storedJob }; + } + const initial = readInitialClaim(workspaceRoot, job.id); + if (initial && initial.status !== "running") { + return fs.existsSync(resolveJobFile(workspaceRoot, job.id)) ? [applyTerminalFence(job, initial)] : []; + } + const admission = initial?.status === "running" ? readAdmissionClaim(workspaceRoot, job.id) : null; + if (admission && admission.status !== "admitted") { + return fs.existsSync(resolveJobFile(workspaceRoot, job.id)) ? [applyTerminalFence(job, admission)] : []; + } + const fence = readTerminalFence(workspaceRoot, job.id); + if (fence) { + return fs.existsSync(resolveJobFile(workspaceRoot, job.id)) ? [applyTerminalFence(job, fence)] : []; + } + job = applyInitialClaim(job, initial); + if (job.status !== "queued" && job.status !== "running") { + return [job]; + } + + if (job.status === "running" && (!Number.isSafeInteger(job.pid) || job.pid <= 0)) { + const failedJob = failTrackedJob(workspaceRoot, job, "Tracked running job has an invalid process id."); + return failedJob ? [failedJob] : []; + } + + const ageMs = now - Date.parse(job.createdAt ?? ""); + if (job.status === "queued" && !Number.isFinite(ageMs)) { + const failedJob = failTrackedJob(workspaceRoot, job, "Tracked queued job has an invalid creation time."); + return failedJob ? [failedJob] : []; + } + if (job.status === "queued" && !Number.isFinite(job.pid) && ageMs >= 5000) { + const failedJob = failTrackedJob(workspaceRoot, job, "Background worker did not start within 5 seconds."); + return failedJob ? [failedJob] : []; + } + if (Number.isFinite(job.pid) && !isProcessAlive(job.pid, { killImpl: options.killImpl })) { + const failedJob = failTrackedJob(workspaceRoot, job, "Background worker exited before completing the job."); + return failedJob ? [failedJob] : []; + } + + return [job]; + }); +} + export async function runTrackedJob(job, runner, options = {}) { + if (isJobRemoved(job.workspaceRoot, job.id)) { + return removedLifecycleRecord(job); + } + const storedJob = readStoredJobOrNull(job.workspaceRoot, job.id); + const initial = readInitialClaim(job.workspaceRoot, job.id); + if (initial && initial.status !== "running") { + return applyTerminalFence(storedJob ?? job, initial); + } + if (initial?.status === "running") { + if (job.gateKey && !storedJob && !isProcessAlive(initial.pid)) { + return terminalizeTrackedJob(job.workspaceRoot, job, { + status: "failed", + phase: "failed", + errorMessage: "Stop-gate worker exited before publishing its job record.", + pid: null, + completedAt: nowIso() + }).job; + } + const terminal = readTerminalFence(job.workspaceRoot, job.id); + if (terminal) { + return applyTerminalFence(storedJob ?? job, terminal); + } + const admission = readAdmissionClaim(job.workspaceRoot, job.id); + if (admission?.status !== "admitted") { + return applyTerminalFence(storedJob ?? job, admission); + } + return applyInitialClaim(storedJob ?? job, initial); + } + const fence = readTerminalFence(job.workspaceRoot, job.id); + if (fence) { + return applyTerminalFence(storedJob ?? job, fence); + } + if (storedJob && storedJob.status !== "queued") { + return storedJob; + } + if (!storedJob && job.request) { + return missingTrackedJobRecord(job); + } + + const startedAt = nowIso(); + const runningClaim = { status: "running", pid: process.pid, startedAt }; + if (!claimFile(resolveInitialClaimFile(job.workspaceRoot, job.id), runningClaim)) { + const winner = readInitialClaim(job.workspaceRoot, job.id); + return applyInitialClaim(storedJob ?? job, winner); + } const runningRecord = { - ...job, + ...(storedJob ?? job), status: "running", - startedAt: nowIso(), + startedAt, phase: "starting", pid: process.pid, logFile: options.logFile ?? job.logFile ?? null }; writeJobFile(job.workspaceRoot, job.id, runningRecord); upsertJob(job.workspaceRoot, runningRecord); + if (isJobRemoved(job.workspaceRoot, job.id)) { + return removedLifecycleRecord(runningRecord); + } + const terminalAfterStart = readTerminalFence(job.workspaceRoot, job.id); + if (terminalAfterStart) { + return applyTerminalFence(runningRecord, terminalAfterStart); + } + let admitted = false; + updateState(job.workspaceRoot, () => { + if (!isJobRemoved(job.workspaceRoot, job.id)) { + admitted = claimFile(resolveAdmissionFile(job.workspaceRoot, job.id), { status: "admitted" }); + } + }); + if (!admitted) { + if (isJobRemoved(job.workspaceRoot, job.id)) { + return removedLifecycleRecord(runningRecord); + } + return applyTerminalFence(runningRecord, readAdmissionClaim(job.workspaceRoot, job.id)); + } try { const execution = await runner(); const completionStatus = execution.exitStatus === 0 ? "completed" : "failed"; const completedAt = nowIso(); - writeJobFile(job.workspaceRoot, job.id, { + const terminal = terminalizeTrackedJob(job.workspaceRoot, runningRecord, { ...runningRecord, status: completionStatus, threadId: execution.threadId ?? null, @@ -166,38 +520,25 @@ export async function runTrackedJob(job, runner, options = {}) { 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; + if (terminal.claimed) { + if (isJobRemovedLocked(job.workspaceRoot, job.id)) { + return removedLifecycleRecord(runningRecord); + } + appendLogBlock(options.logFile ?? job.logFile ?? null, "Final output", execution.rendered); + return execution; + } + return terminal.job; } 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, { - ...existing, + terminalizeTrackedJob(job.workspaceRoot, runningRecord, { + ...runningRecord, status: "failed", phase: "failed", errorMessage, pid: null, completedAt, - logFile: options.logFile ?? job.logFile ?? existing.logFile ?? null - }); - upsertJob(job.workspaceRoot, { - id: job.id, - status: "failed", - phase: "failed", - pid: null, - errorMessage, - completedAt + logFile: options.logFile ?? job.logFile ?? runningRecord.logFile ?? null }); throw error; } diff --git a/plugins/codex/scripts/session-lifecycle-hook.mjs b/plugins/codex/scripts/session-lifecycle-hook.mjs index 778571e6c..d5bf63367 100644 --- a/plugins/codex/scripts/session-lifecycle-hook.mjs +++ b/plugins/codex/scripts/session-lifecycle-hook.mjs @@ -13,7 +13,8 @@ import { sendBrokerShutdown, teardownBrokerSession } from "./lib/broker-lifecycle.mjs"; -import { loadState, resolveStateFile, saveState } from "./lib/state.mjs"; +import { resolveStateFile, updateState } from "./lib/state.mjs"; +import { markTrackedJobRemoved, readEffectiveStoredJob } from "./lib/tracked-jobs.mjs"; import { TRANSCRIPT_PATH_ENV } from "./lib/claude-session-transfer.mjs"; import { resolveWorkspaceRoot } from "./lib/workspace.mjs"; @@ -50,28 +51,25 @@ function cleanupSessionJobs(cwd, sessionId) { return; } - const state = loadState(workspaceRoot); - const removedJobs = state.jobs.filter((job) => job.sessionId === sessionId); - if (removedJobs.length === 0) { - return; - } - - for (const job of removedJobs) { - const stillRunning = job.status === "queued" || job.status === "running"; - if (!stillRunning) { - continue; + const pids = []; + updateState(workspaceRoot, (state) => { + for (const job of state.jobs.filter((candidate) => candidate.sessionId === sessionId)) { + const effectiveJob = { ...job, ...(readEffectiveStoredJob(workspaceRoot, job.id) ?? {}) }; + markTrackedJobRemoved(workspaceRoot, job.id); + if (Number.isSafeInteger(effectiveJob.pid) && effectiveJob.pid > 0) { + pids.push(effectiveJob.pid); + } } + state.jobs = state.jobs.filter((job) => job.sessionId !== sessionId); + }); + + for (const pid of pids) { try { - terminateProcessTree(job.pid ?? Number.NaN); + terminateProcessTree(pid); } catch { // Ignore teardown failures during session shutdown. } } - - saveState(workspaceRoot, { - ...state, - jobs: state.jobs.filter((job) => job.sessionId !== sessionId) - }); } function handleSessionStart(input) { diff --git a/plugins/codex/scripts/stop-review-gate-hook.mjs b/plugins/codex/scripts/stop-review-gate-hook.mjs index 2346bdcf4..e885b2eea 100644 --- a/plugins/codex/scripts/stop-review-gate-hook.mjs +++ b/plugins/codex/scripts/stop-review-gate-hook.mjs @@ -4,13 +4,14 @@ import fs from "node:fs"; import process from "node:process"; import path from "node:path"; import { spawnSync } from "node:child_process"; +import { createHash } from "node:crypto"; import { fileURLToPath } from "node:url"; import { getCodexAvailability } from "./lib/codex.mjs"; import { loadPromptTemplate, interpolateTemplate } from "./lib/prompts.mjs"; -import { getConfig, listJobs } from "./lib/state.mjs"; +import { getConfig } from "./lib/state.mjs"; import { sortJobsNewestFirst } from "./lib/job-control.mjs"; -import { SESSION_ID_ENV } from "./lib/tracked-jobs.mjs"; +import { GATE_KEY_ENV, reconcileTrackedJobs, SESSION_ID_ENV } from "./lib/tracked-jobs.mjs"; import { resolveWorkspaceRoot } from "./lib/workspace.mjs"; const STOP_REVIEW_TIMEOUT_MS = 15 * 60 * 1000; @@ -46,7 +47,7 @@ function filterJobsForCurrentSession(jobs, input = {}) { } function buildStopReviewPrompt(input = {}) { - const lastAssistantMessage = String(input.last_assistant_message ?? "").trim(); + const lastAssistantMessage = String(input.last_assistant_message ?? ""); const template = loadPromptTemplate(ROOT_DIR, "stop-review-gate"); const claudeResponseBlock = lastAssistantMessage ? ["Previous Claude response:", lastAssistantMessage].join("\n") @@ -56,6 +57,37 @@ function buildStopReviewPrompt(input = {}) { }); } +function getGateKey(input = {}) { + const sessionId = input.session_id || process.env[SESSION_ID_ENV] || ""; + const lastAssistantMessage = String(input.last_assistant_message ?? ""); + if (!sessionId || !lastAssistantMessage) { + return null; + } + return createHash("sha256").update(`${sessionId}\0${lastAssistantMessage}`).digest("hex"); +} + +function gateJobNote(job) { + if (job.status === "queued" || job.status === "running") { + return `The stop-time Codex review is already ${job.status} as ${job.id}. Check /codex:status ${job.id} and use /codex:cancel ${job.id} if you want to stop it.`; + } + return `The prior stop-time Codex review ${job.id} is ${job.status}; it will not be rerun automatically. Check /codex:status ${job.id}, then run /codex:review --wait manually or bypass the gate.`; +} + +function getGateJob(jobs, gateKey) { + return gateKey ? jobs.find((job) => job.gateKey === gateKey) ?? null : null; +} + +function parseStoredGateReview(job) { + const rawOutput = job?.result?.rawOutput; + if (typeof rawOutput !== "string") { + return { + ok: false, + reason: `The completed stop-time Codex review ${job.id} has missing or corrupt cached output and will not be rerun automatically. Check /codex:status ${job.id}, then run /codex:review --wait manually or bypass the gate.` + }; + } + return parseStopReviewOutput(rawOutput); +} + function buildSetupNote(cwd) { const availability = getCodexAvailability(cwd); if (availability.available) { @@ -95,12 +127,13 @@ function parseStopReviewOutput(rawOutput) { }; } -function runStopReview(cwd, input = {}) { +function runStopReview(cwd, input = {}, gateKey = null) { const scriptPath = path.join(SCRIPT_DIR, "codex-companion.mjs"); const prompt = buildStopReviewPrompt(input); const childEnv = { ...process.env, - ...(input.session_id ? { [SESSION_ID_ENV]: input.session_id } : {}) + ...(input.session_id ? { [SESSION_ID_ENV]: input.session_id } : {}), + ...(gateKey ? { [GATE_KEY_ENV]: gateKey } : {}) }; const result = spawnSync(process.execPath, [scriptPath, "task", "--json", prompt], { cwd, @@ -129,6 +162,9 @@ function runStopReview(cwd, input = {}) { try { const payload = JSON.parse(result.stdout); + if (payload?.gateDuplicate) { + return { ok: false, reason: gateJobNote({ id: payload.jobId, status: payload.status }) }; + } return parseStopReviewOutput(payload?.rawOutput); } catch { return { @@ -145,7 +181,7 @@ function main() { const workspaceRoot = resolveWorkspaceRoot(cwd); const config = getConfig(workspaceRoot); - const jobs = sortJobsNewestFirst(filterJobsForCurrentSession(listJobs(workspaceRoot), input)); + const jobs = sortJobsNewestFirst(filterJobsForCurrentSession(reconcileTrackedJobs(workspaceRoot), input)); const runningJob = jobs.find((job) => job.status === "queued" || job.status === "running"); const runningTaskNote = runningJob ? `Codex task ${runningJob.id} is still running. Check /codex:status and use /codex:cancel ${runningJob.id} if you want to stop it before ending the session.` @@ -156,14 +192,25 @@ function main() { return; } + const gateKey = getGateKey(input); + const cachedJob = getGateJob(jobs, gateKey); + if (cachedJob) { + const review = cachedJob.status === "completed" ? parseStoredGateReview(cachedJob) : { ok: false, reason: gateJobNote(cachedJob) }; + if (!review.ok) { + emitDecision({ decision: "block", reason: runningTaskNote ? `${runningTaskNote} ${review.reason}` : review.reason }); + } else { + logNote(runningTaskNote); + } + return; + } + const setupNote = buildSetupNote(cwd); if (setupNote) { - logNote(setupNote); - logNote(runningTaskNote); + emitDecision({ decision: "block", reason: setupNote }); return; } - const review = runStopReview(cwd, input); + const review = runStopReview(cwd, input, gateKey); if (!review.ok) { emitDecision({ decision: "block", @@ -179,6 +226,8 @@ try { main(); } catch (error) { const message = error instanceof Error ? error.message : String(error); - process.stderr.write(`${message}\n`); - process.exitCode = 1; + emitDecision({ + decision: "block", + reason: `Codex stop-review gate could not safely continue: ${message}` + }); } diff --git a/tests/process.test.mjs b/tests/process.test.mjs index 80e0715b0..27a4403c7 100644 --- a/tests/process.test.mjs +++ b/tests/process.test.mjs @@ -1,7 +1,51 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; +import { isProcessAlive, terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs"; + +test("isProcessAlive treats successful and permission-denied probes as alive", () => { + assert.equal(isProcessAlive(123, { killImpl() {} }), true); + assert.equal( + isProcessAlive(123, { + killImpl() { + throw Object.assign(new Error("denied"), { code: "EPERM" }); + } + }), + true + ); +}); + +test("isProcessAlive treats a missing process as dead", () => { + assert.equal( + isProcessAlive(123, { + killImpl() { + throw Object.assign(new Error("gone"), { code: "ESRCH" }); + } + }), + false + ); +}); + +test("process helpers reject unsafe process ids", () => { + for (const pid of [0, -1, 1.5, Number.MAX_SAFE_INTEGER + 1]) { + assert.equal(isProcessAlive(pid, { killImpl() { throw new Error("must not probe"); } }), false); + assert.equal(terminateProcessTree(pid, { killImpl() { throw new Error("must not kill"); } }).attempted, false); + } +}); + +test("terminateProcessTree falls back from a missing POSIX group to its leader", () => { + const calls = []; + const outcome = terminateProcessTree(1234, { + platform: "linux", + killImpl(pid) { + calls.push(pid); + throw Object.assign(new Error("gone"), { code: "ESRCH" }); + } + }); + assert.deepEqual(calls, [-1234, 1234]); + assert.equal(outcome.delivered, false); + assert.equal(outcome.method, "process"); +}); test("terminateProcessTree uses taskkill on Windows", () => { let captured = null; diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs index 8f276835b..716a01e4b 100644 --- a/tests/runtime.test.mjs +++ b/tests/runtime.test.mjs @@ -1,4 +1,5 @@ import fs from "node:fs"; +import { createHash } from "node:crypto"; import path from "node:path"; import test from "node:test"; import assert from "node:assert/strict"; @@ -8,7 +9,8 @@ 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, readJobFile, resolveJobFile, resolveJobLogFile, resolveStateDir, saveState, upsertJob, writeJobFile } from "../plugins/codex/scripts/lib/state.mjs"; +import { createJobProgressUpdater, reconcileTrackedJobs, runTrackedJob, terminalizeTrackedJob } from "../plugins/codex/scripts/lib/tracked-jobs.mjs"; const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); const PLUGIN_ROOT = path.join(ROOT, "plugins", "codex"); @@ -28,6 +30,23 @@ async function waitFor(predicate, { timeoutMs = 5000, intervalMs = 50 } = {}) { throw new Error("Timed out waiting for condition."); } +function resolveTerminalFenceFile(workspaceRoot, jobId) { + return resolveJobFile(workspaceRoot, jobId).replace(/\.json$/, ".terminal.json"); +} + +function runHookAsync(cwd, env, input, script = STOP_HOOK) { + return new Promise((resolve, reject) => { + const child = spawn(process.execPath, [script], { cwd, env, stdio: ["pipe", "pipe", "pipe"] }); + let stdout = ""; + let stderr = ""; + child.stdout.on("data", (chunk) => { stdout += chunk; }); + child.stderr.on("data", (chunk) => { stderr += chunk; }); + child.once("error", reject); + child.once("close", (status) => resolve({ status, stdout, stderr })); + child.stdin.end(input); + }); +} + test("setup reports ready when fake codex is installed and authenticated", () => { const binDir = makeTempDir(); installFakeCodex(binDir); @@ -969,6 +988,331 @@ test("task --background enqueues a detached worker and exposes per-job status", assert.match(resultPayload.storedJob.rendered, /Handled the requested task/); }); +test("background task publishes its queued record before spawning the detached worker", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + const markerFile = path.join(makeTempDir(), "task-worker-spawned-at"); + const preloadFile = path.join(makeTempDir(), "record-task-worker-spawn.cjs"); + installFakeCodex(binDir); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + fs.writeFileSync( + preloadFile, + [ + 'const fs = require("node:fs");', + 'const childProcess = require("node:child_process");', + "const originalSpawn = childProcess.spawn;", + "childProcess.spawn = (...args) => {", + ' if (args[1]?.includes("task-worker")) fs.writeFileSync(process.env.CODEX_TASK_WORKER_SPAWN_MARKER, String(Date.now()));', + " return originalSpawn(...args);", + "};" + ].join("\n"), + "utf8" + ); + + const result = run("node", [SCRIPT, "task", "--background", "--json", "check queue publication"], { + cwd: repo, + env: { + ...buildEnv(binDir), + CODEX_TASK_WORKER_SPAWN_MARKER: markerFile, + NODE_OPTIONS: `--require ${preloadFile}` + } + }); + + assert.equal(result.status, 0, result.stderr); + const jobId = JSON.parse(result.stdout).jobId; + const jobFile = resolveJobFile(repo, jobId); + assert.equal(fs.existsSync(jobFile), true); + assert.ok(fs.statSync(jobFile).mtimeMs <= Number(fs.readFileSync(markerFile, "utf8"))); +}); + +test("terminal job fence prevents a queued worker from running", async () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-fenced", workspaceRoot, status: "queued", request: { prompt: "do not run" } }; + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + fs.writeFileSync( + resolveTerminalFenceFile(workspaceRoot, job.id), + JSON.stringify({ status: "cancelled", completedAt: "2026-08-19T12:00:00.000Z" }), + "utf8" + ); + + let runnerInvoked = false; + const result = await runTrackedJob(job, async () => { + runnerInvoked = true; + return { exitStatus: 0 }; + }); + + assert.equal(runnerInvoked, false); + assert.equal(result.status, "cancelled"); + assert.equal(readJobFile(resolveJobFile(workspaceRoot, job.id)).status, "queued"); +}); + +test("terminal initial claim prevents a late worker from publishing running state", async () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-initial-terminal", workspaceRoot, status: "queued", request: { prompt: "do not run" } }; + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + fs.writeFileSync( + resolveJobFile(workspaceRoot, job.id).replace(/\.json$/, ".started.json"), + JSON.stringify({ status: "failed", completedAt: "2026-08-19T12:00:00.000Z" }), + "utf8" + ); + + let runnerInvoked = false; + const result = await runTrackedJob(job, async () => { + runnerInvoked = true; + return { exitStatus: 0 }; + }); + + assert.equal(runnerInvoked, false); + assert.equal(result.status, "failed"); + assert.equal(readJobFile(resolveJobFile(workspaceRoot, job.id)).status, "queued"); +}); + +test("terminalizer that loses the initial claim to running wins the terminal fence", () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-stage-two", workspaceRoot, status: "queued" }; + const jobFile = resolveJobFile(workspaceRoot, job.id); + const initialFile = jobFile.replace(/\.json$/, ".started.json"); + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + const originalOpen = fs.openSync; + let intercepted = false; + fs.openSync = (file, flags, ...rest) => { + if (!intercepted && file === initialFile && flags === "wx") { + intercepted = true; + fs.writeFileSync(initialFile, JSON.stringify({ status: "running", pid: process.pid, startedAt: "2026-08-19T12:00:00.000Z" })); + } + return originalOpen(file, flags, ...rest); + }; + try { + const result = terminalizeTrackedJob(workspaceRoot, job, { status: "cancelled", completedAt: "2026-08-19T12:01:00.000Z" }); + assert.equal(result.claimed, true); + assert.equal(result.job.status, "cancelled"); + } finally { + fs.openSync = originalOpen; + } +}); + +test("terminalization and reconciliation never return null jobs after removal races", () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-mutable-gone", workspaceRoot, status: "running" }; + const jobFile = resolveJobFile(workspaceRoot, job.id); + upsertJob(workspaceRoot, job); + fs.writeFileSync( + jobFile.replace(/\.json$/, ".started.json"), + JSON.stringify({ status: "running", pid: process.pid, startedAt: "2026-08-19T12:00:00.000Z" }), + "utf8" + ); + fs.writeFileSync( + resolveTerminalFenceFile(workspaceRoot, job.id), + JSON.stringify({ status: "failed", completedAt: "2026-08-19T12:01:00.000Z" }), + "utf8" + ); + + const result = terminalizeTrackedJob(workspaceRoot, job, { status: "cancelled", completedAt: "2026-08-19T12:02:00.000Z" }); + assert.equal(result.job.status, "cancelled"); + + fs.writeFileSync(jobFile.replace(/\.json$/, ".removed"), "", "utf8"); + assert.deepEqual(reconcileTrackedJobs(workspaceRoot), []); +}); + +test("terminal admission claim after running publication prevents runner invocation", async () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-admission-terminal", workspaceRoot, status: "queued" }; + const jobFile = resolveJobFile(workspaceRoot, job.id); + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + fs.writeFileSync(jobFile.replace(/\.json$/, ".started.json"), JSON.stringify({ status: "running", pid: process.pid, startedAt: "2026-08-19T12:00:00.000Z" })); + const terminal = terminalizeTrackedJob(workspaceRoot, job, { status: "cancelled", completedAt: "2026-08-19T12:01:00.000Z" }); + let invoked = false; + const result = await runTrackedJob(job, async () => { + invoked = true; + return { exitStatus: 0 }; + }); + assert.equal(terminal.claimed, true); + assert.equal(result.status, "cancelled"); + assert.equal(invoked, false); +}); + +test("admitted running job uses the later terminal fence", () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-admitted", workspaceRoot, status: "running" }; + const jobFile = resolveJobFile(workspaceRoot, job.id); + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + fs.writeFileSync(jobFile.replace(/\.json$/, ".started.json"), JSON.stringify({ status: "running", pid: process.pid, startedAt: "2026-08-19T12:00:00.000Z" })); + fs.writeFileSync(jobFile.replace(/\.json$/, ".admission.json"), JSON.stringify({ status: "admitted" })); + const result = terminalizeTrackedJob(workspaceRoot, job, { status: "cancelled", completedAt: "2026-08-19T12:01:00.000Z" }); + assert.equal(result.claimed, true); + assert.equal(result.job.status, "cancelled"); + assert.equal(fs.existsSync(resolveTerminalFenceFile(workspaceRoot, job.id)), true); +}); + +test("corrupt admission fails closed before runner invocation", async () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-admission-corrupt", workspaceRoot, status: "queued" }; + const jobFile = resolveJobFile(workspaceRoot, job.id); + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + fs.writeFileSync(jobFile.replace(/\.json$/, ".started.json"), JSON.stringify({ status: "running", pid: process.pid, startedAt: "2026-08-19T12:00:00.000Z" })); + fs.writeFileSync(jobFile.replace(/\.json$/, ".admission.json"), "{"); + let invoked = false; + const result = await runTrackedJob(job, async () => { + invoked = true; + return { exitStatus: 0 }; + }); + assert.equal(result.status, "failed"); + assert.equal(invoked, false); +}); + +test("progress updates an unindexed mutable job without making it visible", () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-unindexed-progress", workspaceRoot, status: "running" }; + writeJobFile(workspaceRoot, job.id, job); + + createJobProgressUpdater(workspaceRoot, job.id)({ phase: "investigating" }); + + assert.deepEqual(listJobs(workspaceRoot), []); + assert.equal(readJobFile(resolveJobFile(workspaceRoot, job.id)).phase, "investigating"); +}); + +test("terminal job reconciliation wins over late worker finalization", async () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-race", workspaceRoot, status: "queued" }; + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + + let releaseRunner; + const runnerStarted = new Promise((resolve) => { + releaseRunner = resolve; + }); + const execution = runTrackedJob(job, async () => { + await runnerStarted; + return { exitStatus: 0 }; + }); + + await waitFor(() => listJobs(workspaceRoot).some((candidate) => candidate.id === job.id && candidate.status === "running")); + reconcileTrackedJobs(workspaceRoot, { + killImpl() { + throw Object.assign(new Error("gone"), { code: "ESRCH" }); + } + }); + releaseRunner(); + const result = await execution; + + assert.equal(result.status, "failed"); + assert.equal(listJobs(workspaceRoot).find((candidate) => candidate.id === job.id).status, "failed"); + assert.equal(readJobFile(resolveJobFile(workspaceRoot, job.id)).status, "failed"); +}); + +test("removed fence claimed during completion suppresses stale runner output", async () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-remove-at-terminal", workspaceRoot, status: "queued" }; + const jobFile = writeJobFile(workspaceRoot, job.id, job); + const terminalFile = resolveTerminalFenceFile(workspaceRoot, job.id); + const removedFile = jobFile.replace(/\.json$/, ".removed"); + upsertJob(workspaceRoot, job); + const originalOpen = fs.openSync; + let runnerInvoked = false; + fs.openSync = (file, flags, ...rest) => { + if (file === terminalFile && flags === "wx" && !fs.existsSync(removedFile)) { + fs.writeFileSync(removedFile, "", "utf8"); + } + return originalOpen(file, flags, ...rest); + }; + try { + const result = await runTrackedJob(job, async () => { + runnerInvoked = true; + return { exitStatus: 0, payload: { rawOutput: "ALLOW: stale" }, rendered: "ALLOW: stale" }; + }); + assert.equal(result.removed, true); + } finally { + fs.openSync = originalOpen; + } + assert.equal(runnerInvoked, true); +}); + +test("reconciliation fails a running job with an invalid process id", () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-invalid-pid", workspaceRoot, status: "running", pid: null }; + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + + const [result] = reconcileTrackedJobs(workspaceRoot); + assert.equal(result.status, "failed"); + assert.match(result.errorMessage, /invalid process id/i); +}); + +test("reconciliation fails a queued job without a valid creation time", () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-invalid-queued-time", workspaceRoot, status: "queued", pid: null, createdAt: "not-a-date" }; + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + + const [result] = reconcileTrackedJobs(workspaceRoot); + assert.equal(result.status, "failed"); + assert.match(result.errorMessage, /invalid creation time/i); +}); + +test("removed job terminal fence prevents progress and finalization from recreating it", async () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-removed", workspaceRoot, status: "queued" }; + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + const progress = createJobProgressUpdater(workspaceRoot, job.id); + + let releaseRunner; + const runnerStarted = new Promise((resolve) => { + releaseRunner = resolve; + }); + const execution = runTrackedJob(job, async () => { + await runnerStarted; + return { exitStatus: 0 }; + }); + + await waitFor(() => listJobs(workspaceRoot).some((candidate) => candidate.id === job.id && candidate.status === "running")); + fs.writeFileSync(resolveJobFile(workspaceRoot, job.id).replace(/\.json$/, ".removed"), "", "utf8"); + saveState(workspaceRoot, { config: { stopReviewGate: false }, jobs: [] }); + progress({ phase: "investigating" }); + releaseRunner(); + await execution; + + assert.deepEqual(listJobs(workspaceRoot), []); + assert.equal(fs.existsSync(resolveJobFile(workspaceRoot, job.id)), false); +}); + +test("removal between the pre-admission check and claim prevents runner invocation", async () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-remove-before-admission", workspaceRoot, status: "queued" }; + const jobFile = writeJobFile(workspaceRoot, job.id, job); + const removedFile = jobFile.replace(/\.json$/, ".removed"); + upsertJob(workspaceRoot, job); + const originalExists = fs.existsSync; + let removalChecks = 0; + let invoked = false; + fs.existsSync = (file) => { + if (file === removedFile && ++removalChecks === 2) { + fs.writeFileSync(removedFile, "", "utf8"); + return false; + } + return originalExists(file); + }; + try { + const result = await runTrackedJob(job, async () => { + invoked = true; + return { exitStatus: 0 }; + }); + assert.equal(result.removed, true); + } finally { + fs.existsSync = originalExists; + } + assert.equal(invoked, false); +}); + test("review rejects focus text because it is native-review only", () => { const repo = makeTempDir(); const binDir = makeTempDir(); @@ -1108,6 +1452,7 @@ test("status shows phases, hints, and the latest finished job", () => { kind: "review", kindLabel: "review", status: "running", + pid: process.pid, title: "Codex Review", jobClass: "review", phase: "reviewing", @@ -1251,6 +1596,7 @@ test("status preserves adversarial review kind labels", () => { id: "review-adv-live", kind: "adversarial-review", status: "running", + pid: process.pid, title: "Codex Adversarial Review", jobClass: "review", phase: "reviewing", @@ -1307,6 +1653,7 @@ test("status --wait times out cleanly when a job is still active", () => { id: "task-live", status: "running", title: "Codex Task", + pid: process.pid, logFile }, null, @@ -1325,9 +1672,11 @@ test("status --wait times out cleanly when a job is still active", () => { { id: "task-live", status: "running", + pid: process.pid, title: "Codex Task", jobClass: "task", summary: "Investigate flaky test", + pid: process.pid, logFile, createdAt: "2026-03-18T15:30:00.000Z", startedAt: "2026-03-18T15:30:01.000Z", @@ -1352,6 +1701,105 @@ test("status --wait times out cleanly when a job is still active", () => { assert.equal(payload.waitTimedOut, true); }); +test("status --wait marks a dead worker as failed before timing out", () => { + const workspace = makeTempDir(); + const stateDir = resolveStateDir(workspace); + const jobsDir = path.join(stateDir, "jobs"); + fs.mkdirSync(jobsDir, { recursive: true }); + + const logFile = path.join(jobsDir, "task-dead.log"); + fs.writeFileSync(logFile, "", "utf8"); + fs.writeFileSync( + path.join(jobsDir, "task-dead.json"), + JSON.stringify({ id: "task-dead", status: "running", title: "Codex Task", pid: 999999, logFile }, null, 2), + "utf8" + ); + fs.writeFileSync( + path.join(stateDir, "state.json"), + `${JSON.stringify( + { + version: 1, + config: { stopReviewGate: false }, + jobs: [ + { + id: "task-dead", + status: "running", + title: "Codex Task", + jobClass: "task", + pid: 999999, + logFile, + createdAt: "2026-03-18T15:30:00.000Z", + startedAt: "2026-03-18T15:30:01.000Z", + updatedAt: "2026-03-18T15:30:02.000Z" + } + ] + }, + null, + 2 + )}\n`, + "utf8" + ); + + const result = run("node", [SCRIPT, "status", "task-dead", "--wait", "--timeout-ms", "25", "--json"], { + cwd: workspace + }); + + assert.equal(result.status, 0, result.stderr); + const payload = JSON.parse(result.stdout); + assert.equal(payload.job.status, "failed"); + assert.equal(payload.waitTimedOut, false); + assert.match(payload.job.errorMessage, /worker exited/i); + + const rendered = run("node", [SCRIPT, "status", "task-dead"], { cwd: workspace }); + assert.equal(rendered.status, 0, rendered.stderr); + assert.match(rendered.stdout, /Error: Background worker exited before completing the job\./); +}); + +test("status marks a queued job past startup grace as failed", () => { + const workspace = makeTempDir(); + const stateDir = resolveStateDir(workspace); + const jobsDir = path.join(stateDir, "jobs"); + fs.mkdirSync(jobsDir, { recursive: true }); + + const logFile = path.join(jobsDir, "task-unstarted.log"); + fs.writeFileSync(logFile, "", "utf8"); + fs.writeFileSync( + path.join(jobsDir, "task-unstarted.json"), + JSON.stringify({ id: "task-unstarted", status: "queued", title: "Codex Task", logFile }, null, 2), + "utf8" + ); + fs.writeFileSync( + path.join(stateDir, "state.json"), + `${JSON.stringify( + { + version: 1, + config: { stopReviewGate: false }, + jobs: [ + { + id: "task-unstarted", + status: "queued", + title: "Codex Task", + jobClass: "task", + logFile, + createdAt: "2026-03-18T15:30:00.000Z", + updatedAt: "2026-03-18T15:30:00.000Z" + } + ] + }, + null, + 2 + )}\n`, + "utf8" + ); + + const result = run("node", [SCRIPT, "status", "task-unstarted", "--json"], { cwd: workspace }); + + assert.equal(result.status, 0, result.stderr); + const payload = JSON.parse(result.stdout); + assert.equal(payload.job.status, "failed"); + assert.match(payload.job.errorMessage, /did not start within 5 seconds/i); +}); + test("result returns the stored output for the latest finished job by default", () => { const workspace = makeTempDir(); const stateDir = resolveStateDir(workspace); @@ -1652,6 +2100,7 @@ test("cancel without a job id ignores active jobs from other Claude sessions", ( { id: "task-other", status: "running", + pid: process.pid, title: "Codex Task", jobClass: "task", sessionId: "sess-other", @@ -1689,13 +2138,26 @@ test("cancel without a job id ignores active jobs from other Claude sessions", ( assert.equal(state.jobs[0].status, "running"); }); -test("cancel with a job id can still target an active job from another Claude session", () => { +test("cancel with a job id can still target an active job from another Claude session", (t) => { const workspace = makeTempDir(); const stateDir = resolveStateDir(workspace); const jobsDir = path.join(stateDir, "jobs"); fs.mkdirSync(jobsDir, { recursive: true }); const logFile = path.join(jobsDir, "task-other.log"); + const sleeper = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { detached: true, stdio: "ignore" }); + sleeper.unref(); + t.after(() => { + try { + process.kill(-sleeper.pid, "SIGTERM"); + } catch { + try { + process.kill(sleeper.pid, "SIGTERM"); + } catch { + // Ignore missing process. + } + } + }); fs.writeFileSync(logFile, "", "utf8"); fs.writeFileSync( path.join(stateDir, "state.json"), @@ -1707,6 +2169,7 @@ test("cancel with a job id can still target an active job from another Claude se { id: "task-other", status: "running", + pid: sleeper.pid, title: "Codex Task", jobClass: "task", sessionId: "sess-other", @@ -1737,6 +2200,64 @@ test("cancel with a job id can still target an active job from another Claude se assert.equal(state.jobs[0].status, "cancelled"); }); +test("cancel reports a completed first terminal outcome without killing the worker", async (t) => { + const workspace = makeTempDir(); + const job = { id: "task-cancel-race", status: "running", title: "Codex Task", jobClass: "task" }; + const sleeper = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { detached: true, stdio: "ignore" }); + sleeper.unref(); + t.after(() => { + try { + process.kill(-sleeper.pid, "SIGTERM"); + } catch { + process.kill(sleeper.pid, "SIGTERM"); + } + }); + const preloadFile = path.join(makeTempDir(), "complete-before-cancel-claim.cjs"); + const jobFile = resolveJobFile(workspace, job.id); + const terminalFile = jobFile.replace(/\.json$/, ".terminal.json"); + writeJobFile(workspace, job.id, { ...job, pid: sleeper.pid }); + upsertJob(workspace, { ...job, pid: sleeper.pid }); + fs.writeFileSync( + jobFile.replace(/\.json$/, ".started.json"), + JSON.stringify({ status: "running", pid: sleeper.pid, startedAt: "2026-08-19T12:00:00.000Z" }), + "utf8" + ); + fs.writeFileSync(jobFile.replace(/\.json$/, ".admission.json"), JSON.stringify({ status: "admitted" }), "utf8"); + fs.writeFileSync( + preloadFile, + [ + 'const fs = require("node:fs");', + "const originalOpen = fs.openSync;", + "let armed = true;", + "fs.openSync = (file, flags, ...rest) => {", + " if (armed && file === process.env.CODEX_CANCEL_RACE_FENCE && flags === \"wx\") {", + " armed = false; fs.writeFileSync(file, JSON.stringify({ status: \"completed\", completedAt: \"2026-08-19T12:01:00.000Z\" }));", + " }", + " return originalOpen(file, flags, ...rest);", + "};" + ].join("\n"), + "utf8" + ); + + const result = run("node", [SCRIPT, "cancel", job.id, "--json"], { + cwd: workspace, + env: { ...process.env, CODEX_CANCEL_RACE_FENCE: terminalFile, NODE_OPTIONS: `--require ${preloadFile}` } + }); + + assert.equal(result.status, 0, result.stderr); + assert.equal(JSON.parse(result.stdout).status, "completed"); + assert.doesNotMatch(result.stdout, /Cancelled/i); + fs.unlinkSync(terminalFile); + const rendered = run("node", [SCRIPT, "cancel", job.id], { + cwd: workspace, + env: { ...process.env, CODEX_CANCEL_RACE_FENCE: terminalFile, NODE_OPTIONS: `--require ${preloadFile}` } + }); + assert.equal(rendered.status, 0, rendered.stderr); + assert.match(rendered.stdout, /already completed/i); + assert.doesNotMatch(rendered.stdout, /Cancelled/i); + process.kill(sleeper.pid, 0); +}); + test("cancel sends turn interrupt to the shared app-server before killing a brokered task", async () => { const repo = makeTempDir(); const binDir = makeTempDir(); @@ -1760,8 +2281,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 job = state.jobs.find((candidate) => candidate.id === jobId); + const job = readJobFile(resolveJobFile(repo, jobId)); if (job?.status === "running" && job.threadId && job.turnId) { return job; } @@ -1812,9 +2332,9 @@ test("session end fully cleans up jobs for the ending session", async (t) => { const jobsDir = path.join(stateDir, "jobs"); fs.mkdirSync(jobsDir, { recursive: true }); - const completedLog = path.join(jobsDir, "completed.log"); - const runningLog = path.join(jobsDir, "running.log"); - const otherSessionLog = path.join(jobsDir, "other.log"); + const completedLog = resolveJobLogFile(repo, "review-completed"); + const runningLog = resolveJobLogFile(repo, "review-running"); + const otherSessionLog = resolveJobLogFile(repo, "review-other"); const completedJobFile = path.join(jobsDir, "review-completed.json"); const runningJobFile = path.join(jobsDir, "review-running.json"); const otherJobFile = path.join(jobsDir, "review-other.json"); @@ -1831,6 +2351,11 @@ test("session end fully cleans up jobs for the ending session", async (t) => { }); sleeper.unref(); fs.writeFileSync(runningJobFile, JSON.stringify({ id: "review-running" }, null, 2), "utf8"); + fs.writeFileSync( + runningJobFile.replace(/\.json$/, ".started.json"), + JSON.stringify({ status: "running", pid: sleeper.pid, startedAt: "2026-08-19T12:00:00.000Z" }), + "utf8" + ); t.after(() => { try { @@ -1862,10 +2387,10 @@ test("session end fully cleans up jobs for the ending session", async (t) => { }, { id: "review-running", - status: "running", + status: "queued", title: "Codex Review", sessionId: "sess-current", - pid: sleeper.pid, + pid: null, logFile: runningLog, createdAt: "2026-03-18T15:32:00.000Z", updatedAt: "2026-03-18T15:33:00.000Z" @@ -1905,7 +2430,12 @@ test("session end fully cleans up jobs for the ending session", async (t) => { assert.equal(fs.existsSync(otherJobFile), true); assert.deepEqual( fs.readdirSync(path.dirname(otherJobFile)).sort(), - [path.basename(otherJobFile), path.basename(otherSessionLog)].sort() + [ + path.basename(otherJobFile), + path.basename(otherSessionLog), + "review-completed.removed", + "review-running.removed" + ].sort() ); await waitFor(() => { @@ -1921,6 +2451,82 @@ test("session end fully cleans up jobs for the ending session", async (t) => { assert.deepEqual(state.jobs.map((job) => job.id), ["review-other"]); const otherJob = state.jobs[0]; assert.equal(otherJob.logFile, otherSessionLog); + saveState(repo, state); + assert.deepEqual( + fs.readdirSync(jobsDir).sort(), + [ + path.basename(otherJobFile), + path.basename(otherSessionLog), + "review-completed.removed", + "review-running.removed" + ].sort() + ); +}); + +test("session end preserves an other-session job added after its stale snapshot", async () => { + const repo = makeTempDir(); + initGitRepo(repo); + const stateDir = resolveStateDir(repo); + const jobsDir = path.join(stateDir, "jobs"); + fs.mkdirSync(jobsDir, { recursive: true }); + const staleJob = { id: "stale-current", status: "completed", sessionId: "sess-current", logFile: path.join(jobsDir, "stale-current.log") }; + writeJobFile(repo, staleJob.id, staleJob); + fs.writeFileSync(staleJob.logFile, "stale\n", "utf8"); + saveState(repo, { config: { stopReviewGate: false }, jobs: [staleJob] }); + + const marker = path.join(makeTempDir(), "stale-read"); + const release = path.join(path.dirname(marker), "release"); + const preload = path.join(path.dirname(marker), "pause-state-read.cjs"); + fs.writeFileSync(preload, [ + 'const fs = require("node:fs");', + 'const original = fs.readFileSync;', + 'let paused = false;', + 'fs.readFileSync = function(file, ...args) {', + ' const value = original.call(this, file, ...args);', + ' if (!paused && file === process.env.CODEX_STALE_STATE_FILE) {', + ' paused = true; fs.writeFileSync(process.env.CODEX_STALE_MARKER, "paused");', + ' while (!fs.existsSync(process.env.CODEX_STALE_RELEASE)) Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 10);', + ' }', + ' return value;', + '};' + ].join("\n"), "utf8"); + fs.writeFileSync(marker, "ready", "utf8"); + fs.unlinkSync(marker); + + const hook = runHookAsync(repo, { + ...process.env, + CODEX_COMPANION_SESSION_ID: "sess-current", + CODEX_STALE_STATE_FILE: path.join(stateDir, "state.json"), + CODEX_STALE_MARKER: marker, + CODEX_STALE_RELEASE: release, + NODE_OPTIONS: `--require ${preload}` + }, JSON.stringify({ hook_event_name: "SessionEnd", session_id: "sess-current", cwd: repo }), SESSION_HOOK); + await waitFor(() => fs.existsSync(marker)); + + const freshJob = { id: "fresh-other", status: "completed", sessionId: "sess-other", logFile: path.join(jobsDir, "fresh-other.log") }; + const stateModule = path.join(PLUGIN_ROOT, "scripts", "lib", "state.mjs"); + let writerDone = false; + const writer = new Promise((resolve, reject) => { + const child = spawn(process.execPath, ["--input-type=module", "--eval", [ + 'import fs from "node:fs";', + `import { upsertJob, writeJobFile } from ${JSON.stringify(stateModule)};`, + `const job = ${JSON.stringify(freshJob)};`, + `writeJobFile(${JSON.stringify(repo)}, job.id, job);`, + 'fs.writeFileSync(job.logFile, "fresh\\n", "utf8");', + `upsertJob(${JSON.stringify(repo)}, job);` + ].join("\n")], { stdio: "ignore" }); + child.once("error", reject); + child.once("close", (status) => { writerDone = true; resolve(status); }); + }); + await new Promise((resolve) => setTimeout(resolve, 50)); + assert.equal(writerDone, false); + fs.writeFileSync(release, "go", "utf8"); + const result = await hook; + assert.equal(result.status, 0, result.stderr); + assert.equal(await writer, 0); + assert.deepEqual(listJobs(repo).map((job) => job.id), ["fresh-other"]); + assert.equal(fs.existsSync(resolveJobFile(repo, freshJob.id)), true); + assert.equal(fs.existsSync(freshJob.logFile), true); }); test("stop hook runs a stop-time review task and blocks on findings when the review gate is enabled", () => { @@ -1979,6 +2585,255 @@ test("stop hook runs a stop-time review task and blocks on findings when the rev assert.match(status.stdout, /Codex Stop Gate Review/); }); +test("stop hook reuses the completed review for the same Claude response", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + const fakeStatePath = path.join(binDir, "fake-codex-state.json"); + installFakeCodex(binDir); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + const env = buildEnv(binDir); + const input = JSON.stringify({ + cwd: repo, + session_id: "sess-stop-idempotent", + last_assistant_message: "I completed the exact same change." + }); + + const setup = run("node", [SCRIPT, "setup", "--enable-review-gate", "--json"], { cwd: repo, env }); + assert.equal(setup.status, 0, setup.stderr); + + const first = run("node", [STOP_HOOK], { cwd: repo, env, input }); + assert.equal(first.status, 0, first.stderr); + const nextTurnId = JSON.parse(fs.readFileSync(fakeStatePath, "utf8")).nextTurnId; + + const second = run("node", [STOP_HOOK], { cwd: repo, env, input }); + assert.equal(second.status, 0, second.stderr); + assert.deepEqual(JSON.parse(second.stdout), JSON.parse(first.stdout)); + assert.equal(JSON.parse(fs.readFileSync(fakeStatePath, "utf8")).nextTurnId, nextTurnId); +}); + +test("stop hook keeps cached ALLOW decisions silent", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + const fakeStatePath = path.join(binDir, "fake-codex-state.json"); + installFakeCodex(binDir, "adversarial-clean"); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + const env = buildEnv(binDir); + const input = JSON.stringify({ + cwd: repo, + session_id: "sess-stop-cached-allow", + last_assistant_message: "I completed a clean change." + }); + + const setup = run("node", [SCRIPT, "setup", "--enable-review-gate", "--json"], { cwd: repo, env }); + assert.equal(setup.status, 0, setup.stderr); + const first = run("node", [STOP_HOOK], { cwd: repo, env, input }); + assert.equal(first.status, 0, first.stderr); + assert.equal(first.stdout.trim(), ""); + const nextTurnId = JSON.parse(fs.readFileSync(fakeStatePath, "utf8")).nextTurnId; + const second = run(process.execPath, [STOP_HOOK], { cwd: repo, env: { ...env, PATH: "" }, input }); + assert.equal(second.status, 0, second.stderr); + assert.equal(second.stdout.trim(), ""); + assert.equal(JSON.parse(fs.readFileSync(fakeStatePath, "utf8")).nextTurnId, nextTurnId); +}); + +test("stop hook reuses a cached BLOCK before checking Codex availability", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir); + initGitRepo(repo); + const env = buildEnv(binDir); + const input = JSON.stringify({ cwd: repo, session_id: "sess-stop-cached-block", last_assistant_message: "A blocked change." }); + const setup = run("node", [SCRIPT, "setup", "--enable-review-gate", "--json"], { cwd: repo, env }); + assert.equal(setup.status, 0, setup.stderr); + const first = run("node", [STOP_HOOK], { cwd: repo, env, input }); + assert.equal(first.status, 0, first.stderr); + const second = run(process.execPath, [STOP_HOOK], { cwd: repo, env: { ...env, PATH: "" }, input }); + assert.equal(second.status, 0, second.stderr); + assert.deepEqual(JSON.parse(second.stdout), JSON.parse(first.stdout)); +}); + +test("stop hook blocks every matching non-reusable gate job without starting a turn", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + const fakeStatePath = path.join(binDir, "fake-codex-state.json"); + installFakeCodex(binDir); + initGitRepo(repo); + const env = buildEnv(binDir); + const sessionId = "sess-stop-lifecycle"; + const message = "Exact lifecycle response."; + const gateKey = createHash("sha256").update(`${sessionId}\0${message}`).digest("hex"); + const jobId = `gate-${gateKey}`; + const setup = run("node", [SCRIPT, "setup", "--enable-review-gate", "--json"], { cwd: repo, env }); + assert.equal(setup.status, 0, setup.stderr); + + for (const [status, cachedResult] of [ + ["queued", undefined], + ["running", undefined], + ["failed", undefined], + ["cancelled", undefined], + ["completed", undefined], + ["completed", { rawOutput: null }], + ["completed", "corrupt"] + ]) { + saveState(repo, { + version: 1, + config: { stopReviewGate: true }, + jobs: [{ id: jobId, gateKey, sessionId, status, title: "Codex Stop Gate Review", ...(status === "queued" ? { createdAt: new Date().toISOString() } : {}), ...(status === "running" ? { pid: process.pid } : {}), ...(cachedResult === undefined ? {} : { result: cachedResult }) }] + }); + const beforeTurns = fs.existsSync(fakeStatePath) ? JSON.parse(fs.readFileSync(fakeStatePath, "utf8")).nextTurnId : 1; + const hookResult = run("node", [STOP_HOOK], { cwd: repo, env, input: JSON.stringify({ cwd: repo, session_id: sessionId, last_assistant_message: message }) }); + assert.equal(hookResult.status, 0, hookResult.stderr); + assert.equal(JSON.parse(hookResult.stdout).decision, "block"); + const reason = JSON.parse(hookResult.stdout).reason; + assert.match(reason, new RegExp(jobId)); + if (status === "queued" || status === "running") { + assert.match(reason, new RegExp(`/codex:cancel ${jobId}`)); + } else { + assert.doesNotMatch(reason, /\/codex:cancel/); + assert.match(reason, /\/codex:status/); + assert.match(reason, /manual|bypass/i); + } + const afterTurns = fs.existsSync(fakeStatePath) ? JSON.parse(fs.readFileSync(fakeStatePath, "utf8")).nextTurnId : 1; + assert.equal(afterTurns, beforeTurns); + } +}); + +test("stop hook blocks a parsed state with an invalid schema", () => { + const repo = makeTempDir(); + initGitRepo(repo); + const stateDir = resolveStateDir(repo); + fs.mkdirSync(stateDir, { recursive: true }); + fs.writeFileSync(path.join(stateDir, "state.json"), JSON.stringify({ version: 1, config: {}, jobs: [] }), "utf8"); + + const result = run("node", [STOP_HOOK], { + cwd: repo, + input: JSON.stringify({ cwd: repo, session_id: "sess-invalid-state", last_assistant_message: "done" }) + }); + assert.equal(result.status, 0, result.stderr); + const decision = JSON.parse(result.stdout); + assert.equal(decision.decision, "block"); + assert.match(decision.reason, /could not safely continue.*invalid state schema/i); +}); + +test("stop gate blocks when removal wins during foreground completion", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "adversarial-clean"); + initGitRepo(repo); + const sessionId = "sess-stop-terminal-race"; + const message = "Clean review that loses the terminal race."; + const gateKey = createHash("sha256").update(`${sessionId}\0${message}`).digest("hex"); + const jobId = `gate-${gateKey}`; + const terminalFile = resolveTerminalFenceFile(repo, jobId); + const removedFile = terminalFile.replace(/\.terminal\.json$/, ".removed"); + const preload = path.join(makeTempDir(), "terminal-race.cjs"); + fs.writeFileSync(preload, [ + 'const fs = require("node:fs");', + 'const original = fs.openSync;', + 'let armed = true;', + 'fs.openSync = (file, flags, ...rest) => {', + ' if (armed && file === process.env.CODEX_FOREGROUND_TERMINAL_FENCE && flags === "wx") {', + ' armed = false; fs.writeFileSync(process.env.CODEX_FOREGROUND_REMOVED_FENCE, "");', + ' }', + ' return original(file, flags, ...rest);', + '};' + ].join("\n"), "utf8"); + const env = { + ...buildEnv(binDir), + CODEX_FOREGROUND_TERMINAL_FENCE: terminalFile, + CODEX_FOREGROUND_REMOVED_FENCE: removedFile, + NODE_OPTIONS: `--require ${preload}` + }; + assert.equal(run("node", [SCRIPT, "setup", "--enable-review-gate", "--json"], { cwd: repo, env }).status, 0); + + const result = run("node", [STOP_HOOK], { + cwd: repo, + env, + input: JSON.stringify({ cwd: repo, session_id: sessionId, last_assistant_message: message }) + }); + assert.equal(result.status, 0, result.stderr); + const decision = JSON.parse(result.stdout); + assert.equal(decision.decision, "block"); + assert.match(decision.reason, new RegExp(jobId)); + assert.doesNotMatch(result.stdout, /ALLOW:/); +}); + +test("stop hook without an assistant message runs fresh reviews", () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + const fakeStatePath = path.join(binDir, "fake-codex-state.json"); + installFakeCodex(binDir); + initGitRepo(repo); + const env = buildEnv(binDir); + const input = JSON.stringify({ cwd: repo, session_id: "sess-stop-empty-message", last_assistant_message: "" }); + const setup = run("node", [SCRIPT, "setup", "--enable-review-gate", "--json"], { cwd: repo, env }); + assert.equal(setup.status, 0, setup.stderr); + assert.equal(run("node", [STOP_HOOK], { cwd: repo, env, input }).status, 0); + const nextTurnId = JSON.parse(fs.readFileSync(fakeStatePath, "utf8")).nextTurnId; + assert.equal(run("node", [STOP_HOOK], { cwd: repo, env, input }).status, 0); + assert.equal(JSON.parse(fs.readFileSync(fakeStatePath, "utf8")).nextTurnId, nextTurnId + 1); +}); + +test("concurrent stop hooks start exactly one review for the same Claude response", async () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + const fakeStatePath = path.join(binDir, "fake-codex-state.json"); + installFakeCodex(binDir); + initGitRepo(repo); + fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); + run("git", ["add", "README.md"], { cwd: repo }); + run("git", ["commit", "-m", "init"], { cwd: repo }); + const env = buildEnv(binDir); + const input = JSON.stringify({ + cwd: repo, + session_id: "sess-stop-concurrent", + last_assistant_message: "I completed this concurrent change." + }); + + const setup = run("node", [SCRIPT, "setup", "--enable-review-gate", "--json"], { cwd: repo, env }); + assert.equal(setup.status, 0, setup.stderr); + + const [first, second] = await Promise.all([runHookAsync(repo, env, input), runHookAsync(repo, env, input)]); + assert.equal(first.status, 0, first.stderr); + assert.equal(second.status, 0, second.stderr); + assert.equal(JSON.parse(first.stdout).decision, "block"); + assert.equal(JSON.parse(second.stdout).decision, "block"); + assert.equal(JSON.parse(fs.readFileSync(fakeStatePath, "utf8")).nextTurnId, 2); + const jobs = JSON.parse(fs.readFileSync(path.join(resolveStateDir(repo), "state.json"), "utf8")).jobs; + assert.equal(jobs.filter((job) => job.title === "Codex Stop Gate Review").length, 1); +}); + +test("concurrent stop hooks do not truncate or duplicate the winner log", async () => { + const repo = makeTempDir(); + const binDir = makeTempDir(); + installFakeCodex(binDir, "slow-task"); + initGitRepo(repo); + const env = buildEnv(binDir); + const sessionId = "sess-stop-log-race"; + const message = "Concurrent log response."; + const gateKey = createHash("sha256").update(`${sessionId}\0${message}`).digest("hex"); + const logFile = resolveJobLogFile(repo, `gate-${gateKey}`); + const input = JSON.stringify({ cwd: repo, session_id: sessionId, last_assistant_message: message }); + const setup = run("node", [SCRIPT, "setup", "--enable-review-gate", "--json"], { cwd: repo, env }); + assert.equal(setup.status, 0, setup.stderr); + + const first = runHookAsync(repo, env, input); + await waitFor(() => fs.existsSync(logFile) && fs.readFileSync(logFile, "utf8").includes("Starting Codex Stop Gate Review.")); + fs.appendFileSync(logFile, "winner-marker\n", "utf8"); + const second = runHookAsync(repo, env, input); + await Promise.all([first, second]); + + const log = fs.readFileSync(logFile, "utf8"); + assert.match(log, /winner-marker/); + assert.equal((log.match(/Starting Codex Stop Gate Review\./g) ?? []).length, 1); +}); + test("stop hook logs running tasks to stderr without blocking when the review gate is disabled", () => { const repo = makeTempDir(); initGitRepo(repo); @@ -2005,6 +2860,7 @@ test("stop hook logs running tasks to stderr without blocking when the review ga { id: "task-live", status: "running", + pid: process.pid, title: "Codex Task", jobClass: "task", sessionId: "sess-current", @@ -2061,7 +2917,7 @@ test("stop hook allows the stop when the review gate is enabled and the stop-tim assert.equal(allowed.stdout.trim(), ""); }); -test("stop hook does not block when Codex is unavailable even if the review gate is enabled", () => { +test("stop hook blocks when Codex is unavailable and the review gate is enabled", () => { const repo = makeTempDir(); initGitRepo(repo); fs.writeFileSync(path.join(repo, "README.md"), "hello\n"); @@ -2073,7 +2929,7 @@ test("stop hook does not block when Codex is unavailable even if the review gate }); assert.equal(setup.status, 0, setup.stderr); - const allowed = run(process.execPath, [STOP_HOOK], { + const blocked = run(process.execPath, [STOP_HOOK], { cwd: repo, env: { ...process.env, @@ -2082,10 +2938,30 @@ test("stop hook does not block when Codex is unavailable even if the review gate input: JSON.stringify({ cwd: repo }) }); - assert.equal(allowed.status, 0, allowed.stderr); - assert.equal(allowed.stdout.trim(), ""); - assert.match(allowed.stderr, /Codex is not set up for the review gate/i); - assert.match(allowed.stderr, /Run \/codex:setup/i); + assert.equal(blocked.status, 0, blocked.stderr); + assert.equal(JSON.parse(blocked.stdout).decision, "block"); + assert.match(JSON.parse(blocked.stdout).reason, /not set up/i); +}); + +test("stop hook blocks on corrupt state instead of silently allowing", () => { + const repo = makeTempDir(); + initGitRepo(repo); + const stateFile = path.join(resolveStateDir(repo), "state.json"); + fs.mkdirSync(path.dirname(stateFile), { recursive: true }); + fs.writeFileSync(stateFile, "{invalid json", "utf8"); + + const blocked = run(process.execPath, [STOP_HOOK], { + cwd: repo, + env: process.env, + input: JSON.stringify({ cwd: repo }) + }); + + assert.equal(blocked.status, 0, blocked.stderr); + const payload = JSON.parse(blocked.stdout); + assert.equal(payload.decision, "block"); + assert.match(payload.reason, /Failed to read Codex Companion state at/); + assert.match(payload.reason, new RegExp(stateFile.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"))); + assert.match(payload.reason, /Unexpected|Expected/); }); test("stop hook runs the actual task when auth status looks stale", () => { diff --git a/tests/state.test.mjs b/tests/state.test.mjs index 0f8f57cea..eaa605dfc 100644 --- a/tests/state.test.mjs +++ b/tests/state.test.mjs @@ -5,7 +5,17 @@ import test from "node:test"; import assert from "node:assert/strict"; import { makeTempDir } from "./helpers.mjs"; -import { resolveJobFile, resolveJobLogFile, resolveStateDir, resolveStateFile, saveState } from "../plugins/codex/scripts/lib/state.mjs"; +import { + loadState, + listJobs, + resolveJobFile, + resolveJobLogFile, + resolveStateDir, + resolveStateFile, + saveState, + upsertJob, + writeJobFile +} from "../plugins/codex/scripts/lib/state.mjs"; test("resolveStateDir uses a temp-backed per-workspace directory", () => { const workspace = makeTempDir(); @@ -40,6 +50,100 @@ test("resolveStateDir uses CLAUDE_PLUGIN_DATA when it is provided", () => { } }); +test("loadState rejects invalid state JSON with its absolute path and parse detail", () => { + const workspace = makeTempDir(); + const stateFile = resolveStateFile(workspace); + fs.mkdirSync(path.dirname(stateFile), { recursive: true }); + fs.writeFileSync(stateFile, "{invalid json", "utf8"); + + assert.throws( + () => loadState(workspace), + (error) => + error instanceof Error && + error.message.includes(stateFile) && + error.message.startsWith("Failed to read Codex Companion state at ") && + /Unexpected|Expected/.test(error.message) + ); +}); + +test("loadState rejects parsed state with an invalid schema", () => { + const workspace = makeTempDir(); + const stateFile = resolveStateFile(workspace); + fs.mkdirSync(path.dirname(stateFile), { recursive: true }); + fs.writeFileSync(stateFile, JSON.stringify({ version: 1, config: { stopReviewGate: "yes" }, jobs: {} }), "utf8"); + + assert.throws(() => loadState(workspace), /Failed to read Codex Companion state.*invalid state schema/); +}); + +test("state and job writes leave valid JSON without sibling temporary artifacts", () => { + const workspace = makeTempDir(); + const stateFile = resolveStateFile(workspace); + const jobFile = writeJobFile(workspace, "job-atomic", { id: "job-atomic", status: "queued" }); + + saveState(workspace, { + config: { stopReviewGate: true }, + jobs: [{ id: "job-atomic", status: "queued" }] + }); + + assert.equal(JSON.parse(fs.readFileSync(stateFile, "utf8")).config.stopReviewGate, true); + assert.deepEqual(JSON.parse(fs.readFileSync(jobFile, "utf8")), { id: "job-atomic", status: "queued" }); + assert.deepEqual( + fs.readdirSync(path.dirname(stateFile)).filter((entry) => entry.startsWith(`${path.basename(stateFile)}.`)), + [] + ); + assert.deepEqual( + fs.readdirSync(path.dirname(jobFile)).filter((entry) => entry.startsWith(`${path.basename(jobFile)}.`)), + [] + ); +}); + +test("saveState reaps a lock left by a dead owner", () => { + const workspace = makeTempDir(); + const stateDir = resolveStateDir(workspace); + fs.mkdirSync(stateDir, { recursive: true }); + fs.writeFileSync(path.join(stateDir, ".state.lock"), JSON.stringify({ pid: 999999, token: "dead-owner", createdAt: "2026-08-19T12:00:00.000Z" }), "utf8"); + + saveState(workspace, { config: { stopReviewGate: false }, jobs: [] }); + + assert.equal(fs.existsSync(path.join(stateDir, ".state.lock")), false); +}); + +test("saveState bounds waiting when a dead lock cannot acquire its reap guard", () => { + const workspace = makeTempDir(); + const stateDir = resolveStateDir(workspace); + fs.mkdirSync(stateDir, { recursive: true }); + fs.writeFileSync(path.join(stateDir, ".state.lock"), JSON.stringify({ pid: 999999, token: "dead-owner", createdAt: "2026-08-19T12:00:00.000Z" }), "utf8"); + fs.writeFileSync(path.join(stateDir, ".state.lock.reap"), JSON.stringify({ pid: process.pid, token: "busy-reaper", createdAt: "2026-08-19T12:00:00.000Z" }), "utf8"); + const originalNow = Date.now; + let calls = 0; + Date.now = () => (calls++ === 0 ? 0 : 5001); + try { + assert.throws(() => saveState(workspace, { config: { stopReviewGate: false }, jobs: [] }), /Timed out waiting for Codex Companion state lock/); + } finally { + Date.now = originalNow; + } +}); + +test("loadState rejects unsafe job ids", () => { + const workspace = makeTempDir(); + const stateFile = resolveStateFile(workspace); + fs.mkdirSync(path.dirname(stateFile), { recursive: true }); + fs.writeFileSync(stateFile, JSON.stringify({ version: 1, config: { stopReviewGate: false }, jobs: [{ id: "../escape", status: "completed" }] }), "utf8"); + assert.throws(() => loadState(workspace), /invalid job schema/); +}); + +test("saveState rejects an unsafe requested job id before touching its path", () => { + const workspace = makeTempDir(); + const sentinel = path.join(workspace, "sentinel"); + fs.writeFileSync(sentinel, "keep", "utf8"); + + assert.throws( + () => saveState(workspace, { config: { stopReviewGate: false }, jobs: [{ id: "../sentinel", status: "completed" }] }), + /invalid job schema/ + ); + assert.equal(fs.readFileSync(sentinel, "utf8"), "keep"); +}); + test("saveState prunes dropped job artifacts when indexed jobs exceed the cap", () => { const workspace = makeTempDir(); const stateFile = resolveStateFile(workspace); @@ -52,6 +156,11 @@ test("saveState prunes dropped job artifacts when indexed jobs exceed the cap", const jobFile = resolveJobFile(workspace, jobId); fs.writeFileSync(logFile, `log ${jobId}\n`, "utf8"); fs.writeFileSync(jobFile, JSON.stringify({ id: jobId, status: "completed" }, null, 2), "utf8"); + if (jobId === "job-0") { + fs.writeFileSync(jobFile.replace(/\.json$/, ".started.json"), JSON.stringify({ status: "running", pid: 999999 }), "utf8"); + fs.writeFileSync(jobFile.replace(/\.json$/, ".admission.json"), JSON.stringify({ status: "admitted" }), "utf8"); + fs.writeFileSync(jobFile.replace(/\.json$/, ".terminal.json"), JSON.stringify({ status: "completed" }), "utf8"); + } return { id: jobId, status: "completed", @@ -100,6 +209,63 @@ test("saveState prunes dropped job artifacts when indexed jobs exceed the cap", fs.readdirSync(jobsDir).sort(), Array.from({ length: 50 }, (_, index) => `job-${index + 1}`) .flatMap((jobId) => [`${jobId}.json`, `${jobId}.log`]) + .concat("job-0.removed") .sort() ); }); + +test("saveState retains a removal fence for a pruned job before it starts", () => { + const workspace = makeTempDir(); + const job = { id: "job-prestart", status: "queued", logFile: resolveJobLogFile(workspace, "job-prestart") }; + writeJobFile(workspace, job.id, job); + fs.writeFileSync(job.logFile, "queued\n", "utf8"); + saveState(workspace, { config: { stopReviewGate: false }, jobs: [job] }); + + saveState(workspace, { config: { stopReviewGate: false }, jobs: [] }); + + const jobFile = resolveJobFile(workspace, job.id); + assert.equal(fs.existsSync(jobFile), false); + assert.equal(fs.existsSync(job.logFile), false); + assert.equal(fs.existsSync(jobFile.replace(/\.json$/, ".removed")), true); +}); + +test("saveState does not delete a noncanonical job log path", () => { + const workspace = makeTempDir(); + const externalLog = path.join(makeTempDir(), "outside.log"); + const job = { id: "job-external-log", status: "completed", logFile: externalLog }; + writeJobFile(workspace, job.id, job); + fs.writeFileSync(externalLog, "keep\n", "utf8"); + saveState(workspace, { config: { stopReviewGate: false }, jobs: [job] }); + + saveState(workspace, { config: { stopReviewGate: false }, jobs: [] }); + + assert.equal(fs.existsSync(externalLog), true); +}); + +test("saveState retains active jobs beyond the terminal history cap", () => { + const workspace = makeTempDir(); + const jobs = [ + { id: "active-old", status: "running", pid: process.pid, updatedAt: "2020-01-01T00:00:00.000Z" }, + ...Array.from({ length: 50 }, (_, index) => ({ id: `done-${index}`, status: "completed", updatedAt: new Date(Date.UTC(2026, 0, 1, 0, index, 0)).toISOString() })) + ]; + const saved = saveState(workspace, { config: { stopReviewGate: false }, jobs }); + assert.equal(saved.jobs.length, 50); + assert.equal(saved.jobs.some((job) => job.id === "active-old"), true); +}); + +test("late publication cannot restore a job behind its removal fence", () => { + const workspace = makeTempDir(); + const job = { id: "job-removed-late", status: "queued", logFile: resolveJobLogFile(workspace, "job-removed-late") }; + const jobFile = writeJobFile(workspace, job.id, job); + upsertJob(workspace, job); + fs.writeFileSync(jobFile.replace(/\.json$/, ".started.json"), JSON.stringify({ status: "running", pid: process.pid, startedAt: new Date().toISOString() }), "utf8"); + fs.writeFileSync(jobFile.replace(/\.json$/, ".removed"), "", "utf8"); + + writeJobFile(workspace, job.id, { ...job, status: "running", pid: process.pid }); + upsertJob(workspace, { ...job, status: "running", pid: process.pid }); + + assert.deepEqual(listJobs(workspace), []); + assert.equal(fs.existsSync(jobFile), false); + assert.equal(fs.existsSync(jobFile.replace(/\.json$/, ".started.json")), false); + assert.equal(fs.existsSync(jobFile.replace(/\.json$/, ".removed")), true); +}); diff --git a/tests/tracked-jobs.test.mjs b/tests/tracked-jobs.test.mjs new file mode 100644 index 000000000..6bdcd6b25 --- /dev/null +++ b/tests/tracked-jobs.test.mjs @@ -0,0 +1,77 @@ +import fs from "node:fs"; +import test from "node:test"; +import assert from "node:assert/strict"; + +import { makeTempDir } from "./helpers.mjs"; +import { readJobFile, resolveJobFile, upsertJob, writeJobFile } from "../plugins/codex/scripts/lib/state.mjs"; +import { runTrackedJob } from "../plugins/codex/scripts/lib/tracked-jobs.mjs"; + +test("runTrackedJob does not resurrect a terminal persisted job", async () => { + const workspaceRoot = makeTempDir(); + const job = { + id: "task-terminal", + workspaceRoot, + status: "queued", + request: { prompt: "do not run" } + }; + const terminalJob = { + ...job, + status: "failed", + phase: "failed", + errorMessage: "Background worker exited before completing the job.", + pid: null, + completedAt: "2026-08-19T12:00:00.000Z" + }; + writeJobFile(workspaceRoot, job.id, terminalJob); + upsertJob(workspaceRoot, terminalJob); + + let runnerInvoked = false; + const result = await runTrackedJob(job, async () => { + runnerInvoked = true; + return { exitStatus: 0 }; + }); + + assert.equal(runnerInvoked, false); + assert.deepEqual(result, terminalJob); + assert.deepEqual(readJobFile(resolveJobFile(workspaceRoot, job.id)), terminalJob); +}); + +test("runTrackedJob returns a failed record for missing background state", async () => { + const workspaceRoot = makeTempDir(); + const job = { + id: "task-removed", + workspaceRoot, + status: "queued", + request: { prompt: "do not run" } + }; + const jobFile = resolveJobFile(workspaceRoot, job.id); + + let runnerInvoked = false; + const result = await runTrackedJob(job, async () => { + runnerInvoked = true; + return { exitStatus: 0 }; + }); + + assert.equal(runnerInvoked, false); + assert.equal(result.status, "failed"); + assert.match(result.errorMessage, /record is missing/i); + assert.equal(fs.existsSync(jobFile), false); +}); + +test("runTrackedJob returns a cancelled record after removal", async () => { + const workspaceRoot = makeTempDir(); + const job = { id: "task-removed-fence", workspaceRoot, status: "queued", request: { prompt: "do not run" } }; + writeJobFile(workspaceRoot, job.id, job); + upsertJob(workspaceRoot, job); + fs.writeFileSync(resolveJobFile(workspaceRoot, job.id).replace(/\.json$/, ".removed"), "", "utf8"); + + let runnerInvoked = false; + const result = await runTrackedJob(job, async () => { + runnerInvoked = true; + return { exitStatus: 0 }; + }); + + assert.equal(runnerInvoked, false); + assert.equal(result.status, "cancelled"); + assert.equal(result.removed, true); +});