fix: time out stuck app-server RPCs and cancel before turn start

README.md

@@ -57,6 +57,7 @@ The adapter advertises ACP auth methods during initialization. Clients can authe
 - `INITIAL_AGENT_MODE` - initial mode id: `read-only`, `agent`, or `agent-full-access`.
 - `NO_BROWSER` - hide browser-based ChatGPT auth when set.
 - `APP_SERVER_LOGS` - directory for adapter logs.
+- `CODEX_RPC_TIMEOUT_MS` - override the deadline for Codex app-server JSON-RPC requests. Unset uses 60s, or 15s for `turn/interrupt`.
 
 ## Development
 

readme-dev.md

@@ -12,6 +12,7 @@ Set `CODEX_PATH` to run a different Codex binary; versions other than the one sp
 - `INITIAL_AGENT_MODE` - initial mode id: `read-only`, `agent`, or `agent-full-access`.
 - `NO_BROWSER` - hide browser-based ChatGPT auth when set.
 - `APP_SERVER_LOGS` - directory for adapter logs.
+- `CODEX_RPC_TIMEOUT_MS` - override the deadline for Codex app-server JSON-RPC requests. Unset uses 60s, or 15s for `turn/interrupt`.
 
 ### Quick start
 

src/CodexAcpServer.ts

@@ -2243,16 +2243,9 @@ export class CodexAcpServer {
             return null;
         }
 
-        if (requestName === "Close") {
-            pendingTurnStart.resolve(null);
-            return null;
-        }
-
-        const turnId = await pendingTurnStart.promise;
-        if (!turnId) {
-            logger.log(`${requestName} request rejected: no current turn`, {sessionId: sessionState.sessionId});
-        }
-        return turnId;
+        logger.log(`${requestName} request released: turn has not started`, {sessionId: sessionState.sessionId});
+        pendingTurnStart.resolve(null);
+        return null;
     }
 
     async prompt(
@@ -2793,7 +2786,9 @@ export class CodexAcpServer {
             return;
         }
 
+        this.activePrompts.get(params.sessionId)?.requestCancel();
         // After turnInterrupt(), Codex will send turn/completed, which naturally completes awaitTurnCompleted().
+        // If turn/start never returned, do not wait for it — prompt() already aborted above.
         await this.interruptSessionTurn(sessionState, "Cancel", false);
     }
 }

src/CodexAppServerClient.ts

@@ -1,4 +1,5 @@
 import {type MessageConnection, RequestType} from "vscode-jsonrpc/node";
+import {withDeadline} from "./StdUtils";
 import type {
     ClientRequest,
     InitializeParams,
@@ -129,6 +130,18 @@ const ToolRequestUserInputRequest = new RequestType<
 >('item/tool/requestUserInput');
 
 const GOAL_RUNTIME_EFFECTS_GRACE_MS = 1_000;
+export const DEFAULT_CODEX_RPC_TIMEOUT_MS = 60_000;
+
+export function resolveRpcTimeoutMs(method: string, env: NodeJS.ProcessEnv = process.env): number {
+    const override = Number(env["CODEX_RPC_TIMEOUT_MS"]);
+    if (Number.isFinite(override) && override > 0) {
+        return override;
+    }
+    if (method === "turn/interrupt") {
+        return 15_000;
+    }
+    return DEFAULT_CODEX_RPC_TIMEOUT_MS;
+}
 
 /**
  * A type-safe client over the Codex App Server's JSON-RPC API.
@@ -957,13 +970,10 @@ export class CodexAppServerClient {
         for (const callback of this.codexEventHandlers) {
             callback({ eventType: "request", ...request});
         }
-        let result: any;
-        if (request.params) {
-            result = await this.connection.sendRequest<R>(request.method, request.params)
-        }
-        else {
-            result = await this.connection.sendRequest<R>(request.method);
-        }
+        const send = request.params
+            ? this.connection.sendRequest<R>(request.method, request.params)
+            : this.connection.sendRequest<R>(request.method);
+        const result = await withDeadline(send, resolveRpcTimeoutMs(request.method), `Codex RPC ${request.method}`);
         for (const callback of this.codexEventHandlers) {
             callback({ eventType: "response", ...result});
         }

src/StdUtils.ts

@@ -3,25 +3,90 @@ import {Emitter} from "vscode-jsonrpc/node";
 import type {DataCallback, Disposable, Message, MessageReader, MessageWriter, PartialMessageInfo} from "vscode-jsonrpc/node";
 import * as acp from "@agentclientprotocol/sdk";
 
+export function withDeadline<T>(promise: Promise<T> | T, ms: number, label: string): Promise<T> {
+    const thenable = Promise.resolve(promise);
+    if (!Number.isFinite(ms) || ms <= 0) {
+        return thenable;
+    }
+    return new Promise<T>((resolve, reject) => {
+        const timer = setTimeout(() => {
+            reject(new Error(`${label} timed out after ${ms}ms`));
+        }, ms);
+        thenable.then(
+            (value) => {
+                clearTimeout(timer);
+                resolve(value);
+            },
+            (error) => {
+                clearTimeout(timer);
+                reject(error);
+            },
+        );
+    });
+}
+
+export function writeNdjsonLine(writable: Writable, line: string): Promise<void> {
+    return new Promise<void>((resolve, reject) => {
+        if (!writable.writable) {
+            reject(new Error("Codex app-server stdin is not writable"));
+            return;
+        }
+        let settled = false;
+        const finish = (error?: Error | null) => {
+            if (settled) {
+                return;
+            }
+            settled = true;
+            writable.off("error", onError);
+            writable.off("drain", onDrain);
+            if (error) {
+                reject(error);
+            } else {
+                resolve();
+            }
+        };
+        const onError = (error: Error) => finish(error);
+        const onDrain = () => finish();
+        writable.once("error", onError);
+        const ok = writable.write(line, (error) => {
+            if (error) {
+                finish(error);
+                return;
+            }
+            if (ok) {
+                finish();
+            }
+        });
+        if (!ok) {
+            writable.once("drain", onDrain);
+        }
+    });
+}
+
 //TODO ask to include proper jsonrpc field and remove
 export function createJSONRPCWriter(writable: Writable): MessageWriter {
+    const errorEmitter = new Emitter<[Error, Message | undefined, number | undefined]>();
     return {
         async write(msg: Message) {
+            if (msg && typeof msg === 'object') {
+                // remove jsonrpc for the server
+                msg = {...msg};
+                delete (msg as any).jsonrpc;
+            }
+            const line = JSON.stringify(msg) + '\n';
             try {
-                if (msg && typeof msg === 'object') {
-                    // remove jsonrpc for the server
-                    msg = {...msg};
-                    delete (msg as any).jsonrpc;
-                }
-                writable.write(JSON.stringify(msg) + '\n');
-            } catch {/* ignore */
+                await writeNdjsonLine(writable, line);
+            } catch (error) {
+                const err = error instanceof Error ? error : new Error(String(error));
+                errorEmitter.fire([err, msg, line.length]);
+                throw err;
             }
         },
 
         end() {
             writable.end();
         },
-        onError: new Emitter<[Error, Message | undefined, number | undefined]>().event,
+        onError: errorEmitter.event,
         onClose: new Emitter<void>().event,
 
         dispose() { }

src/__tests__/CodexACPAgent/CodexAcpClient.test.ts

@@ -2529,36 +2529,10 @@ describe('ACP server test', { timeout: 40_000 }, () => {
             .then(() => {
                 cancelResolved = true;
             });
-        await flushAsyncWork();
-        expect(cancelResolved).toBe(false);
-
-        mockFixture.sendServerNotification({
-            method: "thread/goal/updated",
-            params: {
-                threadId: "session-id",
-                turnId: null,
-                goal,
-            },
-        });
-
-        mockFixture.sendServerNotification({
-            method: "item/agentMessage/delta",
-            params: {
-                threadId: "session-id",
-                turnId: "goal-turn-id",
-                itemId: "goal-message-id",
-                delta: "goal output",
-            },
-        });
-
-        await vi.waitFor(() => {
-            expect(turnInterruptSpy).toHaveBeenCalledWith({
-                threadId: "session-id",
-                turnId: "goal-turn-id",
-            });
-        });
         await expect(cancelPromise).resolves.toBeUndefined();
+        expect(cancelResolved).toBe(true);
         await expect(promptPromise).resolves.toMatchObject({stopReason: "cancelled"});
+        expect(turnInterruptSpy).not.toHaveBeenCalled();
     });
 
     it('controls an active goal through the out-of-band session extension', async () => {

src/__tests__/CodexACPAgent/session-close.test.ts

@@ -113,6 +113,35 @@ describe("ACP session close", () => {
         });
     });
 
+    it("does not wait for delayed turn start before cancelling", async () => {
+        const {fixture, codexAcpAgent} = await createSession();
+        const turnStart = deferred<TurnStartResponse>();
+        const turnStartCalled = deferred<void>();
+
+        vi.spyOn(fixture.getCodexAppServerClient(), "turnStart").mockImplementation(async () => {
+            turnStartCalled.resolve();
+            return await turnStart.promise;
+        });
+
+        const promptPromise = codexAcpAgent.prompt({
+            sessionId,
+            prompt: [{type: "text", text: "long running prompt"}],
+        });
+        await turnStartCalled.promise;
+
+        await expect(codexAcpAgent.cancel({sessionId})).resolves.toBeUndefined();
+        await expect(promptPromise).resolves.toMatchObject({stopReason: "cancelled"});
+
+        fixture.clearCodexConnectionDump();
+        turnStart.resolve(createTurnStartResponse("turn-id"));
+
+        await vi.waitFor(() => {
+            const lateRequestMethods = fixture.getCodexConnectionEvents([])
+                .flatMap(event => event.eventType === "request" ? [event.method] : []);
+            expect(lateRequestMethods).toContain("turn/interrupt");
+        });
+    });
+
     it("does not start a turn after close while prompt startup is still refreshing skills", async () => {
         const {fixture, codexAcpAgent} = await createSession();
         const skillRefresh = deferred<{data: []}>();

src/__tests__/StdUtils.test.ts

@@ -0,0 +1,33 @@
+import {PassThrough} from "node:stream";
+import {describe, expect, it} from "vitest";
+import {createJSONRPCWriter, withDeadline, writeNdjsonLine} from "../StdUtils";
+import {resolveRpcTimeoutMs} from "../CodexAppServerClient";
+
+describe("stdio helpers", () => {
+    it("rejects a deadline when the work never finishes", async () => {
+        await expect(withDeadline(new Promise(() => {}), 10, "Codex RPC turn/start"))
+            .rejects.toThrow("Codex RPC turn/start timed out after 10ms");
+    });
+
+    it("writes newline JSON and waits when the pipe asks for drain", async () => {
+        const writable = new PassThrough({highWaterMark: 8});
+        const chunks: string[] = [];
+        writable.on("data", (chunk) => chunks.push(String(chunk)));
+        await writeNdjsonLine(writable, '{"ok":true}\n');
+        expect(chunks.join("")).toBe('{"ok":true}\n');
+    });
+
+    it("fails the JSON-RPC writer instead of swallowing a closed pipe", async () => {
+        const writable = new PassThrough();
+        writable.end();
+        const writer = createJSONRPCWriter(writable);
+        await expect(writer.write({jsonrpc: "2.0", method: "turn/start", id: 1} as any))
+            .rejects.toThrow("Codex app-server stdin is not writable");
+    });
+
+    it("uses a shorter timeout for interrupt and honors CODEX_RPC_TIMEOUT_MS", () => {
+        expect(resolveRpcTimeoutMs("turn/start", {})).toBe(60_000);
+        expect(resolveRpcTimeoutMs("turn/interrupt", {})).toBe(15_000);
+        expect(resolveRpcTimeoutMs("turn/start", {CODEX_RPC_TIMEOUT_MS: "2500"})).toBe(2500);
+    });
+});