diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs
index 1954274f..9f13d0f2 100644
--- a/plugins/codex/scripts/app-server-broker.mjs
+++ b/plugins/codex/scripts/app-server-broker.mjs
@@ -52,7 +52,7 @@ async function main() {
   }
 
   const { options } = parseArgs(argv, {
-    valueOptions: ["cwd", "pid-file", "endpoint"]
+    valueOptions: ["cwd", "pid-file", "endpoint", "instance-token"]
   });
 
   if (!options.endpoint) {
@@ -63,6 +63,7 @@ async function main() {
   const endpoint = String(options.endpoint);
   const listenTarget = parseBrokerEndpoint(endpoint);
   const pidFile = options["pid-file"] ? path.resolve(options["pid-file"]) : null;
+  const instanceToken = options["instance-token"] ? String(options["instance-token"]) : null;
   writePidFile(pidFile);
 
   const appClient = await CodexAppServerClient.connect(cwd, { disableBroker: true });
@@ -158,7 +159,17 @@ async function main() {
         }
 
         if (message.id !== undefined && message.method === "broker/shutdown") {
-          send(socket, { id: message.id, result: {} });
+          if (!instanceToken || message.params?.instanceToken !== instanceToken) {
+            send(socket, {
+              id: message.id,
+              error: buildJsonRpcError(-32003, "Broker shutdown identity did not match this instance.")
+            });
+            continue;
+          }
+          send(socket, {
+            id: message.id,
+            result: { pid: process.pid, instanceToken }
+          });
           await shutdown(server);
           process.exit(0);
         }
diff --git a/plugins/codex/scripts/codex-companion.mjs b/plugins/codex/scripts/codex-companion.mjs
index 83df468a..da57976b 100644
--- a/plugins/codex/scripts/codex-companion.mjs
+++ b/plugins/codex/scripts/codex-companion.mjs
@@ -1,6 +1,7 @@
 #!/usr/bin/env node
 
 import { spawn } from "node:child_process";
+import { randomUUID } from "node:crypto";
 import fs from "node:fs";
 import path from "node:path";
 import process from "node:process";
@@ -24,7 +25,7 @@ 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, processHasLaunchToken, terminateProcessTree, waitForProcessExit } from "./lib/process.mjs";
 import { loadPromptTemplate, interpolateTemplate } from "./lib/prompts.mjs";
 import {
   generateJobId,
@@ -50,7 +51,8 @@ import {
   createProgressReporter,
   nowIso,
   runTrackedJob,
-  SESSION_ID_ENV
+  SESSION_ID_ENV,
+  waitForWorkerJob
 } from "./lib/tracked-jobs.mjs";
 import { resolveWorkspaceRoot } from "./lib/workspace.mjs";
 import {
@@ -668,15 +670,30 @@ async function runForegroundCommand(job, runner, options = {}) {
   return execution;
 }
 
-function spawnDetachedTaskWorker(cwd, jobId) {
+function spawnDetachedTaskWorker(cwd, jobId, workerToken, logFile) {
   const scriptPath = path.join(ROOT_DIR, "scripts", "codex-companion.mjs");
-  const child = spawn(process.execPath, [scriptPath, "task-worker", "--cwd", cwd, "--job-id", jobId], {
-    cwd,
-    env: process.env,
-    detached: true,
-    stdio: "ignore",
-    windowsHide: true
-  });
+  const logFd = fs.openSync(logFile, "a");
+  let child;
+  try {
+    child = spawn(process.execPath, [
+      scriptPath,
+      "task-worker",
+      "--cwd",
+      cwd,
+      "--job-id",
+      jobId,
+      "--worker-token",
+      workerToken
+    ], {
+      cwd,
+      env: process.env,
+      detached: true,
+      stdio: ["ignore", logFd, logFd],
+      windowsHide: true
+    });
+  } finally {
+    fs.closeSync(logFd);
+  }
   child.unref();
   return child;
 }
@@ -685,15 +702,48 @@ function enqueueBackgroundTask(cwd, job, request) {
   const { logFile } = createTrackedProgress(job);
   appendLogLine(logFile, "Queued for background execution.");
 
-  const child = spawnDetachedTaskWorker(cwd, job.id);
-  const queuedRecord = {
+  const workerToken = randomUUID();
+  const pendingRecord = {
     ...job,
     status: "queued",
     phase: "queued",
-    pid: child.pid ?? null,
+    pid: null,
+    workerToken,
     logFile,
     request
   };
+  writeJobFile(job.workspaceRoot, job.id, pendingRecord);
+  upsertJob(job.workspaceRoot, pendingRecord);
+
+  let child;
+  try {
+    child = spawnDetachedTaskWorker(cwd, job.id, workerToken, logFile);
+  } catch (error) {
+    const failedRecord = {
+      ...pendingRecord,
+      status: "failed",
+      phase: "failed",
+      completedAt: nowIso(),
+      errorMessage: error instanceof Error ? error.message : String(error)
+    };
+    writeJobFile(job.workspaceRoot, job.id, failedRecord);
+    upsertJob(job.workspaceRoot, failedRecord);
+    throw error;
+  }
+  if (!Number.isSafeInteger(child.pid) || child.pid <= 0) {
+    const error = new Error(`Detached worker for ${job.id} did not report a valid process ID.`);
+    const failedRecord = {
+      ...pendingRecord,
+      status: "failed",
+      phase: "failed",
+      completedAt: nowIso(),
+      errorMessage: error.message
+    };
+    writeJobFile(job.workspaceRoot, job.id, failedRecord);
+    upsertJob(job.workspaceRoot, failedRecord);
+    throw error;
+  }
+  const queuedRecord = { ...pendingRecord, pid: child.pid };
   writeJobFile(job.workspaceRoot, job.id, queuedRecord);
   upsertJob(job.workspaceRoot, queuedRecord);
 
@@ -837,7 +887,7 @@ async function handleTransfer(argv) {
 
 async function handleTaskWorker(argv) {
   const { options } = parseCommandInput(argv, {
-    valueOptions: ["cwd", "job-id"]
+    valueOptions: ["cwd", "job-id", "worker-token"]
   });
 
   if (!options["job-id"]) {
@@ -846,10 +896,13 @@ async function handleTaskWorker(argv) {
 
   const cwd = resolveCommandCwd(options);
   const workspaceRoot = resolveCommandWorkspace(options);
-  const storedJob = readStoredJob(workspaceRoot, options["job-id"]);
-  if (!storedJob) {
-    throw new Error(`No stored job found for ${options["job-id"]}.`);
-  }
+  if (!options["worker-token"]) throw new Error("Missing required --worker-token for task-worker.");
+  const storedJob = await waitForWorkerJob(
+    workspaceRoot,
+    options["job-id"],
+    options["worker-token"],
+    process.pid
+  );
 
   const request = storedJob.request;
   if (!request || typeof request !== "object") {
@@ -973,6 +1026,10 @@ async function handleCancel(argv) {
   const threadId = existing.threadId ?? job.threadId ?? null;
   const turnId = existing.turnId ?? job.turnId ?? null;
 
+  if (!processHasLaunchToken(job.pid, job.workerToken)) {
+    throw new Error(`Cannot verify that process ${job.pid ?? "unknown"} owns Codex job ${job.id}; state was preserved.`);
+  }
+
   const interrupt = await interruptAppServerTurn(cwd, { threadId, turnId });
   if (interrupt.attempted) {
     appendLogLine(
@@ -983,7 +1040,13 @@ async function handleCancel(argv) {
     );
   }
 
-  terminateProcessTree(job.pid ?? Number.NaN);
+  if (!processHasLaunchToken(job.pid, job.workerToken)) {
+    throw new Error(`Codex job ${job.id} process ownership changed before termination; state was preserved.`);
+  }
+  terminateProcessTree(job.pid);
+  if (!(await waitForProcessExit(job.pid))) {
+    throw new Error(`Codex job ${job.id} process ${job.pid} did not exit; state was preserved.`);
+  }
   appendLogLine(job.logFile, "Cancelled by user.");
 
   const completedAt = nowIso();
diff --git a/plugins/codex/scripts/lib/broker-lifecycle.mjs b/plugins/codex/scripts/lib/broker-lifecycle.mjs
index ef763819..4788e27b 100644
--- a/plugins/codex/scripts/lib/broker-lifecycle.mjs
+++ b/plugins/codex/scripts/lib/broker-lifecycle.mjs
@@ -1,4 +1,5 @@
 import fs from "node:fs";
+import { randomUUID } from "node:crypto";
 import net from "node:net";
 import os from "node:os";
 import path from "node:path";
@@ -6,6 +7,7 @@ import process from "node:process";
 import { spawn } from "node:child_process";
 import { fileURLToPath } from "node:url";
 import { createBrokerEndpoint, parseBrokerEndpoint } from "./broker-endpoint.mjs";
+import { processHasLaunchToken, terminateProcessTree, waitForProcessExit } from "./process.mjs";
 import { resolveStateDir } from "./state.mjs";
 
 export const PID_FILE_ENV = "CODEX_COMPANION_APP_SERVER_PID_FILE";
@@ -40,25 +42,88 @@ export async function waitForBrokerEndpoint(endpoint, timeoutMs = 2000) {
   return false;
 }
 
-export async function sendBrokerShutdown(endpoint) {
-  await new Promise((resolve) => {
+export async function sendBrokerShutdown(endpoint, options = {}) {
+  return await new Promise((resolve) => {
     const socket = connectToEndpoint(endpoint);
+    let settled = false;
+    let buffer = "";
+    const finish = (result = null) => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      socket.destroy();
+      resolve(result);
+    };
     socket.setEncoding("utf8");
+    socket.setTimeout(options.timeoutMs ?? 2000, finish);
     socket.on("connect", () => {
-      socket.write(`${JSON.stringify({ id: 1, method: "broker/shutdown", params: {} })}\n`);
+      socket.write(`${JSON.stringify({
+        id: 1,
+        method: "broker/shutdown",
+        params: { instanceToken: options.instanceToken ?? null }
+      })}\n`);
     });
-    socket.on("data", () => {
-      socket.end();
-      resolve();
+    socket.on("data", (chunk) => {
+      buffer += chunk;
+      const newlineIndex = buffer.indexOf("\n");
+      if (newlineIndex === -1) {
+        return;
+      }
+      try {
+        const response = JSON.parse(buffer.slice(0, newlineIndex));
+        finish({ result: response?.result ?? null, error: response?.error ?? null });
+      } catch {
+        finish(null);
+      }
     });
-    socket.on("error", resolve);
-    socket.on("close", resolve);
+    socket.on("error", () => finish(null));
+    socket.on("close", () => finish(null));
   });
 }
 
-export function spawnBrokerProcess({ scriptPath, cwd, endpoint, pidFile, logFile, env = process.env }) {
+function isValidPid(pid) {
+  return Number.isSafeInteger(pid) && pid > 0;
+}
+
+async function waitForBrokerExit(endpoint, options = {}) {
+  const timeoutMs = options.timeoutMs ?? 2000;
+  const intervalMs = options.intervalMs ?? 25;
+  const target = parseBrokerEndpoint(endpoint);
+  const start = Date.now();
+  let unavailableChecks = 0;
+  while (Date.now() - start < timeoutMs) {
+    if (target.kind === "unix") {
+      if (!fs.existsSync(target.path)) {
+        return true;
+      }
+    } else if (await isBrokerEndpointReady(endpoint)) {
+      unavailableChecks = 0;
+    } else {
+      unavailableChecks += 1;
+      if (unavailableChecks >= 3) {
+        return true;
+      }
+    }
+    await new Promise((resolve) => setTimeout(resolve, intervalMs));
+  }
+  return target.kind === "unix" ? !fs.existsSync(target.path) : !(await isBrokerEndpointReady(endpoint));
+}
+
+export function spawnBrokerProcess({ scriptPath, cwd, endpoint, pidFile, logFile, instanceToken, env = process.env }) {
   const logFd = fs.openSync(logFile, "a");
-  const child = spawn(process.execPath, [scriptPath, "serve", "--endpoint", endpoint, "--cwd", cwd, "--pid-file", pidFile], {
+  const child = spawn(process.execPath, [
+    scriptPath,
+    "serve",
+    "--endpoint",
+    endpoint,
+    "--cwd",
+    cwd,
+    "--pid-file",
+    pidFile,
+    "--instance-token",
+    instanceToken
+  ], {
     cwd,
     env,
     detached: true,
@@ -99,6 +164,146 @@ export function clearBrokerSession(cwd) {
   }
 }
 
+function resolveBrokerPid(session) {
+  const statePid = isValidPid(session.pid) ? session.pid : null;
+  let filePid = null;
+  if (session.pidFile && fs.existsSync(session.pidFile)) {
+    const rawPid = fs.readFileSync(session.pidFile, "utf8").trim();
+    if (/^\d+$/.test(rawPid)) {
+      const parsedPid = Number(rawPid);
+      filePid = isValidPid(parsedPid) ? parsedPid : null;
+    }
+  }
+  if (statePid && filePid && statePid !== filePid) {
+    throw new Error(`Codex app-server broker PID mismatch (${statePid} != ${filePid}).`);
+  }
+  return statePid ?? filePid;
+}
+
+export async function shutdownBrokerSession(cwd, options = {}) {
+  const session = options.session ?? loadBrokerSession(cwd);
+  if (!session) {
+    return { found: false, exited: true, forced: false };
+  }
+
+  const pid = resolveBrokerPid(session);
+  if (session.endpoint && !session.instanceToken) {
+    throw new Error("Codex app-server broker ownership could not be verified; persisted state was preserved.");
+  }
+  let shutdownResponse = null;
+  if (session.endpoint) {
+    shutdownResponse = await sendBrokerShutdown(session.endpoint, {
+      timeoutMs: options.timeoutMs,
+      instanceToken: session.instanceToken
+    });
+  }
+  if (shutdownResponse?.error) {
+    throw new Error(
+      `Codex app-server broker rejected shutdown identity; persisted state was preserved: ${shutdownResponse.error.message ?? "unknown error"}`
+    );
+  }
+  const shutdownAck = shutdownResponse?.result ?? null;
+
+  const acknowledgedPid = isValidPid(shutdownAck?.pid) ? shutdownAck.pid : null;
+  const ownershipVerified =
+    Boolean(session.instanceToken) &&
+    shutdownAck?.instanceToken === session.instanceToken &&
+    acknowledgedPid !== null &&
+    (pid === null || pid === acknowledgedPid);
+  if (shutdownAck && session.instanceToken && !ownershipVerified) {
+    throw new Error("Codex app-server broker shutdown identity did not match persisted state.");
+  }
+  let verifiedPid = ownershipVerified ? acknowledgedPid : null;
+  let pidAlreadyExited = false;
+  if (isValidPid(pid)) {
+    pidAlreadyExited = await waitForProcessExit(pid, {
+      timeoutMs: 0,
+      intervalMs: options.intervalMs,
+      killImpl: options.killImpl,
+      platform: options.platform
+    });
+  }
+  if (!shutdownAck && isValidPid(pid) && !pidAlreadyExited) {
+    const ownsPersistedProcess = options.verifyProcess
+      ? options.verifyProcess(pid, session.instanceToken)
+      : processHasLaunchToken(pid, session.instanceToken, {
+          marker: "--instance-token",
+          platform: options.platform,
+          timeoutMs: options.timeoutMs,
+          runCommandImpl: options.runCommandImpl
+        });
+    if (!ownsPersistedProcess) {
+      pidAlreadyExited = await waitForProcessExit(pid, {
+        timeoutMs: 0,
+        intervalMs: options.intervalMs,
+        killImpl: options.killImpl,
+        platform: options.platform
+      });
+      if (!pidAlreadyExited) {
+        throw new Error("Codex app-server broker ownership could not be verified; persisted state was preserved.");
+      }
+    } else {
+      verifiedPid = pid;
+    }
+  }
+  if (!shutdownAck && session.instanceToken && !isValidPid(pid)) {
+    throw new Error("Codex app-server broker PID is unavailable; persisted ownership state was preserved.");
+  }
+
+  let exited = pidAlreadyExited
+    ? true
+    : isValidPid(verifiedPid)
+    ? await waitForProcessExit(verifiedPid, {
+        timeoutMs: options.timeoutMs,
+        intervalMs: options.intervalMs,
+        killImpl: options.killImpl,
+        platform: options.platform
+      })
+    : session.endpoint
+      ? await waitForBrokerExit(session.endpoint, {
+          timeoutMs: options.timeoutMs,
+          intervalMs: options.intervalMs
+        })
+      : false;
+  let forced = false;
+
+  if (!exited && isValidPid(verifiedPid) && options.killProcess) {
+    const stillOwnsProcess = options.verifyProcess
+      ? options.verifyProcess(verifiedPid, session.instanceToken)
+      : processHasLaunchToken(verifiedPid, session.instanceToken, {
+          marker: "--instance-token",
+          platform: options.platform,
+          timeoutMs: options.timeoutMs,
+          runCommandImpl: options.runCommandImpl
+        });
+    if (!stillOwnsProcess) {
+      throw new Error("Codex app-server broker process ownership changed before forced shutdown.");
+    }
+    options.killProcess(verifiedPid);
+    forced = true;
+    exited = await waitForProcessExit(verifiedPid, {
+      timeoutMs: options.timeoutMs,
+      intervalMs: options.intervalMs,
+      killImpl: options.killImpl,
+      platform: options.platform
+    });
+  }
+
+  if (!exited) {
+    throw new Error(`Codex app-server broker ${verifiedPid ?? session.endpoint ?? "unknown"} did not exit.`);
+  }
+
+  teardownBrokerSession({
+    endpoint: session.endpoint ?? null,
+    pidFile: session.pidFile ?? null,
+    logFile: session.logFile ?? null,
+    sessionDir: session.sessionDir ?? null,
+    pid
+  });
+  clearBrokerSession(cwd);
+  return { found: true, exited: true, forced };
+}
+
 async function isBrokerEndpointReady(endpoint) {
   if (!endpoint) {
     return false;
@@ -117,15 +322,16 @@ export async function ensureBrokerSession(cwd, options = {}) {
   }
 
   if (existing) {
-    teardownBrokerSession({
-      endpoint: existing.endpoint ?? null,
-      pidFile: existing.pidFile ?? null,
-      logFile: existing.logFile ?? null,
-      sessionDir: existing.sessionDir ?? null,
-      pid: existing.pid ?? null,
-      killProcess: options.killProcess ?? null
+    await shutdownBrokerSession(cwd, {
+      session: existing,
+      killProcess: options.killProcess ?? terminateProcessTree,
+      verifyProcess: options.verifyProcess,
+      runCommandImpl: options.runCommandImpl,
+      killImpl: options.killImpl,
+      platform: options.platform,
+      timeoutMs: options.timeoutMs,
+      intervalMs: options.intervalMs
     });
-    clearBrokerSession(cwd);
   }
 
   const sessionDir = createBrokerSessionDir();
@@ -136,6 +342,7 @@ export async function ensureBrokerSession(cwd, options = {}) {
   const scriptPath =
     options.scriptPath ??
     fileURLToPath(new URL("../app-server-broker.mjs", import.meta.url));
+  const instanceToken = options.instanceToken ?? randomUUID();
 
   const child = spawnBrokerProcess({
     scriptPath,
@@ -143,35 +350,40 @@ export async function ensureBrokerSession(cwd, options = {}) {
     endpoint,
     pidFile,
     logFile,
+    instanceToken,
     env: options.env ?? process.env
   });
 
-  const ready = await waitForBrokerEndpoint(endpoint, options.timeoutMs ?? 2000);
-  if (!ready) {
-    teardownBrokerSession({
-      endpoint,
-      pidFile,
-      logFile,
-      sessionDir,
-      pid: child.pid ?? null,
-      killProcess: options.killProcess ?? null
-    });
-    return null;
-  }
-
   const session = {
     endpoint,
     pidFile,
     logFile,
     sessionDir,
-    pid: child.pid ?? null
+    pid: child.pid ?? null,
+    instanceToken
   };
   saveBrokerSession(cwd, session);
+
+  const ready = await waitForBrokerEndpoint(endpoint, options.timeoutMs ?? 2000);
+  if (!ready) {
+    await shutdownBrokerSession(cwd, {
+      session,
+      killProcess: options.killProcess ?? terminateProcessTree,
+      verifyProcess: options.verifyProcess,
+      runCommandImpl: options.runCommandImpl,
+      killImpl: options.killImpl,
+      platform: options.platform,
+      timeoutMs: options.timeoutMs,
+      intervalMs: options.intervalMs
+    });
+    return null;
+  }
+
   return session;
 }
 
 export function teardownBrokerSession({ endpoint = null, pidFile, logFile, sessionDir = null, pid = null, killProcess = null }) {
-  if (Number.isFinite(pid) && killProcess) {
+  if (isValidPid(pid) && killProcess) {
     try {
       killProcess(pid);
     } catch {
diff --git a/plugins/codex/scripts/lib/process.mjs b/plugins/codex/scripts/lib/process.mjs
index dd8fc375..72fe8748 100644
--- a/plugins/codex/scripts/lib/process.mjs
+++ b/plugins/codex/scripts/lib/process.mjs
@@ -1,6 +1,78 @@
 import { spawnSync } from "node:child_process";
 import process from "node:process";
 
+function isValidPid(pid) {
+  return Number.isSafeInteger(pid) && pid > 0;
+}
+
+function isProcessTreeRunning(pid, options = {}) {
+  if (!isValidPid(pid)) {
+    return false;
+  }
+  const platform = options.platform ?? process.platform;
+  const signalTarget = platform === "win32" ? pid : -pid;
+  const killImpl = options.killImpl ?? process.kill.bind(process);
+  try {
+    killImpl(signalTarget, 0);
+    return true;
+  } catch (error) {
+    if (error?.code === "ESRCH") {
+      return false;
+    }
+    if (error?.code === "EPERM") {
+      return true;
+    }
+    throw error;
+  }
+}
+
+export async function waitForProcessExit(pid, options = {}) {
+  if (!isValidPid(pid)) {
+    return false;
+  }
+  const timeoutMs = options.timeoutMs ?? 2000;
+  const intervalMs = options.intervalMs ?? 25;
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    if (!isProcessTreeRunning(pid, options)) {
+      return true;
+    }
+    await new Promise((resolve) => setTimeout(resolve, intervalMs));
+  }
+  return !isProcessTreeRunning(pid, options);
+}
+
+export function processHasLaunchToken(pid, token, options = {}) {
+  if (!isValidPid(pid) || typeof token !== "string" || token.length < 16) {
+    return false;
+  }
+
+  const platform = options.platform ?? process.platform;
+  const runCommandImpl = options.runCommandImpl ?? runCommand;
+  const result =
+    platform === "win32"
+      ? runCommandImpl(
+          "powershell.exe",
+          [
+            "-NoProfile",
+            "-NonInteractive",
+            "-Command",
+            `$p = Get-CimInstance Win32_Process -Filter 'ProcessId = ${pid}'; if ($null -ne $p) { [Console]::Out.Write($p.CommandLine) }`
+          ],
+          { timeout: options.timeoutMs ?? 2000, killSignal: "SIGTERM" }
+        )
+      : runCommandImpl("ps", ["-ww", "-p", String(pid), "-o", "command="], {
+          timeout: options.timeoutMs ?? 2000,
+          killSignal: "SIGTERM"
+        });
+
+  if (result.error || result.status !== 0) {
+    return false;
+  }
+  const commandLine = String(result.stdout ?? "");
+  return commandLine.includes(options.marker ?? "--worker-token") && commandLine.includes(token);
+}
+
 export function runCommand(command, args = [], options = {}) {
   const result = spawnSync(command, args, {
     cwd: options.cwd,
@@ -8,6 +80,8 @@ export function runCommand(command, args = [], options = {}) {
     encoding: "utf8",
     input: options.input,
     maxBuffer: options.maxBuffer,
+    timeout: options.timeout,
+    killSignal: options.killSignal,
     stdio: options.stdio ?? "pipe",
     shell: options.shell ?? (process.platform === "win32" ? (process.env.SHELL || true) : false),
     windowsHide: true
@@ -55,7 +129,7 @@ function looksLikeMissingProcessMessage(text) {
 }
 
 export function terminateProcessTree(pid, options = {}) {
-  if (!Number.isFinite(pid)) {
+  if (!isValidPid(pid)) {
     return { attempted: false, delivered: false, method: null };
   }
 
@@ -66,7 +140,9 @@ export function terminateProcessTree(pid, options = {}) {
   if (platform === "win32") {
     const result = runCommandImpl("taskkill", ["/PID", String(pid), "/T", "/F"], {
       cwd: options.cwd,
-      env: options.env
+      env: options.env,
+      timeout: options.timeoutMs ?? 2000,
+      killSignal: "SIGTERM"
     });
 
     if (!result.error && result.status === 0) {
diff --git a/plugins/codex/scripts/lib/state.mjs b/plugins/codex/scripts/lib/state.mjs
index 2da23498..6e7f313b 100644
--- a/plugins/codex/scripts/lib/state.mjs
+++ b/plugins/codex/scripts/lib/state.mjs
@@ -1,4 +1,4 @@
-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";
@@ -9,8 +9,13 @@ const STATE_VERSION = 1;
 const PLUGIN_DATA_ENV = "CLAUDE_PLUGIN_DATA";
 const FALLBACK_STATE_ROOT_DIR = path.join(os.tmpdir(), "codex-companion");
 const STATE_FILE_NAME = "state.json";
+const STATE_LOCK_DIR_NAME = "state.lock";
+const STATE_LOCK_OWNER_FILE_NAME = "owner";
 const JOBS_DIR_NAME = "jobs";
 const MAX_JOBS = 50;
+const LOCK_TIMEOUT_MS = 2000;
+const STALE_LOCK_MS = 30000;
+const sleepBuffer = new Int32Array(new SharedArrayBuffer(4));
 
 function nowIso() {
   return new Date().toISOString();
@@ -55,6 +60,91 @@ export function ensureStateDir(cwd) {
   fs.mkdirSync(resolveJobsDir(cwd), { recursive: true });
 }
 
+function processExists(pid) {
+  if (!Number.isSafeInteger(pid) || pid <= 0) return false;
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch (error) {
+    return error?.code === "EPERM";
+  }
+}
+
+function readLockOwner(lockDir) {
+  try {
+    return JSON.parse(fs.readFileSync(path.join(lockDir, STATE_LOCK_OWNER_FILE_NAME), "utf8"));
+  } catch {
+    return null;
+  }
+}
+
+function staleLockDescription(lockDir) {
+  try {
+    const ageMs = Date.now() - fs.statSync(lockDir).mtimeMs;
+    if (ageMs < STALE_LOCK_MS) return null;
+    const owner = readLockOwner(lockDir);
+    if (owner && processExists(owner.pid)) return null;
+    return owner?.pid ? `owner PID ${owner.pid} is not running` : "owner metadata is missing or invalid";
+  } catch (error) {
+    if (error?.code === "ENOENT") return null;
+    throw error;
+  }
+}
+
+function tryPublishStateLock(lockDir, owner) {
+  const candidateDir = `${lockDir}.candidate-${owner.token}`;
+  fs.mkdirSync(candidateDir);
+  fs.writeFileSync(
+    path.join(candidateDir, STATE_LOCK_OWNER_FILE_NAME),
+    `${JSON.stringify(owner)}\n`,
+    "utf8"
+  );
+  try {
+    fs.renameSync(candidateDir, lockDir);
+    return true;
+  } catch (error) {
+    fs.rmSync(candidateDir, { recursive: true, force: true });
+    if (error?.code === "EEXIST" || error?.code === "ENOTEMPTY") return false;
+    throw error;
+  }
+}
+
+function releaseStateLock(lockDir, ownerToken) {
+  const owner = readLockOwner(lockDir);
+  if (owner?.token !== ownerToken) {
+    throw new Error(`Codex companion state lock ownership changed before release: ${lockDir}`);
+  }
+  const releaseDir = `${lockDir}.release-${ownerToken}`;
+  fs.renameSync(lockDir, releaseDir);
+  fs.rmSync(releaseDir, { recursive: true, force: true });
+}
+
+function withStateLock(cwd, operation) {
+  ensureStateDir(cwd);
+  const lockDir = path.join(resolveStateDir(cwd), STATE_LOCK_DIR_NAME);
+  const start = Date.now();
+  const owner = { pid: process.pid, token: randomUUID(), createdAt: nowIso() };
+
+  while (!tryPublishStateLock(lockDir, owner)) {
+    const staleDescription = staleLockDescription(lockDir);
+    if (staleDescription) {
+      throw new Error(
+        `Stale Codex companion state lock requires verified manual removal (${staleDescription}): ${lockDir}`
+      );
+    }
+    if (Date.now() - start >= LOCK_TIMEOUT_MS) {
+      throw new Error(`Timed out waiting for Codex companion state lock: ${lockDir}`);
+    }
+    Atomics.wait(sleepBuffer, 0, 0, 10);
+  }
+
+  try {
+    return operation();
+  } finally {
+    releaseStateLock(lockDir, owner.token);
+  }
+}
+
 export function loadState(cwd) {
   const stateFile = resolveStateFile(cwd);
   if (!fs.existsSync(stateFile)) {
@@ -89,7 +179,7 @@ function removeFileIfExists(filePath) {
   }
 }
 
-export function saveState(cwd, state) {
+function saveStateUnlocked(cwd, state) {
   const previousJobs = loadState(cwd).jobs;
   ensureStateDir(cwd);
   const nextJobs = pruneJobs(state.jobs ?? []);
@@ -102,23 +192,30 @@ export function saveState(cwd, state) {
     jobs: nextJobs
   };
 
+  const stateFile = resolveStateFile(cwd);
+  const temporaryFile = `${stateFile}.${process.pid}.${Math.random().toString(36).slice(2)}.tmp`;
+  fs.writeFileSync(temporaryFile, `${JSON.stringify(nextState, null, 2)}\n`, "utf8");
+  fs.renameSync(temporaryFile, stateFile);
+
   const retainedIds = new Set(nextJobs.map((job) => job.id));
   for (const job of previousJobs) {
-    if (retainedIds.has(job.id)) {
-      continue;
-    }
+    if (retainedIds.has(job.id)) continue;
     removeJobFile(resolveJobFile(cwd, job.id));
     removeFileIfExists(job.logFile);
   }
-
-  fs.writeFileSync(resolveStateFile(cwd), `${JSON.stringify(nextState, null, 2)}\n`, "utf8");
   return nextState;
 }
 
+export function saveState(cwd, state) {
+  return withStateLock(cwd, () => saveStateUnlocked(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 saveStateUnlocked(cwd, state);
+  });
 }
 
 export function generateJobId(prefix = "job") {
diff --git a/plugins/codex/scripts/lib/tracked-jobs.mjs b/plugins/codex/scripts/lib/tracked-jobs.mjs
index 90286901..63298343 100644
--- a/plugins/codex/scripts/lib/tracked-jobs.mjs
+++ b/plugins/codex/scripts/lib/tracked-jobs.mjs
@@ -1,7 +1,8 @@
 import fs from "node:fs";
 import process from "node:process";
 
-import { readJobFile, resolveJobFile, resolveJobLogFile, upsertJob, writeJobFile } from "./state.mjs";
+import { processHasLaunchToken, terminateProcessTree, waitForProcessExit } from "./process.mjs";
+import { loadState, readJobFile, resolveJobFile, resolveJobLogFile, updateState, upsertJob, writeJobFile } from "./state.mjs";
 
 export const SESSION_ID_ENV = "CODEX_COMPANION_SESSION_ID";
 
@@ -131,6 +132,112 @@ export function createProgressReporter({ stderr = false, logFile = null, onEvent
   };
 }
 
+export async function waitForWorkerJob(workspaceRoot, jobId, workerToken, workerPid, options = {}) {
+  const timeoutMs = options.timeoutMs ?? 2000;
+  const intervalMs = options.intervalMs ?? 25;
+  const start = Date.now();
+
+  while (Date.now() - start < timeoutMs) {
+    const jobFile = resolveJobFile(workspaceRoot, jobId);
+    if (fs.existsSync(jobFile)) {
+      const storedJob = readJobFile(jobFile);
+      if (storedJob.workerToken !== workerToken) {
+        throw new Error(`Stored job ${jobId} worker identity does not match this process.`);
+      }
+      const indexedJob = loadState(workspaceRoot).jobs.find((job) => job.id === jobId);
+      if (
+        storedJob.pid === workerPid &&
+        indexedJob?.pid === workerPid &&
+        indexedJob.workerToken === workerToken
+      ) {
+        return storedJob;
+      }
+      if (storedJob.pid !== null && storedJob.pid !== undefined && storedJob.pid !== workerPid) {
+        throw new Error(`Stored job ${jobId} belongs to worker process ${storedJob.pid}, not ${workerPid}.`);
+      }
+      if (
+        indexedJob?.pid !== null &&
+        indexedJob?.pid !== undefined &&
+        indexedJob.pid !== workerPid
+      ) {
+        throw new Error(`Indexed job ${jobId} belongs to worker process ${indexedJob.pid}, not ${workerPid}.`);
+      }
+    }
+    await new Promise((resolve) => setTimeout(resolve, intervalMs));
+  }
+
+  throw new Error(`Timed out waiting for stored job ${jobId} to register worker process ${workerPid}.`);
+}
+
+export async function cleanupSessionJobs(workspaceRoot, sessionId, options = {}) {
+  if (!sessionId && !options.all) {
+    return { removed: [], retained: [] };
+  }
+
+  const state = loadState(workspaceRoot);
+  const removedJobs = [];
+  const retainedJobs = [];
+  const failures = [];
+
+  for (const job of state.jobs) {
+    const belongsToSession = options.all || job.sessionId === sessionId;
+    if (!belongsToSession) {
+      continue;
+    }
+
+    const isActive = job.status === "queued" || job.status === "running";
+    if (isActive) {
+      try {
+        const ownsProcess = options.verifyProcess
+          ? options.verifyProcess(job.pid, job.workerToken)
+          : processHasLaunchToken(job.pid, job.workerToken, {
+              platform: options.platform,
+              timeoutMs: options.timeoutMs,
+              runCommandImpl: options.runCommandImpl
+            });
+        if (!ownsProcess) {
+          throw new Error(
+            `Cannot verify that process ${job.pid ?? "unknown"} owns Codex job ${job.id}; state was preserved.`
+          );
+        }
+        terminateProcessTree(job.pid, {
+          platform: options.platform,
+          timeoutMs: options.timeoutMs,
+          runCommandImpl: options.runCommandImpl,
+          killImpl: options.killImpl
+        });
+        const exited = await waitForProcessExit(job.pid, {
+          platform: options.platform,
+          timeoutMs: options.timeoutMs,
+          intervalMs: options.intervalMs,
+          killImpl: options.killImpl
+        });
+        if (!exited) {
+          throw new Error(`Codex job ${job.id} process ${job.pid ?? "unknown"} did not exit.`);
+        }
+      } catch (error) {
+        retainedJobs.push(job);
+        failures.push(error);
+        continue;
+      }
+    }
+
+    removedJobs.push(job);
+  }
+
+  const removedIds = new Set(removedJobs.map((job) => job.id));
+  updateState(workspaceRoot, (freshState) => {
+    freshState.jobs = freshState.jobs.filter((job) => !removedIds.has(job.id));
+  });
+  if (failures.length > 0) {
+    throw new AggregateError(failures, "Failed to stop all Codex session jobs.");
+  }
+  return {
+    removed: removedJobs.map((job) => job.id),
+    retained: retainedJobs.map((job) => job.id)
+  };
+}
+
 function readStoredJobOrNull(workspaceRoot, jobId) {
   const jobFile = resolveJobFile(workspaceRoot, jobId);
   if (!fs.existsSync(jobFile)) {
diff --git a/plugins/codex/scripts/session-lifecycle-hook.mjs b/plugins/codex/scripts/session-lifecycle-hook.mjs
index 778571e6..d2a8a176 100644
--- a/plugins/codex/scripts/session-lifecycle-hook.mjs
+++ b/plugins/codex/scripts/session-lifecycle-hook.mjs
@@ -4,17 +4,12 @@ import fs from "node:fs";
 import process from "node:process";
 
 import { terminateProcessTree } from "./lib/process.mjs";
-import { BROKER_ENDPOINT_ENV } from "./lib/app-server.mjs";
 import {
-  clearBrokerSession,
-  LOG_FILE_ENV,
   loadBrokerSession,
-  PID_FILE_ENV,
-  sendBrokerShutdown,
-  teardownBrokerSession
+  shutdownBrokerSession
 } from "./lib/broker-lifecycle.mjs";
-import { loadState, resolveStateFile, saveState } from "./lib/state.mjs";
 import { TRANSCRIPT_PATH_ENV } from "./lib/claude-session-transfer.mjs";
+import { cleanupSessionJobs } from "./lib/tracked-jobs.mjs";
 import { resolveWorkspaceRoot } from "./lib/workspace.mjs";
 
 export const SESSION_ID_ENV = "CODEX_COMPANION_SESSION_ID";
@@ -39,41 +34,6 @@ function appendEnvVar(name, value) {
   fs.appendFileSync(process.env.CLAUDE_ENV_FILE, `export ${name}=${shellEscape(value)}\n`, "utf8");
 }
 
-function cleanupSessionJobs(cwd, sessionId) {
-  if (!cwd || !sessionId) {
-    return;
-  }
-
-  const workspaceRoot = resolveWorkspaceRoot(cwd);
-  const stateFile = resolveStateFile(workspaceRoot);
-  if (!fs.existsSync(stateFile)) {
-    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;
-    }
-    try {
-      terminateProcessTree(job.pid ?? Number.NaN);
-    } catch {
-      // Ignore teardown failures during session shutdown.
-    }
-  }
-
-  saveState(workspaceRoot, {
-    ...state,
-    jobs: state.jobs.filter((job) => job.sessionId !== sessionId)
-  });
-}
-
 function handleSessionStart(input) {
   appendEnvVar(SESSION_ID_ENV, input.session_id);
   appendEnvVar(TRANSCRIPT_PATH_ENV, input.transcript_path);
@@ -82,35 +42,25 @@ function handleSessionStart(input) {
 
 async function handleSessionEnd(input) {
   const cwd = input.cwd || process.cwd();
-  const brokerSession =
-    loadBrokerSession(cwd) ??
-    (process.env[BROKER_ENDPOINT_ENV]
-      ? {
-          endpoint: process.env[BROKER_ENDPOINT_ENV],
-          pidFile: process.env[PID_FILE_ENV] ?? null,
-          logFile: process.env[LOG_FILE_ENV] ?? null
-        }
-      : null);
-  const brokerEndpoint = brokerSession?.endpoint ?? null;
-  const pidFile = brokerSession?.pidFile ?? null;
-  const logFile = brokerSession?.logFile ?? null;
-  const sessionDir = brokerSession?.sessionDir ?? null;
-  const pid = brokerSession?.pid ?? null;
-
-  if (brokerEndpoint) {
-    await sendBrokerShutdown(brokerEndpoint);
+  const workspaceRoot = resolveWorkspaceRoot(cwd);
+  const brokerSession = loadBrokerSession(cwd);
+  const failures = [];
+  try {
+    await cleanupSessionJobs(workspaceRoot, input.session_id || process.env[SESSION_ID_ENV]);
+  } catch (error) {
+    failures.push(error);
+  }
+  try {
+    await shutdownBrokerSession(cwd, {
+      session: brokerSession,
+      killProcess: terminateProcessTree
+    });
+  } catch (error) {
+    failures.push(error);
+  }
+  if (failures.length > 0) {
+    throw new AggregateError(failures, "Codex session cleanup did not finish.");
   }
-
-  cleanupSessionJobs(cwd, input.session_id || process.env[SESSION_ID_ENV]);
-  teardownBrokerSession({
-    endpoint: brokerEndpoint,
-    pidFile,
-    logFile,
-    sessionDir,
-    pid,
-    killProcess: terminateProcessTree
-  });
-  clearBrokerSession(cwd);
 }
 
 async function main() {
diff --git a/tests/broker-lifecycle.test.mjs b/tests/broker-lifecycle.test.mjs
new file mode 100644
index 00000000..0a489c9b
--- /dev/null
+++ b/tests/broker-lifecycle.test.mjs
@@ -0,0 +1,288 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import net from "node:net";
+import path from "node:path";
+import process from "node:process";
+
+import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs";
+import { initGitRepo, makeTempDir } from "./helpers.mjs";
+import {
+  ensureBrokerSession,
+  loadBrokerSession,
+  saveBrokerSession,
+  shutdownBrokerSession
+} from "../plugins/codex/scripts/lib/broker-lifecycle.mjs";
+import { terminateProcessTree, waitForProcessExit } from "../plugins/codex/scripts/lib/process.mjs";
+
+function isProcessRunning(pid) {
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch (error) {
+    return error?.code === "EPERM";
+  }
+}
+
+function isProcessGroupRunning(pid) {
+  return isProcessRunning(-pid);
+}
+
+test("shutdown waits for the broker and app-server child to exit", { skip: process.platform === "win32" }, async (t) => {
+  const repo = makeTempDir();
+  const binDir = makeTempDir();
+  let session = null;
+  t.after(async () => {
+    if (session?.pid && isProcessGroupRunning(session.pid)) {
+      terminateProcessTree(session.pid);
+      await waitForProcessExit(session.pid);
+    }
+  });
+
+  installFakeCodex(binDir);
+  initGitRepo(repo);
+
+  session = await ensureBrokerSession(repo, {
+    env: buildEnv(binDir),
+    killProcess: terminateProcessTree
+  });
+  assert.ok(session?.pid);
+  const fakeState = JSON.parse(fs.readFileSync(path.join(binDir, "fake-codex-state.json"), "utf8"));
+  assert.equal(fakeState.appServerStarts, 1);
+  assert.equal(isProcessGroupRunning(session.pid), true);
+
+  const outcome = await shutdownBrokerSession(repo, {
+    session: { ...session, pid: null },
+    killProcess: terminateProcessTree
+  });
+
+  assert.equal(outcome.found, true);
+  assert.equal(outcome.exited, true);
+  assert.equal(isProcessRunning(session.pid), false);
+  assert.equal(isProcessGroupRunning(session.pid), false);
+});
+
+test("shutdown refuses endpoint-only state without an instance token", { skip: process.platform === "win32" }, async (t) => {
+  const repo = makeTempDir();
+  const binDir = makeTempDir();
+  let session = null;
+  t.after(async () => {
+    if (session?.pid && isProcessGroupRunning(session.pid)) {
+      terminateProcessTree(session.pid);
+      await waitForProcessExit(session.pid);
+    }
+  });
+
+  installFakeCodex(binDir);
+  initGitRepo(repo);
+  session = await ensureBrokerSession(repo, {
+    env: buildEnv(binDir),
+    killProcess: terminateProcessTree
+  });
+  assert.ok(session?.pid);
+  assert.equal(isProcessGroupRunning(session.pid), true);
+
+  await assert.rejects(
+    shutdownBrokerSession(repo, {
+      session: { ...session, pid: null, pidFile: null, instanceToken: null },
+      killProcess: terminateProcessTree
+    }),
+    /ownership could not be verified/i
+  );
+  assert.equal(isProcessGroupRunning(session.pid), true);
+});
+
+test("shutdown rejects broker PID state that conflicts with the PID file", { skip: process.platform === "win32" }, async (t) => {
+  const repo = makeTempDir();
+  const binDir = makeTempDir();
+  let session = null;
+  t.after(async () => {
+    if (session?.pid && isProcessGroupRunning(session.pid)) {
+      terminateProcessTree(session.pid);
+      await waitForProcessExit(session.pid);
+    }
+  });
+
+  installFakeCodex(binDir);
+  initGitRepo(repo);
+  session = await ensureBrokerSession(repo, {
+    env: buildEnv(binDir),
+    killProcess: terminateProcessTree
+  });
+  assert.ok(session?.pid);
+
+  await assert.rejects(
+    shutdownBrokerSession(repo, {
+      session: { ...session, pid: session.pid + 1 },
+      killProcess: terminateProcessTree
+    }),
+    /PID mismatch/
+  );
+  assert.equal(isProcessGroupRunning(session.pid), true);
+});
+
+test("shutdown never force-kills when the broker token does not match", { skip: process.platform === "win32" }, async (t) => {
+  const repo = makeTempDir();
+  const binDir = makeTempDir();
+  let session = null;
+  t.after(async () => {
+    if (session?.pid && isProcessGroupRunning(session.pid)) {
+      terminateProcessTree(session.pid);
+      await waitForProcessExit(session.pid);
+    }
+  });
+
+  installFakeCodex(binDir);
+  initGitRepo(repo);
+  session = await ensureBrokerSession(repo, {
+    env: buildEnv(binDir),
+    killProcess: terminateProcessTree
+  });
+  assert.ok(session?.pid);
+  let forceKills = 0;
+
+  await assert.rejects(
+    shutdownBrokerSession(repo, {
+      session: { ...session, instanceToken: "wrong-instance-token" },
+      killProcess() {
+        forceKills += 1;
+      }
+    }),
+    /rejected.*identity/i
+  );
+  assert.equal(forceKills, 0);
+  assert.equal(isProcessGroupRunning(session.pid), true);
+});
+
+test("shutdown preserves a live mismatched endpoint when the persisted PID is stale", { skip: process.platform === "win32" }, async (t) => {
+  const repo = makeTempDir();
+  const binDir = makeTempDir();
+  let session = null;
+  t.after(async () => {
+    if (session?.pid && isProcessGroupRunning(session.pid)) {
+      terminateProcessTree(session.pid);
+      await waitForProcessExit(session.pid);
+    }
+  });
+
+  installFakeCodex(binDir);
+  initGitRepo(repo);
+  session = await ensureBrokerSession(repo, {
+    env: buildEnv(binDir),
+    killProcess: terminateProcessTree
+  });
+  assert.ok(session?.pid);
+  const socketPath = session.endpoint.slice("unix:".length);
+
+  await assert.rejects(
+    shutdownBrokerSession(repo, {
+      session: {
+        ...session,
+        pid: 999999,
+        pidFile: null,
+        instanceToken: "wrong-instance-token"
+      },
+      killProcess: terminateProcessTree
+    }),
+    /rejected.*identity/i
+  );
+
+  assert.equal(isProcessGroupRunning(session.pid), true);
+  assert.equal(fs.existsSync(socketPath), true);
+  assert.ok(loadBrokerSession(repo));
+});
+
+test("forced shutdown re-verifies broker ownership before signalling", { skip: process.platform === "win32" }, async (t) => {
+  const sessionDir = makeTempDir("cxc-force-check-");
+  const socketPath = path.join(sessionDir, "broker.sock");
+  const endpoint = `unix:${socketPath}`;
+  const instanceToken = "instance-token-1234567890";
+  const server = net.createServer((socket) => {
+    socket.setEncoding("utf8");
+    socket.on("data", () => {
+      socket.write(`${JSON.stringify({ id: 1, result: { pid: 123, instanceToken } })}\n`);
+    });
+  });
+  await new Promise((resolve) => server.listen(socketPath, resolve));
+  t.after(async () => {
+    await new Promise((resolve) => server.close(resolve));
+    if (fs.existsSync(socketPath)) {
+      fs.unlinkSync(socketPath);
+    }
+  });
+  let forceKills = 0;
+
+  await assert.rejects(
+    shutdownBrokerSession(sessionDir, {
+      session: { endpoint, pid: 123, pidFile: null, logFile: null, sessionDir, instanceToken },
+      timeoutMs: 25,
+      intervalMs: 5,
+      killImpl() {},
+      verifyProcess() {
+        return false;
+      },
+      killProcess() {
+        forceKills += 1;
+      }
+    }),
+    /ownership changed/
+  );
+  assert.equal(forceKills, 0);
+});
+
+test("shutdown rejects an unverifiable broker without deleting its state", async () => {
+  const cwd = makeTempDir();
+  const sessionDir = makeTempDir("cxc-invalid-");
+  const pidFile = path.join(sessionDir, "broker.pid");
+  const logFile = path.join(sessionDir, "broker.log");
+  fs.writeFileSync(pidFile, "123garbage\n", "utf8");
+  fs.writeFileSync(logFile, "keep for diagnosis\n", "utf8");
+  let killCalls = 0;
+
+  await assert.rejects(
+    shutdownBrokerSession(cwd, {
+      session: { endpoint: null, pid: 0, pidFile, logFile, sessionDir },
+      killProcess() {
+        killCalls += 1;
+      },
+      timeoutMs: 25
+    }),
+    /did not exit/
+  );
+  assert.equal(killCalls, 0);
+  assert.equal(fs.existsSync(pidFile), true);
+  assert.equal(fs.existsSync(logFile), true);
+});
+
+test("ensure preserves an unavailable persisted broker when ownership cannot be verified", async () => {
+  const cwd = makeTempDir();
+  const sessionDir = makeTempDir("cxc-unavailable-");
+  const pidFile = path.join(sessionDir, "broker.pid");
+  const logFile = path.join(sessionDir, "broker.log");
+  const session = {
+    endpoint: `unix:${path.join(sessionDir, "missing.sock")}`,
+    pid: 123,
+    pidFile,
+    logFile,
+    sessionDir,
+    instanceToken: "instance-token-unavailable-1234"
+  };
+  fs.writeFileSync(pidFile, "123\n", "utf8");
+  fs.writeFileSync(logFile, "keep for diagnosis\n", "utf8");
+  saveBrokerSession(cwd, session);
+
+  await assert.rejects(
+    ensureBrokerSession(cwd, {
+      timeoutMs: 25,
+      killImpl() {},
+      verifyProcess() {
+        return false;
+      }
+    }),
+    /ownership.*could not be verified/i
+  );
+
+  assert.deepEqual(loadBrokerSession(cwd), session);
+  assert.equal(fs.existsSync(pidFile), true);
+  assert.equal(fs.existsSync(logFile), true);
+});
diff --git a/tests/helpers.mjs b/tests/helpers.mjs
index d6981197..5a9abd71 100644
--- a/tests/helpers.mjs
+++ b/tests/helpers.mjs
@@ -4,8 +4,18 @@ import path from "node:path";
 import process from "node:process";
 import { spawnSync } from "node:child_process";
 
+const tempDirs = new Set();
+
 export function makeTempDir(prefix = "codex-plugin-test-") {
-  return fs.mkdtempSync(path.join(os.tmpdir(), prefix));
+  const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
+  tempDirs.add(tempDir);
+  return tempDir;
+}
+
+export function consumeTempDirs() {
+  const dirs = [...tempDirs];
+  tempDirs.clear();
+  return dirs;
 }
 
 export function writeExecutable(filePath, source) {
diff --git a/tests/process.test.mjs b/tests/process.test.mjs
index 80e0715b..35845a68 100644
--- a/tests/process.test.mjs
+++ b/tests/process.test.mjs
@@ -1,14 +1,107 @@
 import test from "node:test";
 import assert from "node:assert/strict";
 
-import { terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs";
+import { processHasLaunchToken, terminateProcessTree } from "../plugins/codex/scripts/lib/process.mjs";
+
+test("processHasLaunchToken verifies the recorded worker token", () => {
+  let captured = null;
+  const verified = processHasLaunchToken(1234, "worker-token-1234567890", {
+    platform: "darwin",
+    runCommandImpl(command, args, options) {
+      captured = { command, args, options };
+      return {
+        command,
+        args,
+        status: 0,
+        signal: null,
+        stdout: "node worker.mjs --worker-token worker-token-1234567890",
+        stderr: "",
+        error: null
+      };
+    }
+  });
+
+  assert.equal(verified, true);
+  assert.deepEqual(captured, {
+    command: "ps",
+    args: ["-ww", "-p", "1234", "-o", "command="],
+    options: { timeout: 2000, killSignal: "SIGTERM" }
+  });
+});
+
+test("processHasLaunchToken fails closed when the token is absent", () => {
+  const verified = processHasLaunchToken(1234, "worker-token-1234567890", {
+    platform: "darwin",
+    runCommandImpl(command, args) {
+      return {
+        command,
+        args,
+        status: 0,
+        signal: null,
+        stdout: "node unrelated.mjs",
+        stderr: "",
+        error: null
+      };
+    }
+  });
+  assert.equal(verified, false);
+});
+
+test("processHasLaunchToken uses bounded PowerShell lookup on Windows", () => {
+  let captured = null;
+  const verified = processHasLaunchToken(1234, "worker-token-1234567890", {
+    platform: "win32",
+    timeoutMs: 750,
+    runCommandImpl(command, args, options) {
+      captured = { command, args, options };
+      return {
+        command,
+        args,
+        status: 0,
+        signal: null,
+        stdout: "node worker.mjs --worker-token worker-token-1234567890",
+        stderr: "",
+        error: null
+      };
+    }
+  });
+
+  assert.equal(verified, true);
+  assert.equal(captured.command, "powershell.exe");
+  assert.deepEqual(captured.args.slice(0, 3), ["-NoProfile", "-NonInteractive", "-Command"]);
+  assert.match(captured.args[3], /ProcessId = 1234/);
+  assert.deepEqual(captured.options, { timeout: 750, killSignal: "SIGTERM" });
+});
+
+test("processHasLaunchToken fails closed when Windows lookup times out", () => {
+  const verified = processHasLaunchToken(1234, "worker-token-1234567890", {
+    platform: "win32",
+    runCommandImpl(command, args) {
+      const error = new Error("timed out");
+      error.code = "ETIMEDOUT";
+      return { command, args, status: 1, signal: "SIGTERM", stdout: "", stderr: "", error };
+    }
+  });
+  assert.equal(verified, false);
+});
+
+test("terminateProcessTree rejects unsafe process IDs", () => {
+  for (const pid of [Number.NaN, 0, -1, 1.5]) {
+    const outcome = terminateProcessTree(pid, {
+      killImpl() {
+        throw new Error("invalid process IDs must not be signalled");
+      }
+    });
+    assert.deepEqual(outcome, { attempted: false, delivered: false, method: null });
+  }
+});
 
 test("terminateProcessTree uses taskkill on Windows", () => {
   let captured = null;
   const outcome = terminateProcessTree(1234, {
     platform: "win32",
-    runCommandImpl(command, args) {
-      captured = { command, args };
+    runCommandImpl(command, args, options) {
+      captured = { command, args, options };
       return {
         command,
         args,
@@ -26,7 +119,13 @@ test("terminateProcessTree uses taskkill on Windows", () => {
 
   assert.deepEqual(captured, {
     command: "taskkill",
-    args: ["/PID", "1234", "/T", "/F"]
+    args: ["/PID", "1234", "/T", "/F"],
+    options: {
+      cwd: undefined,
+      env: undefined,
+      timeout: 2000,
+      killSignal: "SIGTERM"
+    }
   });
   assert.equal(outcome.delivered, true);
   assert.equal(outcome.method, "taskkill");
diff --git a/tests/runtime.test.mjs b/tests/runtime.test.mjs
index 8f276835..b67d909b 100644
--- a/tests/runtime.test.mjs
+++ b/tests/runtime.test.mjs
@@ -1,14 +1,23 @@
 import fs from "node:fs";
 import path from "node:path";
-import test from "node:test";
+import test, { afterEach } from "node:test";
 import assert from "node:assert/strict";
-import { spawn } from "node:child_process";
 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 { consumeTempDirs, initGitRepo, makeTempDir, run } from "./helpers.mjs";
+import {
+  clearBrokerSession,
+  loadBrokerSession,
+  saveBrokerSession,
+  shutdownBrokerSession
+} from "../plugins/codex/scripts/lib/broker-lifecycle.mjs";
+import {
+  processHasLaunchToken,
+  terminateProcessTree,
+  waitForProcessExit
+} from "../plugins/codex/scripts/lib/process.mjs";
+import { loadState, resolveStateDir } from "../plugins/codex/scripts/lib/state.mjs";
 
 const ROOT = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..");
 const PLUGIN_ROOT = path.join(ROOT, "plugins", "codex");
@@ -16,6 +25,49 @@ const SCRIPT = path.join(PLUGIN_ROOT, "scripts", "codex-companion.mjs");
 const STOP_HOOK = path.join(PLUGIN_ROOT, "scripts", "stop-review-gate-hook.mjs");
 const SESSION_HOOK = path.join(PLUGIN_ROOT, "scripts", "session-lifecycle-hook.mjs");
 
+async function cleanupActiveTestProcesses(cwd) {
+  const failures = [];
+  for (const job of loadState(cwd).jobs) {
+    const isActive = job.status === "queued" || job.status === "running";
+    if (!isActive || !Number.isSafeInteger(job.pid) || job.pid <= 0) {
+      continue;
+    }
+    try {
+      if (!processHasLaunchToken(job.pid, job.workerToken)) {
+        throw new Error(`Cannot verify test job ${job.id} process ${job.pid}; refusing unsafe cleanup.`);
+      }
+      terminateProcessTree(job.pid);
+      if (!(await waitForProcessExit(job.pid))) {
+        throw new Error(`Test job ${job.id} process ${job.pid} did not exit.`);
+      }
+    } catch (error) {
+      failures.push(error);
+    }
+  }
+  if (failures.length > 0) {
+    throw new AggregateError(failures, "Failed to stop all test job processes.");
+  }
+}
+
+afterEach(async () => {
+  const failures = [];
+  for (const cwd of consumeTempDirs()) {
+    try {
+      await cleanupActiveTestProcesses(cwd);
+    } catch (error) {
+      failures.push(error);
+    }
+    try {
+      await shutdownBrokerSession(cwd, { killProcess: terminateProcessTree });
+    } catch (error) {
+      failures.push(error);
+    }
+  }
+  if (failures.length > 0) {
+    throw new AggregateError(failures, "Failed to stop all test broker processes.");
+  }
+});
+
 async function waitFor(predicate, { timeoutMs = 5000, intervalMs = 50 } = {}) {
   const start = Date.now();
   while (Date.now() - start < timeoutMs) {
@@ -28,6 +80,23 @@ async function waitFor(predicate, { timeoutMs = 5000, intervalMs = 50 } = {}) {
   throw new Error("Timed out waiting for condition.");
 }
 
+function launchDetachedSleeper(cwd, workerToken) {
+  const launcher = run(process.execPath, [
+    "-e",
+    `const { spawn } = require("node:child_process");
+const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)", "--", "--worker-token", ${JSON.stringify(workerToken)}], {
+  detached: true,
+  stdio: "ignore"
+});
+child.unref();
+process.stdout.write(String(child.pid));`
+  ], { cwd });
+  assert.equal(launcher.status, 0, launcher.stderr);
+  const pid = Number.parseInt(launcher.stdout, 10);
+  assert.ok(Number.isSafeInteger(pid));
+  return pid;
+}
+
 test("setup reports ready when fake codex is installed and authenticated", () => {
   const binDir = makeTempDir();
   installFakeCodex(binDir);
@@ -907,9 +976,7 @@ test("task using the shared broker still completes when Codex spawns subagents",
   });
   assert.equal(review.status, 0, review.stderr);
 
-  if (!loadBrokerSession(repo)) {
-    return;
-  }
+  assert.ok(loadBrokerSession(repo), "review must start the shared broker");
 
   const result = run("node", [SCRIPT, "task", "challenge the current design"], {
     cwd: repo,
@@ -1545,12 +1612,8 @@ test("cancel stops an active background job and marks it cancelled", async (t) =
   const jobsDir = path.join(stateDir, "jobs");
   fs.mkdirSync(jobsDir, { recursive: true });
 
-  const sleeper = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], {
-    cwd: workspace,
-    detached: true,
-    stdio: "ignore"
-  });
-  sleeper.unref();
+  const workerToken = "worker-token-cancel-1234567890";
+  const sleeper = { pid: launchDetachedSleeper(workspace, workerToken) };
 
   t.after(() => {
     try {
@@ -1595,6 +1658,7 @@ test("cancel stops an active background job and marks it cancelled", async (t) =
             jobClass: "task",
             summary: "Investigate flaky test",
             pid: sleeper.pid,
+            workerToken,
             logFile,
             createdAt: "2026-03-18T15:30:00.000Z",
             startedAt: "2026-03-18T15:30:01.000Z",
@@ -1689,7 +1753,7 @@ 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 preserves an unverifiable active job from another session", () => {
   const workspace = makeTempDir();
   const stateDir = resolveStateDir(workspace);
   const jobsDir = path.join(stateDir, "jobs");
@@ -1730,11 +1794,11 @@ test("cancel with a job id can still target an active job from another Claude se
     cwd: workspace,
     env
   });
-  assert.equal(cancel.status, 0, cancel.stderr);
-  assert.equal(JSON.parse(cancel.stdout).jobId, "task-other");
+  assert.equal(cancel.status, 1);
+  assert.match(cancel.stderr, /Cannot verify that process unknown owns Codex job task-other/);
 
   const state = JSON.parse(fs.readFileSync(path.join(stateDir, "state.json"), "utf8"));
-  assert.equal(state.jobs[0].status, "cancelled");
+  assert.equal(state.jobs[0].status, "running");
 });
 
 test("cancel sends turn interrupt to the shared app-server before killing a brokered task", async () => {
@@ -1824,12 +1888,8 @@ test("session end fully cleans up jobs for the ending session", async (t) => {
   fs.writeFileSync(completedJobFile, JSON.stringify({ id: "review-completed" }, null, 2), "utf8");
   fs.writeFileSync(otherJobFile, JSON.stringify({ id: "review-other" }, null, 2), "utf8");
 
-  const sleeper = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], {
-    cwd: repo,
-    detached: true,
-    stdio: "ignore"
-  });
-  sleeper.unref();
+  const workerToken = "worker-token-session-end-1234";
+  const sleeper = { pid: launchDetachedSleeper(repo, workerToken) };
   fs.writeFileSync(runningJobFile, JSON.stringify({ id: "review-running" }, null, 2), "utf8");
 
   t.after(() => {
@@ -1866,6 +1926,7 @@ test("session end fully cleans up jobs for the ending session", async (t) => {
             title: "Codex Review",
             sessionId: "sess-current",
             pid: sleeper.pid,
+            workerToken,
             logFile: runningLog,
             createdAt: "2026-03-18T15:32:00.000Z",
             updatedAt: "2026-03-18T15:33:00.000Z"
@@ -2137,9 +2198,7 @@ test("commands lazily start and reuse one shared app-server after first use", as
   assert.equal(review.status, 0, review.stderr);
 
   const brokerSession = loadBrokerSession(repo);
-  if (!brokerSession) {
-    return;
-  }
+  assert.ok(brokerSession, "review must start the shared broker");
 
   const adversarial = run("node", [SCRIPT, "adversarial-review"], {
     cwd: repo,
@@ -2182,9 +2241,7 @@ test("setup reuses an existing shared app-server without starting another one",
   assert.equal(review.status, 0, review.stderr);
 
   const brokerSession = loadBrokerSession(repo);
-  if (!brokerSession) {
-    return;
-  }
+  assert.ok(brokerSession, "review must start the shared broker");
 
   const setup = run("node", [SCRIPT, "setup", "--json"], {
     cwd: repo,
@@ -2222,9 +2279,7 @@ test("status reports shared session runtime when a lazy broker is active", () =>
   });
   assert.equal(review.status, 0, review.stderr);
 
-  if (!loadBrokerSession(repo)) {
-    return;
-  }
+  assert.ok(loadBrokerSession(repo), "review must start the shared broker");
 
   const result = run("node", [SCRIPT, "status"], {
     cwd: repo,
@@ -2256,4 +2311,5 @@ test("setup and status honor --cwd when reading shared session runtime", () => {
   const payload = JSON.parse(setup.stdout);
   assert.equal(payload.sessionRuntime.mode, "shared");
   assert.equal(payload.sessionRuntime.endpoint, "unix:/tmp/fake-broker.sock");
+  clearBrokerSession(targetWorkspace);
 });
diff --git a/tests/state.test.mjs b/tests/state.test.mjs
index 0f8f57ce..4c9e347d 100644
--- a/tests/state.test.mjs
+++ b/tests/state.test.mjs
@@ -102,4 +102,25 @@ test("saveState prunes dropped job artifacts when indexed jobs exceed the cap",
       .flatMap((jobId) => [`${jobId}.json`, `${jobId}.log`])
       .sort()
   );
+  assert.equal(fs.existsSync(path.join(resolveStateDir(workspace), "state.lock")), false);
+});
+
+test("saveState fails closed on a stale lock instead of reclaiming it unsafely", () => {
+  const workspace = makeTempDir();
+  const stateDir = resolveStateDir(workspace);
+  const lockDir = path.join(stateDir, "state.lock");
+  fs.mkdirSync(lockDir, { recursive: true });
+  fs.writeFileSync(
+    path.join(lockDir, "owner"),
+    `${JSON.stringify({ pid: 999999, token: "stale-owner-token", createdAt: "2026-01-01T00:00:00.000Z" })}\n`,
+    "utf8"
+  );
+  const staleTime = new Date(Date.now() - 60000);
+  fs.utimesSync(lockDir, staleTime, staleTime);
+
+  assert.throws(
+    () => saveState(workspace, { jobs: [] }),
+    /requires verified manual removal.*owner PID 999999 is not running/i
+  );
+  assert.equal(fs.existsSync(lockDir), true);
 });
diff --git a/tests/tracked-jobs.test.mjs b/tests/tracked-jobs.test.mjs
new file mode 100644
index 00000000..9c24482e
--- /dev/null
+++ b/tests/tracked-jobs.test.mjs
@@ -0,0 +1,155 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import path from "node:path";
+
+import { makeTempDir } from "./helpers.mjs";
+import { loadState, resolveJobFile, saveState, writeJobFile } from "../plugins/codex/scripts/lib/state.mjs";
+import { cleanupSessionJobs, waitForWorkerJob } from "../plugins/codex/scripts/lib/tracked-jobs.mjs";
+
+test("session cleanup waits for active jobs before removing their state", async () => {
+  const workspace = makeTempDir();
+  const logFile = path.join(workspace, "active.log");
+  const otherLogFile = path.join(workspace, "other.log");
+  fs.writeFileSync(logFile, "active\n", "utf8");
+  fs.writeFileSync(otherLogFile, "other\n", "utf8");
+  const activeJob = {
+    id: "task-active",
+    status: "running",
+    sessionId: "session-a",
+    pid: 123,
+    workerToken: "worker-token-1234567890",
+    logFile
+  };
+  const otherJob = {
+    id: "task-other",
+    status: "completed",
+    sessionId: "session-b",
+    pid: null,
+    logFile: otherLogFile
+  };
+  saveState(workspace, { jobs: [activeJob, otherJob] });
+  writeJobFile(workspace, activeJob.id, activeJob);
+  writeJobFile(workspace, otherJob.id, otherJob);
+
+  let running = true;
+  const outcome = await cleanupSessionJobs(workspace, "session-a", {
+    platform: "darwin",
+    verifyProcess(pid, token) {
+      return pid === activeJob.pid && token === activeJob.workerToken;
+    },
+    killImpl(target, signal) {
+      assert.equal(target, -123);
+      if (signal === 0 && !running) {
+        const error = new Error("missing");
+        error.code = "ESRCH";
+        throw error;
+      }
+      if (signal === "SIGTERM") {
+        running = false;
+      }
+    }
+  });
+
+  assert.deepEqual(outcome.removed, [activeJob.id]);
+  assert.deepEqual(loadState(workspace).jobs.map((job) => job.id), [otherJob.id]);
+  assert.equal(fs.existsSync(resolveJobFile(workspace, activeJob.id)), false);
+  assert.equal(fs.existsSync(logFile), false);
+  assert.equal(fs.existsSync(resolveJobFile(workspace, otherJob.id)), true);
+  assert.equal(fs.existsSync(otherLogFile), true);
+});
+
+test("session cleanup retains unverifiable active jobs and diagnostic files", async () => {
+  const workspace = makeTempDir();
+  const logFile = path.join(workspace, "active.log");
+  fs.writeFileSync(logFile, "diagnostic\n", "utf8");
+  const activeJob = {
+    id: "task-unverifiable",
+    status: "running",
+    sessionId: "session-a",
+    pid: 0,
+    workerToken: "worker-token-invalid-1234",
+    logFile
+  };
+  saveState(workspace, { jobs: [activeJob] });
+  writeJobFile(workspace, activeJob.id, activeJob);
+
+  await assert.rejects(cleanupSessionJobs(workspace, "session-a"), /Failed to stop all Codex session jobs/);
+
+  assert.deepEqual(loadState(workspace).jobs.map((job) => job.id), [activeJob.id]);
+  assert.equal(fs.existsSync(resolveJobFile(workspace, activeJob.id)), true);
+  assert.equal(fs.existsSync(logFile), true);
+});
+
+test("session cleanup preserves jobs added while an active job is stopping", async () => {
+  const workspace = makeTempDir();
+  const activeJob = {
+    id: "task-active",
+    status: "running",
+    sessionId: "session-a",
+    pid: 123,
+    workerToken: "worker-token-1234567890"
+  };
+  const concurrentJob = {
+    id: "task-concurrent",
+    status: "queued",
+    sessionId: "session-b",
+    pid: 456,
+    workerToken: "worker-token-concurrent-1234"
+  };
+  saveState(workspace, { jobs: [activeJob] });
+
+  let running = true;
+  await cleanupSessionJobs(workspace, "session-a", {
+    platform: "darwin",
+    verifyProcess() {
+      return true;
+    },
+    killImpl(target, signal) {
+      assert.equal(target, -123);
+      if (signal === "SIGTERM") {
+        running = false;
+        saveState(workspace, { jobs: [activeJob, concurrentJob] });
+      } else if (signal === 0 && !running) {
+        const error = new Error("missing");
+        error.code = "ESRCH";
+        throw error;
+      }
+    }
+  });
+
+  assert.deepEqual(loadState(workspace).jobs.map((job) => job.id), [concurrentJob.id]);
+});
+
+test("task worker waits until its tokenized record contains its own PID", async () => {
+  const workspace = makeTempDir();
+  const jobId = "task-delayed";
+  const workerToken = "worker-token-delayed-1234567890";
+  writeJobFile(workspace, jobId, {
+    id: jobId,
+    status: "queued",
+    pid: null,
+    workerToken,
+    request: { prompt: "delayed" }
+  });
+
+  const waiting = waitForWorkerJob(workspace, jobId, workerToken, 987, {
+    timeoutMs: 500,
+    intervalMs: 10
+  });
+  setTimeout(() => {
+    const registeredJob = {
+      id: jobId,
+      status: "queued",
+      pid: 987,
+      workerToken,
+      request: { prompt: "delayed" }
+    };
+    writeJobFile(workspace, jobId, registeredJob);
+    saveState(workspace, { jobs: [registeredJob] });
+  }, 30);
+
+  const storedJob = await waiting;
+  assert.equal(storedJob.pid, 987);
+  assert.equal(storedJob.workerToken, workerToken);
+});