diff --git a/README.md b/README.md index 2bda2a26..8793ffd8 100644 --- a/README.md +++ b/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 diff --git a/readme-dev.md b/readme-dev.md index bc147807..9b9c935c 100644 --- a/readme-dev.md +++ b/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 diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index f3ebb373..aab1882c 100644 --- a/src/CodexAcpServer.ts +++ b/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); } } diff --git a/src/CodexAppServerClient.ts b/src/CodexAppServerClient.ts index 0f802d68..6871a1d6 100644 --- a/src/CodexAppServerClient.ts +++ b/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(request.method, request.params) - } - else { - result = await this.connection.sendRequest(request.method); - } + const send = request.params + ? this.connection.sendRequest(request.method, request.params) + : this.connection.sendRequest(request.method); + const result = await withDeadline(send, resolveRpcTimeoutMs(request.method), `Codex RPC ${request.method}`); for (const callback of this.codexEventHandlers) { callback({ eventType: "response", ...result}); } diff --git a/src/StdUtils.ts b/src/StdUtils.ts index 3d9e2159..d00d6490 100644 --- a/src/StdUtils.ts +++ b/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(promise: Promise | T, ms: number, label: string): Promise { + const thenable = Promise.resolve(promise); + if (!Number.isFinite(ms) || ms <= 0) { + return thenable; + } + return new Promise((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 { + return new Promise((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().event, dispose() { } diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index 67cd14b3..8fbbd3c5 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/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 () => { diff --git a/src/__tests__/CodexACPAgent/session-close.test.ts b/src/__tests__/CodexACPAgent/session-close.test.ts index a04b7729..05c7a499 100644 --- a/src/__tests__/CodexACPAgent/session-close.test.ts +++ b/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(); + const turnStartCalled = deferred(); + + 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: []}>(); diff --git a/src/__tests__/StdUtils.test.ts b/src/__tests__/StdUtils.test.ts new file mode 100644 index 00000000..139f0b4c --- /dev/null +++ b/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); + }); +});