From b457024d67000a4d00bbae8fc8f42faab5698947 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 19 Aug 2026 23:00:44 +0800 Subject: [PATCH 1/5] feat(api): abort signal support for openai-codex (completePrompt + createMessage) - completePrompt: use a request-local signal built with mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) for the fetch call instead of the handler-wide AbortController; re-throw abort errors as-is so cancellation is detectable by the "AbortError" name - createMessage: pass metadata into executeRequest and bridge metadata?.abortSignal into the internal AbortController (Bedrock pattern: pre-aborted guard + { once: true } listener), covering both the OpenAI SDK streaming path and the manual SSE fetch fallback - specs: port the reference completePrompt coverage (request body, timeoutMs=0, abortSignal/timeoutMs merging, error paths) and add pre-aborted and in-flight abort tests rejecting with name === "AbortError"; port the createMessage abort bridge + pre-aborted tests into the native tool calls spec --- .../openai-codex-native-tool-calls.spec.ts | 89 +++++ .../providers/__tests__/openai-codex.spec.ts | 364 ++++++++++++++++++ src/api/providers/openai-codex.ts | 26 +- 3 files changed, 474 insertions(+), 5 deletions(-) diff --git a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts index d9fcdcb967..b1ded8bbf3 100644 --- a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts +++ b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts @@ -518,4 +518,93 @@ describe("OpenAiCodexHandler native tool calls", () => { }), ) }) + + describe("createMessage abort signal", () => { + it("should bridge the external abortSignal into the internal AbortController", async () => { + vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + const mockCreate = vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + yield { type: "response.text.delta", delta: "test" } + yield { + type: "response.completed", + response: { + id: "resp_1", + status: "completed", + output: [{ type: "message", content: [{ type: "output_text", text: "test" }] }], + usage: { input_tokens: 1, output_tokens: 1 }, + }, + } + }, + }) + Object.assign(handler, { + client: { + responses: { create: mockCreate }, + }, + }) + + const controller = new AbortController() + const stream = handler.createMessage("system", [{ role: "user", content: "hello" }], { + taskId: "t", + abortSignal: controller.signal, + }) + + // Consume the stream to trigger the request + await collectStream(stream) + + expect(mockCreate).toHaveBeenCalled() + const createCallArgs = mockCreate.mock.calls[0][1] as { signal?: AbortSignal } + expect(createCallArgs.signal).toBeDefined() + expect(createCallArgs.signal).toBeInstanceOf(AbortSignal) + + // Verify the signal is not aborted before we abort the external one + expect(createCallArgs.signal?.aborted).toBe(false) + + // Abort the external signal; the bridge aborts the internal controller + controller.abort() + expect(controller.signal.aborted).toBe(true) + }) + + it("should immediately abort when the external signal is already aborted", async () => { + vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + const mockCreate = vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + yield { type: "response.text.delta", delta: "test" } + yield { + type: "response.completed", + response: { + id: "resp_1", + status: "completed", + output: [{ type: "message", content: [{ type: "output_text", text: "test" }] }], + usage: { input_tokens: 1, output_tokens: 1 }, + }, + } + }, + }) + Object.assign(handler, { + client: { + responses: { create: mockCreate }, + }, + }) + + const controller = new AbortController() + controller.abort() // Pre-abort + + const stream = handler.createMessage("system", [{ role: "user", content: "hello" }], { + taskId: "t", + abortSignal: controller.signal, + }) + + // Consume the stream to trigger the request + await collectStream(stream) + + expect(mockCreate).toHaveBeenCalled() + const createCallArgs = mockCreate.mock.calls[0][1] as { signal?: AbortSignal } + // The internal signal should already be aborted since the external one was pre-aborted + expect(createCallArgs.signal?.aborted).toBe(true) + }) + }) }) diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index 9a256535c1..01ff123f0a 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -641,3 +641,367 @@ describe("OpenAiCodexHandler Luna Responses Lite requests", () => { }) }) }) + +describe("OpenAiCodexHandler.completePrompt", () => { + afterEach(() => { + vitest.restoreAllMocks() + vitest.unstubAllGlobals() + }) + + it("should call fetch with correct request body and return text response", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + const mockFetch = vitest.fn().mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "Hello world" }], + }, + ], + }), + text: () => Promise.resolve(""), + }) + vitest.stubGlobal("fetch", mockFetch) + + const result = await handler.completePrompt("test prompt") + + expect(result).toBe("Hello world") + expect(mockFetch).toHaveBeenCalledWith( + expect.stringContaining("/responses"), + expect.objectContaining({ + method: "POST", + headers: expect.objectContaining({ + Authorization: "Bearer test-token", + originator: "zoo-code", + }), + body: expect.stringContaining('"model":"gpt-5.1-codex"'), + }), + ) + }) + + it("should treat timeoutMs=0 as no timeout", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + let fetchInit: RequestInit | undefined + const mockFetch = vitest.fn().mockImplementation(async (_url: string, init?: RequestInit) => { + fetchInit = init + return { + ok: true, + json: () => + Promise.resolve({ + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "response" }], + }, + ], + }), + text: () => Promise.resolve(""), + } + }) + vitest.stubGlobal("fetch", mockFetch) + + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + + expect(mockFetch).toHaveBeenCalled() + expect(fetchInit?.signal).toBeDefined() + expect(fetchInit?.signal?.aborted).toBe(false) + }) + + it("should merge abortSignal with local controller", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + let fetchInit: RequestInit | undefined + const mockFetch = vitest.fn().mockImplementation(async (_url: string, init?: RequestInit) => { + fetchInit = init + return { + ok: true, + json: () => + Promise.resolve({ + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "response" }], + }, + ], + }), + text: () => Promise.resolve(""), + } + }) + vitest.stubGlobal("fetch", mockFetch) + + const controller = new AbortController() + const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal }) + controller.abort() + await promise + + expect(mockFetch).toHaveBeenCalled() + expect(fetchInit?.signal).toBeDefined() + expect(fetchInit?.signal?.aborted).toBe(true) + }) + + it("should merge abortSignal and timeoutMs together", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + let fetchInit: RequestInit | undefined + const mockFetch = vitest.fn().mockImplementation(async (_url: string, init?: RequestInit) => { + fetchInit = init + return { + ok: true, + json: () => + Promise.resolve({ + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "response" }], + }, + ], + }), + text: () => Promise.resolve(""), + } + }) + vitest.stubGlobal("fetch", mockFetch) + + const controller = new AbortController() + const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 5000 }) + controller.abort() + await promise + + expect(mockFetch).toHaveBeenCalled() + expect(fetchInit?.signal).toBeDefined() + expect(fetchInit?.signal?.aborted).toBe(true) + }) + + it("should reject with an AbortError when the signal is already aborted", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + // Emulate fetch semantics: an already-aborted signal rejects immediately + const mockFetch = vitest.fn().mockImplementation((_url: string, init?: RequestInit) => { + if (init?.signal?.aborted) { + const abortError = new Error("The operation was aborted") + abortError.name = "AbortError" + return Promise.reject(abortError) + } + return Promise.resolve({ + ok: true, + json: () => Promise.resolve({}), + text: () => Promise.resolve(""), + }) + }) + vitest.stubGlobal("fetch", mockFetch) + + const controller = new AbortController() + controller.abort() + + await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toSatisfy( + (error: unknown) => error instanceof Error && error.name === "AbortError", + ) + }) + + it("should abort an in-flight request when the external signal aborts", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + // Emulate fetch semantics: the pending request rejects when the signal aborts + const mockFetch = vitest.fn().mockImplementation((_url: string, init?: RequestInit) => { + return new Promise((_resolve, reject) => { + init?.signal?.addEventListener( + "abort", + () => { + const abortError = new Error("The operation was aborted") + abortError.name = "AbortError" + reject(abortError) + }, + { once: true }, + ) + }) + }) + vitest.stubGlobal("fetch", mockFetch) + + const controller = new AbortController() + const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal }) + await vitest.waitFor(() => expect(mockFetch).toHaveBeenCalled()) + controller.abort() + + await expect(promise).rejects.toSatisfy( + (error: unknown) => error instanceof Error && error.name === "AbortError", + ) + }) + + it("should return empty string when no output text found", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + const mockFetch = vitest.fn().mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ output: [{ type: "message", role: "assistant", content: [] }] }), + text: () => Promise.resolve(""), + }) + vitest.stubGlobal("fetch", mockFetch) + + const result = await handler.completePrompt("test prompt") + + expect(result).toBe("") + }) + + it("should handle responseData.text fallback", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + const mockFetch = vitest.fn().mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ text: "fallback response" }), + text: () => Promise.resolve(""), + }) + vitest.stubGlobal("fetch", mockFetch) + + const result = await handler.completePrompt("test prompt") + + expect(result).toBe("fallback response") + }) + + it("should throw error when not authenticated", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue(null) + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + const controller = new AbortController() + await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toThrow() + }) + + it("should throw error when fetch returns non-ok response", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + const mockFetch = vitest.fn().mockResolvedValue({ + ok: false, + status: 401, + text: () => Promise.resolve("Unauthorized"), + }) + vitest.stubGlobal("fetch", mockFetch) + + const controller = new AbortController() + await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toThrow() + }) + + it("should include reasoning config when model has reasoning effort", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + const mockFetch = vitest.fn().mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "response" }], + }, + ], + }), + text: () => Promise.resolve(""), + }) + vitest.stubGlobal("fetch", mockFetch) + + await handler.completePrompt("test prompt") + + expect(mockFetch).toHaveBeenCalled() + const fetchOptions = mockFetch.mock.calls[0][1] + const requestBody = JSON.parse(fetchOptions.body) + expect(requestBody.include).toContain("reasoning.encrypted_content") + expect(requestBody.reasoning).toBeDefined() + expect(requestBody.reasoning.effort).toBe("medium") + }) + + it("should include ChatGPT-Account-Id header when accountId is available", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_12345") + + const mockFetch = vitest.fn().mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "response" }], + }, + ], + }), + text: () => Promise.resolve(""), + }) + vitest.stubGlobal("fetch", mockFetch) + + await handler.completePrompt("test prompt") + + expect(mockFetch).toHaveBeenCalled() + const fetchOptions = mockFetch.mock.calls[0][1] + expect(fetchOptions.headers["ChatGPT-Account-Id"]).toBe("acct_12345") + }) + + it("should work without accountId when not available", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue(null) + + const mockFetch = vitest.fn().mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "response" }], + }, + ], + }), + text: () => Promise.resolve(""), + }) + vitest.stubGlobal("fetch", mockFetch) + + await handler.completePrompt("test prompt") + + expect(mockFetch).toHaveBeenCalled() + const fetchOptions = mockFetch.mock.calls[0][1] + expect(fetchOptions.headers["ChatGPT-Account-Id"]).toBeUndefined() + }) +}) diff --git a/src/api/providers/openai-codex.ts b/src/api/providers/openai-codex.ts index e9bc3bbf5d..3e64609ef2 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -29,6 +29,7 @@ import { isMcpTool } from "../../utils/mcp-name" import { sanitizeOpenAiCallId } from "../../utils/tool-id" import { openAiCodexOAuthManager } from "../../integrations/openai-codex/oauth" import { t } from "../../i18n" +import { mergeAbortSignalAndTimeout } from "./utils/abort-signal" export type OpenAiCodexModel = ReturnType @@ -274,7 +275,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // Make the request with retry on auth failure for (let attempt = 0; attempt < 2; attempt++) { try { - yield* this.executeRequest(requestBody, model, accessToken, effectiveSessionId) + yield* this.executeRequest(requestBody, model, accessToken, effectiveSessionId, metadata) return } catch (error) { const message = error instanceof Error ? error.message : String(error) @@ -438,10 +439,21 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion model: OpenAiCodexModel, accessToken: string, effectiveSessionId: string, + metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { // Create AbortController for cancellation this.abortController = new AbortController() + // Bridge the external abort signal into the internal controller (Bedrock pattern) + const externalAbortSignal = metadata?.abortSignal + if (externalAbortSignal) { + if (externalAbortSignal.aborted) { + this.abortController.abort() + } else { + externalAbortSignal.addEventListener("abort", () => this.abortController?.abort(), { once: true }) + } + } + try { // Prefer OpenAI SDK streaming (same approach as openai-native) so event handling // is consistent across providers. @@ -1257,7 +1269,8 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { - this.abortController = new AbortController() + const requestAbortSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) + const requestSignal = requestAbortSignal ?? new AbortController().signal try { const model = this.getModel() @@ -1318,7 +1331,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion method: "POST", headers, body: JSON.stringify(requestBody), - signal: this.abortController.signal, + signal: requestSignal, }) if (!response.ok) { @@ -1354,12 +1367,15 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion const apiError = new ApiProviderError(errorMessage, this.providerName, errorModel.id, "completePrompt") TelemetryService.instance.captureException(apiError) + // Re-throw abort errors as-is so callers can detect cancellation by the "AbortError" name + if (error instanceof Error && error.name === "AbortError") { + throw error + } + if (error instanceof Error) { throw new Error(t("common:errors.openAiCodex.completionError", { message: error.message })) } throw error - } finally { - this.abortController = undefined } } } From 22da1d1c873469be5fc4c5b4d7c94f316eb2a926 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 00:50:20 +0800 Subject: [PATCH 2/5] fix(api): address CodeRabbit review on openai-codex abort handling - executeRequest: create a request-local AbortController (mirrored to this.abortController for existing abort handling); the external-signal bridge listener now captures the local controller and is removed in finally, so a late abort from an earlier request can no longer abort a newer request and listeners no longer leak - completePrompt: normalize any rejected request whose request-local signal aborted (external abort, AbortSignal.timeout "TimeoutError") to an error with name "AbortError", and throw the same AbortError when the transport quietly completes after cancellation - specs: bridge test now asserts the captured request-local SDK signal aborts mid-flight; merge tests assert AbortError rejection on quiet completion; new tests cover timeout cancellation and quiet completion after abort --- .../openai-codex-native-tool-calls.spec.ts | 60 ++++++++++------ .../providers/__tests__/openai-codex.spec.ts | 70 ++++++++++++++++++- src/api/providers/openai-codex.ts | 57 +++++++++++---- 3 files changed, 148 insertions(+), 39 deletions(-) diff --git a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts index b1ded8bbf3..37037e4adc 100644 --- a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts +++ b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts @@ -524,19 +524,30 @@ describe("OpenAiCodexHandler native tool calls", () => { vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") - const mockCreate = vi.fn().mockResolvedValue({ - async *[Symbol.asyncIterator]() { - yield { type: "response.text.delta", delta: "test" } - yield { - type: "response.completed", - response: { - id: "resp_1", - status: "completed", - output: [{ type: "message", content: [{ type: "output_text", text: "test" }] }], - usage: { input_tokens: 1, output_tokens: 1 }, - }, - } - }, + // The mock transport pauses mid-flight until the request-local signal aborts + const mockCreate = vi.fn().mockImplementation(async (_body: unknown, init?: { signal?: AbortSignal }) => { + return { + async *[Symbol.asyncIterator]() { + yield { type: "response.text.delta", delta: "test" } + await new Promise((resolve) => { + const signal = init?.signal + if (!signal || signal.aborted) { + resolve() + return + } + signal.addEventListener("abort", () => resolve(), { once: true }) + }) + yield { + type: "response.completed", + response: { + id: "resp_1", + status: "completed", + output: [{ type: "message", content: [{ type: "output_text", text: "test" }] }], + usage: { input_tokens: 1, output_tokens: 1 }, + }, + } + }, + } }) Object.assign(handler, { client: { @@ -550,20 +561,25 @@ describe("OpenAiCodexHandler native tool calls", () => { abortSignal: controller.signal, }) - // Consume the stream to trigger the request - await collectStream(stream) + // Consume the stream (the mock transport pauses mid-flight) + const collected = collectStream(stream) + + // Wait until the request has started; the bridge listener is registered before + // the SDK call, so aborting now lands mid-flight + await vi.waitFor(() => expect(mockCreate).toHaveBeenCalled()) + + // Abort the external signal mid-flight; the bridge must abort the request-local controller + controller.abort() + + const chunks = await collected + expect(chunks.length).toBeGreaterThan(0) expect(mockCreate).toHaveBeenCalled() const createCallArgs = mockCreate.mock.calls[0][1] as { signal?: AbortSignal } + // The captured (request-local) signal passed to the SDK must now be aborted expect(createCallArgs.signal).toBeDefined() expect(createCallArgs.signal).toBeInstanceOf(AbortSignal) - - // Verify the signal is not aborted before we abort the external one - expect(createCallArgs.signal?.aborted).toBe(false) - - // Abort the external signal; the bridge aborts the internal controller - controller.abort() - expect(controller.signal.aborted).toBe(true) + expect(createCallArgs.signal?.aborted).toBe(true) }) it("should immediately abort when the external signal is already aborted", async () => { diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index 01ff123f0a..ab80805fb1 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -748,7 +748,11 @@ describe("OpenAiCodexHandler.completePrompt", () => { const controller = new AbortController() const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal }) controller.abort() - await promise + // The mock transport completes without honoring the abort; an aborted request must + // reject with an AbortError rather than return the late response + await expect(promise).rejects.toSatisfy( + (error: unknown) => error instanceof Error && error.name === "AbortError", + ) expect(mockFetch).toHaveBeenCalled() expect(fetchInit?.signal).toBeDefined() @@ -784,7 +788,11 @@ describe("OpenAiCodexHandler.completePrompt", () => { const controller = new AbortController() const promise = handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 5000 }) controller.abort() - await promise + // The mock transport completes without honoring the abort; an aborted request must + // reject with an AbortError rather than return the late response + await expect(promise).rejects.toSatisfy( + (error: unknown) => error instanceof Error && error.name === "AbortError", + ) expect(mockFetch).toHaveBeenCalled() expect(fetchInit?.signal).toBeDefined() @@ -1004,4 +1012,62 @@ describe("OpenAiCodexHandler.completePrompt", () => { const fetchOptions = mockFetch.mock.calls[0][1] expect(fetchOptions.headers["ChatGPT-Account-Id"]).toBeUndefined() }) + + it("should reject with an AbortError when the timeout elapses", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + // Emulate a hung fetch that rejects when its signal aborts (native AbortSignal.timeout) + const mockFetch = vitest.fn().mockImplementation((_url: string, init?: RequestInit) => { + return new Promise((_resolve, reject) => { + init?.signal?.addEventListener( + "abort", + () => { + const abortError = new Error("The operation was aborted") + abortError.name = "AbortError" + reject(abortError) + }, + { once: true }, + ) + }) + }) + vitest.stubGlobal("fetch", mockFetch) + + await expect(handler.completePrompt("test prompt", { timeoutMs: 50 })).rejects.toSatisfy( + (error: unknown) => error instanceof Error && error.name === "AbortError", + ) + }) + + it("should reject with an AbortError when the signal aborts and the transport completes anyway", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.1-codex" }) + + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + + // Emulate a transport that resolves successfully despite the aborted signal + const mockFetch = vitest.fn().mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "late response" }], + }, + ], + }), + text: () => Promise.resolve(""), + }) + vitest.stubGlobal("fetch", mockFetch) + + const controller = new AbortController() + controller.abort() + + await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toSatisfy( + (error: unknown) => error instanceof Error && error.name === "AbortError", + ) + }) }) diff --git a/src/api/providers/openai-codex.ts b/src/api/providers/openai-codex.ts index 3e64609ef2..fb01c9e98d 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -441,16 +441,22 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion effectiveSessionId: string, metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { - // Create AbortController for cancellation - this.abortController = new AbortController() + // Create a request-local AbortController. The class field mirrors it so the + // existing abort handling keeps working, but the bridge listener below captures + // the local controller so a late abort can never hit a newer request's controller. + const requestController = new AbortController() + this.abortController = requestController - // Bridge the external abort signal into the internal controller (Bedrock pattern) + // Bridge the external abort signal into the request controller (Bedrock pattern) const externalAbortSignal = metadata?.abortSignal + const bridgeAbort = () => requestController.abort() + let bridgeCleanup: (() => void) | undefined if (externalAbortSignal) { if (externalAbortSignal.aborted) { - this.abortController.abort() + requestController.abort() } else { - externalAbortSignal.addEventListener("abort", () => this.abortController?.abort(), { once: true }) + externalAbortSignal.addEventListener("abort", bridgeAbort, { once: true }) + bridgeCleanup = () => externalAbortSignal.removeEventListener("abort", bridgeAbort) } } @@ -475,7 +481,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion }) const stream = (await (client as any).responses.create(requestBody, { - signal: this.abortController.signal, + signal: requestController.signal, // If the SDK supports per-request overrides, ensure headers are present. headers: codexHeaders, })) as AsyncIterable @@ -487,7 +493,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } for await (const event of stream) { - if (this.abortController.signal.aborted) { + if (requestController.signal.aborted) { break } @@ -503,7 +509,11 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion yield* this.makeCodexRequest(requestBody, model, accessToken, effectiveSessionId) } } finally { - this.abortController = undefined + bridgeCleanup?.() + // Only clear the field if this request still owns it + if (this.abortController === requestController) { + this.abortController = undefined + } } } @@ -1344,32 +1354,49 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion const responseData = await response.json() + let result: string | undefined if (responseData?.output && Array.isArray(responseData.output)) { for (const outputItem of responseData.output) { if (outputItem.type === "message" && outputItem.content) { for (const content of outputItem.content) { if (content.type === "output_text" && content.text) { - return content.text + result = content.text + break } } + if (result !== undefined) { + break + } } } } - if (responseData?.text) { - return responseData.text + if (result === undefined && responseData?.text) { + result = responseData.text } - return "" + // The request may have been cancelled while the transport was finishing; + // surface it as an abort instead of returning the completed response. + if (requestSignal.aborted) { + const abortError = new Error("This operation was aborted") + abortError.name = "AbortError" + throw abortError + } + + return result ?? "" } catch (error) { const errorModel = this.getModel() const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, errorModel.id, "completePrompt") TelemetryService.instance.captureException(apiError) - // Re-throw abort errors as-is so callers can detect cancellation by the "AbortError" name - if (error instanceof Error && error.name === "AbortError") { - throw error + // An aborted request surfaces as "AbortError" (external signal) or "TimeoutError" + // (AbortSignal.timeout); normalize it so callers can detect cancellation by the + // "AbortError" name. + if (requestSignal.aborted) { + const abortError = new Error("This operation was aborted") + abortError.name = "AbortError" + throw abortError } if (error instanceof Error) { From 0ca6c23278761c314da09d39c92a35ba156b44d9 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 19:00:27 +0800 Subject: [PATCH 3/5] refactor(api): adopt RequestConfigBuilder in feat/abort-r1-openai-codex abort wiring --- .../__tests__/request-config-builder.spec.ts | 40 +++++++++++++++++++ .../config-builder/request-config-builder.ts | 15 +++++++ src/api/providers/openai-codex.ts | 7 +++- 3 files changed, 60 insertions(+), 2 deletions(-) diff --git a/src/api/providers/__tests__/request-config-builder.spec.ts b/src/api/providers/__tests__/request-config-builder.spec.ts index 977b09df6b..f9557a123f 100644 --- a/src/api/providers/__tests__/request-config-builder.spec.ts +++ b/src/api/providers/__tests__/request-config-builder.spec.ts @@ -505,4 +505,44 @@ describe("RequestConfigBuilder", () => { expect(config.maxTokens).toBe(2000) }) }) + + describe("static merge helpers (canonical abort-signal entry points)", () => { + it("returns undefined from mergeAbortSignalAndTimeout when no external signal and no valid timeout", () => { + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(undefined, undefined)).toBeUndefined() + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(undefined, 0)).toBeUndefined() + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(undefined, -5)).toBeUndefined() + }) + + it("returns the external signal directly when no timeout is merged", () => { + const controller = new AbortController() + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(controller.signal, undefined)).toBe( + controller.signal, + ) + expect(RequestConfigBuilder.mergeAbortSignalAndTimeout(controller.signal, 0)).toBe(controller.signal) + }) + + it("returns the primary signal directly from mergeAbortSignals when there is no secondary", () => { + const controller = new AbortController() + expect(RequestConfigBuilder.mergeAbortSignals(controller.signal)).toBe(controller.signal) + expect(RequestConfigBuilder.mergeAbortSignals(controller.signal, undefined)).toBe(controller.signal) + }) + + it("delegates to AbortSignal.any when two distinct signals are merged", () => { + const a = new AbortController() + const b = new AbortController() + const merged = RequestConfigBuilder.mergeAbortSignals(a.signal, b.signal) + expect(merged.aborted).toBe(false) + b.abort() + expect(merged.aborted).toBe(true) + }) + + it("aborts the merged signal when the primary signal aborts", () => { + const a = new AbortController() + const b = new AbortController() + const merged = RequestConfigBuilder.mergeAbortSignals(a.signal, b.signal) + expect(merged.aborted).toBe(false) + a.abort() + expect(merged.aborted).toBe(true) + }) + }) }) diff --git a/src/api/providers/config-builder/request-config-builder.ts b/src/api/providers/config-builder/request-config-builder.ts index 2201d735bc..a3c1ba7ebb 100644 --- a/src/api/providers/config-builder/request-config-builder.ts +++ b/src/api/providers/config-builder/request-config-builder.ts @@ -163,4 +163,19 @@ export class RequestConfigBuilder @@ -1279,7 +1279,10 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { - const requestAbortSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) + const requestAbortSignal = RequestConfigBuilder.mergeAbortSignalAndTimeout( + options?.abortSignal, + options?.timeoutMs, + ) const requestSignal = requestAbortSignal ?? new AbortController().signal try { From 714754cfa78f4918189feec0b2b829425a2786b4 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 14:30:13 +0800 Subject: [PATCH 4/5] fix(api): address CodeRabbit abort-signal findings in openai-codex --- .../openai-codex-native-tool-calls.spec.ts | 10 ++- .../providers/__tests__/openai-codex.spec.ts | 67 ++++++++++++++++ src/api/providers/openai-codex.ts | 80 ++++++++++++++----- 3 files changed, 133 insertions(+), 24 deletions(-) diff --git a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts index b4bdf87c04..133b97bbf0 100644 --- a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts +++ b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts @@ -577,7 +577,9 @@ describe("OpenAiCodexHandler native tool calls", () => { controller.abort() const chunks = await collected - expect(chunks.length).toBeGreaterThan(0) + // Exactly the pre-abort delta: the completed event only arrives once the abort resolves + // the transport's pause, so the loop must break before processing it. + expect(chunks).toEqual([{ type: "text", text: "test" }]) expect(mockCreate).toHaveBeenCalled() const createCallArgs = mockCreate.mock.calls[0][1] as { signal?: AbortSignal } @@ -619,8 +621,10 @@ describe("OpenAiCodexHandler native tool calls", () => { abortSignal: controller.signal, }) - // Consume the stream to trigger the request - await collectStream(stream) + // Consume the stream to trigger the request; the request-local controller is already + // aborted, so the loop must break before the first event is processed. + const chunks = await collectStream(stream) + expect(chunks).toEqual([]) expect(mockCreate).toHaveBeenCalled() const createCallArgs = mockCreate.mock.calls[0][1] as { signal?: AbortSignal } diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index e11b35256f..d88a2b2172 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -9,6 +9,7 @@ vitest.mock("@roo-code/telemetry", () => ({ })) import { Anthropic } from "@anthropic-ai/sdk" +import { TelemetryService } from "@roo-code/telemetry" import { OPEN_AI_CODEX_SERVICE_TIER_KEY, OpenAiCodexServiceTier, SERVICE_TIER_KEY } from "@roo-code/types" import { OpenAiCodexHandler, transformResponsesLiteBody } from "../openai-codex" import { openAiCodexOAuthManager } from "../../../integrations/openai-codex/oauth" @@ -591,6 +592,72 @@ describe("OpenAiCodexHandler.completePrompt streaming", () => { expect(mockFetch).not.toHaveBeenCalled() }) + // The cancellation wins over the auth retry: force-refreshing a token for a request that is + // already gone would spend a network round trip on a dead request and surface the + // cancellation as an authentication failure. + it("fails fast with the abort contract instead of force-refreshing the token after cancellation", async () => { + const handler = createHandler() + const refresh = vitest.spyOn(openAiCodexOAuthManager, "forceRefreshAccessToken") + const controller = new AbortController() + // The SDK keeps the request open until the signal it was handed aborts, then rejects the way + // it rejects aborted requests. + const create = vitest.fn().mockImplementation( + (_body: unknown, options: { signal: AbortSignal }) => + new Promise((_resolve, reject) => { + options.signal.addEventListener("abort", () => reject(new Error("Request was aborted")), { + once: true, + }) + }), + ) + Reflect.set(handler, "client", { responses: { create } }) + const mockFetch = vitest.fn() + vitest.stubGlobal("fetch", mockFetch) + + const completion = handler.completePrompt("Hello", { abortSignal: controller.signal }) + await vitest.waitFor(() => expect(create).toHaveBeenCalled()) + controller.abort() + + await expect(completion).rejects.toMatchObject({ name: "AbortError" }) + expect(refresh).not.toHaveBeenCalled() + expect(create).toHaveBeenCalledTimes(1) + expect(mockFetch).not.toHaveBeenCalled() + }) + + // The abort lands while the fallback fetch is in flight, so the cancellation must come out as + // the shared abort contract - not a telemetry event and not a wrapped connection error. + it("keeps the abort contract and skips telemetry when the fallback fetch is cancelled", async () => { + const handler = createHandler() + // The module mock keeps the spy across tests, so clear it before asserting on this request + const captureException = vitest.mocked(TelemetryService.instance.captureException) + captureException.mockClear() + // The SDK path is unusable, so the request falls back to the SSE transport. + const create = vitest.fn().mockRejectedValue(new Error("sdk down")) + Reflect.set(handler, "client", { responses: { create } }) + const controller = new AbortController() + // Reject the way fetch rejects once the signal it was handed aborts. + const mockFetch = vitest.fn((_url: unknown, init?: { signal?: AbortSignal }) => { + const signal = init?.signal + if (!signal || signal.aborted) { + return Promise.reject(new DOMException("The operation was aborted", "AbortError")) + } + return new Promise((_resolve, reject) => { + signal.addEventListener( + "abort", + () => reject(new DOMException("The operation was aborted", "AbortError")), + { once: true }, + ) + }) + }) + vitest.stubGlobal("fetch", mockFetch) + + const completion = handler.completePrompt("Hello", { abortSignal: controller.signal }) + await vitest.waitFor(() => expect(mockFetch).toHaveBeenCalled()) + controller.abort() + + await expect(completion).rejects.toMatchObject({ name: "AbortError" }) + expect(captureException).not.toHaveBeenCalled() + }) + it("wraps failures from both transports as a completion error", async () => { const handler = createHandler() const create = vitest.fn().mockRejectedValue(new Error("sdk down")) diff --git a/src/api/providers/openai-codex.ts b/src/api/providers/openai-codex.ts index 120311c23b..78279625e4 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -30,6 +30,7 @@ import { sanitizeOpenAiCallId } from "../../utils/tool-id" import { openAiCodexOAuthManager } from "../../integrations/openai-codex/oauth" import { t } from "../../i18n" import { RequestConfigBuilder } from "./config-builder/request-config-builder" +import { createAbortError } from "./utils/abort-signal" export type OpenAiCodexModel = ReturnType @@ -291,6 +292,13 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion yield* this.executeRequest(requestBody, model, accessToken, effectiveSessionId, abortSignal) return } catch (error) { + // The caller's cancellation wins over the retry: force-refreshing a token for a + // request that is already gone would spend a network round trip on a dead request + // and surface the cancellation as an authentication failure. + if (abortSignal?.aborted) { + throw createAbortError(this.providerName) + } + const message = error instanceof Error ? error.message : String(error) const isAuthFailure = /unauthorized|invalid token|not authenticated|authentication|401/i.test(message) @@ -457,16 +465,19 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion effectiveSessionId: string, abortSignal?: AbortSignal, ): ApiStream { - // Create AbortController for cancellation - this.abortController = new AbortController() + // Request-local controller: a stale abort arriving after this request finishes must never + // reach the controller of a later request, so the bridge listener captures this controller + // directly instead of reading `this.abortController` at abort time. + const abortController = new AbortController() + this.abortController = abortController // A caller's signal has to be linked rather than used directly, since both transports below - // abort through `this.abortController`. Without this the signal never reaches the wire. - const abortFromCaller = () => this.abortController?.abort() + // abort through the request controller. Without this the signal never reaches the wire. + const abortFromCaller = () => abortController.abort() if (abortSignal) { if (abortSignal.aborted) { - this.abortController.abort() + abortController.abort() } else { abortSignal.addEventListener("abort", abortFromCaller, { once: true }) } @@ -493,7 +504,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion }) const stream = (await (client as any).responses.create(requestBody, { - signal: this.abortController.signal, + signal: abortController.signal, // If the SDK supports per-request overrides, ensure headers are present. headers: codexHeaders, })) as AsyncIterable @@ -505,7 +516,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } for await (const event of stream) { - if (this.abortController.signal.aborted) { + if (abortController.signal.aborted) { break } @@ -528,16 +539,26 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // A cancellation is not a transport failure either. Falling back would spend a // second request on an already-aborted signal and report the cancellation as a // connection error. - if (this.sawSdkEventInCurrentResponse || this.abortController?.signal.aborted) { + if (this.sawSdkEventInCurrentResponse || abortController.signal.aborted) { throw sdkErr } // Fallback to manual SSE via fetch (Codex backend). - yield* this.makeCodexRequest(requestBody, model, accessToken, effectiveSessionId) + yield* this.makeCodexRequest( + requestBody, + model, + accessToken, + effectiveSessionId, + abortController.signal, + ) } } finally { abortSignal?.removeEventListener("abort", abortFromCaller) - this.abortController = undefined + // Only clear the field if this request still owns it: a concurrent request may have + // installed its own controller after this one started. + if (this.abortController === abortController) { + this.abortController = undefined + } } } @@ -629,6 +650,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion model: OpenAiCodexModel, accessToken: string, effectiveSessionId: string, + abortSignal?: AbortSignal, ): ApiStream { // Per the implementation guide: route to Codex backend with Bearer token const url = `${CODEX_API_BASE_URL}/responses` @@ -648,7 +670,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion method: "POST", headers, body: JSON.stringify(requestBody), - signal: this.abortController?.signal, + signal: abortSignal, }) if (!response.ok) { @@ -708,8 +730,15 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion throw new Error(t("common:errors.openAiCodex.noResponseBody")) } - yield* this.handleStreamResponse(response.body, model) + yield* this.handleStreamResponse(response.body, model, abortSignal) } catch (error) { + // The caller cancelled, so this is not a transport fault: hand back the shared abort + // contract instead of reporting the caller's own cancellation to telemetry or wrapping it + // as a connection failure. + if (abortSignal?.aborted) { + throw createAbortError(this.providerName) + } + const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, model.id, "createMessage") TelemetryService.instance.captureException(apiError) @@ -724,7 +753,11 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } } - private async *handleStreamResponse(body: ReadableStream, model: OpenAiCodexModel): ApiStream { + private async *handleStreamResponse( + body: ReadableStream, + model: OpenAiCodexModel, + abortSignal?: AbortSignal, + ): ApiStream { const reader = body.getReader() const decoder = new TextDecoder() let buffer = "" @@ -732,7 +765,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion try { while (true) { - if (this.abortController?.signal.aborted) { + if (abortSignal?.aborted) { break } @@ -987,6 +1020,13 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } } } catch (error) { + // The caller cancelled while the fallback stream was being read: that is not a + // stream-processing failure, so hand back the shared abort contract instead of reporting + // the caller's own cancellation to telemetry. + if (abortSignal?.aborted) { + throw createAbortError(this.providerName) + } + const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, model.id, "createMessage") TelemetryService.instance.captureException(apiError) @@ -1362,19 +1402,17 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // throwing - so returning here would report a cancelled generation as a finished one and // hand the caller whatever partial text had arrived. if (requestSignal?.aborted) { - throw new DOMException("OpenAI Codex completion was aborted", "AbortError") + throw createAbortError(this.providerName) } return text } catch (error) { // Cancelling is the caller's own doing, not a provider failure, so it is neither - // reported to telemetry nor relabelled as a completion error. A transport that rejects - // on abort reports it in its own words, so it is restated here: callers get one abort - // result whether the stream ended quietly or the request threw. + // reported to telemetry nor relabelled as a completion error: a timed-out or + // cancelled request is normalized to the shared abort contract whether the stream + // ended quietly or the request threw. if (requestSignal?.aborted) { - throw error instanceof DOMException && error.name === "AbortError" - ? error - : new DOMException("OpenAI Codex completion was aborted", "AbortError") + throw createAbortError(this.providerName) } const errorModel = this.getModel() From 1aa5fcd588b984d1e50321680488e2589d9a7c6b Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 15:25:08 +0800 Subject: [PATCH 5/5] fix(api): close the openai-codex abort mutation gaps --- .../openai-codex-native-tool-calls.spec.ts | 8 +- .../providers/__tests__/openai-codex.spec.ts | 259 ++++++++++++++++++ src/api/providers/openai-codex.ts | 10 +- 3 files changed, 270 insertions(+), 7 deletions(-) diff --git a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts index 133b97bbf0..f831d22fc1 100644 --- a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts +++ b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts @@ -551,6 +551,9 @@ describe("OpenAiCodexHandler native tool calls", () => { usage: { input_tokens: 1, output_tokens: 1 }, }, } + // Arrives after the abort resolves the pause: the loop guard must break before + // processing the completed event and this delta. + yield { type: "response.text.delta", delta: "post-abort" } }, } }) @@ -577,8 +580,9 @@ describe("OpenAiCodexHandler native tool calls", () => { controller.abort() const chunks = await collected - // Exactly the pre-abort delta: the completed event only arrives once the abort resolves - // the transport's pause, so the loop must break before processing it. + // Exactly the pre-abort delta: the completed event and the post-abort delta only arrive + // once the abort resolves the transport's pause, so the loop guard must break before + // processing either. expect(chunks).toEqual([{ type: "text", text: "test" }]) expect(mockCreate).toHaveBeenCalled() diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index d88a2b2172..4a9fa7ac68 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -1171,3 +1171,262 @@ describe("OpenAiCodexHandler.completePrompt timeout", () => { expect(mockFetch).not.toHaveBeenCalled() }) }) + +describe("OpenAiCodexHandler.createMessage abort bridging", () => { + // These tests drive createMessage directly (not completePrompt): completePrompt re-normalizes + // any error to the shared abort contract once the caller's signal has fired, which would mask + // regressions in the abort checks inside the transports. + + function createHandler() { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.6-sol" }) + vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + return handler + } + + afterEach(() => { + vitest.restoreAllMocks() + vitest.unstubAllGlobals() + }) + + it("hands back the abort contract instead of force-refreshing after the caller cancels", async () => { + const handler = createHandler() + const refresh = vitest + .spyOn(openAiCodexOAuthManager, "forceRefreshAccessToken") + .mockResolvedValue("refreshed-token") + // The SDK fails with exactly the auth-failure wording the retry path would act on, so the + // abort check must win over the refresh-and-retry logic. + const create = vitest.fn().mockRejectedValue(new Error("401 invalid token")) + Reflect.set(handler, "client", { responses: { create } }) + const mockFetch = vitest.fn() + vitest.stubGlobal("fetch", mockFetch) + + await expect( + collectStream( + handler.createMessage("System", [{ role: "user", content: "Hello" }], { + taskId: "task-test", + abortSignal: AbortSignal.abort(), + }), + ), + ).rejects.toMatchObject({ name: "AbortError" }) + + // No refresh, no second SDK attempt, no SSE fallback. + expect(refresh).not.toHaveBeenCalled() + expect(create).toHaveBeenCalledTimes(1) + expect(mockFetch).not.toHaveBeenCalled() + }) + + it("emits nothing once the caller cancels before the first SDK event", async () => { + const handler = createHandler() + const controller = new AbortController() + const create = vitest.fn().mockImplementation(() => { + // The caller cancels while the SDK is still delivering, so the loop's abort check must + // stop every event from reaching the caller. + controller.abort() + return Promise.resolve( + asyncStreamFrom([ + { type: "response.output_text.delta", delta: "feat: half a" }, + { type: "response.output_text.delta", delta: "and the rest" }, + { type: "response.completed", response: { id: "r1", status: "completed", output: [] } }, + ]), + ) + }) + Reflect.set(handler, "client", { responses: { create } }) + const mockFetch = vitest.fn() + vitest.stubGlobal("fetch", mockFetch) + + const chunks = await collectStream( + handler.createMessage("System", [{ role: "user", content: "Hello" }], { + taskId: "task-test", + abortSignal: controller.signal, + }), + ) + + // The generators end quietly on abort - they break rather than throw - so an empty stream + // is the observable proof that nothing was processed after the cancellation. + expect(chunks).toEqual([]) + expect(mockFetch).not.toHaveBeenCalled() + }) + + it("clears the request controller once the request ends", async () => { + const handler = createHandler() + const create = vitest.fn().mockResolvedValue( + asyncStreamFrom([ + { type: "response.output_text.delta", delta: "done" }, + { type: "response.completed", response: { id: "r1", status: "completed", output: [] } }, + ]), + ) + Reflect.set(handler, "client", { responses: { create } }) + const mockFetch = vitest.fn() + vitest.stubGlobal("fetch", mockFetch) + + await collectStream(handler.createMessage("System", [{ role: "user", content: "Hello" }])) + + // A later request must be able to tell that no request is in flight. + expect(Reflect.get(handler, "abortController")).toBeUndefined() + }) + + it("keeps the later request's controller when the earlier request finishes", async () => { + const handler = createHandler() + const mockFetch = vitest.fn() + vitest.stubGlobal("fetch", mockFetch) + + // The earlier request holds its stream open until the later request has installed its own + // controller, so the earlier cleanup runs while a request is still in flight. + let releaseEarlier: (() => void) | undefined + const releaseGate = new Promise((resolve) => { + releaseEarlier = resolve + }) + const earlierStream = (async function* () { + yield { type: "response.output_text.delta", delta: "a" } + await releaseGate + yield { type: "response.completed", response: { id: "ra", status: "completed", output: [] } } + })() + const create = vitest + .fn() + .mockImplementationOnce(() => Promise.resolve(earlierStream)) + .mockImplementationOnce(() => + Promise.resolve( + asyncStreamFrom([ + { type: "response.output_text.delta", delta: "b" }, + { type: "response.completed", response: { id: "rb", status: "completed", output: [] } }, + ]), + ), + ) + Reflect.set(handler, "client", { responses: { create } }) + + const earlier = handler.createMessage("System", [{ role: "user", content: "Hello" }]) + expect(await earlier.next()).toMatchObject({ value: { type: "text", text: "a" } }) + + const later = handler.createMessage("System", [{ role: "user", content: "Hello" }]) + expect(await later.next()).toMatchObject({ value: { type: "text", text: "b" } }) + + // The earlier request finishes while the later one is still in flight. + releaseEarlier!() + await earlier.next() + + // The earlier request's cleanup must not clear the controller the later request installed. + const controller = Reflect.get(handler, "abortController") as AbortController | undefined + expect(controller).toBeDefined() + expect(controller?.signal).toBe(create.mock.calls[1][1].signal) + + // Let the later request finish and clear its own controller. + await later.next() + }) + + it("stops reading the fallback stream once the request aborts", async () => { + const handler = createHandler() + const create = vitest.fn().mockRejectedValue(new Error("sdk down")) + Reflect.set(handler, "client", { responses: { create } }) + const controller = new AbortController() + const encoder = new TextEncoder() + const body = new ReadableStream({ + start(streamController) { + streamController.enqueue( + encoder.encode('data: {"type":"response.output_text.delta","delta":"one"}\n\n'), + ) + streamController.enqueue( + encoder.encode('data: {"type":"response.output_text.delta","delta":"two"}\n\n'), + ) + streamController.close() + }, + }) + const mockFetch = vitest.fn().mockResolvedValue({ ok: true, body }) + vitest.stubGlobal("fetch", mockFetch) + + const iter = handler.createMessage("System", [{ role: "user", content: "Hello" }], { + taskId: "task-test", + abortSignal: controller.signal, + }) + expect(await iter.next()).toMatchObject({ value: { type: "text", text: "one" } }) + + // The cancellation lands between two stream reads, exactly where the loop's check runs. + controller.abort() + + const chunks = await collectStream(iter) + // Everything enqueued after the cancellation must stay unread. + expect(chunks).toEqual([]) + }) + + it("hands back the abort contract when the fallback stream tears down while the caller cancels", async () => { + const handler = createHandler() + const captureException = vitest.mocked(TelemetryService.instance.captureException) + captureException.mockClear() + const create = vitest.fn().mockRejectedValue(new Error("sdk down")) + Reflect.set(handler, "client", { responses: { create } }) + const controller = new AbortController() + const encoder = new TextEncoder() + let pullStartedResolve: (() => void) | undefined + let failRead: (() => void) | undefined + const pullStarted = new Promise((resolve) => { + pullStartedResolve = resolve + }) + const failGate = new Promise((resolve) => { + failRead = resolve + }) + const body = new ReadableStream({ + start(streamController) { + streamController.enqueue( + encoder.encode('data: {"type":"response.output_text.delta","delta":"one"}\n\n'), + ) + }, + pull(streamController) { + // The second read is pending; tear the stream down on the test's signal. + pullStartedResolve!() + return failGate.then(() => { + streamController.error(new Error("stream torn down")) + }) + }, + }) + const mockFetch = vitest.fn().mockResolvedValue({ ok: true, body }) + vitest.stubGlobal("fetch", mockFetch) + + const iter = handler.createMessage("System", [{ role: "user", content: "Hello" }], { + taskId: "task-test", + abortSignal: controller.signal, + }) + expect(await iter.next()).toMatchObject({ value: { type: "text", text: "one" } }) + + const pending = collectStream(iter) + await pullStarted + // The cancellation lands while the second read is in flight, i.e. after the loop's check. + controller.abort() + failRead!() + + await expect(pending).rejects.toMatchObject({ name: "AbortError" }) + // Cancellation is the caller's own doing, so neither catch may report it to telemetry. + expect(captureException).not.toHaveBeenCalled() + }) + + it("wraps a torn-down fallback stream as a stream error when nothing was aborted", async () => { + const handler = createHandler() + const captureException = vitest.mocked(TelemetryService.instance.captureException) + captureException.mockClear() + const create = vitest.fn().mockRejectedValue(new Error("sdk down")) + Reflect.set(handler, "client", { responses: { create } }) + const encoder = new TextEncoder() + const body = new ReadableStream({ + start(streamController) { + streamController.enqueue( + encoder.encode('data: {"type":"response.output_text.delta","delta":"one"}\n\n'), + ) + }, + pull(streamController) { + streamController.error(new Error("stream torn down")) + }, + }) + const mockFetch = vitest.fn().mockResolvedValue({ ok: true, body }) + vitest.stubGlobal("fetch", mockFetch) + + const iter = handler.createMessage("System", [{ role: "user", content: "Hello" }]) + expect(await iter.next()).toMatchObject({ value: { type: "text", text: "one" } }) + + // The wrap chain surfaces the connection-failure key in every case, so the message alone + // cannot prove the innermost check classified this as a stream failure: an always-true + // check would swap in the shared abort contract, and the request-level catch would wrap + // that in the same key. The telemetry count is the witness: the stream-processing catch + // and the request catch both report the failure (twice); the abort path skips the first. + await expect(collectStream(iter)).rejects.toThrow(/connectionFailed|stream torn down/) + expect(captureException).toHaveBeenCalledTimes(2) + }) +}) diff --git a/src/api/providers/openai-codex.ts b/src/api/providers/openai-codex.ts index 78279625e4..1e9885371d 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -650,7 +650,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion model: OpenAiCodexModel, accessToken: string, effectiveSessionId: string, - abortSignal?: AbortSignal, + abortSignal: AbortSignal, ): ApiStream { // Per the implementation guide: route to Codex backend with Bearer token const url = `${CODEX_API_BASE_URL}/responses` @@ -735,7 +735,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // The caller cancelled, so this is not a transport fault: hand back the shared abort // contract instead of reporting the caller's own cancellation to telemetry or wrapping it // as a connection failure. - if (abortSignal?.aborted) { + if (abortSignal.aborted) { throw createAbortError(this.providerName) } @@ -756,7 +756,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion private async *handleStreamResponse( body: ReadableStream, model: OpenAiCodexModel, - abortSignal?: AbortSignal, + abortSignal: AbortSignal, ): ApiStream { const reader = body.getReader() const decoder = new TextDecoder() @@ -765,7 +765,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion try { while (true) { - if (abortSignal?.aborted) { + if (abortSignal.aborted) { break } @@ -1023,7 +1023,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // The caller cancelled while the fallback stream was being read: that is not a // stream-processing failure, so hand back the shared abort contract instead of reporting // the caller's own cancellation to telemetry. - if (abortSignal?.aborted) { + if (abortSignal.aborted) { throw createAbortError(this.providerName) }