From 74f367a706f0fe7aac770b36958bde9041899460 Mon Sep 17 00:00:00 2001 From: Serhii Vecherenko Date: Tue, 18 Aug 2026 11:21:28 -0700 Subject: [PATCH] fix(codex): release active thread writers on dispose --- src/supervisor/agents/codex/acp.ts | 37 +++++- src/supervisor/agents/codex/codex.test.ts | 134 ++++++++++++++++++++++ 2 files changed, 166 insertions(+), 5 deletions(-) diff --git a/src/supervisor/agents/codex/acp.ts b/src/supervisor/agents/codex/acp.ts index d8a0db854..6269ba1fa 100644 --- a/src/supervisor/agents/codex/acp.ts +++ b/src/supervisor/agents/codex/acp.ts @@ -866,6 +866,7 @@ export class CodexStructuredSession implements StructuredSessionHandle { threadId: string, turnId: string, timeoutMs?: number, + canRetry: () => boolean = () => true, ): Promise { try { await this.rpc.request("turn/interrupt", { threadId, turnId }, timeoutMs); @@ -874,6 +875,9 @@ export class CodexStructuredSession implements StructuredSessionHandle { if (!retryTurnId) { throw error; } + if (!canRetry()) { + return; + } this.activeTurnIds.delete(turnId); this.activeTurnIds.add(retryTurnId); this.activeTurnId = retryTurnId; @@ -1003,12 +1007,35 @@ export class CodexStructuredSession implements StructuredSessionHandle { this.isDisposed = true; this.clearPendingSystemErrorFallback(); - if (this.remoteThreadId) { + const remoteThreadId = this.remoteThreadId; + if (remoteThreadId) { + const activeTurnIds = new Set(this.activeTurnIds); if (this.activeTurnId) { + activeTurnIds.add(this.activeTurnId); + } + if (this.currentThreadStatus.type === "active") { + const result = await this.rpc + .request( + "thread/read", + { threadId: remoteThreadId, includeTurns: true }, + CODEX_DISPOSE_INTERRUPT_TIMEOUT_MS, + ) + .catch(() => undefined); + for (const turn of result?.thread?.turns ?? []) { + if (turn.status === "inProgress") { + activeTurnIds.add(turn.id); + } + } + } + for (const activeTurnId of activeTurnIds) { + if (!this.rpc.ownsThread(remoteThreadId)) { + break; + } await this.interruptActiveTurn( - this.remoteThreadId, - this.activeTurnId, + remoteThreadId, + activeTurnId, CODEX_DISPOSE_INTERRUPT_TIMEOUT_MS, + () => this.rpc.ownsThread(remoteThreadId), ).catch(() => undefined); } // Re-check ownership *after* the interrupt round-trip: a force-stopped @@ -1016,11 +1043,11 @@ export class CodexStructuredSession implements StructuredSessionHandle { // resubscribes to the same provider thread on the shared app-server. // Unsubscribing then would silence the live session's notifications and // strand it on "working" with no output. - if (this.rpc.ownsThread(this.remoteThreadId)) { + if (this.rpc.ownsThread(remoteThreadId)) { await this.rpc .request( "thread/unsubscribe", - { threadId: this.remoteThreadId }, + { threadId: remoteThreadId }, CODEX_DISPOSE_INTERRUPT_TIMEOUT_MS, ) .catch(() => undefined); diff --git a/src/supervisor/agents/codex/codex.test.ts b/src/supervisor/agents/codex/codex.test.ts index fcf0ba8f9..bc9642d38 100644 --- a/src/supervisor/agents/codex/codex.test.ts +++ b/src/supervisor/agents/codex/codex.test.ts @@ -1042,6 +1042,103 @@ describe("CodexStructuredSession", () => { expect(releaseAppServer).toHaveBeenCalledOnce(); }); + it("merges tracked and authoritative active provider turns during dispose", async () => { + const structuredSession = makeStructuredSession([]); + const requests: Array<{ + method: string; + params: Record; + timeoutMs?: number; + }> = []; + (structuredSession as unknown as Record)["currentThreadStatus"] = { + type: "active", + activeFlags: [], + }; + (structuredSession as unknown as Record)["activeTurnId"] = "turn-live-1"; + (structuredSession as unknown as Record)["activeTurnIds"] = new Set([ + "turn-live-1", + ]); + (structuredSession as unknown as Record)["releaseAppServer"] = () => {}; + (structuredSession as unknown as Record)["rpc"] = { + ownsThread: () => true, + request: (method: string, params: Record, timeoutMs?: number) => { + requests.push({ method, params, ...(timeoutMs !== undefined ? { timeoutMs } : {}) }); + if (method === "thread/read") { + return Promise.resolve({ + thread: { + turns: [ + { id: "turn-complete", status: "completed" }, + { id: "turn-live-1", status: "inProgress" }, + { id: "turn-live-2", status: "inProgress" }, + ], + }, + }); + } + return Promise.resolve({}); + }, + dispose: () => {}, + }; + + await structuredSession.dispose(); + + expect(requests).toEqual([ + { + method: "thread/read", + params: { threadId: "provider-thread", includeTurns: true }, + timeoutMs: 2_000, + }, + { + method: "turn/interrupt", + params: { threadId: "provider-thread", turnId: "turn-live-1" }, + timeoutMs: 2_000, + }, + { + method: "turn/interrupt", + params: { threadId: "provider-thread", turnId: "turn-live-2" }, + timeoutMs: 2_000, + }, + { + method: "thread/unsubscribe", + params: { threadId: "provider-thread" }, + timeoutMs: 2_000, + }, + ]); + }); + + it("does not interrupt a replacement that claims the thread during the active-turn read", async () => { + const structuredSession = makeStructuredSession([]); + const requests: Array<{ method: string; params: Record }> = []; + let ownsThread = true; + let resolveRead!: (result: unknown) => void; + (structuredSession as unknown as Record)["currentThreadStatus"] = { + type: "active", + activeFlags: [], + }; + (structuredSession as unknown as Record)["releaseAppServer"] = () => {}; + (structuredSession as unknown as Record)["rpc"] = { + ownsThread: () => ownsThread, + request: (method: string, params: Record) => { + requests.push({ method, params }); + if (method === "thread/read") { + return new Promise((resolve) => { + resolveRead = resolve; + }); + } + return Promise.resolve({}); + }, + dispose: () => {}, + }; + + const dispose = structuredSession.dispose(); + await vi.waitFor(() => + expect(requests.map((request) => request.method)).toEqual(["thread/read"]), + ); + ownsThread = false; + resolveRead({ thread: { turns: [{ id: "replacement-turn", status: "inProgress" }] } }); + await dispose; + + expect(requests.map((request) => request.method)).toEqual(["thread/read"]); + }); + it("keeps a replacement session subscribed when a superseded session disposes", async () => { const structuredSession = makeStructuredSession([]); const requests: Array<{ method: string; params: Record }> = []; @@ -1063,6 +1160,43 @@ describe("CodexStructuredSession", () => { expect(requests.map((request) => request.method)).toEqual(["turn/interrupt"]); }); + it("does not retry an interrupt after a replacement claims the thread", async () => { + const structuredSession = makeStructuredSession([]); + const requests: Array<{ method: string; params: Record }> = []; + let ownsThread = true; + let rejectInterrupt!: (error: Error) => void; + (structuredSession as unknown as Record)["activeTurnId"] = "turn-stale"; + (structuredSession as unknown as Record)["releaseAppServer"] = () => {}; + (structuredSession as unknown as Record)["rpc"] = { + ownsThread: () => ownsThread, + request: (method: string, params: Record) => { + requests.push({ method, params }); + if (method === "turn/interrupt") { + return new Promise((_, reject) => { + rejectInterrupt = reject; + }); + } + return Promise.resolve({}); + }, + dispose: () => {}, + }; + + const dispose = structuredSession.dispose(); + await vi.waitFor(() => + expect(requests.map((request) => request.method)).toEqual(["turn/interrupt"]), + ); + ownsThread = false; + rejectInterrupt(new Error("expected active turn id replacement-turn but found turn-stale")); + await dispose; + + expect(requests).toEqual([ + { + method: "turn/interrupt", + params: { threadId: "provider-thread", turnId: "turn-stale" }, + }, + ]); + }); + it("interrupts the active Codex app-server turn", async () => { const requests: Array<{ method: string; params: Record }> = []; const structuredSession = makeStructuredSession(requests);