Register broker ownership and safely reclaim abandoned helper trees

plugins/codex/scripts/app-server-child.mjs

@@ -0,0 +1,87 @@
+#!/usr/bin/env node
+
+import fs from "node:fs";
+import process from "node:process";
+import { spawn } from "node:child_process";
+
+const activation = fs.createReadStream(null, { fd: 3, autoClose: true });
+let activated = false;
+let activationBuffer = "";
+let child = null;
+
+function exitWithoutChild(code = 1) {
+  if (child || process.exitCode != null) {
+    return;
+  }
+  process.exitCode = code;
+  process.stdin.resume();
+}
+
+function startAppServer() {
+  if (activated || child) {
+    return;
+  }
+  activated = true;
+  child = spawn("codex", ["app-server"], {
+    cwd: process.cwd(),
+    env: process.env,
+    detached: false,
+    stdio: ["pipe", "pipe", "pipe"],
+    shell: process.platform === "win32" ? (process.env.SHELL || true) : false,
+    windowsHide: true
+  });
+  child.stdout.pipe(process.stdout);
+  child.stderr.pipe(process.stderr);
+  process.stdin.pipe(child.stdin);
+  child.on("error", (error) => {
+    process.stderr.write(`Unable to start codex app-server: ${error.message}\n`);
+    process.exit(1);
+  });
+  // `close` follows both process exit and stdio closure, so the wrapper does
+  // not truncate the JSONL stream while forwarding the child's final bytes.
+  child.on("close", (code, signal) => {
+    process.exit(Number.isInteger(code) ? code : signal ? 1 : 0);
+  });
+}
+
+activation.setEncoding("utf8");
+activation.on("data", (chunk) => {
+  if (activated) {
+    return;
+  }
+  activationBuffer += chunk;
+  const newlineIndex = activationBuffer.indexOf("\n");
+  if (newlineIndex === -1) {
+    return;
+  }
+  if (activationBuffer.slice(0, newlineIndex).trim() !== "activate") {
+    exitWithoutChild(1);
+    return;
+  }
+  startAppServer();
+});
+activation.on("end", () => {
+  if (!activated) {
+    exitWithoutChild(1);
+  }
+});
+activation.on("error", () => {
+  if (!activated) {
+    exitWithoutChild(1);
+  }
+});
+
+process.stdin.on("end", () => {
+  if (!activated) {
+    exitWithoutChild(1);
+    return;
+  }
+  child?.stdin.end();
+});
+process.stdin.on("error", () => {
+  child?.stdin.destroy();
+});
+// Keep protocol bytes buffered until activation has created the real app
+// server and its stdin pipe. Flowing stdin here can discard an initialize
+// request that races the activation-control pipe.
+process.stdin.pause();

plugins/codex/scripts/codex-companion.mjs

@@ -4,7 +4,7 @@ import { spawn } from "node:child_process";
 import fs from "node:fs";
 import path from "node:path";
 import process from "node:process";
-import { fileURLToPath } from "node:url";
+import { fileURLToPath, pathToFileURL } from "node:url";
 
 import { parseArgs, splitRawArgumentString } from "./lib/args.mjs";
 import {
@@ -24,14 +24,15 @@ import {
 import { resolveClaudeSessionPath } from "./lib/claude-session-transfer.mjs";
 import { readStdinIfPiped } from "./lib/fs.mjs";
 import { collectReviewContext, ensureGitRepository, resolveReviewTarget } from "./lib/git.mjs";
-import { binaryAvailable, terminateProcessTree } from "./lib/process.mjs";
+import { binaryAvailable, getProcessIdentity, terminateProcessTree } from "./lib/process.mjs";
 import { loadPromptTemplate, interpolateTemplate } from "./lib/prompts.mjs";
 import {
   generateJobId,
   getConfig,
   listJobs,
   setConfig,
   upsertJob,
+  writeCancelFlag,
   writeJobFile
 } from "./lib/state.mjs";
 import {
@@ -681,22 +682,61 @@ function spawnDetachedTaskWorker(cwd, jobId) {
   return child;
 }
 
-function enqueueBackgroundTask(cwd, job, request) {
+function recordTaskWorkerSpawnFailure(workspaceRoot, jobId, error) {
+  const storedJob = readStoredJob(workspaceRoot, jobId);
+  if (!storedJob || storedJob.status !== "queued" || storedJob.pid != null) {
+    return;
+  }
+
+  const errorMessage = error instanceof Error ? error.message : String(error);
+  const completedAt = nowIso();
+  const failedRecord = {
+    ...storedJob,
+    status: "failed",
+    phase: "failed",
+    pid: null,
+    errorMessage,
+    completedAt
+  };
+  writeJobFile(workspaceRoot, jobId, failedRecord);
+  upsertJob(workspaceRoot, {
+    id: jobId,
+    status: "failed",
+    phase: "failed",
+    pid: null,
+    errorMessage,
+    completedAt
+  });
+  appendLogLine(storedJob.logFile, `Worker spawn failed: ${errorMessage}`);
+}
+
+export function enqueueBackgroundTask(cwd, job, request, dependencies = {}) {
   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);
 
+  const spawnWorker = dependencies.spawnDetachedTaskWorkerImpl ?? spawnDetachedTaskWorker;
+  let worker;
+  try {
+    worker = spawnWorker(cwd, job.id);
+  } catch (error) {
+    recordTaskWorkerSpawnFailure(job.workspaceRoot, job.id, error);
+    throw error;
+  }
+  worker?.once?.("error", (error) => {
+    recordTaskWorkerSpawnFailure(job.workspaceRoot, job.id, error);
+  });
+
   return {
     payload: {
       jobId: job.id,
@@ -835,7 +875,7 @@ async function handleTransfer(argv) {
   outputCommandResult(payload, rendered, options.json);
 }
 
-async function handleTaskWorker(argv) {
+export async function handleTaskWorker(argv, dependencies = {}) {
   const { options } = parseCommandInput(argv, {
     valueOptions: ["cwd", "job-id"]
   });
@@ -851,25 +891,40 @@ async function handleTaskWorker(argv) {
     throw new Error(`No stored job found for ${options["job-id"]}.`);
   }
 
+  const {
+    processIdentity: _storedProcessIdentity,
+    ownershipSnapshot: _storedOwnershipSnapshot,
+    ownershipCaptureFailed: _storedOwnershipCaptureFailed,
+    ...storedTask
+  } = storedJob;
+  let workerOwnership;
+  try {
+    const processIdentity = (dependencies.getProcessIdentityImpl ?? getProcessIdentity)(process.pid);
+    workerOwnership = processIdentity ? { processIdentity } : { ownershipCaptureFailed: true };
+  } catch {
+    workerOwnership = { ownershipCaptureFailed: true };
+  }
+
   const request = storedJob.request;
   if (!request || typeof request !== "object") {
     throw new Error(`Stored job ${options["job-id"]} is missing its task request payload.`);
   }
 
   const { logFile, progress } = createTrackedProgress(
     {
-      ...storedJob,
+      ...storedTask,
       workspaceRoot
     },
     {
       logFile: storedJob.logFile ?? null
     }
   );
-  await runTrackedJob(
+  await (dependencies.runTrackedJobImpl ?? runTrackedJob)(
     {
-      ...storedJob,
+      ...storedTask,
       workspaceRoot,
-      logFile
+      logFile,
+      ...workerOwnership
     },
     () =>
       executeTaskRun({
@@ -960,49 +1015,25 @@ function handleTaskResumeCandidate(argv) {
   outputCommandResult(payload, rendered, options.json);
 }
 
-async function handleCancel(argv) {
-  const { options, positionals } = parseCommandInput(argv, {
-    valueOptions: ["cwd"],
-    booleanOptions: ["json"]
-  });
-
-  const cwd = resolveCommandCwd(options);
-  const reference = positionals[0] ?? "";
-  const { workspaceRoot, job } = resolveCancelableJob(cwd, reference, { env: process.env });
-  const existing = readStoredJob(workspaceRoot, job.id) ?? {};
-  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.");
+function finishCancelledJob(workspaceRoot, record, interrupt, options) {
+  appendLogLine(record.logFile, "Cancelled by user.");
 
   const completedAt = nowIso();
   const nextJob = {
-    ...job,
+    ...record,
     status: "cancelled",
     phase: "cancelled",
     pid: null,
     completedAt,
     errorMessage: "Cancelled by user."
   };
 
-  writeJobFile(workspaceRoot, job.id, {
-    ...existing,
+  writeJobFile(workspaceRoot, record.id, {
     ...nextJob,
     cancelledAt: completedAt
   });
   upsertJob(workspaceRoot, {
-    id: job.id,
+    id: record.id,
     status: "cancelled",
     phase: "cancelled",
     pid: null,
@@ -1011,16 +1042,93 @@ async function handleCancel(argv) {
   });
 
   const payload = {
-    jobId: job.id,
+    jobId: record.id,
     status: "cancelled",
-    title: job.title,
+    title: record.title,
     turnInterruptAttempted: interrupt.attempted,
     turnInterrupted: interrupt.interrupted
   };
 
   outputCommandResult(payload, renderCancelReport(nextJob), options.json);
 }
 
+export async function handleCancel(argv, dependencies = {}) {
+  const { options, positionals } = parseCommandInput(argv, {
+    valueOptions: ["cwd"],
+    booleanOptions: ["json"]
+  });
+
+  const cwd = resolveCommandCwd(options);
+  const reference = positionals[0] ?? "";
+  const { workspaceRoot, job } = resolveCancelableJob(cwd, reference, { env: process.env });
+  let existing = readStoredJob(workspaceRoot, job.id) ?? {};
+  let record = { ...job, ...existing };
+
+  if (!Number.isFinite(record.pid)) {
+    writeCancelFlag(workspaceRoot, job.id);
+    existing = readStoredJob(workspaceRoot, job.id) ?? existing;
+    record = { ...job, ...existing };
+    if (!Number.isFinite(record.pid)) {
+      finishCancelledJob(
+        workspaceRoot,
+        record,
+        { attempted: false, interrupted: false },
+        options
+      );
+      return;
+    }
+  }
+
+  const threadId = record.threadId ?? null;
+  const turnId = record.turnId ?? null;
+
+  const interrupt = await (dependencies.interruptAppServerTurnImpl ?? interruptAppServerTurn)(cwd, { threadId, turnId });
+  if (interrupt.attempted) {
+    appendLogLine(
+      record.logFile,
+      interrupt.interrupted
+        ? `Requested Codex turn interrupt for ${turnId} on ${threadId}.`
+        : `Codex turn interrupt failed${interrupt.detail ? `: ${interrupt.detail}` : "."}`
+    );
+  }
+
+  const expectedRootIdentity = existing.processIdentity ?? null;
+  const ownershipCaptureFailed = existing.ownershipCaptureFailed === true;
+  const cleanupOutcome = await (dependencies.terminateProcessTreeImpl ?? terminateProcessTree)(record.pid, {
+    expectedRootIdentity,
+    ownershipSnapshot: null,
+    requireVerifiedOwnership: ownershipCaptureFailed,
+    priorCleanupDegraded: existing.cleanupOutcome?.degraded === true
+  });
+  if (cleanupOutcome?.verified !== true) {
+    const failureMessage =
+      ownershipCaptureFailed && !expectedRootIdentity
+        ? `Job ${job.id} could not be verified as owned and was left alone.`
+        : `Unable to verify cleanup for ${job.id}; ownership records were preserved for retry.`;
+    appendLogLine(record.logFile, failureMessage);
+    const recoveryRecord = {
+      ...record,
+      status: record.status,
+      phase: "cleanup-pending",
+      pid: record.pid,
+      cleanupOutcome,
+      cleanupFailure: failureMessage
+    };
+    writeJobFile(workspaceRoot, job.id, recoveryRecord);
+    upsertJob(workspaceRoot, {
+      id: job.id,
+      status: record.status,
+      phase: "cleanup-pending",
+      pid: record.pid,
+      cleanupOutcome,
+      cleanupFailure: failureMessage
+    });
+    throw new Error(failureMessage);
+  }
+
+  finishCancelledJob(workspaceRoot, record, interrupt, options);
+}
+
 async function main() {
   const [subcommand, ...argv] = process.argv.slice(2);
   if (!subcommand || subcommand === "help" || subcommand === "--help") {
@@ -1066,8 +1174,10 @@ async function main() {
   }
 }
 
-main().catch((error) => {
-  const message = error instanceof Error ? error.message : String(error);
-  process.stderr.write(`${message}\n`);
-  process.exitCode = 1;
-});
+if (process.argv[1] && pathToFileURL(path.resolve(process.argv[1])).href === import.meta.url) {
+  main().catch((error) => {
+    const message = error instanceof Error ? error.message : String(error);
+    process.stderr.write(`${message}\n`);
+    process.exitCode = 1;
+  });
+}