From b457024d67000a4d00bbae8fc8f42faab5698947 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 19 Aug 2026 23:00:44 +0800 Subject: [PATCH 1/8] 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/8] 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/8] 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/8] 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/8] 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) } From 0183b943abf523c2ed1c3dff7b9fcea1ac215cd1 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Tue, 15 Sep 2026 23:09:05 +0800 Subject: [PATCH 6/8] fix(api): import mergeAbortSignalAndTimeout directly; drop builder static wrappers Address edelauna's review nits on the abort-signal API surface: openai-codex's completePrompt now imports mergeAbortSignalAndTimeout straight from utils/abort-signal (same pattern as Bedrock) instead of going through RequestConfigBuilder, and the two static pass-through wrappers (mergeAbortSignalAndTimeout, mergeAbortSignals) are dropped - they had no production callers once the direct import landed, and their behavior is covered by the abort-signal util's own spec. The request-config-builder files revert to main, shrinking the PR diff from 5 files to 3. --- .../__tests__/request-config-builder.spec.ts | 40 ------------------- .../config-builder/request-config-builder.ts | 15 ------- src/api/providers/openai-codex.ts | 5 +-- 3 files changed, 2 insertions(+), 58 deletions(-) diff --git a/src/api/providers/__tests__/request-config-builder.spec.ts b/src/api/providers/__tests__/request-config-builder.spec.ts index f9557a123f..977b09df6b 100644 --- a/src/api/providers/__tests__/request-config-builder.spec.ts +++ b/src/api/providers/__tests__/request-config-builder.spec.ts @@ -505,44 +505,4 @@ 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 a3c1ba7ebb..2201d735bc 100644 --- a/src/api/providers/config-builder/request-config-builder.ts +++ b/src/api/providers/config-builder/request-config-builder.ts @@ -163,19 +163,4 @@ export class RequestConfigBuilder @@ -1372,7 +1371,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { // Merge an optional timeout into the caller's abort signal so a timeout cancels the // completion the same way an external abort does (timeoutMs <= 0 disables it). - const requestSignal = RequestConfigBuilder.mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) + const requestSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) try { const model = this.getModel() From 73b2b237476064810fbddb783f45854801aee973 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 16 Sep 2026 00:02:52 +0800 Subject: [PATCH 7/8] fix(api): fast-fail and guard the openai-codex completePrompt stream on abort Close the abort-signal contract gaps the series alignment check flags in completePrompt (in this unit's delta): (1) throwIfAborted fast-fail before the first await, matching the sibling units - a pre-aborted request no longer spends the OAuth token/account setup or an SDK request on a dead completion (the regression test defers the OAuth resolution to prove the fast-fail does not wait on it); (2) top-of-loop abort break on the streaming consumer loop so a buffered post-abort chunk is never joined into the completion; (3) the catch normalizes via isRequestAborted(error, requestSignal) like the sibling units, so an SDK abort error is normalized to the shared abort contract even when the signal has not marked itself aborted. --- .../providers/__tests__/openai-codex.spec.ts | 24 ++++++++++++++----- src/api/providers/openai-codex.ts | 22 +++++++++++++---- 2 files changed, 36 insertions(+), 10 deletions(-) diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index 4a9fa7ac68..5f089692f5 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -492,17 +492,29 @@ describe("OpenAiCodexHandler.completePrompt streaming", () => { expect(signalDuringRequest!.aborted).toBe(true) }) - it("rejects when the caller's signal is already aborted", async () => { - const handler = createHandler() - const create = injectStream(handler, [ - { type: "response.completed", response: { id: "r1", status: "completed", output: [] } }, - ]) + // A request cancelled before it starts must not spend the provider setup on it: the + // fast-fail rejects before the (here deferred, never resolving) token and account fetch + // and before the SDK request, so a pending OAuth flow cannot hold a cancelled completion + // hostage. + it("fast-fails before the OAuth setup when the caller's signal is already aborted", async () => { + const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.6-sol" }) + const getAccessToken = vitest + .spyOn(openAiCodexOAuthManager, "getAccessToken") + .mockReturnValue(new Promise(() => {})) + const getAccountId = vitest + .spyOn(openAiCodexOAuthManager, "getAccountId") + .mockReturnValue(new Promise(() => {})) + const create = vitest.fn() + Reflect.set(handler, "client", { responses: { create } }) await expect(handler.completePrompt("Hello", { abortSignal: AbortSignal.abort() })).rejects.toMatchObject({ name: "AbortError", + message: "This operation was aborted", }) - expect(create.mock.calls[0][1].signal.aborted).toBe(true) + expect(create).not.toHaveBeenCalled() + expect(getAccessToken).not.toHaveBeenCalled() + expect(getAccountId).not.toHaveBeenCalled() }) // The SSE fallback is for an SDK that could not be used at all. Replaying the request after the diff --git a/src/api/providers/openai-codex.ts b/src/api/providers/openai-codex.ts index 7c8a68cdfd..ddf0393b71 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -29,7 +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 { createAbortError, mergeAbortSignalAndTimeout } from "./utils/abort-signal" +import { createAbortError, isRequestAborted, mergeAbortSignalAndTimeout, throwIfAborted } from "./utils/abort-signal" export type OpenAiCodexModel = ReturnType @@ -1369,6 +1369,11 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion * from having to be duplicated here. */ async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { + // Fast-fail if the caller's stop signal already fired before we started: a cancelled + // request must not spend the OAuth setup (token and account fetch) or an SDK request on + // a completion that is already gone. + throwIfAborted(options?.abortSignal) + // Merge an optional timeout into the caller's abort signal so a timeout cancels the // completion the same way an external abort does (timeoutMs <= 0 disables it). const requestSignal = mergeAbortSignalAndTimeout(options?.abortSignal, options?.timeoutMs) @@ -1380,7 +1385,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // the prompt enhancer writes this straight into the input box. let text = "" - for await (const chunk of this.handleResponsesApiMessage( + const stream = this.handleResponsesApiMessage( model, "", [{ role: "user", content: prompt }], @@ -1388,7 +1393,16 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // directly, so `prompt_cache_key` is unchanged. { taskId: this.sessionId }, requestSignal, - )) { + ) + + for await (const chunk of stream) { + // A buffered chunk can still be pulled in the window between the abort and the + // inner generator's own stop, so break here: post-abort output must never be + // joined into the completion. + if (requestSignal?.aborted) { + break + } + // Refusals are streamed as text for the chat, but they are not output: the // non-streaming request this replaced read `output_text`, which never carries them. // Keeping them would paste "[Refusal] ..." into the input box as if it were an answer. @@ -1410,7 +1424,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // 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) { + if (isRequestAborted(error, requestSignal)) { throw createAbortError(this.providerName) } From 641400265dddf961b9d16231547bf5a194901f95 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 16 Sep 2026 08:34:24 +0800 Subject: [PATCH 8/8] test(api): structural kill test for the completePrompt top-of-loop abort guard The guard at the top of completePrompt's streaming consumer loop is the only thing that stops the loop from pulling the SDK stream once more after the abort rides in on an in-flight chunk; the post-loop check throws the same AbortError either way, so the kill test asserts the pull count (event2's delta getter fires the abort after executeRequest's own check has passed, before completePrompt's) - the shape the class-h rule requires for consumer-loop guards. --- .../providers/__tests__/openai-codex.spec.ts | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index 5f089692f5..63c72403ae 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -517,6 +517,51 @@ describe("OpenAiCodexHandler.completePrompt streaming", () => { expect(getAccountId).not.toHaveBeenCalled() }) + // The abort lands in the only window the consumer guard can still act: event2's `delta` + // getter fires it while processEvent builds the chunk - after executeRequest's + // top-of-loop check has passed, before completePrompt's own. The post-loop check throws + // the same AbortError whether or not the guard breaks, so what distinguishes the guarded + // loop from a mutated one is the pull count: the guard stops pulling after the chunk that + // the abort rode in on, a loop that keeps running pulls the SDK stream a third time. + it("breaks the streaming consumer loop at the top-of-loop guard and stops pulling", async () => { + const handler = createHandler() + const controller = new AbortController() + let sdkPulls = 0 + + const event1 = { type: "response.output_text.delta", delta: "pre-abort" } + const event2 = { + type: "response.output_text.delta", + get delta() { + controller.abort() + return "post-abort" + }, + } + + const create = vitest.fn().mockImplementation(() => { + return Promise.resolve({ + [Symbol.asyncIterator]() { + return { + next: async () => { + sdkPulls++ + return { value: sdkPulls === 1 ? event1 : event2, done: false } + }, + return: async () => ({ value: undefined, done: true }), + } + }, + }) + }) + Reflect.set(handler, "client", { responses: { create } }) + + await expect(handler.completePrompt("Hello", { abortSignal: controller.signal })).rejects.toMatchObject({ + name: "AbortError", + }) + + // event1 is pulled and joined; event2 is pulled (its getter fires the abort) but the + // guard breaks before it is joined - a third pull only happens when the mutated + // top-of-loop check keeps the loop running. + expect(sdkPulls).toBe(2) + }) + // The SSE fallback is for an SDK that could not be used at all. Replaying the request after the // SDK has already produced output would append a second generation to the first. it("does not replay over SSE when the SDK fails after emitting", async () => {