From 8e624225412d137cfbed2f6c76d4cb26e18f4a17 Mon Sep 17 00:00:00 2001 From: thossullivan Date: Sat, 29 Aug 2026 09:52:58 -0500 Subject: [PATCH 1/9] fix(app-server): unsubscribe task threads after client disconnect --- plugins/codex/scripts/app-server-broker.mjs | 232 ++++++++++- .../scripts/lib/app-server-protocol.d.ts | 3 + tests/broker-subscriptions.test.mjs | 380 ++++++++++++++++++ tests/fake-codex-fixture.mjs | 59 ++- 4 files changed, 672 insertions(+), 2 deletions(-) create mode 100644 tests/broker-subscriptions.test.mjs diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 1954274fe..0f420e91f 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -10,6 +10,48 @@ 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"]); + +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 buildNotificationSubscriptionThreadIds(message) { + const threadIds = new Set(); + const params = message?.params; + if (message?.method === "thread/started" && params?.thread?.id) { + threadIds.add(params.thread.id); + } + if (params?.threadId) { + threadIds.add(params.threadId); + } + if (Array.isArray(params?.item?.receiverThreadIds)) { + for (const threadId of params.item.receiverThreadIds) { + if (threadId) { + threadIds.add(threadId); + } + } + } + return threadIds; +} function buildStreamThreadIds(method, params, result) { const threadIds = new Set(); @@ -70,6 +112,175 @@ 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 socketUnsubscribingThreadIds = new Map(); + + function addThreadOwner(socket, 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 setSocketThreadUnsubscribing(socket, threadId, isUnsubscribing) { + let threadIds = socketUnsubscribingThreadIds.get(socket); + if (isUnsubscribing) { + if (!threadIds) { + threadIds = new Set(); + socketUnsubscribingThreadIds.set(socket, threadIds); + } + threadIds.add(threadId); + return; + } + threadIds?.delete(threadId); + if (threadIds?.size === 0) { + socketUnsubscribingThreadIds.delete(socket); + } + } + + function requestThreadUnsubscribe(threadId) { + const pending = pendingUnsubscribes.get(threadId); + if (pending) { + return pending; + } + const request = appClient.request("thread/unsubscribe", { threadId }).then( + (result) => ({ result, error: null }), + (error) => { + process.stderr.write( + `Failed to unsubscribe Codex thread ${threadId}: ${error instanceof Error ? error.message : String(error)}\n` + ); + return { result: null, error }; + } + ); + pendingUnsubscribes.set(threadId, request); + void request.finally(() => { + if (pendingUnsubscribes.get(threadId) === request) { + pendingUnsubscribes.delete(threadId); + } + }); + return request; + } + + async function unsubscribeIfUnowned(threadId) { + if (threadSockets.has(threadId) || appClient.closed) { + return null; + } + return requestThreadUnsubscribe(threadId); + } + + 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))); + } + + function trackSubscriptionResults(socket, method, result, provisionalThreadIds) { + for (const threadId of buildSubscriptionThreadIds(method, result)) { + if (provisionalThreadIds.has(threadId)) { + continue; + } + if (socket.destroyed || !sockets.has(socket)) { + void unsubscribeIfUnowned(threadId); + continue; + } + addThreadOwner(socket, threadId); + } + } + + function trackNotificationSubscriptions(socket, message) { + // App-server auto-subscribes its connection to child threads created by subagents. + // Attribute those notification-only subscriptions to the active downstream client. + for (const threadId of buildNotificationSubscriptionThreadIds(message)) { + if (socket && !socket.destroyed && sockets.has(socket)) { + if (!socketUnsubscribingThreadIds.get(socket)?.has(threadId)) { + addThreadOwner(socket, threadId); + } + } else { + void unsubscribeIfUnowned(threadId); + } + } + } + + 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" }; + } + setSocketThreadUnsubscribing(socket, threadId, true); + try { + const outcome = await requestThreadUnsubscribe(threadId); + if (outcome.error) { + throw outcome.error; + } + return outcome.result; + } finally { + setSocketThreadUnsubscribing(socket, threadId, false); + } + } + + removeThreadOwner(socket, threadId); + if (threadSockets.has(threadId)) { + return { status: "unsubscribed" }; + } + + setSocketThreadUnsubscribing(socket, threadId, true); + try { + 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" }; + } finally { + setSocketThreadUnsubscribing(socket, threadId, false); + } + } function clearSocketOwnership(socket) { if (activeRequestSocket === socket) { @@ -83,6 +294,7 @@ async function main() { function routeNotification(message) { const target = activeRequestSocket ?? activeStreamSocket; + trackNotificationSubscriptions(target, message); if (!target) { return; } @@ -195,10 +407,23 @@ async function main() { } 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); + } + } activeRequestSocket = socket; try { - const result = await appClient.request(message.method, message.params ?? {}); + const result = + message.method === "thread/unsubscribe" + ? await handleThreadUnsubscribe(socket, message.params ?? {}) + : await appClient.request(message.method, message.params ?? {}); + trackSubscriptionResults(socket, message.method, result, provisionalThreadIds); send(socket, { id: message.id, result }); if (isStreaming) { activeStreamSocket = socket; @@ -208,6 +433,7 @@ async function main() { activeRequestSocket = null; } } catch (error) { + await releaseThreadOwners(socket, addedProvisionalThreadIds); send(socket, { id: message.id, error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) @@ -225,11 +451,15 @@ async function main() { socket.on("close", () => { sockets.delete(socket); clearSocketOwnership(socket); + socketUnsubscribingThreadIds.delete(socket); + void releaseThreadOwners(socket); }); socket.on("error", () => { sockets.delete(socket); clearSocketOwnership(socket); + socketUnsubscribingThreadIds.delete(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..3277b611d --- /dev/null +++ b/tests/broker-subscriptions.test.mjs @@ -0,0 +1,380 @@ +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 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 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() { + if (child.exitCode !== null || child.signalCode !== null) { + await exited; + return; + } + child.kill("SIGTERM"); + const result = await Promise.race([exited, delay(5000).then(() => null)]); + if (!result) { + child.kill("SIGKILL"); + await exited; + } + } + + 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)) { + notificationWaiters.delete(waiter); + waiter.resolve(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; + } + const notification = new Promise((resolve) => { + notificationWaiters.add({ predicate, resolve }); + }); + return Promise.race([notification, delay(timeoutMs).then(() => null)]); + } + + 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 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 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 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"); + assert.deepEqual(readState(broker.statePath).subscriptions, [threadId]); +}); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index f83c96a0d..db0efc189 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: [], capabilities: null, lastInterrupt: null }; } return JSON.parse(fs.readFileSync(STATE_PATH, "utf8")); } @@ -313,6 +313,8 @@ 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 } } }); break; @@ -346,11 +348,47 @@ rl.on("line", (line) => { } const thread = ensureThread(state, message.params.threadId); thread.updatedAt = now(); + 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); + state.subscriptions = [...new Set([...(state.subscriptions || []), thread.id])]; + saveState(state); + send({ id: message.id, result: { thread: buildThread(thread) } }); + send({ method: "thread/started", params: { thread: { id: thread.id } } }); + 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 === "unsubscribe-fails") { + saveState(state); + send({ id: message.id, error: { code: -32000, message: "thread unsubscribe failed" } }); + 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 +447,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,6 +498,22 @@ 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); + 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") } }); + }, 100); + break; + } + if ( BEHAVIOR === "with-subagent" || BEHAVIOR === "with-late-subagent-message" || @@ -466,6 +522,7 @@ rl.on("line", (line) => { const subThread = nextThread(state, thread.cwd, true); const subThreadRecord = ensureThread(state, subThread.id); subThreadRecord.name = "design-challenger"; + state.subscriptions = [...new Set([...(state.subscriptions || []), subThread.id])]; saveState(state); const subTurnId = nextTurnId(state); From 1203f5d5228f5fa274163e54c433ce628a710377 Mon Sep 17 00:00:00 2001 From: thossullivan Date: Tue, 1 Sep 2026 13:00:28 -0500 Subject: [PATCH 2/9] fix(broker): harden thread subscription ownership --- plugins/codex/scripts/app-server-broker.mjs | 355 +++++++++++--------- tests/broker-subscriptions.test.mjs | 173 +++++++++- tests/fake-codex-fixture.mjs | 32 +- 3 files changed, 371 insertions(+), 189 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 0f420e91f..dbcb068d4 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -11,6 +11,7 @@ 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]; function buildSubscriptionThreadIds(method, result) { const threadIds = new Set(); @@ -34,23 +35,22 @@ function buildProvisionalSubscriptionThreadIds(method, params) { return threadIds; } -function buildNotificationSubscriptionThreadIds(message) { - const threadIds = new Set(); - const params = message?.params; - if (message?.method === "thread/started" && params?.thread?.id) { - threadIds.add(params.thread.id); - } - if (params?.threadId) { - threadIds.add(params.threadId); +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 }); } - if (Array.isArray(params?.item?.receiverThreadIds)) { - for (const threadId of params.item.receiverThreadIds) { + + const item = message?.params?.item; + if (item?.type === "collabAgentToolCall" && item?.senderThreadId && Array.isArray(item.receiverThreadIds)) { + for (const threadId of item.receiverThreadIds) { if (threadId) { - threadIds.add(threadId); + relationships.push({ sourceThreadId: item.senderThreadId, subscribedThreadId: threadId }); } } } - return threadIds; + return relationships; } function buildStreamThreadIds(method, params, result) { @@ -117,9 +117,19 @@ async function main() { const socketThreadIds = new Map(); const threadSockets = new Map(); const pendingUnsubscribes = new Map(); - const socketUnsubscribingThreadIds = new Map(); + const unsubscribeRetryTimers = 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(); @@ -156,22 +166,6 @@ async function main() { return true; } - function setSocketThreadUnsubscribing(socket, threadId, isUnsubscribing) { - let threadIds = socketUnsubscribingThreadIds.get(socket); - if (isUnsubscribing) { - if (!threadIds) { - threadIds = new Set(); - socketUnsubscribingThreadIds.set(socket, threadIds); - } - threadIds.add(threadId); - return; - } - threadIds?.delete(threadId); - if (threadIds?.size === 0) { - socketUnsubscribingThreadIds.delete(socket); - } - } - function requestThreadUnsubscribe(threadId) { const pending = pendingUnsubscribes.get(threadId); if (pending) { @@ -195,11 +189,35 @@ async function main() { return request; } - async function unsubscribeIfUnowned(threadId) { + 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; } - return requestThreadUnsubscribe(threadId); + 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()) { @@ -209,32 +227,35 @@ async function main() { releasedThreadIds.push(threadId); } } - await Promise.all(releasedThreadIds.map((threadId) => unsubscribeIfUnowned(threadId))); + await Promise.all( + releasedThreadIds.map((threadId) => unsubscribeIfUnowned(threadId, { retryOnFailure: true })) + ); } - function trackSubscriptionResults(socket, method, result, provisionalThreadIds) { + function trackSubscriptionResults(socket, method, result) { for (const threadId of buildSubscriptionThreadIds(method, result)) { - if (provisionalThreadIds.has(threadId)) { - continue; - } if (socket.destroyed || !sockets.has(socket)) { - void unsubscribeIfUnowned(threadId); + void unsubscribeIfUnowned(threadId, { retryOnFailure: true }); continue; } addThreadOwner(socket, threadId); } } - function trackNotificationSubscriptions(socket, message) { - // App-server auto-subscribes its connection to child threads created by subagents. - // Attribute those notification-only subscriptions to the active downstream client. - for (const threadId of buildNotificationSubscriptionThreadIds(message)) { - if (socket && !socket.destroyed && sockets.has(socket)) { - if (!socketUnsubscribingThreadIds.get(socket)?.has(threadId)) { - addThreadOwner(socket, threadId); - } - } else { - void unsubscribeIfUnowned(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; + } + for (const socket of sourceOwners) { + addThreadOwner(socket, subscribedThreadId); } } } @@ -250,16 +271,11 @@ async function main() { if (threadSockets.has(threadId)) { return { status: "notSubscribed" }; } - setSocketThreadUnsubscribing(socket, threadId, true); - try { - const outcome = await requestThreadUnsubscribe(threadId); - if (outcome.error) { - throw outcome.error; - } - return outcome.result; - } finally { - setSocketThreadUnsubscribing(socket, threadId, false); + const outcome = await unsubscribeIfUnowned(threadId); + if (outcome?.error) { + throw outcome.error; } + return outcome?.result ?? { status: "notSubscribed" }; } removeThreadOwner(socket, threadId); @@ -267,19 +283,14 @@ async function main() { return { status: "unsubscribed" }; } - setSocketThreadUnsubscribing(socket, threadId, true); - try { - const outcome = await unsubscribeIfUnowned(threadId); - if (outcome?.error) { - if (!socket.destroyed && sockets.has(socket)) { - addThreadOwner(socket, threadId); - } - throw outcome.error; + const outcome = await unsubscribeIfUnowned(threadId); + if (outcome?.error) { + if (!socket.destroyed && sockets.has(socket)) { + addThreadOwner(socket, threadId); } - return outcome?.result ?? { status: "unsubscribed" }; - } finally { - setSocketThreadUnsubscribing(socket, threadId, false); + throw outcome.error; } + return outcome?.result ?? { status: "unsubscribed" }; } function clearSocketOwnership(socket) { @@ -294,7 +305,7 @@ async function main() { function routeNotification(message) { const target = activeRequestSocket ?? activeStreamSocket; - trackNotificationSubscriptions(target, message); + trackNotificationSubscriptions(message); if (!target) { return; } @@ -312,6 +323,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(); } @@ -331,134 +346,140 @@ 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; - } + async function handleLine(line) { + if (!line.trim() || socket.destroyed || !sockets.has(socket)) { + return; + } - let message; - try { - message = JSON.parse(line); - } catch (error) { - send(socket, { - id: null, - error: buildJsonRpcError(-32700, `Invalid JSON: ${error.message}`) - }); - continue; - } + 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 && message.method === "initialize") { - send(socket, { - id: message.id, - result: { - userAgent: "codex-companion-broker" - } - }); - continue; - } + if (message.id !== undefined && message.method === "initialize") { + send(socket, { + id: message.id, + result: { + userAgent: "codex-companion-broker" + } + }); + return; + } - if (message.method === "initialized" && message.id === undefined) { - continue; - } + if (message.method === "initialized" && message.id === undefined) { + return; + } - if (message.id !== undefined && message.method === "broker/shutdown") { - send(socket, { id: message.id, result: {} }); - await shutdown(server); - process.exit(0); - } + if (message.id !== undefined && message.method === "broker/shutdown") { + send(socket, { id: message.id, result: {} }); + await shutdown(server); + process.exit(0); + } - if (message.id === undefined) { - continue; - } + if (message.id === undefined) { + return; + } - const allowInterruptDuringActiveStream = - isInterruptRequest(message) && activeStreamSocket && activeStreamSocket !== socket && !activeRequestSocket; + 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 ( - ((activeRequestSocket && activeRequestSocket !== socket) || (activeStreamSocket && activeStreamSocket !== socket)) && - !allowInterruptDuringActiveStream - ) { + 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(BROKER_BUSY_RPC_CODE, "Shared Codex broker is busy.") + error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) }); - continue; } + return; + } - 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; + 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); } + } + activeRequestSocket = socket; - 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); - } + try { + const result = + message.method === "thread/unsubscribe" + ? await handleThreadUnsubscribe(socket, message.params ?? {}) + : await appClient.request(message.method, message.params ?? {}); + trackSubscriptionResults(socket, message.method, result); + send(socket, { id: message.id, result }); + if (isStreaming && !socket.destroyed && sockets.has(socket)) { + activeStreamSocket = socket; + activeStreamThreadIds = buildStreamThreadIds(message.method, message.params ?? {}, result); } - activeRequestSocket = socket; - - try { - const result = - message.method === "thread/unsubscribe" - ? await handleThreadUnsubscribe(socket, message.params ?? {}) - : await appClient.request(message.method, message.params ?? {}); - trackSubscriptionResults(socket, message.method, result, provisionalThreadIds); - send(socket, { id: message.id, result }); - if (isStreaming) { - activeStreamSocket = socket; - activeStreamThreadIds = buildStreamThreadIds(message.method, message.params ?? {}, result); - } - if (activeRequestSocket === socket) { - activeRequestSocket = null; - } - } catch (error) { - await releaseThreadOwners(socket, addedProvisionalThreadIds); - send(socket, { - id: message.id, - error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) - }); - if (activeRequestSocket === socket) { - activeRequestSocket = null; - } - if (activeStreamSocket === socket && !isStreaming) { - activeStreamSocket = null; - } + if (activeRequestSocket === socket) { + activeRequestSocket = null; } + } catch (error) { + await releaseThreadOwners(socket, addedProvisionalThreadIds); + send(socket, { + id: message.id, + error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) + }); + if (activeRequestSocket === socket) { + activeRequestSocket = null; + } + if (activeStreamSocket === socket && !isStreaming) { + activeStreamSocket = null; + } + } + } + + 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); - socketUnsubscribingThreadIds.delete(socket); void releaseThreadOwners(socket); }); socket.on("error", () => { sockets.delete(socket); clearSocketOwnership(socket); - socketUnsubscribingThreadIds.delete(socket); void releaseThreadOwners(socket); }); }); diff --git a/tests/broker-subscriptions.test.mjs b/tests/broker-subscriptions.test.mjs index 3277b611d..88fa48a9e 100644 --- a/tests/broker-subscriptions.test.mjs +++ b/tests/broker-subscriptions.test.mjs @@ -16,6 +16,20 @@ 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) { @@ -43,6 +57,7 @@ function startBroker(behavior = "review-ok") { 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, @@ -58,15 +73,21 @@ function startBroker(behavior = "review-ok") { }); async function stop() { - if (child.exitCode !== null || child.signalCode !== null) { - await exited; - return; - } - child.kill("SIGTERM"); - const result = await Promise.race([exited, delay(5000).then(() => null)]); - if (!result) { - child.kill("SIGKILL"); - await exited; + 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 }); + } } } @@ -97,8 +118,7 @@ async function connectClient(socketPath) { function notifyWaiters(message) { for (const waiter of [...notificationWaiters]) { if (waiter.predicate(message)) { - notificationWaiters.delete(waiter); - waiter.resolve(message); + waiter.settle(message); } } } @@ -152,10 +172,19 @@ async function connectClient(socketPath) { if (existing) { return existing; } - const notification = new Promise((resolve) => { - notificationWaiters.add({ predicate, resolve }); + 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); }); - return Promise.race([notification, delay(timeoutMs).then(() => null)]); } await request("initialize", {}); @@ -296,6 +325,30 @@ test("broker unsubscribes auto-subscribed subagent threads", async (t) => { 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()); @@ -321,6 +374,80 @@ test("broker unsubscribes a child thread created after its client disconnects", 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 keeps shared upstream subscriptions when one client explicitly unsubscribes", async (t) => { const broker = startBroker(); t.after(() => broker.stop()); @@ -378,3 +505,21 @@ test("broker logs upstream unsubscribe failures", async (t) => { assert.equal(warningObserved, true, "unsubscribe failure was not logged"); 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}`)); +}); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index db0efc189..125f1b897 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -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: "", @@ -316,7 +320,7 @@ rl.on("line", (line) => { 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; } @@ -343,6 +347,12 @@ rl.on("line", (line) => { } case "thread/resume": { + 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"); } @@ -356,11 +366,11 @@ rl.on("line", (line) => { case "thread/fork": { const sourceThread = ensureThread(state, message.params.threadId); - const thread = nextThread(state, sourceThread.cwd, message.params.ephemeral); + 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: { id: thread.id } } }); + send({ method: "thread/started", params: { thread: buildThread(thread) } }); break; } @@ -369,7 +379,10 @@ rl.on("line", (line) => { 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 === "unsubscribe-fails") { + 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; @@ -501,7 +514,7 @@ rl.on("line", (line) => { if (BEHAVIOR === "with-delayed-subagent") { setTimeout(() => { const delayedState = loadState(); - const subThread = nextThread(delayedState, thread.cwd, true); + 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])]; @@ -516,17 +529,20 @@ rl.on("line", (line) => { 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", From fc227197c7a8de63c1a0cbc8185e0286499876a9 Mon Sep 17 00:00:00 2001 From: thossullivan Date: Wed, 2 Sep 2026 17:11:45 -0500 Subject: [PATCH 3/9] fix(broker): serialize thread unsubscribe against reacquisition A thread resumed while its automatic thread/unsubscribe was still in flight was never released again: the later release reused the stale pending request, so the shared app-server connection stayed subscribed after every client had closed. If the app-server had processed the unsubscribe after the resume, the new owner would also have lost its subscription. Never reuse a pending unsubscribe. A release chains a fresh request behind any in-flight one and re-checks ownership before sending. A resume or review that provisionally claims a thread waits, bounded to five seconds, for an in-flight unsubscribe of that thread before it is dispatched, so a late unsubscribe cannot overtake the new subscription and a hung cleanup cannot wedge the broker. Reply to a failed request and clear the busy state before releasing provisional owners for the same reason. Add the unsubscribe-delayed and resume-fails-unsubscribe-hangs fixture behaviors, four regression tests, and assert the retry bound in the existing failure test. --- plugins/codex/scripts/app-server-broker.mjs | 66 +++++++++-- tests/broker-subscriptions.test.mjs | 118 ++++++++++++++++++++ tests/fake-codex-fixture.mjs | 29 ++++- 3 files changed, 200 insertions(+), 13 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index dbcb068d4..714454f39 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -12,6 +12,23 @@ 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; + +function settleWithin(promise, timeoutMs) { + let timer; + return Promise.race([ + promise.then( + () => {}, + () => {} + ), + new Promise((resolve) => { + timer = setTimeout(resolve, timeoutMs); + timer.unref?.(); + }) + ]).finally(() => clearTimeout(timer)); +} function buildSubscriptionThreadIds(method, result) { const threadIds = new Set(); @@ -167,13 +184,32 @@ async function main() { } function requestThreadUnsubscribe(threadId) { - const pending = pendingUnsubscribes.get(threadId); - if (pending) { - return pending; - } - const request = appClient.request("thread/unsubscribe", { threadId }).then( - (result) => ({ result, error: null }), + // Never reuse an earlier request: the thread may have been reacquired and + // released again while that request was in flight. Chain behind it so the + // app-server sees one unsubscribe at a time, then re-check ownership. + const previous = pendingUnsubscribes.get(threadId); + const execute = async () => { + if (previous) { + await settleWithin(previous, UNSUBSCRIBE_WAIT_TIMEOUT_MS); + } + if (threadSockets.has(threadId)) { + return { result: null, error: null, skipped: true }; + } + const result = await appClient.request("thread/unsubscribe", { threadId }); + return { result, error: null }; + }; + let request; + request = execute().then( + (outcome) => { + if (pendingUnsubscribes.get(threadId) === request) { + pendingUnsubscribes.delete(threadId); + } + return outcome; + }, (error) => { + if (pendingUnsubscribes.get(threadId) === request) { + pendingUnsubscribes.delete(threadId); + } process.stderr.write( `Failed to unsubscribe Codex thread ${threadId}: ${error instanceof Error ? error.message : String(error)}\n` ); @@ -181,11 +217,6 @@ async function main() { } ); pendingUnsubscribes.set(threadId, request); - void request.finally(() => { - if (pendingUnsubscribes.get(threadId) === request) { - pendingUnsubscribes.delete(threadId); - } - }); return request; } @@ -426,6 +457,15 @@ async function main() { } } activeRequestSocket = socket; + // Let an in-flight unsubscribe for the same thread settle first so it cannot + // overtake the new subscription. The wait is bounded: a hung cleanup request + // must not block this client or keep the broker busy for everyone else. + await Promise.all( + [...provisionalThreadIds].map((threadId) => { + const pending = pendingUnsubscribes.get(threadId); + return pending ? settleWithin(pending, UNSUBSCRIBE_WAIT_TIMEOUT_MS) : null; + }) + ); try { const result = @@ -442,7 +482,6 @@ async function main() { activeRequestSocket = null; } } catch (error) { - await releaseThreadOwners(socket, addedProvisionalThreadIds); send(socket, { id: message.id, error: buildJsonRpcError(error.rpcCode ?? -32000, error.message) @@ -453,6 +492,9 @@ async function main() { 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. + void releaseThreadOwners(socket, addedProvisionalThreadIds); } } diff --git a/tests/broker-subscriptions.test.mjs b/tests/broker-subscriptions.test.mjs index 88fa48a9e..46750d121 100644 --- a/tests/broker-subscriptions.test.mjs +++ b/tests/broker-subscriptions.test.mjs @@ -250,6 +250,53 @@ test("broker keeps a resumed thread subscribed until its final client closes", a 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()); @@ -448,6 +495,40 @@ test("broker serializes subscription requests from one downstream client", async 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()); @@ -503,6 +584,13 @@ test("broker logs upstream unsubscribe failures", async (t) => { () => 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]); }); @@ -523,3 +611,33 @@ test("broker retries a transient upstream unsubscribe failure", async (t) => { assert.deepEqual(readState(broker.statePath).unsubscribeRequests, [threadId, threadId]); assert.match(broker.stderr(), new RegExp(`Failed to unsubscribe Codex thread ${threadId}`)); }); + +test("broker bounds the wait for a hung unsubscribe before a resume proceeds", 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 }), 8000); + assert.notEqual(resumed, null, "resume never completed while the upstream unsubscribe hung"); + assert.ok(Date.now() - startedAt >= 4000, "resume did not wait for the in-flight unsubscribe"); + 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(); +}); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index 125f1b897..3dd3ff7ec 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: [], subscriptions: [], unsubscribeRequests: [], 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")); } @@ -347,6 +347,12 @@ rl.on("line", (line) => { } case "thread/resume": { + 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" } }); @@ -358,6 +364,9 @@ rl.on("line", (line) => { } const thread = ensureThread(state, message.params.threadId); thread.updatedAt = now(); + if (BEHAVIOR === "unsubscribe-delayed") { + 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 } }); @@ -379,6 +388,10 @@ rl.on("line", (line) => { 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) @@ -387,6 +400,20 @@ rl.on("line", (line) => { 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", From f0ddd6b95be216042a23517dc530a9073a641a77 Mon Sep 17 00:00:00 2001 From: thossullivan Date: Wed, 2 Sep 2026 17:26:54 -0500 Subject: [PATCH 4/9] fix(broker): fail a claim whose in-flight unsubscribe outlives the wait When the bounded wait for an in-flight thread/unsubscribe expires, do not send the resume anyway: that would race the outstanding unsubscribe and could leave a tracked owner without an upstream subscription. Reject the request with a retryable error, release the provisional claim, and clear the busy state so other clients continue. --- plugins/codex/scripts/app-server-broker.mjs | 28 ++++++++++++++++----- tests/broker-subscriptions.test.mjs | 15 ++++++++--- 2 files changed, 34 insertions(+), 9 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 714454f39..55ccde248 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -16,15 +16,16 @@ const UNSUBSCRIBE_RETRY_DELAYS_MS = [100, 500, 2000]; // 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, timeoutMs); + timer = setTimeout(() => resolve(false), timeoutMs); timer.unref?.(); }) ]).finally(() => clearTimeout(timer)); @@ -459,13 +460,28 @@ async function main() { activeRequestSocket = socket; // Let an in-flight unsubscribe for the same thread settle first so it cannot // overtake the new subscription. The wait is bounded: a hung cleanup request - // must not block this client or keep the broker busy for everyone else. - await Promise.all( + // must not block this client or keep the broker busy for everyone else. If + // it expires, fail the request instead of racing the outstanding unsubscribe. + const settled = await Promise.all( [...provisionalThreadIds].map((threadId) => { const pending = pendingUnsubscribes.get(threadId); - return pending ? settleWithin(pending, UNSUBSCRIBE_WAIT_TIMEOUT_MS) : null; + return pending ? settleWithin(pending, 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; + } + void releaseThreadOwners(socket, addedProvisionalThreadIds); + return; + } try { const result = diff --git a/tests/broker-subscriptions.test.mjs b/tests/broker-subscriptions.test.mjs index 46750d121..04b2a69f8 100644 --- a/tests/broker-subscriptions.test.mjs +++ b/tests/broker-subscriptions.test.mjs @@ -612,7 +612,7 @@ test("broker retries a transient upstream unsubscribe failure", async (t) => { assert.match(broker.stderr(), new RegExp(`Failed to unsubscribe Codex thread ${threadId}`)); }); -test("broker bounds the wait for a hung unsubscribe before a resume proceeds", async (t) => { +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()}`); @@ -627,9 +627,18 @@ test("broker bounds the wait for a hung unsubscribe before a resume proceeds", a const secondClient = await connectClient(broker.socketPath); const startedAt = Date.now(); - const resumed = await waitWithTimeout(secondClient.request("thread/resume", { threadId }), 8000); - assert.notEqual(resumed, null, "resume never completed while the upstream unsubscribe hung"); + 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); From 93b5660d21dff1384c46c214dba25fc042fed25d Mon Sep 17 00:00:00 2001 From: thossullivan Date: Wed, 2 Sep 2026 17:40:55 -0500 Subject: [PATCH 5/9] fix(broker): keep a pending unsubscribe outstanding until it settles A release that followed a timed-out claim installed a fresh cleanup wrapper that waited for the original unsubscribe for only another five seconds, then settled as skipped once a retried claim owned the thread. The retry could then resume upstream while the original unsubscribe was still in flight. Make a wrapper wait for its predecessor without a bound so the pending entry never settles before every underlying request has, and reject retried claims until then. --- plugins/codex/scripts/app-server-broker.mjs | 8 ++++- tests/broker-subscriptions.test.mjs | 38 +++++++++++++++++++++ tests/fake-codex-fixture.mjs | 2 +- 3 files changed, 46 insertions(+), 2 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 55ccde248..5c36d0632 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -191,7 +191,13 @@ async function main() { const previous = pendingUnsubscribes.get(threadId); const execute = async () => { if (previous) { - await settleWithin(previous, UNSUBSCRIBE_WAIT_TIMEOUT_MS); + // 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.then( + () => {}, + () => {} + ); } if (threadSockets.has(threadId)) { return { result: null, error: null, skipped: true }; diff --git a/tests/broker-subscriptions.test.mjs b/tests/broker-subscriptions.test.mjs index 04b2a69f8..f5bd9c665 100644 --- a/tests/broker-subscriptions.test.mjs +++ b/tests/broker-subscriptions.test.mjs @@ -650,3 +650,41 @@ test("broker fails a resume when an in-flight unsubscribe outlives the bounded w 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]); +}); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index 3dd3ff7ec..89b74def5 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -364,7 +364,7 @@ rl.on("line", (line) => { } const thread = ensureThread(state, message.params.threadId); thread.updatedAt = now(); - if (BEHAVIOR === "unsubscribe-delayed") { + if (BEHAVIOR === "unsubscribe-delayed" || BEHAVIOR === "resume-fails-unsubscribe-hangs") { state.requestOrder = [...(state.requestOrder || []), "thread/resume"]; } state.subscriptions = [...new Set([...(state.subscriptions || []), thread.id])]; From 68b402952445f577747795cf309a9b2e2d30b7f9 Mon Sep 17 00:00:00 2001 From: thossullivan Date: Wed, 2 Sep 2026 18:25:56 -0500 Subject: [PATCH 6/9] fix(broker): gate every path that waits on pending thread cleanup An explicit thread/unsubscribe for an unowned thread queued behind a hung automatic unsubscribe without a bound while its socket held the busy slot, so every other client stayed busy. Apply the same bounded wait and retryable rejection to explicit unsubscribes that the provisional claims already use. Two related gaps closed in the same sweep: - A release that arrives while an earlier cleanup is still queued now shares that queued request instead of adding another link to the chain. A queued request re-checks ownership when it sends, so this is safe, and repeated retries against a hung upstream no longer grow an unbounded chain of duplicate unsubscribes. - A child thread whose cleanup is still outstanding is no longer handed to new owners of its parent by a later notification. --- plugins/codex/scripts/app-server-broker.mjs | 46 ++++++++++++++------- tests/broker-subscriptions.test.mjs | 37 +++++++++++++++++ 2 files changed, 68 insertions(+), 15 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 5c36d0632..e241d77ce 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -185,16 +185,21 @@ async function main() { } function requestThreadUnsubscribe(threadId) { - // Never reuse an earlier request: the thread may have been reacquired and - // released again while that request was in flight. Chain behind it so the - // app-server sees one unsubscribe at a time, then re-check ownership. + // 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.then( + await previous.request.then( () => {}, () => {} ); @@ -202,19 +207,19 @@ async function main() { 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 }; }; - let request; - request = execute().then( + entry.request = execute().then( (outcome) => { - if (pendingUnsubscribes.get(threadId) === request) { + if (pendingUnsubscribes.get(threadId) === entry) { pendingUnsubscribes.delete(threadId); } return outcome; }, (error) => { - if (pendingUnsubscribes.get(threadId) === request) { + if (pendingUnsubscribes.get(threadId) === entry) { pendingUnsubscribes.delete(threadId); } process.stderr.write( @@ -223,8 +228,8 @@ async function main() { return { result: null, error }; } ); - pendingUnsubscribes.set(threadId, request); - return request; + pendingUnsubscribes.set(threadId, entry); + return entry.request; } function scheduleUnsubscribeRetry(threadId, retryIndex) { @@ -292,6 +297,11 @@ async function main() { 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) { addThreadOwner(socket, subscribedThreadId); } @@ -465,13 +475,19 @@ async function main() { } activeRequestSocket = socket; // Let an in-flight unsubscribe for the same thread settle first so it cannot - // overtake the new subscription. 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 the outstanding unsubscribe. + // 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( - [...provisionalThreadIds].map((threadId) => { + [...gatedThreadIds].map((threadId) => { const pending = pendingUnsubscribes.get(threadId); - return pending ? settleWithin(pending, UNSUBSCRIBE_WAIT_TIMEOUT_MS) : true; + return pending ? settleWithin(pending.request, UNSUBSCRIBE_WAIT_TIMEOUT_MS) : true; }) ); if (settled.includes(false)) { diff --git a/tests/broker-subscriptions.test.mjs b/tests/broker-subscriptions.test.mjs index f5bd9c665..5f98f4a41 100644 --- a/tests/broker-subscriptions.test.mjs +++ b/tests/broker-subscriptions.test.mjs @@ -688,3 +688,40 @@ test("broker keeps rejecting claims until a hung unsubscribe settles", async (t) 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(); +}); From fd45841d782c1e86965f8cea375b3c869254d9ec Mon Sep 17 00:00:00 2001 From: thossullivan Date: Wed, 2 Sep 2026 19:40:01 -0500 Subject: [PATCH 7/9] fix(broker): roll back children inherited through a failed claim A child notification that arrived while a provisional resume or review was waiting or in flight attributed the child to the claiming socket, and a rejected or failed claim released only the parent. The child then stayed owned until the socket closed. Record children inherited through an open claim and release them together with the claim when it is rejected or fails. --- plugins/codex/scripts/app-server-broker.mjs | 29 +++++++++++++++++-- tests/broker-subscriptions.test.mjs | 32 +++++++++++++++++++++ tests/fake-codex-fixture.mjs | 6 ++++ 3 files changed, 64 insertions(+), 3 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index e241d77ce..27ec62317 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -136,6 +136,9 @@ async function main() { 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); @@ -303,7 +306,14 @@ async function main() { continue; } for (const socket of sourceOwners) { - addThreadOwner(socket, subscribedThreadId); + if (addThreadOwner(socket, subscribedThreadId)) { + // Ownership inherited through a claim that has not succeeded yet is + // rolled back with that claim. + const claim = provisionalClaims.get(socket); + if (claim?.threadIds.has(sourceThreadId)) { + claim.inheritedThreadIds.add(subscribedThreadId); + } + } } } } @@ -473,6 +483,16 @@ async function main() { 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 @@ -501,7 +521,7 @@ async function main() { if (activeRequestSocket === socket) { activeRequestSocket = null; } - void releaseThreadOwners(socket, addedProvisionalThreadIds); + rollBackClaim(); return; } @@ -511,6 +531,9 @@ async function main() { ? 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; @@ -532,7 +555,7 @@ async function main() { } // Release after replying: a hung upstream unsubscribe must not withhold // the error or leave the broker busy for other clients. - void releaseThreadOwners(socket, addedProvisionalThreadIds); + rollBackClaim(); } } diff --git a/tests/broker-subscriptions.test.mjs b/tests/broker-subscriptions.test.mjs index 5f98f4a41..6dc28e749 100644 --- a/tests/broker-subscriptions.test.mjs +++ b/tests/broker-subscriptions.test.mjs @@ -725,3 +725,35 @@ test("broker rejects an explicit unsubscribe that would queue behind a hung clea 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"); + + // Both the parent and the inherited child are released while the client stays connected. + await waitForUnsubscribes(broker.statePath, [threadId, threadId, childThread.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 89b74def5..5172b8352 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -347,6 +347,12 @@ rl.on("line", (line) => { } case "thread/resume": { + if (BEHAVIOR === "with-delayed-subagent" && message.params.persistFullHistory === true) { + 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" } }); From e7fba1e2d0e7992d824dab984803195ac543aa6c Mon Sep 17 00:00:00 2001 From: thossullivan Date: Wed, 2 Sep 2026 21:47:32 -0500 Subject: [PATCH 8/9] fix(broker): roll back nested descendants inherited through a failed claim A grandchild spawned while a provisional claim was open has an inherited child as its source, not the claimed root, so it was owned but omitted from the rollback. Record descendants whose source is either a claimed root or an already inherited thread. --- plugins/codex/scripts/app-server-broker.mjs | 5 +++-- tests/broker-subscriptions.test.mjs | 12 ++++++++++-- tests/fake-codex-fixture.mjs | 13 +++++++++++++ 3 files changed, 26 insertions(+), 4 deletions(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index 27ec62317..cd4682b57 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -308,9 +308,10 @@ async function main() { 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. + // 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?.threadIds.has(sourceThreadId)) { + if (claim && (claim.threadIds.has(sourceThreadId) || claim.inheritedThreadIds.has(sourceThreadId))) { claim.inheritedThreadIds.add(subscribedThreadId); } } diff --git a/tests/broker-subscriptions.test.mjs b/tests/broker-subscriptions.test.mjs index 6dc28e749..3fc171683 100644 --- a/tests/broker-subscriptions.test.mjs +++ b/tests/broker-subscriptions.test.mjs @@ -751,9 +751,17 @@ test("broker rolls back child threads inherited through a failed provisional cla (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" + ); - // Both the parent and the inherited child are released while the client stays connected. - await waitForUnsubscribes(broker.statePath, [threadId, threadId, childThread.id]); + // 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 5172b8352..73ed121f8 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -348,6 +348,8 @@ 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); @@ -556,6 +558,17 @@ rl.on("line", (line) => { 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; } From 162f7e8d5ca87a547555e7defd143a9c6ced832a Mon Sep 17 00:00:00 2001 From: thossullivan Date: Wed, 2 Sep 2026 22:29:36 -0500 Subject: [PATCH 9/9] fix(broker): retry cleanup when an explicit unsubscribe's requester leaves When the final owner sends thread/unsubscribe and disconnects before the upstream request fails, nothing retried the cleanup: the owner was already removed, so the close path had nothing to release, and the explicit path restored ownership only to a still-open socket. Schedule the bounded automatic retry in that case. Test fixture: write state atomically so a test never reads a partial document, and add a delayed single-failure unsubscribe behavior for the new regression test. --- plugins/codex/scripts/app-server-broker.mjs | 4 ++++ tests/broker-subscriptions.test.mjs | 17 +++++++++++++++++ tests/fake-codex-fixture.mjs | 12 +++++++++++- 3 files changed, 32 insertions(+), 1 deletion(-) diff --git a/plugins/codex/scripts/app-server-broker.mjs b/plugins/codex/scripts/app-server-broker.mjs index cd4682b57..635c8ed0f 100644 --- a/plugins/codex/scripts/app-server-broker.mjs +++ b/plugins/codex/scripts/app-server-broker.mjs @@ -346,6 +346,10 @@ async function main() { if (outcome?.error) { if (!socket.destroyed && sockets.has(socket)) { addThreadOwner(socket, threadId); + } else { + // The requester is gone, so nobody will retry on its behalf. Fall back + // to the automatic cleanup path for the now-unowned thread. + scheduleUnsubscribeRetry(threadId, 0); } throw outcome.error; } diff --git a/tests/broker-subscriptions.test.mjs b/tests/broker-subscriptions.test.mjs index 3fc171683..93b1998da 100644 --- a/tests/broker-subscriptions.test.mjs +++ b/tests/broker-subscriptions.test.mjs @@ -765,3 +765,20 @@ test("broker rolls back child threads inherited through a failed provisional cla assert.deepEqual(readState(broker.statePath).subscriptions, []); await secondClient.end(); }); + +test("broker retries cleanup when a client disconnects during a failing explicit unsubscribe", async (t) => { + const broker = startBroker("unsubscribe-fails-once-delayed"); + 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: false })).thread.id; + // Send the explicit unsubscribe and drop the socket before the upstream reply. + void client.request("thread/unsubscribe", { threadId }).catch(() => {}); + await waitFor(() => readState(broker.statePath)?.unsubscribeRequests?.length === 1); + client.destroy(); + + // The first attempt fails 300 ms later, after the requester is gone; the broker must retry on its own. + await waitForUnsubscribes(broker.statePath, [threadId, threadId]); + assert.deepEqual(readState(broker.statePath).subscriptions, []); +}); diff --git a/tests/fake-codex-fixture.mjs b/tests/fake-codex-fixture.mjs index 73ed121f8..278736e6d 100644 --- a/tests/fake-codex-fixture.mjs +++ b/tests/fake-codex-fixture.mjs @@ -25,7 +25,10 @@ const readline = require("node:readline"); } function saveState(state) { - fs.writeFileSync(STATE_PATH, JSON.stringify(state, null, 2)); + // Write atomically so a test reading the file never sees a partial document. + const tmpPath = STATE_PATH + ".tmp"; + fs.writeFileSync(tmpPath, JSON.stringify(state, null, 2)); + fs.renameSync(tmpPath, STATE_PATH); } function requiresExperimental(field, message, state) { @@ -400,6 +403,13 @@ rl.on("line", (line) => { saveState(state); break; } + if (BEHAVIOR === "unsubscribe-fails-once-delayed" && state.unsubscribeRequests.length === 1) { + saveState(state); + setTimeout(() => { + send({ id: message.id, error: { code: -32000, message: "thread unsubscribe failed" } }); + }, 300); + break; + } if ( BEHAVIOR === "unsubscribe-fails" || (BEHAVIOR === "unsubscribe-fails-once" && state.unsubscribeRequests.length === 1)