diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs
index 1954274fe..cd4682b57 100644
--- a/plugins/codex/scripts/app-server-broker.mjs
+++ b/plugins/codex/scripts/app-server-broker.mjs
@@ -10,6 +10,66 @@ import { BROKER_BUSY_RPC_CODE, CodexAppServerClient } from "./lib/app-server.mjs
 import { parseBrokerEndpoint } from "./lib/broker-endpoint.mjs";
 
 const STREAMING_METHODS = new Set(["turn/start", "review/start", "thread/compact/start"]);
+const SUBSCRIBING_METHODS = new Set(["thread/start", "thread/resume", "thread/fork"]);
+const UNSUBSCRIBE_RETRY_DELAYS_MS = [100, 500, 2000];
+// Upper bound on how long a request waits for an in-flight thread/unsubscribe of
+// the same thread. A hung cleanup request must not wedge the shared broker.
+const UNSUBSCRIBE_WAIT_TIMEOUT_MS = 5000;
+
+// Resolves true once the promise settles, or false if the timeout expires first.
+function settleWithin(promise, timeoutMs) {
+  let timer;
+  return Promise.race([
+    promise.then(
+      () => true,
+      () => true
+    ),
+    new Promise((resolve) => {
+      timer = setTimeout(() => resolve(false), timeoutMs);
+      timer.unref?.();
+    })
+  ]).finally(() => clearTimeout(timer));
+}
+
+function buildSubscriptionThreadIds(method, result) {
+  const threadIds = new Set();
+  if (SUBSCRIBING_METHODS.has(method) && result?.thread?.id) {
+    threadIds.add(result.thread.id);
+  }
+  if (method === "review/start" && result?.reviewThreadId) {
+    threadIds.add(result.reviewThreadId);
+  }
+  return threadIds;
+}
+
+function buildProvisionalSubscriptionThreadIds(method, params) {
+  const threadIds = new Set();
+  if (method === "thread/resume" && params?.threadId) {
+    threadIds.add(params.threadId);
+  }
+  if (method === "review/start" && params?.threadId) {
+    threadIds.add(params.threadId);
+  }
+  return threadIds;
+}
+
+function buildNotificationSubscriptionRelationships(message) {
+  const relationships = [];
+  const thread = message?.method === "thread/started" ? message.params?.thread : null;
+  if (thread?.id && thread?.parentThreadId) {
+    relationships.push({ sourceThreadId: thread.parentThreadId, subscribedThreadId: thread.id });
+  }
+
+  const item = message?.params?.item;
+  if (item?.type === "collabAgentToolCall" && item?.senderThreadId && Array.isArray(item.receiverThreadIds)) {
+    for (const threadId of item.receiverThreadIds) {
+      if (threadId) {
+        relationships.push({ sourceThreadId: item.senderThreadId, subscribedThreadId: threadId });
+      }
+    }
+  }
+  return relationships;
+}
 
 function buildStreamThreadIds(method, params, result) {
   const threadIds = new Set();
@@ -70,6 +130,227 @@ async function main() {
   let activeStreamSocket = null;
   let activeStreamThreadIds = null;
   const sockets = new Set();
+  // App-server subscriptions belong to the broker's single upstream connection.
+  // Mirror downstream ownership so one client cannot release another client's thread.
+  const socketThreadIds = new Map();
+  const threadSockets = new Map();
+  const pendingUnsubscribes = new Map();
+  const unsubscribeRetryTimers = new Map();
+  // Threads a socket claimed for a request that has not succeeded yet, plus any
+  // child threads it inherited through those claims while the request was open.
+  const provisionalClaims = new Map();
+
+  function cancelUnsubscribeRetry(threadId) {
+    const retry = unsubscribeRetryTimers.get(threadId);
+    if (!retry) {
+      return;
+    }
+    clearTimeout(retry.timer);
+    unsubscribeRetryTimers.delete(threadId);
+  }
+
+  function addThreadOwner(socket, threadId) {
+    cancelUnsubscribeRetry(threadId);
+    let ownedThreadIds = socketThreadIds.get(socket);
+    if (!ownedThreadIds) {
+      ownedThreadIds = new Set();
+      socketThreadIds.set(socket, ownedThreadIds);
+    }
+    if (ownedThreadIds.has(threadId)) {
+      return false;
+    }
+    ownedThreadIds.add(threadId);
+
+    let owners = threadSockets.get(threadId);
+    if (!owners) {
+      owners = new Set();
+      threadSockets.set(threadId, owners);
+    }
+    owners.add(socket);
+    return true;
+  }
+
+  function removeThreadOwner(socket, threadId) {
+    const ownedThreadIds = socketThreadIds.get(socket);
+    if (!ownedThreadIds?.delete(threadId)) {
+      return false;
+    }
+    if (ownedThreadIds.size === 0) {
+      socketThreadIds.delete(socket);
+    }
+
+    const owners = threadSockets.get(threadId);
+    owners?.delete(socket);
+    if (owners?.size === 0) {
+      threadSockets.delete(threadId);
+    }
+    return true;
+  }
+
+  function requestThreadUnsubscribe(threadId) {
+    // A request that has already been sent is never reused: the thread may have
+    // been reacquired and released again while it was in flight. A request that
+    // is still queued behind an earlier one re-checks ownership when it sends, so
+    // every release until then can share it instead of growing the chain.
+    const previous = pendingUnsubscribes.get(threadId);
+    if (previous && !previous.sent) {
+      return previous.request;
+    }
+    const entry = { request: null, sent: false };
+    const execute = async () => {
+      if (previous) {
+        // Wait without a bound. The pending entry must not settle while any
+        // earlier upstream unsubscribe for this thread is still outstanding,
+        // otherwise a retried claim could slip past a hung cleanup request.
+        await previous.request.then(
+          () => {},
+          () => {}
+        );
+      }
+      if (threadSockets.has(threadId)) {
+        return { result: null, error: null, skipped: true };
+      }
+      entry.sent = true;
+      const result = await appClient.request("thread/unsubscribe", { threadId });
+      return { result, error: null };
+    };
+    entry.request = execute().then(
+      (outcome) => {
+        if (pendingUnsubscribes.get(threadId) === entry) {
+          pendingUnsubscribes.delete(threadId);
+        }
+        return outcome;
+      },
+      (error) => {
+        if (pendingUnsubscribes.get(threadId) === entry) {
+          pendingUnsubscribes.delete(threadId);
+        }
+        process.stderr.write(
+          `Failed to unsubscribe Codex thread ${threadId}: ${error instanceof Error ? error.message : String(error)}\n`
+        );
+        return { result: null, error };
+      }
+    );
+    pendingUnsubscribes.set(threadId, entry);
+    return entry.request;
+  }
+
+  function scheduleUnsubscribeRetry(threadId, retryIndex) {
+    if (
+      retryIndex >= UNSUBSCRIBE_RETRY_DELAYS_MS.length ||
+      unsubscribeRetryTimers.has(threadId) ||
+      threadSockets.has(threadId) ||
+      appClient.closed
+    ) {
+      return;
+    }
+    const timer = setTimeout(() => {
+      unsubscribeRetryTimers.delete(threadId);
+      void unsubscribeIfUnowned(threadId, { retryOnFailure: true, retryIndex: retryIndex + 1 });
+    }, UNSUBSCRIBE_RETRY_DELAYS_MS[retryIndex]);
+    timer.unref?.();
+    unsubscribeRetryTimers.set(threadId, { timer, retryIndex });
+  }
+
+  async function unsubscribeIfUnowned(threadId, { retryOnFailure = false, retryIndex = 0 } = {}) {
+    if (threadSockets.has(threadId) || appClient.closed) {
+      cancelUnsubscribeRetry(threadId);
+      return null;
+    }
+    const outcome = await requestThreadUnsubscribe(threadId);
+    if (outcome.error && retryOnFailure) {
+      scheduleUnsubscribeRetry(threadId, retryIndex);
+    } else if (!outcome.error) {
+      cancelUnsubscribeRetry(threadId);
+    }
+    return outcome;
+  }
+
+  async function releaseThreadOwners(socket, threadIds = socketThreadIds.get(socket) ?? new Set()) {
+    const releasedThreadIds = [];
+    for (const threadId of [...threadIds]) {
+      if (removeThreadOwner(socket, threadId) && !threadSockets.has(threadId)) {
+        releasedThreadIds.push(threadId);
+      }
+    }
+    await Promise.all(
+      releasedThreadIds.map((threadId) => unsubscribeIfUnowned(threadId, { retryOnFailure: true }))
+    );
+  }
+
+  function trackSubscriptionResults(socket, method, result) {
+    for (const threadId of buildSubscriptionThreadIds(method, result)) {
+      if (socket.destroyed || !sockets.has(socket)) {
+        void unsubscribeIfUnowned(threadId, { retryOnFailure: true });
+        continue;
+      }
+      addThreadOwner(socket, threadId);
+    }
+  }
+
+  function trackNotificationSubscriptions(message) {
+    // App-server can auto-subscribe its connection to subagent threads. Attribute
+    // each child to the downstream owners of its causal parent, not the client
+    // that happens to be active when a delayed notification arrives.
+    for (const { sourceThreadId, subscribedThreadId } of buildNotificationSubscriptionRelationships(message)) {
+      const sourceOwners = [...(threadSockets.get(sourceThreadId) ?? [])].filter(
+        (socket) => !socket.destroyed && sockets.has(socket)
+      );
+      if (sourceOwners.length === 0) {
+        void unsubscribeIfUnowned(subscribedThreadId, { retryOnFailure: true });
+        continue;
+      }
+      if (pendingUnsubscribes.has(subscribedThreadId)) {
+        // A cleanup request for this child is still outstanding, so its upstream
+        // subscription is going away. Do not hand it to new owners.
+        continue;
+      }
+      for (const socket of sourceOwners) {
+        if (addThreadOwner(socket, subscribedThreadId)) {
+          // Ownership inherited through a claim that has not succeeded yet is
+          // rolled back with that claim. The source may itself be an inherited
+          // child, so nested descendants are recorded too.
+          const claim = provisionalClaims.get(socket);
+          if (claim && (claim.threadIds.has(sourceThreadId) || claim.inheritedThreadIds.has(sourceThreadId))) {
+            claim.inheritedThreadIds.add(subscribedThreadId);
+          }
+        }
+      }
+    }
+  }
+
+  async function handleThreadUnsubscribe(socket, params) {
+    const threadId = params?.threadId;
+    if (typeof threadId !== "string") {
+      return appClient.request("thread/unsubscribe", params ?? {});
+    }
+
+    const ownedThreadIds = socketThreadIds.get(socket);
+    if (!ownedThreadIds?.has(threadId)) {
+      if (threadSockets.has(threadId)) {
+        return { status: "notSubscribed" };
+      }
+      const outcome = await unsubscribeIfUnowned(threadId);
+      if (outcome?.error) {
+        throw outcome.error;
+      }
+      return outcome?.result ?? { status: "notSubscribed" };
+    }
+
+    removeThreadOwner(socket, threadId);
+    if (threadSockets.has(threadId)) {
+      return { status: "unsubscribed" };
+    }
+
+    const outcome = await unsubscribeIfUnowned(threadId);
+    if (outcome?.error) {
+      if (!socket.destroyed && sockets.has(socket)) {
+        addThreadOwner(socket, threadId);
+      }
+      throw outcome.error;
+    }
+    return outcome?.result ?? { status: "unsubscribed" };
+  }
 
   function clearSocketOwnership(socket) {
     if (activeRequestSocket === socket) {
@@ -83,6 +364,7 @@ async function main() {
 
   function routeNotification(message) {
     const target = activeRequestSocket ?? activeStreamSocket;
+    trackNotificationSubscriptions(message);
     if (!target) {
       return;
     }
@@ -100,6 +382,10 @@ async function main() {
   }
 
   async function shutdown(server) {
+    for (const { timer } of unsubscribeRetryTimers.values()) {
+      clearTimeout(timer);
+    }
+    unsubscribeRetryTimers.clear();
     for (const socket of sockets) {
       socket.end();
     }
@@ -119,117 +405,186 @@ async function main() {
     sockets.add(socket);
     socket.setEncoding("utf8");
     let buffer = "";
+    let processing = Promise.resolve();
 
-    socket.on("data", async (chunk) => {
-      buffer += chunk;
-      let newlineIndex = buffer.indexOf("\n");
-      while (newlineIndex !== -1) {
-        const line = buffer.slice(0, newlineIndex);
-        buffer = buffer.slice(newlineIndex + 1);
-        newlineIndex = buffer.indexOf("\n");
-
-        if (!line.trim()) {
-          continue;
-        }
-
-        let message;
-        try {
-          message = JSON.parse(line);
-        } catch (error) {
-          send(socket, {
-            id: null,
-            error: buildJsonRpcError(-32700, `Invalid JSON: ${error.message}`)
-          });
-          continue;
-        }
-
-        if (message.id !== undefined && message.method === "initialize") {
-          send(socket, {
-            id: message.id,
-            result: {
-              userAgent: "codex-companion-broker"
-            }
-          });
-          continue;
-        }
-
-        if (message.method === "initialized" && message.id === undefined) {
-          continue;
-        }
+    async function handleLine(line) {
+      if (!line.trim() || socket.destroyed || !sockets.has(socket)) {
+        return;
+      }
 
-        if (message.id !== undefined && message.method === "broker/shutdown") {
-          send(socket, { id: message.id, result: {} });
-          await shutdown(server);
-          process.exit(0);
-        }
+      let message;
+      try {
+        message = JSON.parse(line);
+      } catch (error) {
+        send(socket, {
+          id: null,
+          error: buildJsonRpcError(-32700, `Invalid JSON: ${error.message}`)
+        });
+        return;
+      }
 
-        if (message.id === undefined) {
-          continue;
-        }
+      if (message.id !== undefined && message.method === "initialize") {
+        send(socket, {
+          id: message.id,
+          result: {
+            userAgent: "codex-companion-broker"
+          }
+        });
+        return;
+      }
 
-        const allowInterruptDuringActiveStream =
-          isInterruptRequest(message) && activeStreamSocket && activeStreamSocket !== socket && !activeRequestSocket;
+      if (message.method === "initialized" && message.id === undefined) {
+        return;
+      }
 
-        if (
-          ((activeRequestSocket && activeRequestSocket !== socket) || (activeStreamSocket && activeStreamSocket !== socket)) &&
-          !allowInterruptDuringActiveStream
-        ) {
-          send(socket, {
-            id: message.id,
-            error: buildJsonRpcError(BROKER_BUSY_RPC_CODE, "Shared Codex broker is busy.")
-          });
-          continue;
-        }
+      if (message.id !== undefined && message.method === "broker/shutdown") {
+        send(socket, { id: message.id, result: {} });
+        await shutdown(server);
+        process.exit(0);
+      }
 
-        if (allowInterruptDuringActiveStream) {
-          try {
-            const result = await appClient.request(message.method, message.params ?? {});
-            send(socket, { id: message.id, result });
-          } catch (error) {
-            send(socket, {
-              id: message.id,
-              error: buildJsonRpcError(error.rpcCode ?? -32000, error.message)
-            });
-          }
-          continue;
-        }
+      if (message.id === undefined) {
+        return;
+      }
 
-        const isStreaming = STREAMING_METHODS.has(message.method);
-        activeRequestSocket = socket;
+      const allowInterruptDuringActiveStream =
+        isInterruptRequest(message) && activeStreamSocket && activeStreamSocket !== socket && !activeRequestSocket;
+
+      if (
+        ((activeRequestSocket && activeRequestSocket !== socket) || (activeStreamSocket && activeStreamSocket !== socket)) &&
+        !allowInterruptDuringActiveStream
+      ) {
+        send(socket, {
+          id: message.id,
+          error: buildJsonRpcError(BROKER_BUSY_RPC_CODE, "Shared Codex broker is busy.")
+        });
+        return;
+      }
 
+      if (allowInterruptDuringActiveStream) {
         try {
           const result = await appClient.request(message.method, message.params ?? {});
           send(socket, { id: message.id, result });
-          if (isStreaming) {
-            activeStreamSocket = socket;
-            activeStreamThreadIds = buildStreamThreadIds(message.method, message.params ?? {}, result);
-          }
-          if (activeRequestSocket === socket) {
-            activeRequestSocket = null;
-          }
         } catch (error) {
           send(socket, {
             id: message.id,
             error: buildJsonRpcError(error.rpcCode ?? -32000, error.message)
           });
-          if (activeRequestSocket === socket) {
-            activeRequestSocket = null;
-          }
-          if (activeStreamSocket === socket && !isStreaming) {
-            activeStreamSocket = null;
-          }
         }
+        return;
+      }
+
+      const isStreaming = STREAMING_METHODS.has(message.method);
+      // Claim known thread ids before awaiting app-server. This prevents another
+      // client's close handler from unsubscribing a concurrently resumed thread.
+      const provisionalThreadIds = buildProvisionalSubscriptionThreadIds(message.method, message.params ?? {});
+      const addedProvisionalThreadIds = new Set();
+      for (const threadId of provisionalThreadIds) {
+        if (addThreadOwner(socket, threadId)) {
+          addedProvisionalThreadIds.add(threadId);
+        }
+      }
+      const claim = { threadIds: addedProvisionalThreadIds, inheritedThreadIds: new Set() };
+      if (addedProvisionalThreadIds.size > 0) {
+        provisionalClaims.set(socket, claim);
+      }
+      const rollBackClaim = () => {
+        if (provisionalClaims.get(socket) === claim) {
+          provisionalClaims.delete(socket);
+        }
+        void releaseThreadOwners(socket, new Set([...claim.threadIds, ...claim.inheritedThreadIds]));
+      };
+      activeRequestSocket = socket;
+      // Let an in-flight unsubscribe for the same thread settle first so it cannot
+      // overtake the new subscription, and so an explicit unsubscribe does not
+      // queue behind it while this socket holds the busy slot. The wait is
+      // bounded: a hung cleanup request must not block this client or keep the
+      // broker busy for everyone else. If it expires, fail the request instead of
+      // racing or waiting on the outstanding unsubscribe.
+      const gatedThreadIds = new Set(provisionalThreadIds);
+      if (message.method === "thread/unsubscribe" && typeof message.params?.threadId === "string") {
+        gatedThreadIds.add(message.params.threadId);
+      }
+      const settled = await Promise.all(
+        [...gatedThreadIds].map((threadId) => {
+          const pending = pendingUnsubscribes.get(threadId);
+          return pending ? settleWithin(pending.request, UNSUBSCRIBE_WAIT_TIMEOUT_MS) : true;
+        })
+      );
+      if (settled.includes(false)) {
+        send(socket, {
+          id: message.id,
+          error: buildJsonRpcError(
+            -32000,
+            "Codex thread is still being released upstream; retry the request shortly."
+          )
+        });
+        if (activeRequestSocket === socket) {
+          activeRequestSocket = null;
+        }
+        rollBackClaim();
+        return;
+      }
+
+      try {
+        const result =
+          message.method === "thread/unsubscribe"
+            ? await handleThreadUnsubscribe(socket, message.params ?? {})
+            : await appClient.request(message.method, message.params ?? {});
+        trackSubscriptionResults(socket, message.method, result);
+        if (provisionalClaims.get(socket) === claim) {
+          provisionalClaims.delete(socket);
+        }
+        send(socket, { id: message.id, result });
+        if (isStreaming && !socket.destroyed && sockets.has(socket)) {
+          activeStreamSocket = socket;
+          activeStreamThreadIds = buildStreamThreadIds(message.method, message.params ?? {}, result);
+        }
+        if (activeRequestSocket === socket) {
+          activeRequestSocket = null;
+        }
+      } catch (error) {
+        send(socket, {
+          id: message.id,
+          error: buildJsonRpcError(error.rpcCode ?? -32000, error.message)
+        });
+        if (activeRequestSocket === socket) {
+          activeRequestSocket = null;
+        }
+        if (activeStreamSocket === socket && !isStreaming) {
+          activeStreamSocket = null;
+        }
+        // Release after replying: a hung upstream unsubscribe must not withhold
+        // the error or leave the broker busy for other clients.
+        rollBackClaim();
+      }
+    }
+
+    socket.on("data", (chunk) => {
+      buffer += chunk;
+      let newlineIndex = buffer.indexOf("\n");
+      while (newlineIndex !== -1) {
+        const line = buffer.slice(0, newlineIndex);
+        buffer = buffer.slice(newlineIndex + 1);
+        newlineIndex = buffer.indexOf("\n");
+        processing = processing.then(() => handleLine(line)).catch((error) => {
+          process.stderr.write(
+            `Failed to process broker request: ${error instanceof Error ? error.message : String(error)}\n`
+          );
+        });
       }
     });
 
     socket.on("close", () => {
       sockets.delete(socket);
       clearSocketOwnership(socket);
+      void releaseThreadOwners(socket);
     });
 
     socket.on("error", () => {
       sockets.delete(socket);
       clearSocketOwnership(socket);
+      void releaseThreadOwners(socket);
     });
   });
 
diff --git a/plugins/codex/scripts/lib/app-server-protocol.d.ts b/plugins/codex/scripts/lib/app-server-protocol.d.ts
index f61a4588e..e023b324f 100644
--- a/plugins/codex/scripts/lib/app-server-protocol.d.ts
+++ b/plugins/codex/scripts/lib/app-server-protocol.d.ts
@@ -21,6 +21,8 @@ import type {
   ThreadSetNameResponse,
   ThreadStartParams as RawThreadStartParams,
   ThreadStartResponse,
+  ThreadUnsubscribeParams,
+  ThreadUnsubscribeResponse,
   Turn,
   TurnInterruptParams,
   TurnInterruptResponse,
@@ -63,6 +65,7 @@ export interface AppServerMethodMap {
   "thread/resume": { params: ThreadResumeParams; result: ThreadResumeResponse };
   "thread/name/set": { params: ThreadSetNameParams; result: ThreadSetNameResponse };
   "thread/list": { params: ThreadListParams; result: ThreadListResponse };
+  "thread/unsubscribe": { params: ThreadUnsubscribeParams; result: ThreadUnsubscribeResponse };
   "review/start": { params: ReviewStartParams; result: ReviewStartResponse };
   "turn/start": { params: TurnStartParams; result: TurnStartResponse };
   "turn/interrupt": { params: TurnInterruptParams; result: TurnInterruptResponse };
diff --git a/tests/broker-subscriptions.test.mjs b/tests/broker-subscriptions.test.mjs
new file mode 100644
index 000000000..3fc171683
--- /dev/null
+++ b/tests/broker-subscriptions.test.mjs
@@ -0,0 +1,767 @@
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import net from "node:net";
+import path from "node:path";
+import test from "node:test";
+import { spawn } from "node:child_process";
+import { fileURLToPath } from "node:url";
+
+import { buildEnv, installFakeCodex } from "./fake-codex-fixture.mjs";
+import { makeTempDir } from "./helpers.mjs";
+
+const ROOT = path.resolve(fileURLToPath(new URL("..", import.meta.url)));
+const BROKER = path.join(ROOT, "plugins", "codex", "scripts", "app-server-broker.mjs");
+
+function delay(ms) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+async function waitWithTimeout(promise, timeoutMs) {
+  let timer;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise((resolve) => {
+        timer = setTimeout(() => resolve(null), timeoutMs);
+      })
+    ]);
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
+async function waitFor(predicate, { timeoutMs = 10000, intervalMs = 25 } = {}) {
+  const startedAt = Date.now();
+  while (Date.now() - startedAt < timeoutMs) {
+    if (await predicate()) {
+      return true;
+    }
+    await delay(intervalMs);
+  }
+  return false;
+}
+
+function readState(statePath) {
+  try {
+    return JSON.parse(fs.readFileSync(statePath, "utf8"));
+  } catch {
+    return null;
+  }
+}
+
+function startBroker(behavior = "review-ok") {
+  const binDir = makeTempDir("codex-broker-bin-");
+  installFakeCodex(binDir, behavior);
+  const sessionDir = makeTempDir("codex-broker-subscriptions-");
+  const cwd = makeTempDir("codex-broker-cwd-");
+  const socketPath = path.join(sessionDir, "broker.sock");
+  const pidFile = path.join(sessionDir, "broker.pid");
+  const statePath = path.join(binDir, "fake-codex-state.json");
+  const tempDirs = [binDir, sessionDir, cwd];
+
+  const child = spawn(
+    process.execPath,
+    [BROKER, "serve", "--endpoint", `unix:${socketPath}`, "--cwd", cwd, "--pid-file", pidFile],
+    { env: buildEnv(binDir), stdio: ["ignore", "pipe", "pipe"] }
+  );
+  let stderr = "";
+  child.stderr.on("data", (chunk) => {
+    stderr += chunk;
+  });
+  const exited = new Promise((resolve) => {
+    child.on("exit", (code, signal) => resolve({ code, signal }));
+  });
+
+  async function stop() {
+    try {
+      if (child.exitCode !== null || child.signalCode !== null) {
+        await exited;
+        return;
+      }
+      child.kill("SIGTERM");
+      const result = await waitWithTimeout(exited, 5000);
+      if (!result) {
+        child.kill("SIGKILL");
+        await exited;
+      }
+    } finally {
+      for (const tempDir of tempDirs) {
+        fs.rmSync(tempDir, { recursive: true, force: true });
+      }
+    }
+  }
+
+  return {
+    socketPath,
+    statePath,
+    stderr: () => stderr,
+    listening: () => waitFor(() => fs.existsSync(socketPath)),
+    stop
+  };
+}
+
+async function connectClient(socketPath) {
+  const socket = await new Promise((resolve, reject) => {
+    const candidate = net.createConnection({ path: socketPath });
+    candidate.on("connect", () => resolve(candidate));
+    candidate.on("error", reject);
+  });
+  socket.setEncoding("utf8");
+
+  let nextId = 1;
+  let buffer = "";
+  const pending = new Map();
+  const notifications = [];
+  const notificationWaiters = new Set();
+  const closed = new Promise((resolve) => socket.on("close", resolve));
+
+  function notifyWaiters(message) {
+    for (const waiter of [...notificationWaiters]) {
+      if (waiter.predicate(message)) {
+        waiter.settle(message);
+      }
+    }
+  }
+
+  socket.on("data", (chunk) => {
+    buffer += chunk;
+    let newlineIndex = buffer.indexOf("\n");
+    while (newlineIndex !== -1) {
+      const line = buffer.slice(0, newlineIndex);
+      buffer = buffer.slice(newlineIndex + 1);
+      newlineIndex = buffer.indexOf("\n");
+      if (!line.trim()) {
+        continue;
+      }
+      const message = JSON.parse(line);
+      if (message.id !== undefined) {
+        const request = pending.get(message.id);
+        if (request) {
+          pending.delete(message.id);
+          if (message.error) {
+            request.reject(new Error(message.error.message));
+          } else {
+            request.resolve(message.result);
+          }
+        }
+        continue;
+      }
+      notifications.push(message);
+      notifyWaiters(message);
+    }
+  });
+
+  socket.on("error", (error) => {
+    for (const request of pending.values()) {
+      request.reject(error);
+    }
+    pending.clear();
+  });
+
+  function request(method, params = {}) {
+    const id = nextId++;
+    const response = new Promise((resolve, reject) => {
+      pending.set(id, { resolve, reject });
+    });
+    socket.write(`${JSON.stringify({ id, method, params })}\n`);
+    return response;
+  }
+
+  async function waitForNotification(predicate, timeoutMs = 10000) {
+    const existing = notifications.find(predicate);
+    if (existing) {
+      return existing;
+    }
+    return new Promise((resolve) => {
+      let timer;
+      const waiter = {
+        predicate,
+        settle(message) {
+          clearTimeout(timer);
+          notificationWaiters.delete(waiter);
+          resolve(message);
+        }
+      };
+      timer = setTimeout(() => waiter.settle(null), timeoutMs);
+      notificationWaiters.add(waiter);
+    });
+  }
+
+  await request("initialize", {});
+  return {
+    request,
+    waitForNotification,
+    async end() {
+      socket.end();
+      await closed;
+    },
+    destroy() {
+      socket.destroy();
+    }
+  };
+}
+
+async function waitForUnsubscribes(statePath, expectedThreadIds) {
+  const expected = [...expectedThreadIds].sort();
+  const found = await waitFor(() => {
+    const actual = [...(readState(statePath)?.unsubscribeRequests ?? [])].sort();
+    return actual.length === expected.length && actual.every((threadId, index) => threadId === expected[index]);
+  });
+  assert.equal(found, true, `expected unsubscribe requests for ${expected.join(", ")}`);
+}
+
+test("broker unsubscribes a completed task thread when its client closes", async (t) => {
+  const broker = startBroker();
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const started = await client.request("thread/start", { cwd: process.cwd(), ephemeral: true });
+  const threadId = started.thread.id;
+  await client.request("turn/start", {
+    threadId,
+    input: [{ type: "text", text: "test normal completion" }]
+  });
+  const completed = await client.waitForNotification(
+    (message) => message.method === "turn/completed" && message.params?.threadId === threadId
+  );
+  assert.ok(completed, "task never completed");
+
+  await client.end();
+  await waitForUnsubscribes(broker.statePath, [threadId]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, []);
+});
+
+test("broker keeps a resumed thread subscribed until its final client closes", async (t) => {
+  const broker = startBroker();
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const threadId = (await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  const secondClient = await connectClient(broker.socketPath);
+  await secondClient.request("thread/resume", { threadId });
+
+  await firstClient.end();
+  await delay(250);
+  assert.deepEqual(readState(broker.statePath).unsubscribeRequests, []);
+
+  await secondClient.end();
+  await waitForUnsubscribes(broker.statePath, [threadId]);
+});
+
+test("broker serializes a resume behind an in-flight unsubscribe", async (t) => {
+  const broker = startBroker("unsubscribe-delayed");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const threadId = (await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  await firstClient.end();
+  const unsubscribeStarted = await waitFor(() =>
+    readState(broker.statePath)?.unsubscribeRequests?.includes(threadId)
+  );
+  assert.equal(unsubscribeStarted, true, "unsubscribe request was not observed");
+
+  const secondClient = await connectClient(broker.socketPath);
+  await secondClient.request("thread/resume", { threadId });
+  assert.deepEqual(readState(broker.statePath).requestOrder, ["unsubscribe:response", "thread/resume"]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, [threadId]);
+
+  await secondClient.end();
+  await waitForUnsubscribes(broker.statePath, [threadId, threadId]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, []);
+});
+
+test("broker cancels a retry when a client reacquires the thread", async (t) => {
+  const broker = startBroker("unsubscribe-fails-once");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const threadId = (await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  await firstClient.end();
+  const firstAttemptObserved = await waitFor(
+    () => readState(broker.statePath)?.unsubscribeRequests?.length === 1
+  );
+  assert.equal(firstAttemptObserved, true, "initial unsubscribe request was not observed");
+
+  const secondClient = await connectClient(broker.socketPath);
+  await secondClient.request("thread/resume", { threadId });
+  await delay(3000);
+  assert.deepEqual(readState(broker.statePath).subscriptions, [threadId]);
+
+  await secondClient.end();
+  const unsubscribed = await waitFor(() => readState(broker.statePath)?.subscriptions?.length === 0);
+  assert.equal(unsubscribed, true, "reacquired thread was not unsubscribed after its final owner closed");
+  assert.equal(readState(broker.statePath).unsubscribeRequests.at(-1), threadId);
+});
+
+test("broker unsubscribes source and detached review threads", async (t) => {
+  const broker = startBroker();
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const sourceThreadId = (await client.request("thread/start", { cwd: process.cwd(), ephemeral: true })).thread.id;
+  const review = await client.request("review/start", {
+    threadId: sourceThreadId,
+    delivery: "detached",
+    target: { type: "uncommittedChanges" }
+  });
+  assert.notEqual(review.reviewThreadId, sourceThreadId);
+  const completed = await client.waitForNotification(
+    (message) => message.method === "turn/completed" && message.params?.threadId === review.reviewThreadId
+  );
+  assert.ok(completed, "review never completed");
+
+  await client.end();
+  await waitForUnsubscribes(broker.statePath, [sourceThreadId, review.reviewThreadId]);
+});
+
+test("broker unsubscribes a forked thread when its client closes", async (t) => {
+  const broker = startBroker();
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const sourceThreadId = (await client.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  const forkThreadId = (await client.request("thread/fork", { threadId: sourceThreadId, ephemeral: true })).thread.id;
+
+  await client.end();
+  await waitForUnsubscribes(broker.statePath, [sourceThreadId, forkThreadId]);
+});
+
+test("broker unsubscribes when a client disconnects during an active turn", async (t) => {
+  const broker = startBroker("slow-task");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const threadId = (await client.request("thread/start", { cwd: process.cwd(), ephemeral: true })).thread.id;
+  await client.request("turn/start", {
+    threadId,
+    input: [{ type: "text", text: "disconnect this client" }]
+  });
+  client.destroy();
+
+  await waitForUnsubscribes(broker.statePath, [threadId]);
+});
+
+test("broker unsubscribes auto-subscribed subagent threads", async (t) => {
+  const broker = startBroker("with-subagent");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const threadId = (await client.request("thread/start", { cwd: process.cwd(), ephemeral: true })).thread.id;
+  await client.request("turn/start", {
+    threadId,
+    input: [{ type: "text", text: "delegate this task" }]
+  });
+  const completed = await client.waitForNotification(
+    (message) => message.method === "turn/completed" && message.params?.threadId === threadId
+  );
+  assert.ok(completed, "task never completed");
+
+  const subagentThread = readState(broker.statePath).threads.find((thread) => thread.name === "design-challenger");
+  assert.ok(subagentThread, "subagent thread was not created");
+
+  await client.end();
+  await waitForUnsubscribes(broker.statePath, [threadId, subagentThread.id]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, []);
+});
+
+test("broker tracks subagent subscriptions from collaboration items", async (t) => {
+  const broker = startBroker("with-receiver-only-subagent");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const threadId = (await client.request("thread/start", { cwd: process.cwd(), ephemeral: true })).thread.id;
+  await client.request("turn/start", {
+    threadId,
+    input: [{ type: "text", text: "delegate without a thread-started notification" }]
+  });
+  const completed = await client.waitForNotification(
+    (message) => message.method === "turn/completed" && message.params?.threadId === threadId
+  );
+  assert.ok(completed, "task never completed");
+
+  const subagentThread = readState(broker.statePath).threads.find((thread) => thread.name === "design-challenger");
+  assert.ok(subagentThread, "subagent thread was not created");
+
+  await client.end();
+  await waitForUnsubscribes(broker.statePath, [threadId, subagentThread.id]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, []);
+});
+
+test("broker unsubscribes a child thread created after its client disconnects", async (t) => {
+  const broker = startBroker("with-delayed-subagent");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const threadId = (await client.request("thread/start", { cwd: process.cwd(), ephemeral: true })).thread.id;
+  await client.request("turn/start", {
+    threadId,
+    input: [{ type: "text", text: "disconnect before delegation" }]
+  });
+  client.destroy();
+
+  const childCreated = await waitFor(() =>
+    readState(broker.statePath)?.threads.some((thread) => thread.name === "delayed-design-challenger")
+  );
+  assert.equal(childCreated, true, "delayed child thread was not created");
+  const childThread = readState(broker.statePath).threads.find(
+    (thread) => thread.name === "delayed-design-challenger"
+  );
+
+  await waitForUnsubscribes(broker.statePath, [threadId, childThread.id]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, []);
+});
+
+test("broker does not assign a delayed child thread to an unrelated active client", async (t) => {
+  const broker = startBroker("with-delayed-subagent");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const firstThreadId = (
+    await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: true })
+  ).thread.id;
+  await firstClient.request("turn/start", {
+    threadId: firstThreadId,
+    input: [{ type: "text", text: "create a delayed child" }]
+  });
+  firstClient.destroy();
+
+  const secondClient = await connectClient(broker.socketPath);
+  const secondThreadId = (
+    await secondClient.request("thread/start", { cwd: process.cwd(), ephemeral: true })
+  ).thread.id;
+  await secondClient.request("turn/start", {
+    threadId: secondThreadId,
+    input: [{ type: "text", text: "remain active while the first child arrives" }]
+  });
+
+  const childrenCreated = await waitFor(
+    () => readState(broker.statePath)?.threads.filter((thread) => thread.parentThreadId).length === 2
+  );
+  assert.equal(childrenCreated, true, "delayed child threads were not created");
+
+  const state = readState(broker.statePath);
+  const firstChild = state.threads.find((thread) => thread.parentThreadId === firstThreadId);
+  const secondChild = state.threads.find((thread) => thread.parentThreadId === secondThreadId);
+  assert.ok(firstChild, "the first client's child thread was not recorded");
+  assert.ok(secondChild, "the second client's child thread was not recorded");
+
+  const firstReleased = await waitFor(() => {
+    const requests = readState(broker.statePath)?.unsubscribeRequests ?? [];
+    return requests.includes(firstThreadId) && requests.includes(firstChild.id);
+  });
+  assert.equal(firstReleased, true, "the disconnected client's subscriptions were not released");
+  const requestsBeforeSecondClientCloses = readState(broker.statePath).unsubscribeRequests;
+  assert.equal(requestsBeforeSecondClientCloses.includes(secondThreadId), false);
+  assert.equal(requestsBeforeSecondClientCloses.includes(secondChild.id), false);
+
+  await secondClient.end();
+  await waitForUnsubscribes(broker.statePath, [firstThreadId, firstChild.id, secondThreadId, secondChild.id]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, []);
+});
+
+test("broker serializes subscription requests from one downstream client", async (t) => {
+  const broker = startBroker("overlapping-resume");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const threadId = (await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  const secondClient = await connectClient(broker.socketPath);
+  const results = await Promise.allSettled([
+    secondClient.request("thread/resume", { threadId, persistFullHistory: true }),
+    secondClient.request("thread/resume", { threadId })
+  ]);
+  assert.equal(results[0].status, "rejected");
+  assert.match(results[0].reason.message, /forced delayed resume failure/);
+  assert.equal(results[1].status, "fulfilled");
+
+  await firstClient.end();
+  await delay(250);
+  assert.deepEqual(readState(broker.statePath).unsubscribeRequests, []);
+  assert.deepEqual(readState(broker.statePath).subscriptions, [threadId]);
+
+  await secondClient.end();
+  await waitForUnsubscribes(broker.statePath, [threadId]);
+});
+
+test("broker replies to a failed resume even when the upstream unsubscribe hangs", async (t) => {
+  const broker = startBroker("resume-fails-unsubscribe-hangs");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  // Nobody owns this thread, so the failed provisional claim releases it upstream.
+  const threadId = "thr_unowned";
+  const firstClient = await connectClient(broker.socketPath);
+  const failedResume = await waitWithTimeout(
+    firstClient.request("thread/resume", { threadId, persistFullHistory: true }).then(
+      () => ({ status: "fulfilled" }),
+      (error) => ({ status: "rejected", error })
+    ),
+    2000
+  );
+  assert.notEqual(failedResume, null, "failed resume did not receive a response");
+  assert.equal(failedResume.status, "rejected");
+  assert.match(failedResume.error.message, /forced resume failure/);
+
+  const secondClient = await connectClient(broker.socketPath);
+  const secondStarted = await waitWithTimeout(
+    secondClient.request("thread/start", { cwd: process.cwd(), ephemeral: false }),
+    2000
+  );
+  assert.notEqual(secondStarted, null, "broker remained busy after the failed resume");
+
+  const unsubscribeStarted = await waitFor(() =>
+    readState(broker.statePath)?.unsubscribeRequests?.includes(threadId)
+  );
+  assert.equal(unsubscribeStarted, true, "the released thread was not unsubscribed upstream");
+  await secondClient.end();
+  await firstClient.end();
+});
+
+test("broker keeps shared upstream subscriptions when one client explicitly unsubscribes", async (t) => {
+  const broker = startBroker();
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const threadId = (await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  const secondClient = await connectClient(broker.socketPath);
+  await secondClient.request("thread/resume", { threadId });
+
+  assert.deepEqual(await firstClient.request("thread/unsubscribe", { threadId }), { status: "unsubscribed" });
+  assert.deepEqual(readState(broker.statePath).unsubscribeRequests, []);
+  assert.deepEqual(readState(broker.statePath).subscriptions, [threadId]);
+
+  const thirdClient = await connectClient(broker.socketPath);
+  assert.deepEqual(await thirdClient.request("thread/unsubscribe", { threadId }), { status: "notSubscribed" });
+  assert.deepEqual(readState(broker.statePath).unsubscribeRequests, []);
+
+  await firstClient.end();
+  await thirdClient.end();
+  await secondClient.end();
+  await waitForUnsubscribes(broker.statePath, [threadId]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, []);
+});
+
+test("broker does not reclaim ownership from notifications during explicit unsubscribe", async (t) => {
+  const broker = startBroker("unsubscribe-notifies");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const threadId = (await client.request("thread/start", { cwd: process.cwd(), ephemeral: true })).thread.id;
+  assert.deepEqual(await client.request("thread/unsubscribe", { threadId }), { status: "unsubscribed" });
+  await client.end();
+  await delay(250);
+
+  assert.deepEqual(readState(broker.statePath).unsubscribeRequests, [threadId]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, []);
+});
+
+test("broker logs upstream unsubscribe failures", async (t) => {
+  const broker = startBroker("unsubscribe-fails");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const threadId = (await client.request("thread/start", { cwd: process.cwd(), ephemeral: true })).thread.id;
+  await client.end();
+
+  const requestObserved = await waitFor(() => readState(broker.statePath)?.unsubscribeRequests?.includes(threadId));
+  assert.equal(requestObserved, true, "unsubscribe request was not observed");
+  const warningObserved = await waitFor(
+    () => broker.stderr().includes(`Failed to unsubscribe Codex thread ${threadId}: thread unsubscribe failed`)
+  );
+  assert.equal(warningObserved, true, "unsubscribe failure was not logged");
+  const retryBoundReached = await waitFor(
+    () => readState(broker.statePath)?.unsubscribeRequests?.length === 4,
+    { timeoutMs: 6000 }
+  );
+  assert.equal(retryBoundReached, true, "unsubscribe retries did not reach the expected bound");
+  await delay(500);
+  assert.equal(readState(broker.statePath).unsubscribeRequests.length, 4);
+  assert.deepEqual(readState(broker.statePath).subscriptions, [threadId]);
+});
+
+test("broker retries a transient upstream unsubscribe failure", async (t) => {
+  const broker = startBroker("unsubscribe-fails-once");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const client = await connectClient(broker.socketPath);
+  const threadId = (await client.request("thread/start", { cwd: process.cwd(), ephemeral: true })).thread.id;
+  await client.end();
+
+  const retried = await waitFor(() => {
+    const state = readState(broker.statePath);
+    return state?.unsubscribeRequests?.length === 2 && state.subscriptions.length === 0;
+  });
+  assert.equal(retried, true, "the failed unsubscribe was not retried");
+  assert.deepEqual(readState(broker.statePath).unsubscribeRequests, [threadId, threadId]);
+  assert.match(broker.stderr(), new RegExp(`Failed to unsubscribe Codex thread ${threadId}`));
+});
+
+test("broker fails a resume when an in-flight unsubscribe outlives the bounded wait", async (t) => {
+  const broker = startBroker("resume-fails-unsubscribe-hangs");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const threadId = (await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  await firstClient.end();
+  const unsubscribeStarted = await waitFor(() =>
+    readState(broker.statePath)?.unsubscribeRequests?.includes(threadId)
+  );
+  assert.equal(unsubscribeStarted, true, "unsubscribe request was not observed");
+
+  const secondClient = await connectClient(broker.socketPath);
+  const startedAt = Date.now();
+  const resumed = await waitWithTimeout(
+    secondClient.request("thread/resume", { threadId }).then(
+      () => ({ status: "fulfilled" }),
+      (error) => ({ status: "rejected", error })
+    ),
+    8000
+  );
+  assert.notEqual(resumed, null, "resume never settled while the upstream unsubscribe hung");
+  assert.equal(resumed.status, "rejected");
+  assert.match(resumed.error.message, /still being released upstream/);
+  assert.ok(Date.now() - startedAt >= 4000, "resume did not wait for the in-flight unsubscribe");
+  // The resume was never sent upstream, so the fake still lists only the original subscription.
+  assert.deepEqual(readState(broker.statePath).subscriptions, [threadId]);
+
+  const thirdClient = await connectClient(broker.socketPath);
+  const thirdStarted = await waitWithTimeout(
+    thirdClient.request("thread/start", { cwd: process.cwd(), ephemeral: false }),
+    2000
+  );
+  assert.notEqual(thirdStarted, null, "broker remained busy after the bounded wait");
+  await thirdClient.end();
+  await secondClient.end();
+});
+
+test("broker keeps rejecting claims until a hung unsubscribe settles", async (t) => {
+  const broker = startBroker("resume-fails-unsubscribe-hangs");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const threadId = (await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  await firstClient.end();
+  const unsubscribeStarted = await waitFor(() =>
+    readState(broker.statePath)?.unsubscribeRequests?.includes(threadId)
+  );
+  assert.equal(unsubscribeStarted, true, "unsubscribe request was not observed");
+
+  const attempt = async () => {
+    const client = await connectClient(broker.socketPath);
+    const outcome = await waitWithTimeout(
+      client.request("thread/resume", { threadId }).then(
+        () => ({ status: "fulfilled" }),
+        (error) => ({ status: "rejected", error })
+      ),
+      8000
+    );
+    await client.end();
+    return outcome;
+  };
+
+  // The first claim times out on the hung unsubscribe. Its release installs a
+  // follow-up cleanup that must stay pending; a retry must still be rejected.
+  const first = await attempt();
+  assert.equal(first?.status, "rejected");
+  assert.match(first.error.message, /still being released upstream/);
+  const retry = await attempt();
+  assert.equal(retry?.status, "rejected", "a retry slipped past the still-outstanding unsubscribe");
+  assert.match(retry.error.message, /still being released upstream/);
+  assert.deepEqual(readState(broker.statePath).requestOrder, []);
+  assert.deepEqual(readState(broker.statePath).subscriptions, [threadId]);
+});
+
+test("broker rejects an explicit unsubscribe that would queue behind a hung cleanup", async (t) => {
+  const broker = startBroker("resume-fails-unsubscribe-hangs");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const threadId = (await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  await firstClient.end();
+  const unsubscribeStarted = await waitFor(() =>
+    readState(broker.statePath)?.unsubscribeRequests?.includes(threadId)
+  );
+  assert.equal(unsubscribeStarted, true, "unsubscribe request was not observed");
+
+  const secondClient = await connectClient(broker.socketPath);
+  const explicit = await waitWithTimeout(
+    secondClient.request("thread/unsubscribe", { threadId }).then(
+      () => ({ status: "fulfilled" }),
+      (error) => ({ status: "rejected", error })
+    ),
+    8000
+  );
+  assert.notEqual(explicit, null, "explicit unsubscribe never settled behind the hung cleanup");
+  assert.equal(explicit.status, "rejected");
+  assert.match(explicit.error.message, /still being released upstream/);
+
+  const thirdClient = await connectClient(broker.socketPath);
+  const thirdStarted = await waitWithTimeout(
+    thirdClient.request("thread/start", { cwd: process.cwd(), ephemeral: false }),
+    2000
+  );
+  assert.notEqual(thirdStarted, null, "broker remained busy after the rejected explicit unsubscribe");
+  // Only the original hung request reached upstream; nothing else was queued.
+  assert.deepEqual(readState(broker.statePath).unsubscribeRequests, [threadId]);
+  await thirdClient.end();
+  await secondClient.end();
+});
+
+test("broker rolls back child threads inherited through a failed provisional claim", async (t) => {
+  const broker = startBroker("with-delayed-subagent");
+  t.after(() => broker.stop());
+  assert.equal(await broker.listening(), true, `broker never listened: ${broker.stderr()}`);
+
+  const firstClient = await connectClient(broker.socketPath);
+  const threadId = (await firstClient.request("thread/start", { cwd: process.cwd(), ephemeral: false })).thread.id;
+  await firstClient.request("turn/start", {
+    threadId,
+    input: [{ type: "text", text: "spawn a child while another client is resuming" }]
+  });
+  firstClient.destroy();
+  await waitForUnsubscribes(broker.statePath, [threadId]);
+
+  // The resume fails after 250 ms; the delayed child arrives at 100 ms while the
+  // claim is still open, so the claiming socket inherits it.
+  const secondClient = await connectClient(broker.socketPath);
+  await assert.rejects(
+    secondClient.request("thread/resume", { threadId, persistFullHistory: true }),
+    /forced resume failure after child arrival/
+  );
+  const childThread = readState(broker.statePath).threads.find(
+    (thread) => thread.name === "delayed-design-challenger"
+  );
+  assert.ok(childThread, "delayed child thread was not created");
+  const grandchildCreated = await waitFor(() =>
+    readState(broker.statePath)?.threads.some((thread) => thread.name === "delayed-design-grandchild")
+  );
+  assert.equal(grandchildCreated, true, "delayed grandchild thread was not created");
+  const grandchildThread = readState(broker.statePath).threads.find(
+    (thread) => thread.name === "delayed-design-grandchild"
+  );
+
+  // The parent, the inherited child, and the nested grandchild are all released
+  // while the client stays connected.
+  await waitForUnsubscribes(broker.statePath, [threadId, threadId, childThread.id, grandchildThread.id]);
+  assert.deepEqual(readState(broker.statePath).subscriptions, []);
+  await secondClient.end();
+});
diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs
index f83c96a0d..73ed121f8 100644
--- a/tests/fake-codex-fixture.mjs
+++ b/tests/fake-codex-fixture.mjs
@@ -19,7 +19,7 @@ const readline = require("node:readline");
 
 	function loadState() {
 	  if (!fs.existsSync(STATE_PATH)) {
-	    return { nextThreadId: 1, nextTurnId: 1, appServerStarts: 0, threads: [], capabilities: null, lastInterrupt: null };
+	    return { nextThreadId: 1, nextTurnId: 1, appServerStarts: 0, threads: [], subscriptions: [], unsubscribeRequests: [], requestOrder: [], capabilities: null, lastInterrupt: null };
 	  }
 	  return JSON.parse(fs.readFileSync(STATE_PATH, "utf8"));
 	}
@@ -42,6 +42,8 @@ function now() {
 function buildThread(thread) {
   return {
     id: thread.id,
+    forkedFromId: thread.forkedFromId || null,
+    parentThreadId: thread.parentThreadId || null,
     preview: thread.preview || "",
     ephemeral: Boolean(thread.ephemeral),
     modelProvider: "openai",
@@ -116,9 +118,11 @@ function send(message) {
   process.stdout.write(JSON.stringify(message) + "\\n");
 }
 
-function nextThread(state, cwd, ephemeral) {
+function nextThread(state, cwd, ephemeral, { forkedFromId = null, parentThreadId = null } = {}) {
   const thread = {
     id: "thr_" + state.nextThreadId++,
+    forkedFromId,
+    parentThreadId,
     cwd: cwd || process.cwd(),
     name: null,
     preview: "",
@@ -313,8 +317,10 @@ rl.on("line", (line) => {
           throw new Error("thread/start.persistFullHistory requires experimentalApi capability");
         }
         const thread = nextThread(state, message.params.cwd, message.params.ephemeral);
+        state.subscriptions = [...new Set([...(state.subscriptions || []), thread.id])];
+        saveState(state);
         send({ id: message.id, result: { thread: buildThread(thread), model: message.params.model || "gpt-5.4", modelProvider: "openai", serviceTier: null, cwd: thread.cwd, approvalPolicy: "never", sandbox: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false }, reasoningEffort: null } });
-        send({ method: "thread/started", params: { thread: { id: thread.id } } });
+        send({ method: "thread/started", params: { thread: buildThread(thread) } });
         break;
       }
 
@@ -341,16 +347,96 @@ rl.on("line", (line) => {
       }
 
       case "thread/resume": {
+        if (BEHAVIOR === "with-delayed-subagent" && message.params.persistFullHistory === true) {
+          state.nestedSubagentRequested = true;
+          saveState(state);
+          setTimeout(() => {
+            send({ id: message.id, error: { code: -32000, message: "forced resume failure after child arrival" } });
+          }, 250);
+          break;
+        }
+        if (BEHAVIOR === "resume-fails-unsubscribe-hangs" && message.params.persistFullHistory === true) {
+          setTimeout(() => {
+            send({ id: message.id, error: { code: -32000, message: "forced resume failure" } });
+          }, 50);
+          break;
+        }
+        if (BEHAVIOR === "overlapping-resume" && message.params.persistFullHistory === true) {
+          setTimeout(() => {
+            send({ id: message.id, error: { code: -32000, message: "forced delayed resume failure" } });
+          }, 100);
+          break;
+        }
         if (requiresExperimental("persistExtendedHistory", message, state) || requiresExperimental("persistFullHistory", message, state)) {
           throw new Error("thread/resume.persistFullHistory requires experimentalApi capability");
         }
         const thread = ensureThread(state, message.params.threadId);
         thread.updatedAt = now();
+        if (BEHAVIOR === "unsubscribe-delayed" || BEHAVIOR === "resume-fails-unsubscribe-hangs") {
+          state.requestOrder = [...(state.requestOrder || []), "thread/resume"];
+        }
+        state.subscriptions = [...new Set([...(state.subscriptions || []), thread.id])];
         saveState(state);
         send({ id: message.id, result: { thread: buildThread(thread), model: message.params.model || "gpt-5.4", modelProvider: "openai", serviceTier: null, cwd: thread.cwd, approvalPolicy: "never", sandbox: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false }, reasoningEffort: null } });
         break;
       }
 
+      case "thread/fork": {
+        const sourceThread = ensureThread(state, message.params.threadId);
+        const thread = nextThread(state, sourceThread.cwd, message.params.ephemeral, { forkedFromId: sourceThread.id });
+        state.subscriptions = [...new Set([...(state.subscriptions || []), thread.id])];
+        saveState(state);
+        send({ id: message.id, result: { thread: buildThread(thread) } });
+        send({ method: "thread/started", params: { thread: buildThread(thread) } });
+        break;
+      }
+
+      case "thread/unsubscribe": {
+        const subscriptions = state.subscriptions || [];
+        const wasSubscribed = subscriptions.includes(message.params.threadId);
+        const wasLoaded = state.threads.some((thread) => thread.id === message.params.threadId);
+        state.unsubscribeRequests = [...(state.unsubscribeRequests || []), message.params.threadId];
+        if (BEHAVIOR === "resume-fails-unsubscribe-hangs") {
+          saveState(state);
+          break;
+        }
+        if (
+          BEHAVIOR === "unsubscribe-fails" ||
+          (BEHAVIOR === "unsubscribe-fails-once" && state.unsubscribeRequests.length === 1)
+        ) {
+          saveState(state);
+          send({ id: message.id, error: { code: -32000, message: "thread unsubscribe failed" } });
+          break;
+        }
+        if (BEHAVIOR === "unsubscribe-delayed") {
+          state.subscriptions = subscriptions.filter((threadId) => threadId !== message.params.threadId);
+          saveState(state);
+          setTimeout(() => {
+            const delayedState = loadState();
+            delayedState.requestOrder = [...(delayedState.requestOrder || []), "unsubscribe:response"];
+            saveState(delayedState);
+            send({
+              id: message.id,
+              result: { status: wasSubscribed ? "unsubscribed" : wasLoaded ? "notSubscribed" : "notLoaded" }
+            });
+          }, 300);
+          break;
+        }
+        if (BEHAVIOR === "unsubscribe-notifies") {
+          send({
+            method: "thread/status/changed",
+            params: { threadId: message.params.threadId, status: { type: "idle" } }
+          });
+        }
+        state.subscriptions = subscriptions.filter((threadId) => threadId !== message.params.threadId);
+        saveState(state);
+        send({
+          id: message.id,
+          result: { status: wasSubscribed ? "unsubscribed" : wasLoaded ? "notSubscribed" : "notLoaded" }
+        });
+        break;
+      }
+
       case "externalAgentConfig/import": {
         if (BEHAVIOR === "external-import-unsupported") {
           send({ id: message.id, error: { code: -32601, message: "Unsupported method: externalAgentConfig/import" } });
@@ -409,6 +495,8 @@ rl.on("line", (line) => {
         let reviewThread = thread;
         if (message.params.delivery === "detached") {
           reviewThread = nextThread(state, thread.cwd, true);
+          state.subscriptions = [...new Set([...(state.subscriptions || []), reviewThread.id])];
+          saveState(state);
           send({ method: "thread/started", params: { thread: { id: reviewThread.id } } });
         }
         const turnId = nextTurnId(state);
@@ -458,18 +546,49 @@ rl.on("line", (line) => {
           ? structuredReviewPayload(prompt)
           : taskPayload(prompt, thread.name && thread.name.startsWith("Codex Companion Task") && prompt.includes("Continue from the current thread state"));
 
+        if (BEHAVIOR === "with-delayed-subagent") {
+          setTimeout(() => {
+            const delayedState = loadState();
+            const subThread = nextThread(delayedState, thread.cwd, true, { parentThreadId: thread.id });
+            const subThreadRecord = ensureThread(delayedState, subThread.id);
+            subThreadRecord.name = "delayed-design-challenger";
+            delayedState.subscriptions = [...new Set([...(delayedState.subscriptions || []), subThread.id])];
+            saveState(delayedState);
+            const subTurnId = nextTurnId(delayedState);
+            send({ method: "thread/started", params: { thread: { ...buildThread(subThreadRecord), name: subThreadRecord.name, agentNickname: subThreadRecord.name } } });
+            send({ method: "turn/started", params: { threadId: subThread.id, turn: buildTurn(subTurnId) } });
+            send({ method: "turn/completed", params: { threadId: subThread.id, turn: buildTurn(subTurnId, "completed") } });
+            if (delayedState.nestedSubagentRequested) {
+              setTimeout(() => {
+                const nestedState = loadState();
+                const grandchild = nextThread(nestedState, thread.cwd, true, { parentThreadId: subThread.id });
+                const grandchildRecord = ensureThread(nestedState, grandchild.id);
+                grandchildRecord.name = "delayed-design-grandchild";
+                nestedState.subscriptions = [...new Set([...(nestedState.subscriptions || []), grandchild.id])];
+                saveState(nestedState);
+                send({ method: "thread/started", params: { thread: { ...buildThread(grandchildRecord), name: grandchildRecord.name, agentNickname: grandchildRecord.name } } });
+              }, 100);
+            }
+          }, 100);
+          break;
+        }
+
         if (
           BEHAVIOR === "with-subagent" ||
+          BEHAVIOR === "with-receiver-only-subagent" ||
           BEHAVIOR === "with-late-subagent-message" ||
           BEHAVIOR === "with-subagent-no-main-turn-completed"
         ) {
-          const subThread = nextThread(state, thread.cwd, true);
+          const subThread = nextThread(state, thread.cwd, true, { parentThreadId: thread.id });
           const subThreadRecord = ensureThread(state, subThread.id);
           subThreadRecord.name = "design-challenger";
+          state.subscriptions = [...new Set([...(state.subscriptions || []), subThread.id])];
           saveState(state);
           const subTurnId = nextTurnId(state);
 
-          send({ method: "thread/started", params: { thread: { ...buildThread(subThreadRecord), name: "design-challenger", agentNickname: "design-challenger" } } });
+          if (BEHAVIOR !== "with-receiver-only-subagent") {
+            send({ method: "thread/started", params: { thread: { ...buildThread(subThreadRecord), name: "design-challenger", agentNickname: "design-challenger" } } });
+          }
           send({ method: "turn/started", params: { threadId: thread.id, turn: buildTurn(turnId) } });
           send({
             method: "item/started",