fix(codex): release active thread writers on dispose

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<void> {
     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,24 +1007,47 @@ 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
       // session is replaced while this teardown drains, and the replacement
       // 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);

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<string, unknown>;
+      timeoutMs?: number;
+    }> = [];
+    (structuredSession as unknown as Record<string, unknown>)["currentThreadStatus"] = {
+      type: "active",
+      activeFlags: [],
+    };
+    (structuredSession as unknown as Record<string, unknown>)["activeTurnId"] = "turn-live-1";
+    (structuredSession as unknown as Record<string, unknown>)["activeTurnIds"] = new Set([
+      "turn-live-1",
+    ]);
+    (structuredSession as unknown as Record<string, unknown>)["releaseAppServer"] = () => {};
+    (structuredSession as unknown as Record<string, unknown>)["rpc"] = {
+      ownsThread: () => true,
+      request: (method: string, params: Record<string, unknown>, 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<string, unknown> }> = [];
+    let ownsThread = true;
+    let resolveRead!: (result: unknown) => void;
+    (structuredSession as unknown as Record<string, unknown>)["currentThreadStatus"] = {
+      type: "active",
+      activeFlags: [],
+    };
+    (structuredSession as unknown as Record<string, unknown>)["releaseAppServer"] = () => {};
+    (structuredSession as unknown as Record<string, unknown>)["rpc"] = {
+      ownsThread: () => ownsThread,
+      request: (method: string, params: Record<string, unknown>) => {
+        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<string, unknown> }> = [];
@@ -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<string, unknown> }> = [];
+    let ownsThread = true;
+    let rejectInterrupt!: (error: Error) => void;
+    (structuredSession as unknown as Record<string, unknown>)["activeTurnId"] = "turn-stale";
+    (structuredSession as unknown as Record<string, unknown>)["releaseAppServer"] = () => {};
+    (structuredSession as unknown as Record<string, unknown>)["rpc"] = {
+      ownsThread: () => ownsThread,
+      request: (method: string, params: Record<string, unknown>) => {
+        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<string, unknown> }> = [];
     const structuredSession = makeStructuredSession(requests);