From f5d990c54f883d9467d9e300e7d842d2aa5608b6 Mon Sep 17 00:00:00 2001 From: Rafael-Silva-Oliveira Date: Fri, 14 Aug 2026 11:46:14 +0200 Subject: [PATCH 1/4] fix(openai-codex): complete prompts over the streaming transport The Codex subscription endpoint only accepts streaming requests, so `completePrompt` sending `stream: false` was rejected outright with HTTP 400 `Stream must be set to true`. That made commit-message generation, prompt enhancement and condensing unusable on Codex. Rather than issue its own request, `completePrompt` now runs the existing streaming path and joins the text chunks. That inherits the OAuth refresh-and-retry, the SDK-then-SSE fallback and the Luna body instead of duplicating a second, subtly different request builder. Reasoning chunks are deliberately dropped: a commit message is written straight into the Source Control box. A caller's abort signal also never reached the wire, since both transports abort through the handler's own controller. It is now linked to that controller, so stopping a generation actually cancels it. The spec asserted `stream: false`, pinning the bug in place; it now asserts the opposite and covers chunk joining, reasoning exclusion, auth retry and signal propagation. Co-Authored-By: Claude Opus 5 --- .../openai-codex-native-tool-calls.spec.ts | 19 +- .../providers/__tests__/openai-codex.spec.ts | 191 ++++++++++++++++-- src/api/providers/openai-codex.ts | 131 ++++-------- src/eslint-suppressions.json | 2 +- 4 files changed, 230 insertions(+), 113 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..a985fc35d3 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 @@ -475,15 +475,20 @@ describe("OpenAiCodexHandler native tool calls", () => { vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + // Completions stream like everything else, so the SDK path is forced to fail and the + // hand-built SSE request is what these assertions inspect. + Reflect.set(handler, "client", { + responses: { create: vi.fn().mockRejectedValue(new Error("SDK unavailable")) }, + }) const mockFetch = vi.fn().mockResolvedValue({ ok: true, - json: vi.fn().mockResolvedValue({ - output: [ - { - type: "message", - content: [{ type: "output_text", text: "done" }], - }, - ], + body: new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode('data: {"type":"response.output_text.delta","delta":"done"}\n\n'), + ) + controller.close() + }, }), }) global.fetch = mockFetch as any diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index 9a256535c1..1df16f671f 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -247,23 +247,26 @@ describe("OpenAiCodexHandler.completePrompt service tier", () => { it.each<[string, OpenAiCodexServiceTier | undefined, typeof OpenAiCodexServiceTier.Priority | undefined]>([ ["Fast", OpenAiCodexServiceTier.Priority, OpenAiCodexServiceTier.Priority], ["Standard", undefined, undefined], - ])("uses the %s preference in non-streaming requests", async (_mode, configuredTier, expectedTier) => { + ])("uses the %s preference in completion requests", async (_mode, configuredTier, expectedTier) => { const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.6-sol", ...(configuredTier ? { [OPEN_AI_CODEX_SERVICE_TIER_KEY]: configuredTier } : {}), }) vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") - const mockFetch = vitest.fn().mockResolvedValue({ - ok: true, - json: vitest.fn().mockResolvedValue({ text: "Complete" }), - }) - vitest.stubGlobal("fetch", mockFetch) + const create = vitest.fn().mockResolvedValue( + asyncStreamFrom([ + { type: "response.output_text.delta", delta: "Complete" }, + { type: "response.completed", response: { id: "r1", status: "completed", output: [] } }, + ]), + ) + Reflect.set(handler, "client", { responses: { create } }) await expect(handler.completePrompt("Hello")).resolves.toBe("Complete") - const body = JSON.parse(mockFetch.mock.calls[0][1].body) - expect(body.stream).toBe(false) + const body = create.mock.calls[0][0] + // The Codex subscription endpoint rejects `stream: false` outright. + expect(body.stream).toBe(true) if (expectedTier) { expect(body[SERVICE_TIER_KEY]).toBe(expectedTier) } else { @@ -272,6 +275,157 @@ describe("OpenAiCodexHandler.completePrompt service tier", () => { }) }) +describe("OpenAiCodexHandler.completePrompt streaming", () => { + 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 + } + + function injectStream(handler: OpenAiCodexHandler, events: unknown[]) { + const create = vitest.fn().mockResolvedValue(asyncStreamFrom(events)) + Reflect.set(handler, "client", { responses: { create } }) + return create + } + + afterEach(() => { + vitest.restoreAllMocks() + vitest.unstubAllGlobals() + }) + + it("joins consecutive text deltas into one string", async () => { + const handler = createHandler() + injectStream(handler, [ + { type: "response.output_text.delta", delta: "feat: " }, + { type: "response.output_text.delta", delta: "add commit messages" }, + { type: "response.completed", response: { id: "r1", status: "completed", output: [] } }, + ]) + + await expect(handler.completePrompt("Hello")).resolves.toBe("feat: add commit messages") + }) + + // A commit message is written straight into the Source Control box, so reasoning must never + // become part of it. + it("omits reasoning from the completion", async () => { + const handler = createHandler() + injectStream(handler, [ + { type: "response.reasoning_summary_text.delta", delta: "Thinking about the diff" }, + { type: "response.output_text.delta", delta: "fix: correct the parser" }, + { type: "response.completed", response: { id: "r1", status: "completed", output: [] } }, + ]) + + const result = await handler.completePrompt("Hello") + + expect(result).toBe("fix: correct the parser") + expect(result).not.toContain("Thinking") + }) + + it("omits usage and tool calls from the completion", async () => { + const handler = createHandler() + injectStream(handler, [ + { type: "response.output_text.delta", delta: "chore: tidy" }, + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_1", name: "read_file", arguments: "{}" }, + }, + { + type: "response.completed", + response: { + id: "r1", + status: "completed", + output: [], + usage: { input_tokens: 10, output_tokens: 5 }, + }, + }, + ]) + + await expect(handler.completePrompt("Hello")).resolves.toBe("chore: tidy") + }) + + // The SDK path swallows its own errors into the SSE fallback, so an auth failure only reaches + // the retry loop from the fallback - the same shape the streaming Luna retry test relies on. + it("retries once with a refreshed token when the first attempt is unauthorized", async () => { + const handler = createHandler() + const refresh = vitest + .spyOn(openAiCodexOAuthManager, "forceRefreshAccessToken") + .mockResolvedValue("fresh-token") + Reflect.set(handler, "client", { + responses: { create: vitest.fn().mockRejectedValue(new Error("SDK unavailable")) }, + }) + const mockFetch = vitest + .fn() + .mockResolvedValueOnce({ + ok: false, + status: 401, + text: vitest.fn().mockResolvedValue('{"error":{"message":"Codex API invalid token"}}'), + }) + .mockResolvedValueOnce({ + ok: true, + body: new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + 'data: {"type":"response.output_text.delta","delta":"docs: update"}\n\n', + ), + ) + controller.close() + }, + }), + }) + vitest.stubGlobal("fetch", mockFetch) + + await expect(handler.completePrompt("Hello")).resolves.toBe("docs: update") + expect(refresh).toHaveBeenCalled() + expect(mockFetch).toHaveBeenCalledTimes(2) + }) + + // The caller's signal is linked to the internal controller rather than passed through, so what + // matters is that aborting the caller's one aborts the signal the request is actually using. + it("passes the caller's abort signal down to the request", async () => { + const handler = createHandler() + const controller = new AbortController() + let signalDuringRequest: AbortSignal | undefined + + const create = vitest.fn().mockImplementation((_body: unknown, options: { signal: AbortSignal }) => { + signalDuringRequest = options.signal + // Abort mid-flight, while `executeRequest`'s listener is still attached. + controller.abort() + return Promise.resolve( + asyncStreamFrom([ + { type: "response.completed", response: { id: "r1", status: "completed", output: [] } }, + ]), + ) + }) + Reflect.set(handler, "client", { responses: { create } }) + + await handler.completePrompt("Hello", { abortSignal: controller.signal }) + + expect(signalDuringRequest).toBeInstanceOf(AbortSignal) + expect(signalDuringRequest!.aborted).toBe(true) + }) + + it("aborts immediately 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: [] } }, + ]) + + await handler.completePrompt("Hello", { abortSignal: AbortSignal.abort() }) + + expect(create.mock.calls[0][1].signal.aborted).toBe(true) + }) + + it("wraps failures from both transports as a completion error", async () => { + const handler = createHandler() + const create = vitest.fn().mockRejectedValue(new Error("sdk down")) + Reflect.set(handler, "client", { responses: { create } }) + vitest.stubGlobal("fetch", vitest.fn().mockRejectedValue(new Error("network down"))) + + await expect(handler.completePrompt("Hello")).rejects.toThrow(/completionError|network down/) + }) +}) + describe("transformLunaResponsesLiteBody", () => { it("creates the exact Responses Lite body while preserving unrelated fields and reasoning", () => { const tools = [{ type: "function", name: "read_file", parameters: { type: "object" } }] @@ -601,15 +755,20 @@ describe("OpenAiCodexHandler Luna Responses Lite requests", () => { const handler = new OpenAiCodexHandler({ apiModelId: "gpt-5.6-luna", reasoningEffort: "disable" }) vitest.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") vitest.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + // Forcing the SDK path to fail exercises the SSE fallback, which is where the request body + // and the Codex-specific headers are assembled by hand. + Reflect.set(handler, "client", { + responses: { create: vitest.fn().mockRejectedValue(new Error("SDK unavailable")) }, + }) const mockFetch = vitest.fn().mockResolvedValue({ ok: true, - json: vitest.fn().mockResolvedValue({ - output: [ - { - type: "message", - content: [{ type: "output_text", text: "Complete" }], - }, - ], + body: new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode('data: {"type":"response.output_text.delta","delta":"Complete"}\n\n'), + ) + controller.close() + }, }), }) vitest.stubGlobal("fetch", mockFetch) @@ -622,7 +781,7 @@ describe("OpenAiCodexHandler Luna Responses Lite requests", () => { expect(sessionId).toMatch(/^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i) expect(body).toMatchObject({ model: "gpt-5.6-luna", - stream: false, + stream: true, tool_choice: "auto", parallel_tool_calls: false, reasoning: { context: "all_turns" }, diff --git a/src/api/providers/openai-codex.ts b/src/api/providers/openai-codex.ts index e9bc3bbf5d..83570debc1 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -217,7 +217,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion metadata?: ApiHandlerCreateMessageMetadata, ): ApiStream { const model = this.getModel() - yield* this.handleResponsesApiMessage(model, systemPrompt, messages, metadata) + yield* this.handleResponsesApiMessage(model, systemPrompt, messages, metadata, metadata?.abortSignal) } private async *handleResponsesApiMessage( @@ -225,6 +225,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion systemPrompt: string, messages: Anthropic.Messages.MessageParam[], metadata?: ApiHandlerCreateMessageMetadata, + abortSignal?: AbortSignal, ): ApiStream { // Reset state for this request this.lastResponseOutput = undefined @@ -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, abortSignal) return } catch (error) { const message = error instanceof Error ? error.message : String(error) @@ -438,10 +439,23 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion model: OpenAiCodexModel, accessToken: string, effectiveSessionId: string, + abortSignal?: AbortSignal, ): ApiStream { // Create AbortController for cancellation this.abortController = new 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() + + if (abortSignal) { + if (abortSignal.aborted) { + this.abortController.abort() + } else { + abortSignal.addEventListener("abort", abortFromCaller, { once: true }) + } + } + try { // Prefer OpenAI SDK streaming (same approach as openai-native) so event handling // is consistent across providers. @@ -491,6 +505,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion yield* this.makeCodexRequest(requestBody, model, accessToken, effectiveSessionId) } } finally { + abortSignal?.removeEventListener("abort", abortFromCaller) this.abortController = undefined } } @@ -1256,98 +1271,38 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion return this.lastResponseId } + /** + * The Codex subscription endpoint only accepts streaming requests - a body with `stream: false` + * is rejected with `Stream must be set to true` - so a one-shot completion is the streaming + * request with its text chunks joined back together. + * + * Going through `handleResponsesApiMessage` rather than issuing its own request is what keeps + * the OAuth refresh-and-retry, the SDK-then-SSE fallback, the Luna body and the service tier + * from having to be duplicated here. + */ async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { - this.abortController = new AbortController() - try { const model = this.getModel() - // Get access token - const accessToken = await openAiCodexOAuthManager.getAccessToken() - if (!accessToken) { - throw new Error( - t("common:errors.openAiCodex.notAuthenticated", { - defaultValue: - "Not authenticated with OpenAI Codex. Please sign in using the OpenAI Codex OAuth flow.", - }), - ) - } - - const reasoningEffort = this.getReasoningEffort(model) - const serviceTier = getOpenAiCodexServiceTier(this.options) - - const baseRequestBody: any = { - model: model.id, - input: [ - { - role: "user", - content: [{ type: "input_text", text: prompt }], - }, - ], - stream: false, - store: false, - ...(serviceTier ? { [SERVICE_TIER_KEY]: serviceTier } : {}), - ...(reasoningEffort ? { include: ["reasoning.encrypted_content"] } : {}), - } - - if (reasoningEffort) { - baseRequestBody.reasoning = { - effort: reasoningEffort, - summary: "auto" as const, + // Only the answer is wanted. Reasoning is deliberately dropped rather than concatenated: + // callers such as commit-message generation write this straight into the editor. + let text = "" + + for await (const chunk of this.handleResponsesApiMessage( + model, + "", + [{ role: "user", content: prompt }], + // `taskId` is required, and resolves to the same session id this used to send + // directly, so `prompt_cache_key` is unchanged. + { taskId: this.sessionId }, + options?.abortSignal, + )) { + if (chunk.type === "text") { + text += chunk.text } } - const requestBody = - model.id === LUNA_MODEL_ID - ? this.buildLunaRequestBody(baseRequestBody, this.sessionId) - : baseRequestBody - - const url = `${CODEX_API_BASE_URL}/responses` - - // Get ChatGPT account ID for organization subscriptions - const accountId = await openAiCodexOAuthManager.getAccountId() - - // Build headers with required Codex-specific fields - const headers: Record = { - ...this.buildCodexHeaders(model, this.sessionId, accountId), - "Content-Type": "application/json", - Authorization: `Bearer ${accessToken}`, - } - - const response = await fetch(url, { - method: "POST", - headers, - body: JSON.stringify(requestBody), - signal: this.abortController.signal, - }) - - if (!response.ok) { - const errorText = await response.text() - throw new Error( - t("common:errors.openAiCodex.genericError", { status: response.status }) + - (errorText ? `: ${errorText}` : ""), - ) - } - - const responseData = await response.json() - - 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 - } - } - } - } - } - - if (responseData?.text) { - return responseData.text - } - - return "" + return text } catch (error) { const errorModel = this.getModel() const errorMessage = error instanceof Error ? error.message : String(error) @@ -1358,8 +1313,6 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion throw new Error(t("common:errors.openAiCodex.completionError", { message: error.message })) } throw error - } finally { - this.abortController = undefined } } } diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index 569c846c29..0fcf33e149 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -396,7 +396,7 @@ }, "api/providers/openai-codex.ts": { "@typescript-eslint/no-explicit-any": { - "count": 35 + "count": 34 } }, "api/providers/openai-native.ts": { From 979269b01e3213d0d8c3149aececa2b4ed21e212 Mon Sep 17 00:00:00 2001 From: Rafael-Silva-Oliveira Date: Sat, 15 Aug 2026 13:18:14 +0200 Subject: [PATCH 2/4] fix(openai-codex): stop replaying the stream and treat an abort as one Two problems in the streaming completePrompt path. The SDK stream and its consumption loop sat inside the same try, so an error raised part way through the stream was handled as "the SDK could not be used at all" and the request was replayed over SSE. Whatever the SDK had already yielded stayed with the caller, so the replay appended a second generation to the first. The fallback now closes after the first SDK event, which is the point where the request has been accepted and its output is already out. It is set before processEvent runs, since that mutates response state too, so a throw from it must not replay either. This covers the chat path as well, since both go through executeRequest. Both transports also end quietly on abort, breaking out of their loops rather than throwing, so completePrompt returned whatever partial text had arrived and callers read a cancelled generation as a finished one. It now rejects with an AbortError instead, and a cancellation is passed through rather than reported to telemetry or relabelled as a completion error, since stopping is the caller's own doing. The two abort specs asserted that a cancelled call resolves, which held the bug in place. They now assert the rejection and cover the pre aborted and mid stream cases, and a new spec covers the mid stream SDK failure and checks that SSE is never reached. Co-Authored-By: Claude Opus 5 --- .../providers/__tests__/openai-codex.spec.ts | 33 ++++++++++++++++--- src/api/providers/openai-codex.ts | 28 +++++++++++++++- 2 files changed, 56 insertions(+), 5 deletions(-) diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index 1df16f671f..b4be574ed3 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -382,7 +382,9 @@ describe("OpenAiCodexHandler.completePrompt streaming", () => { // The caller's signal is linked to the internal controller rather than passed through, so what // matters is that aborting the caller's one aborts the signal the request is actually using. - it("passes the caller's abort signal down to the request", async () => { + // The stream then ends quietly, so rejecting is the only thing that tells the caller apart a + // cancelled generation from a finished one. + it("rejects and stops the request when cancelled mid-stream", async () => { const handler = createHandler() const controller = new AbortController() let signalDuringRequest: AbortSignal | undefined @@ -393,29 +395,52 @@ describe("OpenAiCodexHandler.completePrompt streaming", () => { controller.abort() return Promise.resolve( asyncStreamFrom([ + { type: "response.output_text.delta", delta: "feat: half a" }, { type: "response.completed", response: { id: "r1", status: "completed", output: [] } }, ]), ) }) Reflect.set(handler, "client", { responses: { create } }) - await handler.completePrompt("Hello", { abortSignal: controller.signal }) + await expect(handler.completePrompt("Hello", { abortSignal: controller.signal })).rejects.toMatchObject({ + name: "AbortError", + }) expect(signalDuringRequest).toBeInstanceOf(AbortSignal) expect(signalDuringRequest!.aborted).toBe(true) }) - it("aborts immediately when the caller's signal is already aborted", async () => { + 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: [] } }, ]) - await handler.completePrompt("Hello", { abortSignal: AbortSignal.abort() }) + await expect(handler.completePrompt("Hello", { abortSignal: AbortSignal.abort() })).rejects.toMatchObject({ + name: "AbortError", + }) expect(create.mock.calls[0][1].signal.aborted).toBe(true) }) + // 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 () => { + const handler = createHandler() + const create = vitest.fn().mockResolvedValue( + (async function* () { + yield { type: "response.output_text.delta", delta: "feat: add" } + throw new Error("stream broke") + })(), + ) + Reflect.set(handler, "client", { responses: { create } }) + const mockFetch = vitest.fn() + vitest.stubGlobal("fetch", mockFetch) + + await expect(handler.completePrompt("Hello")).rejects.toThrow(/completionError|stream broke/) + expect(mockFetch).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 83570debc1..de33b20f28 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -456,6 +456,11 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } } + // Once the SDK stream has produced an event the request has been accepted and its output is + // already with the caller, so replaying it over SSE would append a second generation to the + // first. The fallback below is only for an SDK that could not be used at all. + let sawSdkEvent = false + try { // Prefer OpenAI SDK streaming (same approach as openai-native) so event handling // is consistent across providers. @@ -493,6 +498,10 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion break } + // Set before the event is processed, not after: `processEvent` also mutates + // response state, so a throw from it must not replay either. + sawSdkEvent = true + for await (const outChunk of this.processEvent(event, model)) { if (outChunk.type === "text") { this.sawTextOutputInCurrentResponse = true @@ -500,7 +509,11 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion yield outChunk } } - } catch (_sdkErr) { + } catch (sdkErr) { + if (sawSdkEvent) { + throw sdkErr + } + // Fallback to manual SSE via fetch (Codex backend). yield* this.makeCodexRequest(requestBody, model, accessToken, effectiveSessionId) } @@ -1302,8 +1315,21 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } } + // Both transports end quietly on abort - they break out of their loops rather than + // throwing - so returning here would report a cancelled generation as a finished one and + // hand the caller whatever partial text had arrived. + if (options?.abortSignal?.aborted) { + throw new DOMException("OpenAI Codex completion was aborted", "AbortError") + } + 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. + if (options?.abortSignal?.aborted) { + throw error + } + const errorModel = this.getModel() const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, errorModel.id, "completePrompt") From c68dae15f08d28624f308c3bb412a733550a82ba Mon Sep 17 00:00:00 2001 From: Rafael-Silva-Oliveira Date: Sat, 15 Aug 2026 14:04:33 +0200 Subject: [PATCH 3/4] fix(openai-codex): close the last two replay routes into a live request Follow-ups to the previous commit, both raised by CodeRabbit. Closing the SSE fallback after the first SDK event left the OAuth retry loop as a second way back into a request the service had already accepted. A mid stream error that reads as an auth failure would refresh the token and send the whole thing again, so completePrompt could concatenate two generations and the chat path could repeat streamed effects. The flag moved onto the handler, reset per request alongside the other response state, and the retry is now skipped once the SDK has emitted. A refresh before any event still retries as before, which is the case that loop exists for. An abort also still reached the fallback, since the SDK rejects when the caller cancels and that read as a transport failure. It spent a second request on an already aborted signal and reported the cancellation as a connection error. The fallback now rethrows instead. Specs for both, plus a check that the pre event refresh path is untouched. --- .../providers/__tests__/openai-codex.spec.ts | 34 +++++++++++++++++++ src/api/providers/openai-codex.ts | 26 +++++++++----- 2 files changed, 52 insertions(+), 8 deletions(-) diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index b4be574ed3..96769f0301 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -441,6 +441,40 @@ describe("OpenAiCodexHandler.completePrompt streaming", () => { expect(mockFetch).not.toHaveBeenCalled() }) + // The service has accepted the request by the time it emits, so refreshing the token and + // sending it again would bill a second generation and hand the caller both. + it("does not retry with a refreshed token when the SDK fails after emitting", async () => { + const handler = createHandler() + const refresh = vitest.spyOn(openAiCodexOAuthManager, "forceRefreshAccessToken") + const create = vitest.fn().mockResolvedValue( + (async function* () { + yield { type: "response.output_text.delta", delta: "feat: add" } + throw new Error("Codex API invalid token") + })(), + ) + Reflect.set(handler, "client", { responses: { create } }) + const mockFetch = vitest.fn() + vitest.stubGlobal("fetch", mockFetch) + + await expect(handler.completePrompt("Hello")).rejects.toThrow(/completionError|invalid token/) + expect(refresh).not.toHaveBeenCalled() + expect(create).toHaveBeenCalledTimes(1) + expect(mockFetch).not.toHaveBeenCalled() + }) + + // An abort is not a transport failure, so spending a second request on an already-aborted + // signal only turns the cancellation into a connection error. + it("does not fall back to SSE when the SDK fails because the caller aborted", async () => { + const handler = createHandler() + const create = vitest.fn().mockRejectedValue(new Error("Request was aborted")) + Reflect.set(handler, "client", { responses: { create } }) + const mockFetch = vitest.fn() + vitest.stubGlobal("fetch", mockFetch) + + await expect(handler.completePrompt("Hello", { abortSignal: AbortSignal.abort() })).rejects.toThrow() + expect(mockFetch).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 de33b20f28..2ba90bf137 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -142,6 +142,10 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion private sawTextDeltaInCurrentResponse = false // Tracks tool call IDs emitted via streaming partial events to prevent done-event duplicates. private streamedToolCallIds = new Set() + // Tracks whether the SDK stream produced an event, which is the point where the service has + // accepted the request and its output is already with the caller. Neither transport may be + // replayed after that: doing so appends a second generation to the first. + private sawSdkEventInCurrentResponse = false // Event types handled by the shared event processor private readonly coreHandledEventTypes = new Set([ @@ -234,6 +238,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion this.pendingToolCallName = undefined this.sawTextOutputInCurrentResponse = false this.sawTextDeltaInCurrentResponse = false + this.sawSdkEventInCurrentResponse = false this.streamedToolCallIds.clear() // Get access token from OAuth manager @@ -281,7 +286,10 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion const message = error instanceof Error ? error.message : String(error) const isAuthFailure = /unauthorized|invalid token|not authenticated|authentication|401/i.test(message) - if (attempt === 0 && isAuthFailure) { + // Retrying is only safe while nothing has come back yet. Once the SDK has emitted, + // the service has accepted the request, so a refreshed-token retry would replay it + // and append a second generation to output the caller already has. + if (attempt === 0 && isAuthFailure && !this.sawSdkEventInCurrentResponse) { // Force refresh the token for retry const refreshed = await openAiCodexOAuthManager.forceRefreshAccessToken() if (!refreshed) { @@ -456,11 +464,6 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } } - // Once the SDK stream has produced an event the request has been accepted and its output is - // already with the caller, so replaying it over SSE would append a second generation to the - // first. The fallback below is only for an SDK that could not be used at all. - let sawSdkEvent = false - try { // Prefer OpenAI SDK streaming (same approach as openai-native) so event handling // is consistent across providers. @@ -500,7 +503,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // Set before the event is processed, not after: `processEvent` also mutates // response state, so a throw from it must not replay either. - sawSdkEvent = true + this.sawSdkEventInCurrentResponse = true for await (const outChunk of this.processEvent(event, model)) { if (outChunk.type === "text") { @@ -510,7 +513,14 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } } } catch (sdkErr) { - if (sawSdkEvent) { + // The fallback is only for an SDK that could not be used at all. Once the stream has + // emitted, the request has been accepted and its output is already with the caller, + // so replaying it over SSE would append a second generation to the first. + // + // 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) { throw sdkErr } From bb2676a815ae6aadf28920db661117060b19dbd1 Mon Sep 17 00:00:00 2001 From: Rafael-Silva-Oliveira Date: Sat, 15 Aug 2026 14:16:48 +0200 Subject: [PATCH 4/4] fix(openai-codex): report one abort result however the request ended Chasing a review comment about the cancellation tests turned up a real inconsistency behind it. A stream that ended quietly threw an AbortError, but a transport that rejected on abort had its own error passed straight through, so what a cancelled completePrompt rejected with depended on how far the request had got. Callers cannot key off that. An abort is now restated as an AbortError unless it already is one. The in flight case, where the caller cancels after the request is away but before any event, had no coverage. It only works if the caller signal is genuinely linked to the internal controller, so it is the case that would catch that link breaking. Added, and it waits for the request to be in flight before aborting, since the token lookup and the listener are both async and a synchronous abort fires before anything is listening. Also tightened the existing abort assertion from a bare rejects.toThrow, which passed on any error at all, to the same AbortError check the other cancellation tests use. --- .../providers/__tests__/openai-codex.spec.ts | 37 ++++++++++++++++++- src/api/providers/openai-codex.ts | 8 +++- 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/src/api/providers/__tests__/openai-codex.spec.ts b/src/api/providers/__tests__/openai-codex.spec.ts index 96769f0301..a0736b1d4f 100644 --- a/src/api/providers/__tests__/openai-codex.spec.ts +++ b/src/api/providers/__tests__/openai-codex.spec.ts @@ -471,7 +471,42 @@ describe("OpenAiCodexHandler.completePrompt streaming", () => { const mockFetch = vitest.fn() vitest.stubGlobal("fetch", mockFetch) - await expect(handler.completePrompt("Hello", { abortSignal: AbortSignal.abort() })).rejects.toThrow() + await expect(handler.completePrompt("Hello", { abortSignal: AbortSignal.abort() })).rejects.toMatchObject({ + name: "AbortError", + }) + expect(mockFetch).not.toHaveBeenCalled() + }) + + // The abort arrives while the request is still in flight, before any event, which is the case + // that only works if the caller's signal is genuinely linked to the internal controller. A + // broken link would leave the request hanging on a signal that never fires. + it("aborts the in-flight request when the caller cancels before any event", async () => { + const handler = createHandler() + const controller = new AbortController() + let signalDuringRequest: AbortSignal | undefined + + const create = vitest.fn().mockImplementation( + (_body: unknown, options: { signal: AbortSignal }) => + new Promise((_resolve, reject) => { + signalDuringRequest = options.signal + // Reject the way the SDK does once the signal it was handed aborts. + 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 }) + // The token lookup and the listener that links the two signals are both async, so aborting + // synchronously here would fire before anything is listening. + await vitest.waitFor(() => expect(create).toHaveBeenCalled()) + controller.abort() + + await expect(completion).rejects.toMatchObject({ name: "AbortError" }) + expect(signalDuringRequest!.aborted).toBe(true) expect(mockFetch).not.toHaveBeenCalled() }) diff --git a/src/api/providers/openai-codex.ts b/src/api/providers/openai-codex.ts index 2ba90bf137..853d11feea 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -1335,9 +1335,13 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion 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. + // 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. if (options?.abortSignal?.aborted) { - throw error + throw error instanceof DOMException && error.name === "AbortError" + ? error + : new DOMException("OpenAI Codex completion was aborted", "AbortError") } const errorModel = this.getModel()