fix: let /codex:cancel reach turns orphaned by a SIGKILLed companion

plugins/codex/scripts/codex-companion.mjs

@@ -35,6 +35,8 @@ import {
 import {
   buildSingleJobSnapshot,
   buildStatusSnapshot,
+  CANCELLATION_INTERRUPT_REQUIRED_MESSAGE,
+  isOrphanedTurn,
   readStoredJob,
   resolveCancelableJob,
   resolveResultJob,
@@ -1135,6 +1137,44 @@ async function handleCancel(argv) {
     return cancelledJob;
   };
 
+  const isOrphan = isOrphanedTurn(job);
+
+  // An orphaned job has no local pid to fall back on — the remote interrupt is
+  // the ONLY mechanism that can actually stop it, so cancellation must not be
+  // persisted until that interrupt is confirmed. Do this BEFORE the optimistic
+  // persistCancellation() below: interruptAppServerTurn is bounded (see
+  // DEFAULT_INTERRUPT_TIMEOUT_MS), but persisting "cancelled" first and rolling
+  // back only after the await returns leaves a false-cancelled state on disk if
+  // the process is killed or crashes while still awaiting it.
+  if (isOrphan) {
+    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}` : "."}`
+      );
+    }
+    if (!interrupt.interrupted) {
+      const detail = interrupt.detail ? `: ${interrupt.detail}` : ".";
+      const message = `${CANCELLATION_INTERRUPT_REQUIRED_MESSAGE}${detail}`;
+      appendLogLine(job.logFile, message);
+      throw new Error(message);
+    }
+    const nextJob = persistCancellation(null);
+    appendLogLine(job.logFile, "Cancelled by user.");
+    const payload = {
+      jobId: job.id,
+      status: "cancelled",
+      title: job.title,
+      turnInterruptAttempted: interrupt.attempted,
+      turnInterrupted: interrupt.interrupted
+    };
+    outputCommandResult(payload, renderCancelReport(nextJob), options.json);
+    return;
+  }
+
   persistCancellation(job.pid ?? null);
   appendLogLine(job.logFile, "Cancelled by user.");
 

plugins/codex/scripts/lib/codex.mjs

@@ -813,7 +813,8 @@ async function captureTurn(client, threadId, startRequest, options = {}) {
       if (state.threadId && state.turnId && options.cwd) {
         await interruptAppServerTurn(options.cwd, {
           threadId: state.threadId,
-          turnId: state.turnId
+          turnId: state.turnId,
+          timeoutMs: TURN_INTERRUPT_TIMEOUT_MS
         }).catch(() => {});
       }
       throw error;
@@ -1186,7 +1187,26 @@ export async function getCodexAuthStatus(cwd, options = {}) {
   }
 }
 
-export async function interruptAppServerTurn(cwd, { threadId, turnId }) {
+// A control-plane RPC (send-interrupt-and-acknowledge), not agent work — it
+// must not hang the way a turn can. Callers that gate a state transition on
+// the result (e.g. /codex:cancel finalizing an orphaned job) need a bounded
+// wait so a stuck broker/transport fails fast instead of leaving the caller
+// stalled indefinitely on this await. CODEX_INTERRUPT_TIMEOUT_MS lets tests
+// shrink this the same way CODEX_TURN_TIMEOUT_MS overrides the turn budget.
+const DEFAULT_INTERRUPT_TIMEOUT_MS = 15000;
+
+function resolveInterruptTimeoutMs(timeoutMs) {
+  if (Number.isFinite(timeoutMs) && timeoutMs > 0) {
+    return timeoutMs;
+  }
+  const fromEnv = Number(process.env.CODEX_INTERRUPT_TIMEOUT_MS);
+  if (Number.isFinite(fromEnv) && fromEnv > 0) {
+    return fromEnv;
+  }
+  return DEFAULT_INTERRUPT_TIMEOUT_MS;
+}
+
+export async function interruptAppServerTurn(cwd, { threadId, turnId, timeoutMs } = {}) {
   if (!threadId || !turnId) {
     return {
       attempted: false,
@@ -1206,17 +1226,19 @@ export async function interruptAppServerTurn(cwd, { threadId, turnId }) {
     };
   }
 
+  const resolvedTimeoutMs = resolveInterruptTimeoutMs(timeoutMs);
+
   let client = null;
   try {
     client = await CodexAppServerClient.connect(cwd, {
       reuseExistingBroker: true,
       allowBusyStaleBroker: true
     });
     await client.request("turn/interrupt", { threadId, turnId }, {
-      timeoutMs: TURN_INTERRUPT_TIMEOUT_MS,
+      timeoutMs: resolvedTimeoutMs,
       // A wedged peer that accepted the connection but goes silent must not
       // linger past the timeout: force-close so callers aren't blocked
-      // beyond TURN_INTERRUPT_TIMEOUT_MS regardless of transport.
+      // beyond resolvedTimeoutMs regardless of transport.
       onTimeout: () => client?.close().catch(() => {})
     });
     return {

plugins/codex/scripts/lib/job-control.mjs

@@ -2,14 +2,24 @@ import fs from "node:fs";
 
 import { getSessionRuntimeStatus } from "./codex.mjs";
 import { isProcessAlive } from "./process.mjs";
-import { getConfig, listJobs, readJobFile, resolveJobFile, upsertJob, writeJobFile } from "./state.mjs";
+import {
+  getConfig,
+  listJobs,
+  readJobFile,
+  resolveJobFile,
+  UNREPORTED_PROCESS_EXIT_MESSAGE,
+  upsertJob,
+  writeJobFile
+} from "./state.mjs";
 import { SESSION_ID_ENV } from "./tracked-jobs.mjs";
 import { resolveWorkspaceRoot } from "./workspace.mjs";
 
 export const DEFAULT_MAX_STATUS_JOBS = 8;
 export const DEFAULT_MAX_PROGRESS_LINES = 4;
 export const CANCELLATION_TERMINATION_FAILED_MESSAGE =
   "Cancellation requested but process termination failed; retry /codex:cancel.";
+export const CANCELLATION_INTERRUPT_REQUIRED_MESSAGE =
+  "Cancellation requested but the remote Codex turn interrupt failed; retry /codex:cancel.";
 
 export function sortJobsNewestFirst(jobs) {
   return [...jobs].sort((left, right) => String(right.updatedAt ?? "").localeCompare(String(left.updatedAt ?? "")));
@@ -323,10 +333,21 @@ export function resolveResultJob(cwd, reference) {
   throw new Error("No finished Codex jobs found for this repository yet.");
 }
 
+export function isOrphanedTurn(job) {
+  return (
+    job.status === "failed" &&
+    job.errorMessage === UNREPORTED_PROCESS_EXIT_MESSAGE &&
+    Boolean(job.threadId) &&
+    Boolean(job.turnId)
+  );
+}
+
 export function resolveCancelableJob(cwd, reference, options = {}) {
   const workspaceRoot = resolveWorkspaceRoot(cwd);
   const jobs = sortJobsNewestFirst(listJobs(workspaceRoot));
-  const activeJobs = jobs.filter((job) => job.status === "queued" || job.status === "running");
+  const activeJobs = jobs.filter(
+    (job) => job.status === "queued" || job.status === "running" || isOrphanedTurn(job)
+  );
 
   if (reference) {
     const selected = matchJobReference(activeJobs, reference);

plugins/codex/scripts/lib/state.mjs

@@ -12,7 +12,7 @@ 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 UNREPORTED_PROCESS_EXIT_MESSAGE = "Process exited without reporting.";
+export const UNREPORTED_PROCESS_EXIT_MESSAGE = "Process exited without reporting.";
 
 function nowIso() {
   return new Date().toISOString();

tests/fake-codex-fixture.mjs

@@ -927,6 +927,14 @@ rl.on("line", (line) => {
 	      }
 
 	      case "turn/interrupt": {
+	        if (BEHAVIOR === "stalled-interrupt") {
+	          // Never respond — simulates a wedged/hung broker or app-server during
+	          // turn/interrupt. interruptAppServerTurn() must bound this request
+	          // (DEFAULT_INTERRUPT_TIMEOUT_MS) instead of hanging forever, so a
+	          // caller gating a state transition on the result (e.g. /codex:cancel
+	          // finalizing an orphaned job) is never left stuck.
+	          break;
+	        }
 	        state.lastInterrupt = {
 	          threadId: message.params.threadId,
 	          turnId: message.params.turnId
@@ -1007,6 +1015,9 @@ export function buildEnv(binDir) {
     CLAUDE_PLUGIN_DATA: getTestPluginDataDir(),
     // Production keeps an idle broker warm for 15 minutes. Tests only need a
     // brief reuse window and should not leave dozens of detached helpers.
-    CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS: "2000"
+    CODEX_COMPANION_BROKER_IDLE_TIMEOUT_MS: "2000",
+    // Production bounds a stuck turn/interrupt request at 15s. Tests exercising
+    // a hung interrupt (BEHAVIOR "stalled-interrupt") should not wait that long.
+    CODEX_INTERRUPT_TIMEOUT_MS: "1000"
   };
 }

tests/runtime.test.mjs

@@ -2262,6 +2262,179 @@ test("cancel stops an active background job and marks it cancelled", async (t) =
   assert.match(fs.readFileSync(logFile, "utf8"), /Cancelled by user/);
 });
 
+function writeOrphanedJobFixture(workspace, { threadId = "thr_orphaned", turnId = "turn_orphaned" } = {}) {
+  const stateDir = resolveStateDir(workspace);
+  const jobsDir = path.join(stateDir, "jobs");
+  fs.mkdirSync(jobsDir, { recursive: true });
+
+  const logFile = path.join(jobsDir, "task-orphaned.log");
+  const jobFile = path.join(jobsDir, "task-orphaned.json");
+  fs.writeFileSync(logFile, "[2026-03-18T15:30:00.000Z] Starting Codex Task.\n", "utf8");
+  fs.writeFileSync(
+    jobFile,
+    JSON.stringify(
+      {
+        id: "task-orphaned",
+        status: "running",
+        title: "Codex Task",
+        threadId,
+        turnId,
+        logFile
+      },
+      null,
+      2
+    ),
+    "utf8"
+  );
+  fs.writeFileSync(
+    path.join(stateDir, "state.json"),
+    `${JSON.stringify(
+      {
+        version: 1,
+        config: { stopReviewGate: false },
+        jobs: [
+          {
+            id: "task-orphaned",
+            status: "running",
+            title: "Codex Task",
+            jobClass: "task",
+            summary: "Investigate flaky test",
+            threadId,
+            turnId,
+            // A pid that is guaranteed to be dead simulates the SIGKILLed foreground process.
+            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"
+  );
+
+  return { stateDir, jobsDir, logFile, jobFile };
+}
+
+test("cancel reaches a job orphaned by a SIGKILLed companion process once the remote turn interrupt succeeds (closes #42)", () => {
+  const workspace = makeTempDir();
+  const binDir = makeTempDir();
+  installFakeCodex(binDir);
+  // buildEnv() pins process.env.CLAUDE_PLUGIN_DATA for this worker (see its doc
+  // comment) — resolve it before writeOrphanedJobFixture computes resolveStateDir,
+  // so the in-process read and the spawned companion's state dir agree.
+  const env = buildEnv(binDir);
+  const { stateDir, logFile, jobFile } = writeOrphanedJobFixture(workspace);
+
+  // Reconciliation (triggered by /codex:status) flips the dead-pid job to "failed"
+  // before cancel ever runs, mirroring the real orphan sequence from the issue.
+  const statusResult = run("node", [SCRIPT, "status", "--json"], { cwd: workspace, env });
+  assert.equal(statusResult.status, 0, statusResult.stderr);
+  const reconciled = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")).jobs.find(
+    (job) => job.id === "task-orphaned"
+  );
+  assert.equal(reconciled.status, "failed");
+  assert.equal(reconciled.errorMessage, "Process exited without reporting.");
+
+  const cancelResult = run("node", [SCRIPT, "cancel", "task-orphaned", "--json"], {
+    cwd: workspace,
+    env
+  });
+
+  assert.equal(cancelResult.status, 0, cancelResult.stderr);
+  const payload = JSON.parse(cancelResult.stdout);
+  assert.equal(payload.status, "cancelled");
+  assert.equal(payload.turnInterrupted, true);
+
+  const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8"));
+  const cancelled = state.jobs.find((job) => job.id === "task-orphaned");
+  assert.equal(cancelled.status, "cancelled");
+
+  const stored = JSON.parse(fs.readFileSync(jobFile, "utf8"));
+  assert.equal(stored.status, "cancelled");
+  assert.match(fs.readFileSync(logFile, "utf8"), /Cancelled by user/);
+});
+
+test("cancel does not finalize an orphaned job as cancelled when the remote turn interrupt fails, and stays retryable", () => {
+  const workspace = makeTempDir();
+  const { stateDir, logFile, jobFile } = writeOrphanedJobFixture(workspace);
+
+  // No fake codex on PATH: interruptAppServerTurn cannot reach a real broker/thread,
+  // so the interrupt attempt fails (or is never attempted) — this must NOT be
+  // treated as a successful cancellation, since nothing actually stopped the
+  // orphaned remote turn.
+  const statusResult = run("node", [SCRIPT, "status", "--json"], { cwd: workspace });
+  assert.equal(statusResult.status, 0, statusResult.stderr);
+  const reconciled = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")).jobs.find(
+    (job) => job.id === "task-orphaned"
+  );
+  assert.equal(reconciled.status, "failed");
+
+  const cancelResult = run("node", [SCRIPT, "cancel", "task-orphaned", "--json"], {
+    cwd: workspace
+  });
+
+  assert.notEqual(cancelResult.status, 0);
+  assert.match(cancelResult.stderr, /remote Codex turn interrupt failed/i);
+
+  // The job must remain in its original orphaned-failed state so a retried
+  // /codex:cancel can still select and re-attempt it via isOrphanedTurn.
+  const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8"));
+  const stillOrphaned = state.jobs.find((job) => job.id === "task-orphaned");
+  assert.equal(stillOrphaned.status, "failed");
+  assert.equal(stillOrphaned.errorMessage, "Process exited without reporting.");
+  assert.equal(stillOrphaned.pid, null);
+
+  const stored = JSON.parse(fs.readFileSync(jobFile, "utf8"));
+  assert.equal(stored.status, "failed");
+  assert.equal(stored.errorMessage, "Process exited without reporting.");
+
+  const retryResult = run("node", [SCRIPT, "cancel", "task-orphaned", "--json"], {
+    cwd: workspace
+  });
+  assert.notEqual(retryResult.status, 0);
+  assert.match(retryResult.stderr, /remote Codex turn interrupt failed/i);
+});
+
+test("cancel on an orphaned job does not persist cancelled while the remote interrupt is still hung, and stays retryable once it times out", () => {
+  const workspace = makeTempDir();
+  const binDir = makeTempDir();
+  installFakeCodex(binDir, "stalled-interrupt");
+  const env = buildEnv(binDir);
+  const { stateDir, logFile, jobFile } = writeOrphanedJobFixture(workspace);
+
+  const statusResult = run("node", [SCRIPT, "status", "--json"], { cwd: workspace, env });
+  assert.equal(statusResult.status, 0, statusResult.stderr);
+  const reconciled = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8")).jobs.find(
+    (job) => job.id === "task-orphaned"
+  );
+  assert.equal(reconciled.status, "failed");
+
+  // The fake broker never answers turn/interrupt. Without a bounded timeout on
+  // that request, /codex:cancel would hang here; with one, it must return
+  // (non-zero) once the timeout fires rather than persisting "cancelled" before
+  // the interrupt is confirmed.
+  const cancelResult = run("node", [SCRIPT, "cancel", "task-orphaned", "--json"], {
+    cwd: workspace,
+    env
+  });
+
+  assert.notEqual(cancelResult.status, 0);
+  assert.match(cancelResult.stderr, /remote Codex turn interrupt failed/i);
+
+  const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8"));
+  const stillOrphaned = state.jobs.find((job) => job.id === "task-orphaned");
+  assert.equal(stillOrphaned.status, "failed");
+  assert.equal(stillOrphaned.errorMessage, "Process exited without reporting.");
+
+  const stored = JSON.parse(fs.readFileSync(jobFile, "utf8"));
+  assert.equal(stored.status, "failed");
+  assert.match(fs.readFileSync(logFile, "utf8"), /remote Codex turn interrupt failed/i);
+});
+
 test("failed cancellation restores a live job so cancellation can be retried", () => {
   const workspace = makeTempDir();
   const job = {