From 0e967d45a45721ea75cecc89baa90399308c38d5 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 01:33:39 +0800 Subject: [PATCH 01/18] feat(api): abort signal and timeout support for anthropic, anthropic-vertex, xai, minimax - completePrompt: forward CompletePromptOptions abortSignal/timeoutMs into the SDK request for AnthropicHandler, AnthropicVertexHandler, XAIHandler, and MiniMaxHandler (request options built only when a signal/timeout is provided, preserving existing behavior) - createMessage: bridge metadata?.abortSignal into a per-request AbortController using the Bedrock pattern (pre-aborted guard + { once: true } listener) and pass the internal signal as the SDK request signal; existing client-level timeout mechanisms are untouched - xai: rethrow AbortError unmodified from createMessage/completePrompt so callers can detect error.name === 'AbortError' - tests: port reference completePrompt signal/timeout propagation tests and add per-provider createMessage bridging tests (pre-aborted signal rejects with AbortError; mid-flight external abort cancels the request) --- .../__tests__/anthropic-vertex.spec.ts | 164 ++++++++++++++++-- src/api/providers/__tests__/anthropic.spec.ts | 160 ++++++++++++++++- src/api/providers/__tests__/minimax.spec.ts | 120 +++++++++++++ src/api/providers/__tests__/xai.spec.ts | 117 +++++++++++++ src/api/providers/anthropic-vertex.ts | 44 ++++- src/api/providers/anthropic.ts | 53 ++++-- src/api/providers/minimax.ts | 48 ++++- src/api/providers/xai.ts | 60 ++++++- 8 files changed, 716 insertions(+), 50 deletions(-) diff --git a/src/api/providers/__tests__/anthropic-vertex.spec.ts b/src/api/providers/__tests__/anthropic-vertex.spec.ts index 3e98f3ec5b..032d1b0940 100644 --- a/src/api/providers/__tests__/anthropic-vertex.spec.ts +++ b/src/api/providers/__tests__/anthropic-vertex.spec.ts @@ -8,6 +8,7 @@ import { VERTEX_1M_CONTEXT_MODEL_IDS } from "@roo-code/types" import { AnthropicVertexHandler } from "../anthropic-vertex" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" +import { makeCreateMessageMetadata } from "../../../test-utils/api" vitest.mock("../utils/timeout-config", () => ({ getApiRequestTimeout: vitest.fn().mockReturnValue(300_000), @@ -746,6 +747,69 @@ describe("VertexHandler", () => { expect(calledMessages).toHaveLength(2) // Only the two user messages expect(calledMessages.every((m: any) => m.role === "user")).toBe(true) }) + + it("should reject with AbortError when createMessage is called with an already-aborted signal", async () => { + const abortedController = new AbortController() + abortedController.abort() + + const mockCreate = vitest + .spyOn(handler["client"].messages, "create") + .mockImplementation((_params: unknown, options?: { signal?: AbortSignal | null }) => { + if (options?.signal?.aborted) { + const error = new Error("The operation was aborted") + error.name = "AbortError" + throw error + } + return asyncStreamFrom([]) as never + }) + + const stream = handler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: abortedController.signal }), + ) + + await expect(stream.next()).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("should abort the request when the external signal aborts mid-flight", async () => { + const controller = new AbortController() + + const mockCreate = vitest.spyOn(handler["client"].messages, "create").mockImplementation( + (_params: unknown, options?: { signal?: AbortSignal | null }) => + new Promise((_resolve, reject) => { + const signal = options?.signal + if (!signal) { + return + } + if (signal.aborted) { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + return + } + signal.addEventListener( + "abort", + () => { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + }, + { once: true }, + ) + }) as never, + ) + + const stream = handler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const promise = stream.next() + controller.abort() + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + }) }) describe("completePrompt", () => { @@ -758,18 +822,22 @@ describe("VertexHandler", () => { const result = await handler.completePrompt("Test prompt") expect(result).toBe("Test response") - expect(handler["client"].messages.create).toHaveBeenCalledWith({ - model: "claude-3-5-sonnet-v2@20241022", - max_tokens: 8192, - temperature: 0, - messages: [ - { - role: "user", - content: [{ type: "text", text: "Test prompt", cache_control: { type: "ephemeral" } }], - }, - ], - stream: false, - }) + expect(handler["client"].messages.create).toHaveBeenCalledWith( + { + model: "claude-3-5-sonnet-v2@20241022", + max_tokens: 8192, + temperature: 0, + messages: [ + { + role: "user", + content: [{ type: "text", text: "Test prompt", cache_control: { type: "ephemeral" } }], + }, + ], + stream: false, + thinking: undefined, + }, + undefined, + ) }) it("should handle API errors for Claude", async () => { @@ -820,6 +888,78 @@ describe("VertexHandler", () => { expect(result).toBe("") }) + it("should pass abort signal through to client", async () => { + handler = new AnthropicVertexHandler({ + apiModelId: "claude-3-5-sonnet-v2@20241022", + vertexProjectId: "test-project", + vertexRegion: "us-central1", + }) + + const controller = new AbortController() + const mockCreate = vitest + .spyOn(handler["client"].messages, "create") + .mockResolvedValue({ content: [{ type: "text", text: "response" }] } as never) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + + const [, requestOptions] = mockCreate.mock.calls[0] + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.any(Object), + ) + expect(requestOptions?.signal).toBe(controller.signal) + }) + + it("should work without options (backward compatible)", async () => { + handler = new AnthropicVertexHandler({ + apiModelId: "claude-3-5-sonnet-v2@20241022", + vertexProjectId: "test-project", + vertexRegion: "us-central1", + }) + + const mockCreate = vitest + .spyOn(handler["client"].messages, "create") + .mockResolvedValue({ content: [{ type: "text", text: "response" }] } as never) + + const result = await handler.completePrompt("test prompt") + expect(result).toBe("response") + expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), undefined) + }) + + it("completePrompt should pass signal through to client", async () => { + handler = new AnthropicVertexHandler({ + apiModelId: "claude-3-5-sonnet-v2@20241022", + vertexProjectId: "test-project", + vertexRegion: "us-central1", + }) + + const controller = new AbortController() + const mockCreate = vitest + .spyOn(handler["client"].messages, "create") + .mockResolvedValue({ content: [{ type: "text", text: "response" }] } as never) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 5000 }) + + const [, requestOptions] = mockCreate.mock.calls[0] + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ timeout: 5000 }), + ) + expect(requestOptions?.signal).toBe(controller.signal) + }) + + it("completePrompt should pass timeoutMs when provided", async () => { + const mockCreate = vitest + .spyOn(handler["client"].messages, "create") + .mockResolvedValue({ content: [{ type: "text", text: "response" }] } as never) + + await handler.completePrompt("test prompt", { timeoutMs: 3000 }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ timeout: 3000 }), + ) + }) + it("should handle empty content array for Claude", async () => { handler = new AnthropicVertexHandler({ apiModelId: "claude-3-5-sonnet-v2@20241022", diff --git a/src/api/providers/__tests__/anthropic.spec.ts b/src/api/providers/__tests__/anthropic.spec.ts index 21d2816ec7..ae775968c9 100644 --- a/src/api/providers/__tests__/anthropic.spec.ts +++ b/src/api/providers/__tests__/anthropic.spec.ts @@ -3,6 +3,7 @@ import { AnthropicHandler } from "../anthropic" import { ApiHandlerOptions } from "../../../shared/api" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" +import { makeCreateMessageMetadata } from "../../../test-utils/api" import { clearAllMocks } from "../../../test-utils/reset" // Mock TelemetryService @@ -476,20 +477,83 @@ describe("AnthropicHandler", () => { expect(requestBody?.model).toBe("claude-sonnet-5-bf") expect(requestBody?.thinking).toEqual({ type: "adaptive" }) }) + + it("should reject with AbortError when createMessage is called with an already-aborted signal", async () => { + const abortedController = new AbortController() + abortedController.abort() + + mockCreate.mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + const error = new Error("The operation was aborted") + error.name = "AbortError" + throw error + } + return asyncStreamFrom([]) + }) + + const stream = handler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: abortedController.signal }), + ) + + await expect(stream.next()).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("should abort the request when the external signal aborts mid-flight", async () => { + const controller = new AbortController() + + mockCreate.mockImplementation((_params: unknown, options?: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + const signal = options?.signal + if (!signal) { + return + } + if (signal.aborted) { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + return + } + signal.addEventListener( + "abort", + () => { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + }, + { once: true }, + ) + }) + }) + + const stream = handler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const promise = stream.next() + controller.abort() + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + }) }) describe("completePrompt", () => { it("should complete prompt successfully", async () => { const result = await handler.completePrompt("Test prompt") expect(result).toBe("Test response") - expect(mockCreate).toHaveBeenCalledWith({ - model: mockOptions.apiModelId, - messages: [{ role: "user", content: "Test prompt" }], - max_tokens: 8192, - temperature: 0, - thinking: undefined, - stream: false, - }) + expect(mockCreate).toHaveBeenCalledWith( + { + model: mockOptions.apiModelId, + messages: [{ role: "user", content: "Test prompt" }], + max_tokens: 8192, + temperature: 0, + thinking: undefined, + stream: false, + }, + undefined, + ) }) it("should handle API errors", async () => { @@ -512,6 +576,86 @@ describe("AnthropicHandler", () => { const result = await handler.completePrompt("Test prompt") expect(result).toBe("") }) + + it("should pass abort signal through to client", async () => { + const controller = new AbortController() + mockCreate.mockResolvedValueOnce({ content: [{ type: "text", text: "response" }] }) + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + expect(mockCreate).toHaveBeenCalledWith( + { + model: mockOptions.apiModelId, + messages: [{ role: "user", content: "test prompt" }], + max_tokens: 8192, + temperature: 0, + thinking: undefined, + stream: false, + }, + { signal: controller.signal }, + ) + }) + + it("should work without options (backward compatible)", async () => { + mockCreate.mockResolvedValueOnce({ content: [{ type: "text", text: "response" }] }) + const result = await handler.completePrompt("test prompt") + expect(result).toBe("response") + expect(mockCreate).toHaveBeenCalledWith( + { + model: mockOptions.apiModelId, + messages: [{ role: "user", content: "test prompt" }], + max_tokens: 8192, + temperature: 0, + thinking: undefined, + stream: false, + }, + undefined, + ) + }) + + it("should merge signal and timeout together", async () => { + const controller = new AbortController() + mockCreate.mockResolvedValueOnce({ content: [{ type: "text", text: "response" }] }) + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 10000 }) + expect(mockCreate).toHaveBeenCalledWith( + { + model: mockOptions.apiModelId, + messages: [{ role: "user", content: "test prompt" }], + max_tokens: 8192, + temperature: 0, + thinking: undefined, + stream: false, + }, + expect.objectContaining({ signal: controller.signal, timeout: 10000 }), + ) + }) + + it("should pass timeoutMs through to client alongside abortSignal", async () => { + const controller = new AbortController() + mockCreate.mockResolvedValueOnce({ content: [{ type: "text", text: "response" }] }) + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 5000 }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: mockOptions.apiModelId }), + expect.objectContaining({ signal: controller.signal, timeout: 5000 }), + ) + }) + + it("should pass the same signal instance", async () => { + const controller = new AbortController() + mockCreate.mockResolvedValueOnce({ content: [{ type: "text", text: "response" }] }) + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + expect(mockCreate).toHaveBeenCalledWith( + expect.any(Object), + expect.objectContaining({ signal: controller.signal }), + ) + // Verify it's the exact same instance, not just equal + const callOptions = mockCreate.mock.calls[0][1] + expect(callOptions?.signal).toBe(controller.signal) + }) + + it("should not include signal-related options when not provided", async () => { + mockCreate.mockResolvedValueOnce({ content: [{ type: "text", text: "response" }] }) + await handler.completePrompt("test prompt") + expect(mockCreate).toHaveBeenCalledWith(expect.any(Object), undefined) + }) }) describe("getModel", () => { diff --git a/src/api/providers/__tests__/minimax.spec.ts b/src/api/providers/__tests__/minimax.spec.ts index 01102b0457..08ac58591c 100644 --- a/src/api/providers/__tests__/minimax.spec.ts +++ b/src/api/providers/__tests__/minimax.spec.ts @@ -14,6 +14,7 @@ import { type MinimaxModelId, minimaxDefaultModelId, minimaxModels } from "@roo- import { MiniMaxHandler } from "../minimax" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" +import { makeCreateMessageMetadata } from "../../../test-utils/api" import { clearAllMocks } from "../../../test-utils/reset" vitest.mock("@anthropic-ai/sdk", () => { @@ -239,6 +240,63 @@ describe("MiniMaxHandler", () => { await expect(handler.completePrompt("test prompt")).rejects.toThrow() }) + it("should pass abort signal through to client", async () => { + const controller = new AbortController() + mockCreate.mockResolvedValueOnce({ + content: [{ type: "text", text: "response" }], + }) + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + { signal: controller.signal }, // second arg (options) + ) + }) + + it("should work without options (backward compatible)", async () => { + mockCreate.mockResolvedValueOnce({ + content: [{ type: "text", text: "response" }], + }) + const result = await handler.completePrompt("test prompt") + expect(result).toBe("response") + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + undefined, // second arg (options) + ) + }) + + it("should pass timeout through to client", async () => { + const controller = new AbortController() + mockCreate.mockResolvedValueOnce({ + content: [{ type: "text", text: "response" }], + }) + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 5000 }) + expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + signal: controller.signal, + timeout: 5000, + }) + }) + + it("should pass only timeoutMs when no signal provided", async () => { + mockCreate.mockResolvedValueOnce({ + content: [{ type: "text", text: "response" }], + }) + await handler.completePrompt("test prompt", { timeoutMs: 3000 }) + expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + timeout: 3000, + }) + }) + + it("should pass timeout when timeoutMs=0 (defined check)", async () => { + mockCreate.mockResolvedValueOnce({ + content: [{ type: "text", text: "response" }], + }) + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + { timeout: 0 }, // !== undefined check means 0 is passed through + ) + }) + it("createMessage should yield text content from stream", async () => { const testContent = "This is test content from MiniMax stream" @@ -306,6 +364,7 @@ describe("MiniMaxHandler", () => { messages: expect.any(Array), stream: true, }), + undefined, ) }) @@ -319,6 +378,7 @@ describe("MiniMaxHandler", () => { expect.objectContaining({ temperature: 1, }), + undefined, ) }) @@ -375,6 +435,66 @@ describe("MiniMaxHandler", () => { arguments: undefined, }) }) + + it("should reject with AbortError when createMessage is called with an already-aborted signal", async () => { + const abortedController = new AbortController() + abortedController.abort() + + mockCreate.mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + const error = new Error("The operation was aborted") + error.name = "AbortError" + throw error + } + return asyncStreamFrom([]) + }) + + const stream = handler.createMessage( + "system prompt", + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: abortedController.signal }), + ) + + await expect(stream.next()).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("should abort the request when the external signal aborts mid-flight", async () => { + const controller = new AbortController() + + mockCreate.mockImplementation((_params: unknown, options?: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + const signal = options?.signal + if (!signal) { + return + } + if (signal.aborted) { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + return + } + signal.addEventListener( + "abort", + () => { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + }, + { once: true }, + ) + }) + }) + + const stream = handler.createMessage( + "system prompt", + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const promise = stream.next() + controller.abort() + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + }) }) describe("Model Configuration", () => { diff --git a/src/api/providers/__tests__/xai.spec.ts b/src/api/providers/__tests__/xai.spec.ts index a0427b6fe0..f6d5c331d8 100644 --- a/src/api/providers/__tests__/xai.spec.ts +++ b/src/api/providers/__tests__/xai.spec.ts @@ -24,6 +24,7 @@ import { xaiDefaultModelId, xaiModels } from "@roo-code/types" import { XAIHandler } from "../xai" import { asyncStreamFrom } from "../../../test-utils/stream" +import { makeCreateMessageMetadata } from "../../../test-utils/api" import { clearAllMocks } from "../../../test-utils/reset" describe("XAIHandler", () => { @@ -83,6 +84,7 @@ describe("XAIHandler", () => { store: false, include: ["reasoning.encrypted_content"], }), + undefined, ) }) @@ -212,6 +214,7 @@ describe("XAIHandler", () => { tool_choice: "auto", parallel_tool_calls: true, }), + undefined, ) }) @@ -232,6 +235,57 @@ describe("XAIHandler", () => { await expect(handler.completePrompt("test prompt")).rejects.toThrow(`xAI completion error: ${errorMessage}`) }) + it("completePrompt should pass abort signal through to client", async () => { + const controller = new AbortController() + mockResponsesCreate.mockResolvedValueOnce({ output_text: "response" }) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal }) + expect(mockResponsesCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + signal: controller.signal, + }) + }) + + it("completePrompt should work without options (backward compatible)", async () => { + mockResponsesCreate.mockResolvedValueOnce({ output_text: "response" }) + + const result = await handler.completePrompt("test prompt") + expect(result).toBe("response") + expect(mockResponsesCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + undefined, + ) + }) + + it("completePrompt should pass timeout through to client", async () => { + const controller = new AbortController() + mockResponsesCreate.mockResolvedValueOnce({ output_text: "response" }) + + await handler.completePrompt("test prompt", { abortSignal: controller.signal, timeoutMs: 5000 }) + expect(mockResponsesCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + signal: controller.signal, + timeout: 5000, + }) + }) + + it("completePrompt should pass only timeoutMs when no signal provided", async () => { + mockResponsesCreate.mockResolvedValueOnce({ output_text: "response" }) + + await handler.completePrompt("test prompt", { timeoutMs: 3000 }) + expect(mockResponsesCreate).toHaveBeenCalledWith(expect.objectContaining({ model: expect.any(String) }), { + timeout: 3000, + }) + }) + + it("completePrompt should pass timeout when timeoutMs=0 (defined check)", async () => { + mockResponsesCreate.mockResolvedValueOnce({ output_text: "response" }) + + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + expect(mockResponsesCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + { timeout: 0 }, // !== undefined check means 0 is passed through + ) + }) + it("should include reasoning effort for mini models in Responses API format", async () => { const miniModelHandler = new XAIHandler({ apiModelId: "grok-3-mini", @@ -249,6 +303,7 @@ describe("XAIHandler", () => { effort: "high", }), }), + undefined, ) }) @@ -269,6 +324,7 @@ describe("XAIHandler", () => { effort: "high", }), }), + undefined, ) }) @@ -290,6 +346,7 @@ describe("XAIHandler", () => { effort: "low", }), }), + undefined, ) }) @@ -315,4 +372,64 @@ describe("XAIHandler", () => { const stream = handler.createMessage("test prompt", []) await expect(stream.next()).rejects.toThrow(`xAI completion error: ${errorMessage}`) }) + + it("should reject with AbortError when createMessage is called with an already-aborted signal", async () => { + const abortedController = new AbortController() + abortedController.abort() + + mockResponsesCreate.mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + if (options?.signal?.aborted) { + const error = new Error("The operation was aborted") + error.name = "AbortError" + throw error + } + return asyncStreamFrom([]) + }) + + const stream = handler.createMessage( + "test prompt", + [], + makeCreateMessageMetadata({ abortSignal: abortedController.signal }), + ) + + await expect(stream.next()).rejects.toMatchObject({ name: "AbortError" }) + }) + + it("should abort the request when the external signal aborts mid-flight", async () => { + const controller = new AbortController() + + mockResponsesCreate.mockImplementation((_params: unknown, options?: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + const signal = options?.signal + if (!signal) { + return + } + if (signal.aborted) { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + return + } + signal.addEventListener( + "abort", + () => { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + }, + { once: true }, + ) + }) + }) + + const stream = handler.createMessage( + "test prompt", + [], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const promise = stream.next() + controller.abort() + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + }) }) diff --git a/src/api/providers/anthropic-vertex.ts b/src/api/providers/anthropic-vertex.ts index 7b72b1100b..b5be64eff4 100644 --- a/src/api/providers/anthropic-vertex.ts +++ b/src/api/providers/anthropic-vertex.ts @@ -86,6 +86,23 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple tool_choice: convertOpenAIToolChoiceToAnthropic(metadata?.tool_choice, metadata?.parallelToolCalls), } + // Bridge the external abort signal from request metadata into a per-request + // controller so the SDK call is cancelled when the owning request is aborted + // (or when the signal is already aborted). Without an external signal the + // client-level timeout configured in the constructor remains the only + // cancellation mechanism, preserving the existing behavior. + const externalAbortSignal = metadata?.abortSignal + let abortSignal: AbortSignal | undefined + if (externalAbortSignal) { + const controller = new AbortController() + if (externalAbortSignal.aborted) { + controller.abort() + } else { + externalAbortSignal.addEventListener("abort", () => controller.abort(), { once: true }) + } + abortSignal = controller.signal + } + /** * Vertex API has specific limitations for prompt caching: * 1. Maximum of 4 blocks can have cache_control @@ -114,9 +131,18 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple } as Anthropic.Messages.MessageCreateParamsStreaming // and prompt caching - const requestOptions = betas?.length ? { headers: { "anthropic-beta": betas.join(",") } } : undefined + const requestOptions: Anthropic.RequestOptions = {} + if (betas?.length) { + requestOptions.headers = { "anthropic-beta": betas.join(",") } + } + if (abortSignal) { + requestOptions.signal = abortSignal + } - const stream = await this.client.messages.create(params, requestOptions) + const stream = await this.client.messages.create( + params, + Object.keys(requestOptions).length > 0 ? requestOptions : undefined, + ) for await (const chunk of stream) { switch (chunk.type) { @@ -297,7 +323,19 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple stream: false, } as Anthropic.Messages.MessageCreateParamsNonStreaming - const response = await this.client.messages.create(params) + // Build request options with abortSignal and/or timeout handling + const requestOptions: Anthropic.RequestOptions = {} + if (options?.abortSignal) { + requestOptions.signal = options.abortSignal + } + if (options?.timeoutMs !== undefined) { + requestOptions.timeout = options.timeoutMs + } + + const response = await this.client.messages.create( + params, + Object.keys(requestOptions).length > 0 ? requestOptions : undefined, + ) const content = response.content.find(({ type }) => type === "text") return content?.type === "text" ? content.text : "" diff --git a/src/api/providers/anthropic.ts b/src/api/providers/anthropic.ts index b55c8b3089..7e2333fb22 100644 --- a/src/api/providers/anthropic.ts +++ b/src/api/providers/anthropic.ts @@ -98,6 +98,23 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa tool_choice: convertOpenAIToolChoiceToAnthropic(metadata?.tool_choice, metadata?.parallelToolCalls), } + // Bridge the external abort signal from request metadata into a per-request + // controller so the SDK call is cancelled when the owning request is aborted + // (or when the signal is already aborted). Without an external signal the + // client-level timeout configured in the constructor remains the only + // cancellation mechanism, preserving the existing behavior. + const externalAbortSignal = metadata?.abortSignal + let abortSignal: AbortSignal | undefined + if (externalAbortSignal) { + const controller = new AbortController() + if (externalAbortSignal.aborted) { + controller.abort() + } else { + externalAbortSignal.addEventListener("abort", () => controller.abort(), { once: true }) + } + abortSignal = controller.signal + } + switch (modelId) { case "claude-sonnet-5": case "claude-sonnet-4-6": @@ -190,9 +207,12 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa case "claude-haiku-4-5-20251001": case "claude-3-haiku-20240307": betas.push("prompt-caching-2024-07-31") - return { headers: { "anthropic-beta": betas.join(",") } } + return { + headers: { "anthropic-beta": betas.join(",") }, + ...(abortSignal && { signal: abortSignal }), + } default: - return undefined + return abortSignal ? { signal: abortSignal } : undefined } })(), ) @@ -223,6 +243,7 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa } stream = (await this.client.messages.create( requestParams as Anthropic.Messages.MessageCreateParamsStreaming, + abortSignal ? { signal: abortSignal } : undefined, )) as any } catch (error) { TelemetryService.instance.captureException( @@ -436,14 +457,26 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa let message try { - message = await this.client.messages.create({ - model, - max_tokens: ANTHROPIC_DEFAULT_MAX_TOKENS, - thinking: undefined, - temperature, - messages: [{ role: "user", content: prompt }], - stream: false, - }) + // Build request options with both abortSignal and timeout handling + const requestOptions: Anthropic.RequestOptions = {} + if (options?.abortSignal) { + requestOptions.signal = options.abortSignal + } + if (options?.timeoutMs) { + requestOptions.timeout = options.timeoutMs + } + + message = await this.client.messages.create( + { + model, + max_tokens: ANTHROPIC_DEFAULT_MAX_TOKENS, + thinking: undefined, + temperature, + messages: [{ role: "user", content: prompt }], + stream: false, + }, + Object.keys(requestOptions).length > 0 ? requestOptions : undefined, + ) } catch (error) { TelemetryService.instance.captureException( new ApiProviderError( diff --git a/src/api/providers/minimax.ts b/src/api/providers/minimax.ts index e209add72d..9e9c219a25 100644 --- a/src/api/providers/minimax.ts +++ b/src/api/providers/minimax.ts @@ -85,6 +85,23 @@ export class MiniMaxHandler extends BaseProvider implements SingleCompletionHand const cacheControl: CacheControlEphemeral = { type: "ephemeral" } const { id: modelId, info, maxTokens, temperature } = this.getModel() + // Bridge the external abort signal from request metadata into a per-request + // controller so the SDK call is cancelled when the owning request is aborted + // (or when the signal is already aborted). Without an external signal the + // client-level timeout configured in the constructor remains the only + // cancellation mechanism, preserving the existing behavior. + const externalAbortSignal = metadata?.abortSignal + let abortSignal: AbortSignal | undefined + if (externalAbortSignal) { + const controller = new AbortController() + if (externalAbortSignal.aborted) { + controller.abort() + } else { + externalAbortSignal.addEventListener("abort", () => controller.abort(), { once: true }) + } + abortSignal = controller.signal + } + // MiniMax M2 models support prompt caching const supportsPromptCache = info.supportsPromptCache ?? false @@ -113,7 +130,10 @@ export class MiniMaxHandler extends BaseProvider implements SingleCompletionHand tool_choice: convertOpenAIToolChoice(metadata?.tool_choice), } - const stream = await this.client.messages.create(requestParams) + const stream = await this.client.messages.create( + requestParams, + abortSignal ? { signal: abortSignal } : undefined, + ) let inputTokens = 0 let outputTokens = 0 @@ -292,13 +312,25 @@ export class MiniMaxHandler extends BaseProvider implements SingleCompletionHand async completePrompt(prompt: string, options?: CompletePromptOptions) { const { id: model, temperature } = this.getModel() - const message = await this.client.messages.create({ - model, - max_tokens: 16_384, - temperature: temperature ?? 1.0, - messages: [{ role: "user", content: prompt }], - stream: false, - }) + // Build request options with abortSignal and/or timeout handling + const requestOptions: Anthropic.RequestOptions = {} + if (options?.abortSignal) { + requestOptions.signal = options.abortSignal + } + if (options?.timeoutMs !== undefined) { + requestOptions.timeout = options.timeoutMs + } + + const message = await this.client.messages.create( + { + model, + max_tokens: 16_384, + temperature: temperature ?? 1.0, + messages: [{ role: "user", content: prompt }], + stream: false, + }, + Object.keys(requestOptions).length > 0 ? requestOptions : undefined, + ) const content = message.content.find(({ type }) => type === "text") return content?.type === "text" ? content.text : "" diff --git a/src/api/providers/xai.ts b/src/api/providers/xai.ts index 189ec4e9ed..3f677f06b5 100644 --- a/src/api/providers/xai.ts +++ b/src/api/providers/xai.ts @@ -95,6 +95,23 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler const input = convertToResponsesApiInput(messages) const responseTools = this.mapResponseTools(metadata?.tools) + // Bridge the external abort signal from request metadata into a per-request + // controller so the SDK call is cancelled when the owning request is aborted + // (or when the signal is already aborted). Without an external signal the + // client-level timeout configured in the constructor remains the only + // cancellation mechanism, preserving the existing behavior. + const externalAbortSignal = metadata?.abortSignal + let abortSignal: AbortSignal | undefined + if (externalAbortSignal) { + const controller = new AbortController() + if (externalAbortSignal.aborted) { + controller.abort() + } else { + externalAbortSignal.addEventListener("abort", () => controller.abort(), { once: true }) + } + abortSignal = controller.signal + } + // Build request options const requestBody: Record = { model: model.id, @@ -129,11 +146,19 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler let stream: AsyncIterable try { - stream = (await this.client.responses.create({ - ...requestBody, - stream: true, - } as any)) as unknown as AsyncIterable + stream = (await this.client.responses.create( + { + ...requestBody, + stream: true, + } as any, + abortSignal ? { signal: abortSignal } : undefined, + )) as unknown as AsyncIterable } catch (error) { + // Let abort errors propagate unmodified so callers can recognize them + // via error.name === "AbortError". + if (error instanceof Error && error.name === "AbortError") { + throw error + } const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, model.id, "createMessage") TelemetryService.instance.captureException(apiError) @@ -148,15 +173,32 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler const model = this.getModel() try { - const response = await this.client.responses.create({ - model: model.id, - input: [{ role: "user", content: [{ type: "input_text", text: prompt }] }], - store: false, - }) + // Build request options with abortSignal and/or timeout handling + const requestOptions: OpenAI.RequestOptions = {} + if (options?.abortSignal) { + requestOptions.signal = options.abortSignal + } + if (options?.timeoutMs !== undefined) { + requestOptions.timeout = options.timeoutMs + } + + const response = await this.client.responses.create( + { + model: model.id, + input: [{ role: "user", content: [{ type: "input_text", text: prompt }] }], + store: false, + }, + Object.keys(requestOptions).length > 0 ? requestOptions : undefined, + ) // output_text is a convenience field on the Responses API response return response.output_text || "" } catch (error) { + // Let abort errors propagate unmodified so callers can recognize them + // via error.name === "AbortError". + if (error instanceof Error && error.name === "AbortError") { + throw error + } const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, model.id, "completePrompt") TelemetryService.instance.captureException(apiError) From dfbc1d8ce6072dbb7a3605410bee7bf9fefc2e29 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 02:38:01 +0800 Subject: [PATCH 02/18] ci: re-run e2e-mock on flaky restart-persistence (uses openrouter; this PR only touches anthropic family) From 81a75d4df91986bde6e57aa141577549b6974e77 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 05:02:35 +0800 Subject: [PATCH 03/18] fix(api): address CodeRabbit review on anthropic family abort handling - anthropic.ts: use options?.timeoutMs !== undefined (was truthy) so a caller-supplied timeoutMs: 0 is forwarded to the SDK instead of silently dropped; all 4 family providers now share the same defined-check - xai.ts: rethrow the OpenAI SDK's APIUserAbortError (exported from openai v5) unmodified from createMessage/completePrompt alongside native AbortError, since the SDK throws it when the request signal aborts and it would otherwise be mangled by handleOpenAIError - tests: anthropic.spec.ts regression test asserting timeoutMs: 0 reaches the SDK as timeout: 0; xai.spec.ts exposes the real APIUserAbortError in the openai mock and asserts the SDK abort error surfaces as the same instance (unwrapped) through both createMessage and completePrompt --- src/api/providers/__tests__/anthropic.spec.ts | 9 ++++ src/api/providers/__tests__/xai.spec.ts | 46 ++++++++++++++++++- src/api/providers/anthropic.ts | 2 +- src/api/providers/xai.ts | 20 +++++--- 4 files changed, 67 insertions(+), 10 deletions(-) diff --git a/src/api/providers/__tests__/anthropic.spec.ts b/src/api/providers/__tests__/anthropic.spec.ts index ae775968c9..de23a5205a 100644 --- a/src/api/providers/__tests__/anthropic.spec.ts +++ b/src/api/providers/__tests__/anthropic.spec.ts @@ -656,6 +656,15 @@ describe("AnthropicHandler", () => { await handler.completePrompt("test prompt") expect(mockCreate).toHaveBeenCalledWith(expect.any(Object), undefined) }) + + it("should pass timeout when timeoutMs=0 (defined check)", async () => { + mockCreate.mockResolvedValueOnce({ content: [{ type: "text", text: "response" }] }) + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + // timeoutMs=0 must be forwarded as an explicit 0 timeout, not dropped as if unset + expect(mockCreate).toHaveBeenCalledWith(expect.objectContaining({ model: mockOptions.apiModelId }), { + timeout: 0, + }) + }) }) describe("getModel", () => { diff --git a/src/api/providers/__tests__/xai.spec.ts b/src/api/providers/__tests__/xai.spec.ts index f6d5c331d8..316dec872a 100644 --- a/src/api/providers/__tests__/xai.spec.ts +++ b/src/api/providers/__tests__/xai.spec.ts @@ -14,10 +14,16 @@ const mockResponsesCreate = vitest.hoisted(() => vitest.fn()) vitest.mock("openai", async () => { const { mockOpenAiResponsesClient } = await import("../../../test-utils/api") - return mockOpenAiResponsesClient(mockResponsesCreate) + const actual = await vi.importActual("openai") + return { + ...mockOpenAiResponsesClient(mockResponsesCreate), + // Expose the real SDK error class so the mock transport can throw it exactly + // the way the OpenAI SDK does when a request signal aborts. + APIUserAbortError: actual.APIUserAbortError, + } }) -import OpenAI from "openai" +import OpenAI, { APIUserAbortError } from "openai" import type { Anthropic } from "@anthropic-ai/sdk" import { xaiDefaultModelId, xaiModels } from "@roo-code/types" @@ -235,6 +241,18 @@ describe("XAIHandler", () => { await expect(handler.completePrompt("test prompt")).rejects.toThrow(`xAI completion error: ${errorMessage}`) }) + it("completePrompt should surface the SDK APIUserAbortError unmodified on abort", async () => { + const controller = new AbortController() + controller.abort() + const sdkAbortError = new APIUserAbortError() + mockResponsesCreate.mockRejectedValueOnce(sdkAbortError) + + // The error must surface as the same SDK instance, not wrapped by handleOpenAIError + await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toBe( + sdkAbortError, + ) + }) + it("completePrompt should pass abort signal through to client", async () => { const controller = new AbortController() mockResponsesCreate.mockResolvedValueOnce({ output_text: "response" }) @@ -373,6 +391,30 @@ describe("XAIHandler", () => { await expect(stream.next()).rejects.toThrow(`xAI completion error: ${errorMessage}`) }) + it("createMessage should surface the SDK APIUserAbortError unmodified on abort", async () => { + const abortedController = new AbortController() + abortedController.abort() + const sdkAbortError = new APIUserAbortError() + + mockResponsesCreate.mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + // Mimic the OpenAI SDK: reject with its own APIUserAbortError when the + // request signal is aborted. + if (options?.signal?.aborted) { + throw sdkAbortError + } + return asyncStreamFrom([]) + }) + + const stream = handler.createMessage( + "test prompt", + [], + makeCreateMessageMetadata({ abortSignal: abortedController.signal }), + ) + + // The SDK error must surface as the same instance, not wrapped by handleOpenAIError + await expect(stream.next()).rejects.toBe(sdkAbortError) + }) + it("should reject with AbortError when createMessage is called with an already-aborted signal", async () => { const abortedController = new AbortController() abortedController.abort() diff --git a/src/api/providers/anthropic.ts b/src/api/providers/anthropic.ts index 7e2333fb22..1253e3b55a 100644 --- a/src/api/providers/anthropic.ts +++ b/src/api/providers/anthropic.ts @@ -462,7 +462,7 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa if (options?.abortSignal) { requestOptions.signal = options.abortSignal } - if (options?.timeoutMs) { + if (options?.timeoutMs !== undefined) { requestOptions.timeout = options.timeoutMs } diff --git a/src/api/providers/xai.ts b/src/api/providers/xai.ts index 3f677f06b5..95a86b7137 100644 --- a/src/api/providers/xai.ts +++ b/src/api/providers/xai.ts @@ -1,5 +1,5 @@ import { Anthropic } from "@anthropic-ai/sdk" -import OpenAI from "openai" +import OpenAI, { APIUserAbortError } from "openai" import { type XAIModelId, xaiDefaultModelId, xaiModels, ApiProviderError } from "@roo-code/types" import { TelemetryService } from "@roo-code/telemetry" @@ -154,9 +154,12 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler abortSignal ? { signal: abortSignal } : undefined, )) as unknown as AsyncIterable } catch (error) { - // Let abort errors propagate unmodified so callers can recognize them - // via error.name === "AbortError". - if (error instanceof Error && error.name === "AbortError") { + // Let abort errors propagate unmodified so callers can recognize them: + // native AbortError (error.name === "AbortError") and the OpenAI SDK's + // APIUserAbortError, which the SDK throws when the request signal aborts + // (the SDK class does not set a distinctive error.name in v5, so use + // instanceof). + if ((error instanceof Error && error.name === "AbortError") || error instanceof APIUserAbortError) { throw error } const errorMessage = error instanceof Error ? error.message : String(error) @@ -194,9 +197,12 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler // output_text is a convenience field on the Responses API response return response.output_text || "" } catch (error) { - // Let abort errors propagate unmodified so callers can recognize them - // via error.name === "AbortError". - if (error instanceof Error && error.name === "AbortError") { + // Let abort errors propagate unmodified so callers can recognize them: + // native AbortError (error.name === "AbortError") and the OpenAI SDK's + // APIUserAbortError, which the SDK throws when the request signal aborts + // (the SDK class does not set a distinctive error.name in v5, so use + // instanceof). + if ((error instanceof Error && error.name === "AbortError") || error instanceof APIUserAbortError) { throw error } const errorMessage = error instanceof Error ? error.message : String(error) From af279dd6226f1d86e047593a36f4cf48816fffd4 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 20 Aug 2026 17:05:51 +0800 Subject: [PATCH 04/18] fix(api): type xAI streaming request body with the OpenAI SDK type - xai.ts: declare the streaming request body as OpenAI.Responses.ResponseCreateParamsStreaming instead of Record, so the full request shape (incl. include/reasoning) is typechecked against the SDK and the as any on the create() call is no longer needed - xai.ts: type the stream as AsyncIterable (matching the codebase pattern in mimo.ts/openai.ts) and drop the as unknown as AsyncIterable double cast, since the SDK create() streaming overload already returns an AsyncIterable stream - eslint-suppressions.json: reduce @typescript-eslint/no-explicit-any count for api/providers/xai.ts from 7 to 3 (four any usages removed) --- src/api/providers/xai.ts | 10 +++++----- src/eslint-suppressions.json | 2 +- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/api/providers/xai.ts b/src/api/providers/xai.ts index 95a86b7137..9b08745310 100644 --- a/src/api/providers/xai.ts +++ b/src/api/providers/xai.ts @@ -113,7 +113,7 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler } // Build request options - const requestBody: Record = { + const requestBody: OpenAI.Responses.ResponseCreateParamsStreaming = { model: model.id, instructions: systemPrompt, input: input, @@ -144,15 +144,15 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler requestBody.reasoning = { effort: model.reasoning.reasoning_effort } } - let stream: AsyncIterable + let stream: AsyncIterable try { - stream = (await this.client.responses.create( + stream = await this.client.responses.create( { ...requestBody, stream: true, - } as any, + }, abortSignal ? { signal: abortSignal } : undefined, - )) as unknown as AsyncIterable + ) } catch (error) { // Let abort errors propagate unmodified so callers can recognize them: // native AbortError (error.name === "AbortError") and the OpenAI SDK's diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index f790fba436..1085a4f965 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -456,7 +456,7 @@ }, "api/providers/xai.ts": { "@typescript-eslint/no-explicit-any": { - "count": 7 + "count": 3 } }, "api/transform/__tests__/ai-sdk.spec.ts": { From afed0850d9da04126bd9c4059739212e88f87c27 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 3 Sep 2026 03:35:45 +0800 Subject: [PATCH 05/18] fix(api): detach external abort listeners and cover timeoutMs 0 - anthropic-vertex and xai createMessage retain the external abort listener callback and remove it in a finally block, so a long-lived caller signal no longer accumulates one listener (and closed-over controller) per completed request; specs assert the same callback is detached after stream completion. - anthropic-vertex completePrompt spec adds the timeoutMs: 0 case asserting { timeout: 0 } reaches messages.create, guarding the defined-value check against a truthiness regression. --- .../__tests__/anthropic-vertex.spec.ts | 47 +++++ src/api/providers/__tests__/xai.spec.ts | 28 ++- src/api/providers/anthropic-vertex.ts | 174 ++++++++++-------- src/api/providers/xai.ts | 116 ++++++------ 4 files changed, 231 insertions(+), 134 deletions(-) diff --git a/src/api/providers/__tests__/anthropic-vertex.spec.ts b/src/api/providers/__tests__/anthropic-vertex.spec.ts index 032d1b0940..ba983f1fcd 100644 --- a/src/api/providers/__tests__/anthropic-vertex.spec.ts +++ b/src/api/providers/__tests__/anthropic-vertex.spec.ts @@ -810,6 +810,33 @@ describe("VertexHandler", () => { controller.abort() await expect(promise).rejects.toMatchObject({ name: "AbortError" }) }) + + it("should remove the external abort listener when the stream completes", async () => { + const handlerWithSignal = new AnthropicVertexHandler({ + apiModelId: "claude-3-5-sonnet-v2@20241022", + vertexProjectId: "test-project", + vertexRegion: "us-central1", + }) + + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + const stream = handlerWithSignal.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await collectStream(stream) + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // The same retained callback must be detached once the stream is done. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) }) describe("completePrompt", () => { @@ -960,6 +987,26 @@ describe("VertexHandler", () => { ) }) + it("completePrompt should pass timeout when timeoutMs=0 (defined check)", async () => { + handler = new AnthropicVertexHandler({ + apiModelId: "claude-3-5-sonnet-v2@20241022", + vertexProjectId: "test-project", + vertexRegion: "us-central1", + }) + + const mockCreate = vitest + .spyOn(handler["client"].messages, "create") + .mockResolvedValue({ content: [{ type: "text", text: "response" }] } as never) + + await handler.completePrompt("test prompt", { timeoutMs: 0 }) + // 0 is a defined value: it must reach the client as `timeout: 0`, + // not be dropped by a truthiness check. + expect(mockCreate).toHaveBeenCalledWith( + expect.objectContaining({ model: expect.any(String) }), + expect.objectContaining({ timeout: 0 }), + ) + }) + it("should handle empty content array for Claude", async () => { handler = new AnthropicVertexHandler({ apiModelId: "claude-3-5-sonnet-v2@20241022", diff --git a/src/api/providers/__tests__/xai.spec.ts b/src/api/providers/__tests__/xai.spec.ts index 316dec872a..4cb135973b 100644 --- a/src/api/providers/__tests__/xai.spec.ts +++ b/src/api/providers/__tests__/xai.spec.ts @@ -29,7 +29,7 @@ import type { Anthropic } from "@anthropic-ai/sdk" import { xaiDefaultModelId, xaiModels } from "@roo-code/types" import { XAIHandler } from "../xai" -import { asyncStreamFrom } from "../../../test-utils/stream" +import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" import { makeCreateMessageMetadata } from "../../../test-utils/api" import { clearAllMocks } from "../../../test-utils/reset" @@ -474,4 +474,30 @@ describe("XAIHandler", () => { controller.abort() await expect(promise).rejects.toMatchObject({ name: "AbortError" }) }) + + it("should remove the external abort listener when the stream completes", async () => { + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + mockResponsesCreate.mockResolvedValueOnce( + asyncStreamFrom([{ type: "response.output_text.delta", delta: "done" }]), + ) + + const stream = handler.createMessage( + "test prompt", + [], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const chunks = await collectStream(stream) + expect(chunks).toEqual([{ type: "text", text: "done" }]) + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // The same retained callback must be detached once the stream is done. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) }) diff --git a/src/api/providers/anthropic-vertex.ts b/src/api/providers/anthropic-vertex.ts index b5be64eff4..3f36720055 100644 --- a/src/api/providers/anthropic-vertex.ts +++ b/src/api/providers/anthropic-vertex.ts @@ -93,12 +93,18 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple // cancellation mechanism, preserving the existing behavior. const externalAbortSignal = metadata?.abortSignal let abortSignal: AbortSignal | undefined + let removeExternalAbortListener: (() => void) | undefined if (externalAbortSignal) { const controller = new AbortController() if (externalAbortSignal.aborted) { controller.abort() } else { - externalAbortSignal.addEventListener("abort", () => controller.abort(), { once: true }) + // Retain the listener so it can be removed again once streaming + // finishes; otherwise a long-lived external signal would keep one + // listener (and its closed-over controller) per completed request. + const onExternalAbort = () => controller.abort() + externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) + removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) } abortSignal = controller.signal } @@ -139,101 +145,107 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple requestOptions.signal = abortSignal } - const stream = await this.client.messages.create( - params, - Object.keys(requestOptions).length > 0 ? requestOptions : undefined, - ) + try { + const stream = await this.client.messages.create( + params, + Object.keys(requestOptions).length > 0 ? requestOptions : undefined, + ) - for await (const chunk of stream) { - switch (chunk.type) { - case "message_start": { - const usage = chunk.message!.usage - - yield { - type: "usage", - inputTokens: usage.input_tokens || 0, - outputTokens: usage.output_tokens || 0, - cacheWriteTokens: usage.cache_creation_input_tokens || undefined, - cacheReadTokens: usage.cache_read_input_tokens || undefined, - } + for await (const chunk of stream) { + switch (chunk.type) { + case "message_start": { + const usage = chunk.message!.usage + + yield { + type: "usage", + inputTokens: usage.input_tokens || 0, + outputTokens: usage.output_tokens || 0, + cacheWriteTokens: usage.cache_creation_input_tokens || undefined, + cacheReadTokens: usage.cache_read_input_tokens || undefined, + } - break - } - case "message_delta": { - yield { - type: "usage", - inputTokens: 0, - outputTokens: chunk.usage!.output_tokens || 0, + break } + case "message_delta": { + yield { + type: "usage", + inputTokens: 0, + outputTokens: chunk.usage!.output_tokens || 0, + } - break - } - case "content_block_start": { - switch (chunk.content_block!.type) { - case "text": { - if (chunk.index! > 0) { - yield { type: "text", text: "\n" } + break + } + case "content_block_start": { + switch (chunk.content_block!.type) { + case "text": { + if (chunk.index! > 0) { + yield { type: "text", text: "\n" } + } + + yield { type: "text", text: chunk.content_block!.text } + break } + case "thinking": { + if (chunk.index! > 0) { + yield { type: "reasoning", text: "\n" } + } - yield { type: "text", text: chunk.content_block!.text } - break - } - case "thinking": { - if (chunk.index! > 0) { - yield { type: "reasoning", text: "\n" } + yield { type: "reasoning", text: (chunk.content_block as any).thinking } + break } - - yield { type: "reasoning", text: (chunk.content_block as any).thinking } - break - } - case "tool_use": { - // Emit initial tool call partial with id and name - yield { - type: "tool_call_partial", - index: chunk.index, - id: chunk.content_block!.id, - name: chunk.content_block!.name, - arguments: undefined, + case "tool_use": { + // Emit initial tool call partial with id and name + yield { + type: "tool_call_partial", + index: chunk.index, + id: chunk.content_block!.id, + name: chunk.content_block!.name, + arguments: undefined, + } + break } - break } - } - break - } - case "content_block_delta": { - switch (chunk.delta!.type) { - case "text_delta": { - yield { type: "text", text: chunk.delta!.text } - break - } - case "thinking_delta": { - yield { type: "reasoning", text: (chunk.delta as any).thinking } - break - } - case "input_json_delta": { - // Emit tool call partial chunks as arguments stream in - yield { - type: "tool_call_partial", - index: chunk.index, - id: undefined, - name: undefined, - arguments: (chunk.delta as any).partial_json, + break + } + case "content_block_delta": { + switch (chunk.delta!.type) { + case "text_delta": { + yield { type: "text", text: chunk.delta!.text } + break + } + case "thinking_delta": { + yield { type: "reasoning", text: (chunk.delta as any).thinking } + break + } + case "input_json_delta": { + // Emit tool call partial chunks as arguments stream in + yield { + type: "tool_call_partial", + index: chunk.index, + id: undefined, + name: undefined, + arguments: (chunk.delta as any).partial_json, + } + break } - break } - } - break - } - case "content_block_stop": { - // Block complete - no action needed for now. - // NativeToolCallParser handles tool call completion - // Note: Signature for multi-turn thinking would require using stream.finalMessage() - // after iteration completes, which requires restructuring the streaming approach. - break + break + } + case "content_block_stop": { + // Block complete - no action needed for now. + // NativeToolCallParser handles tool call completion + // Note: Signature for multi-turn thinking would require using stream.finalMessage() + // after iteration completes, which requires restructuring the streaming approach. + break + } } } + } finally { + // Release the listener once the stream is consumed, whether the + // request completed, failed, or the generator was closed early. + removeExternalAbortListener?.() } } diff --git a/src/api/providers/xai.ts b/src/api/providers/xai.ts index 9b08745310..8afbc66a4f 100644 --- a/src/api/providers/xai.ts +++ b/src/api/providers/xai.ts @@ -102,74 +102,86 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler // cancellation mechanism, preserving the existing behavior. const externalAbortSignal = metadata?.abortSignal let abortSignal: AbortSignal | undefined + let removeExternalAbortListener: (() => void) | undefined if (externalAbortSignal) { const controller = new AbortController() if (externalAbortSignal.aborted) { controller.abort() } else { - externalAbortSignal.addEventListener("abort", () => controller.abort(), { once: true }) + // Retain the listener so it can be removed again once streaming + // finishes; otherwise a long-lived external signal would keep one + // listener (and its closed-over controller) per completed request. + const onExternalAbort = () => controller.abort() + externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) + removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) } abortSignal = controller.signal } - // Build request options - const requestBody: OpenAI.Responses.ResponseCreateParamsStreaming = { - model: model.id, - instructions: systemPrompt, - input: input, - stream: true, - store: false, // Don't store responses server-side for privacy - include: ["reasoning.encrypted_content"], - } + try { + // Build request options + const requestBody: OpenAI.Responses.ResponseCreateParamsStreaming = { + model: model.id, + instructions: systemPrompt, + input: input, + stream: true, + store: false, // Don't store responses server-side for privacy + include: ["reasoning.encrypted_content"], + } - if (model.maxTokens) { - requestBody.max_output_tokens = model.maxTokens - } + if (model.maxTokens) { + requestBody.max_output_tokens = model.maxTokens + } - if (model.temperature !== undefined) { - requestBody.temperature = model.temperature - } + if (model.temperature !== undefined) { + requestBody.temperature = model.temperature + } - if (responseTools) { - requestBody.tools = responseTools - // Cast tool_choice since metadata uses Chat Completions types but Responses API has its own type - requestBody.tool_choice = (metadata?.tool_choice ?? "auto") as any - requestBody.parallel_tool_calls = metadata?.parallelToolCalls ?? true - } + if (responseTools) { + requestBody.tools = responseTools + // Cast tool_choice since metadata uses Chat Completions types but Responses API has its own type + requestBody.tool_choice = (metadata?.tool_choice ?? "auto") as any + requestBody.parallel_tool_calls = metadata?.parallelToolCalls ?? true + } - // Pass reasoning effort for models that support it (e.g., grok-4.5, grok-3-mini). - // The xAI Responses API uses `reasoning: { effort }` format (not `reasoning_effort` - // which is the Chat Completions format), so we convert from the OpenAI params shape. - if (model.reasoning) { - requestBody.reasoning = { effort: model.reasoning.reasoning_effort } - } + // Pass reasoning effort for models that support it (e.g., grok-4.5, grok-3-mini). + // The xAI Responses API uses `reasoning: { effort }` format (not `reasoning_effort` + // which is the Chat Completions format), so we convert from the OpenAI params shape. + if (model.reasoning) { + requestBody.reasoning = { effort: model.reasoning.reasoning_effort } + } - let stream: AsyncIterable - try { - stream = await this.client.responses.create( - { - ...requestBody, - stream: true, - }, - abortSignal ? { signal: abortSignal } : undefined, - ) - } catch (error) { - // Let abort errors propagate unmodified so callers can recognize them: - // native AbortError (error.name === "AbortError") and the OpenAI SDK's - // APIUserAbortError, which the SDK throws when the request signal aborts - // (the SDK class does not set a distinctive error.name in v5, so use - // instanceof). - if ((error instanceof Error && error.name === "AbortError") || error instanceof APIUserAbortError) { - throw error + let stream: AsyncIterable + try { + stream = await this.client.responses.create( + { + ...requestBody, + stream: true, + }, + abortSignal ? { signal: abortSignal } : undefined, + ) + } catch (error) { + // Let abort errors propagate unmodified so callers can recognize them: + // native AbortError (error.name === "AbortError") and the OpenAI SDK's + // APIUserAbortError, which the SDK throws when the request signal aborts + // (the SDK class does not set a distinctive error.name in v5, so use + // instanceof). + if ((error instanceof Error && error.name === "AbortError") || error instanceof APIUserAbortError) { + throw error + } + const errorMessage = error instanceof Error ? error.message : String(error) + const apiError = new ApiProviderError(errorMessage, this.providerName, model.id, "createMessage") + TelemetryService.instance.captureException(apiError) + throw handleOpenAIError(error, 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) - throw handleOpenAIError(error, this.providerName) - } - const normalizeUsage = createUsageNormalizer() - yield* processResponsesApiStream(stream, normalizeUsage) + const normalizeUsage = createUsageNormalizer() + yield* processResponsesApiStream(stream, normalizeUsage) + } finally { + // Release the listener once the stream is consumed, whether the + // request completed, failed, or the generator was closed early. + removeExternalAbortListener?.() + } } async completePrompt(prompt: string, options?: CompletePromptOptions): Promise { From 8afead4e6938bf0b74a4267aa35909bb28a953a4 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 3 Sep 2026 05:04:44 +0800 Subject: [PATCH 06/18] refactor(api): map xai tool_choice to Responses API shape instead of casting --- src/api/providers/__tests__/xai.spec.ts | 29 ++++++++++++++++++++++ src/api/providers/xai.ts | 32 +++++++++++++++++++++++-- src/eslint-suppressions.json | 2 +- 3 files changed, 60 insertions(+), 3 deletions(-) diff --git a/src/api/providers/__tests__/xai.spec.ts b/src/api/providers/__tests__/xai.spec.ts index 4cb135973b..5c88539e5b 100644 --- a/src/api/providers/__tests__/xai.spec.ts +++ b/src/api/providers/__tests__/xai.spec.ts @@ -224,6 +224,35 @@ describe("XAIHandler", () => { ) }) + it("createMessage should map a forced tool_choice to the Responses API shape", async () => { + const testTools = [ + { + type: "function" as const, + function: { + name: "test_tool", + description: "A test tool", + parameters: { type: "object", properties: { arg1: { type: "string" } }, required: ["arg1"] }, + }, + }, + ] + + mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) + + const stream = handler.createMessage("test prompt", [], { + taskId: "test-task-id", + tools: testTools, + tool_choice: { type: "function", function: { name: "test_tool" } }, + }) + await stream.next() + + expect(mockResponsesCreate).toHaveBeenCalledWith( + expect.objectContaining({ + tool_choice: { type: "function", name: "test_tool" }, + }), + undefined, + ) + }) + it("completePrompt should return text from Responses API", async () => { const expectedResponse = "This is a test response" mockResponsesCreate.mockResolvedValueOnce({ diff --git a/src/api/providers/xai.ts b/src/api/providers/xai.ts index 8afbc66a4f..e01a124f3f 100644 --- a/src/api/providers/xai.ts +++ b/src/api/providers/xai.ts @@ -63,6 +63,33 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler * Uses base provider's convertToolSchemaForOpenAI() for schema hardening * (additionalProperties: false, ensureAllRequired) and handles MCP tools. */ + /** + * Map a Chat Completions tool choice to the Responses API shape so TypeScript + * validates the provider payload (the APIs use different object forms for the + * named-tool choices: Chat Completions nests the name under `function`/`custom`, + * Responses API puts it at the top level). String options (auto/required/none) + * are identical in both APIs and pass through unchanged. + */ + private mapToolChoice( + toolChoice: NonNullable, + ): OpenAI.Responses.ResponseCreateParamsStreaming["tool_choice"] { + if (typeof toolChoice === "string") { + return toolChoice + } + switch (toolChoice.type) { + case "function": + return { type: "function", name: toolChoice.function.name } + case "custom": + return { type: "custom", name: toolChoice.custom.name } + case "allowed_tools": + return { + type: "allowed_tools", + mode: toolChoice.allowed_tools.mode, + tools: toolChoice.allowed_tools.tools, + } + } + } + private mapResponseTools(tools?: any[]): any[] | undefined { const converted = this.convertToolsForOpenAI(tools) if (!converted?.length) { @@ -139,8 +166,9 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler if (responseTools) { requestBody.tools = responseTools - // Cast tool_choice since metadata uses Chat Completions types but Responses API has its own type - requestBody.tool_choice = (metadata?.tool_choice ?? "auto") as any + // Metadata carries a Chat Completions tool choice; the Responses API + // uses its own shape, so map it explicitly instead of casting. + requestBody.tool_choice = this.mapToolChoice(metadata?.tool_choice ?? "auto") requestBody.parallel_tool_calls = metadata?.parallelToolCalls ?? true } diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index 97d5076829..4ccae43858 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -451,7 +451,7 @@ }, "api/providers/xai.ts": { "@typescript-eslint/no-explicit-any": { - "count": 3 + "count": 2 } }, "api/transform/__tests__/ai-sdk.spec.ts": { From 9b23d306e82b66fa6151e1ffd5db3381221ac24d Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Thu, 3 Sep 2026 06:28:50 +0800 Subject: [PATCH 07/18] fix(api): flatten xai allowed_tools entries for the Responses API mapToolChoice forwarded Chat Completions allowlist entries ({ type: "function", function: { name } }) to responses.create, but the Responses API expects { type: "function", name }. Flatten function-shaped entries and leave other entry types untouched; add a regression test. --- src/api/providers/__tests__/xai.spec.ts | 45 +++++++++++++++++++++++++ src/api/providers/xai.ts | 22 ++++++++++-- 2 files changed, 65 insertions(+), 2 deletions(-) diff --git a/src/api/providers/__tests__/xai.spec.ts b/src/api/providers/__tests__/xai.spec.ts index 5c88539e5b..3365886ae1 100644 --- a/src/api/providers/__tests__/xai.spec.ts +++ b/src/api/providers/__tests__/xai.spec.ts @@ -253,6 +253,51 @@ describe("XAIHandler", () => { ) }) + it("createMessage should flatten allowed_tools entries to the Responses API shape", async () => { + const testTools = [ + { + type: "function" as const, + function: { + name: "test_tool", + description: "A test tool", + parameters: { type: "object", properties: { arg1: { type: "string" } }, required: ["arg1"] }, + }, + }, + ] + + mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) + + const stream = handler.createMessage("test prompt", [], { + taskId: "test-task-id", + tools: testTools, + tool_choice: { + type: "allowed_tools", + allowed_tools: { + mode: "required", + tools: [ + { type: "function", function: { name: "test_tool" } }, + { type: "mcp", server_label: "deepwiki" }, + ], + }, + }, + }) + await stream.next() + + expect(mockResponsesCreate).toHaveBeenCalledWith( + expect.objectContaining({ + tool_choice: { + type: "allowed_tools", + mode: "required", + tools: [ + { type: "function", name: "test_tool" }, + { type: "mcp", server_label: "deepwiki" }, + ], + }, + }), + undefined, + ) + }) + it("completePrompt should return text from Responses API", async () => { const expectedResponse = "This is a test response" mockResponsesCreate.mockResolvedValueOnce({ diff --git a/src/api/providers/xai.ts b/src/api/providers/xai.ts index e01a124f3f..edbc89a10c 100644 --- a/src/api/providers/xai.ts +++ b/src/api/providers/xai.ts @@ -68,7 +68,11 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler * validates the provider payload (the APIs use different object forms for the * named-tool choices: Chat Completions nests the name under `function`/`custom`, * Responses API puts it at the top level). String options (auto/required/none) - * are identical in both APIs and pass through unchanged. + * are identical in both APIs and pass through unchanged. `allowed_tools` + * entries keep their allowlist mode, but Chat Completions function references + * ({ type: "function", function: { name } }) are flattened to the Responses + * API shape ({ type: "function", name }); other entry types pass through + * unchanged. */ private mapToolChoice( toolChoice: NonNullable, @@ -85,7 +89,21 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler return { type: "allowed_tools", mode: toolChoice.allowed_tools.mode, - tools: toolChoice.allowed_tools.tools, + tools: toolChoice.allowed_tools.tools.map((entry) => { + // Chat Completions allowlist entries nest the function reference + // ({ type: "function", function: { name } }); the Responses API + // expects the name at the top level ({ type: "function", name }). + const functionRef = entry["function"] + if ( + entry["type"] === "function" && + functionRef != null && + typeof functionRef === "object" && + typeof (functionRef as Record)["name"] === "string" + ) { + return { type: "function", name: (functionRef as Record)["name"] } + } + return entry + }), } } } From a74161594eebdc7785559ff955f99f82cbe34517 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 12:54:21 +0800 Subject: [PATCH 08/18] test(api): cover abort-bridge and request-body mutation gaps Add focused tests that kill the surviving and uncovered changed-code mutants reported by the mutation-diff gate on this branch: - xai: custom tool_choice mapping, allowed_tools passthrough for unmappable entries, default-model max_output_tokens/temperature, omitted tool parameters without tools, native AbortError passthrough in completePrompt, exact Responses API request body, telemetry capture for createMessage failures, and the once option on the external abort listener. - anthropic: a pending external signal does not pre-abort the bridged signal, the external abort listener is registered with the once option, and the default-model beta header is comma-joined. - anthropic-vertex: message_start usage defaults outputTokens to zero when output_tokens is omitted. - minimax: completePrompt sends the configured temperature, user message and stream flag; the abort bridge does not pre-abort pending external signals and registers its listener with the once option. Local mutation-diff gate: 7 surviving and 1 uncovered changed-code mutants remain; all are unkillable without source changes (dead inner default in anthropic.ts, spread-overridden stream flag and always-true guards in xai.ts, single-element beta join and inert case label in anthropic-vertex.ts). --- .../__tests__/anthropic-vertex.spec.ts | 33 ++++ src/api/providers/__tests__/anthropic.spec.ts | 49 ++++++ src/api/providers/__tests__/minimax.spec.ts | 62 ++++++++ src/api/providers/__tests__/xai.spec.ts | 145 ++++++++++++++++++ 4 files changed, 289 insertions(+) diff --git a/src/api/providers/__tests__/anthropic-vertex.spec.ts b/src/api/providers/__tests__/anthropic-vertex.spec.ts index ba983f1fcd..7f21ef761f 100644 --- a/src/api/providers/__tests__/anthropic-vertex.spec.ts +++ b/src/api/providers/__tests__/anthropic-vertex.spec.ts @@ -837,6 +837,39 @@ describe("VertexHandler", () => { expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) }) + + it("should default message_start outputTokens to zero when output_tokens is omitted", async () => { + handler = new AnthropicVertexHandler({ + apiModelId: "claude-3-5-sonnet-v2@20241022", + vertexProjectId: "test-project", + vertexRegion: "us-central1", + }) + + const mockCreate = vitest.fn().mockImplementation(async () => + asyncStreamFrom([ + { + type: "message_start", + message: { + usage: { + input_tokens: 10, + }, + }, + }, + ]), + ) + handler["client"].messages.create = mockCreate + + const stream = handler.createMessage(systemPrompt, [{ role: "user", content: "Hello" }]) + const chunks = await collectStream(stream) + + expect(chunks[0]).toEqual({ + type: "usage", + inputTokens: 10, + outputTokens: 0, + cacheWriteTokens: undefined, + cacheReadTokens: undefined, + }) + }) }) describe("completePrompt", () => { diff --git a/src/api/providers/__tests__/anthropic.spec.ts b/src/api/providers/__tests__/anthropic.spec.ts index de23a5205a..2351c8063d 100644 --- a/src/api/providers/__tests__/anthropic.spec.ts +++ b/src/api/providers/__tests__/anthropic.spec.ts @@ -537,6 +537,55 @@ describe("AnthropicHandler", () => { controller.abort() await expect(promise).rejects.toMatchObject({ name: "AbortError" }) }) + + it("should not pre-abort the bridged signal for a pending external signal", async () => { + const controller = new AbortController() + + mockCreate.mockImplementationOnce(async () => asyncStreamFrom([])) + + const stream = handler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await collectStream(stream) + + const requestOptions = mockCreate.mock.calls[mockCreate.mock.calls.length - 1]?.[1] + expect(requestOptions?.signal?.aborted).toBe(false) + }) + + it("should register the external abort listener with the once option", async () => { + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + + mockCreate.mockImplementationOnce(async () => asyncStreamFrom([])) + + const stream = handler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await collectStream(stream) + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + // The listener is registered once so it detaches itself when the signal aborts. + expect(addEventListenerSpy.mock.calls[0][2]).toEqual({ once: true }) + }) + + it("should join beta headers with a comma for the default model", async () => { + mockCreate.mockImplementationOnce(async () => asyncStreamFrom([])) + + const stream = handler.createMessage(systemPrompt, [{ role: "user", content: "Hello" }]) + + await collectStream(stream) + + const requestOptions = mockCreate.mock.calls[mockCreate.mock.calls.length - 1]?.[1] + expect(requestOptions?.headers?.["anthropic-beta"]).toBe( + "fine-grained-tool-streaming-2025-05-14,prompt-caching-2024-07-31", + ) + }) }) describe("completePrompt", () => { diff --git a/src/api/providers/__tests__/minimax.spec.ts b/src/api/providers/__tests__/minimax.spec.ts index 08ac58591c..41dcd9bd0d 100644 --- a/src/api/providers/__tests__/minimax.spec.ts +++ b/src/api/providers/__tests__/minimax.spec.ts @@ -297,6 +297,29 @@ describe("MiniMaxHandler", () => { ) }) + it("completePrompt should send the configured temperature, user message and stream flag", async () => { + const temperatureHandler = new MiniMaxHandler({ + minimaxApiKey: "test-minimax-api-key", + modelTemperature: 0.5, + }) + mockCreate.mockResolvedValueOnce({ + content: [{ type: "text", text: "response" }], + }) + + await temperatureHandler.completePrompt("test prompt") + + expect(mockCreate).toHaveBeenCalledWith( + { + model: minimaxDefaultModelId, + max_tokens: 16_384, + temperature: 0.5, + messages: [{ role: "user", content: "test prompt" }], + stream: false, + }, + undefined, + ) + }) + it("createMessage should yield text content from stream", async () => { const testContent = "This is test content from MiniMax stream" @@ -495,6 +518,45 @@ describe("MiniMaxHandler", () => { controller.abort() await expect(promise).rejects.toMatchObject({ name: "AbortError" }) }) + + it("should not pre-abort the bridged signal for a pending external signal", async () => { + const controller = new AbortController() + let observedSignal: AbortSignal | undefined + + mockCreate.mockImplementation(async (_params: unknown, options?: { signal?: AbortSignal }) => { + observedSignal = options?.signal + return asyncStreamFrom([]) + }) + + const stream = handler.createMessage( + "system prompt", + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await expect(collectStream(stream)).resolves.toEqual([]) + + expect(observedSignal?.aborted).toBe(false) + }) + + it("should register the external abort listener with the once option", async () => { + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + + mockCreate.mockResolvedValueOnce(asyncStreamFrom([])) + + const stream = handler.createMessage( + "system prompt", + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await expect(collectStream(stream)).resolves.toEqual([]) + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + // The listener is registered once so it detaches itself when the signal aborts. + expect(addEventListenerSpy.mock.calls[0][2]).toEqual({ once: true }) + }) }) describe("Model Configuration", () => { diff --git a/src/api/providers/__tests__/xai.spec.ts b/src/api/providers/__tests__/xai.spec.ts index 3365886ae1..ea6a846bbd 100644 --- a/src/api/providers/__tests__/xai.spec.ts +++ b/src/api/providers/__tests__/xai.spec.ts @@ -298,6 +298,107 @@ describe("XAIHandler", () => { ) }) + it("createMessage should map a custom tool_choice to the Responses API shape", async () => { + const testTools = [ + { + type: "function" as const, + function: { + name: "test_tool", + description: "A test tool", + parameters: { type: "object", properties: { arg1: { type: "string" } }, required: ["arg1"] }, + }, + }, + ] + + mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) + + const stream = handler.createMessage("test prompt", [], { + taskId: "test-task-id", + tools: testTools, + tool_choice: { type: "custom", custom: { name: "custom_tool" } }, + }) + await stream.next() + + expect(mockResponsesCreate).toHaveBeenCalledWith( + expect.objectContaining({ + tool_choice: { type: "custom", name: "custom_tool" }, + }), + undefined, + ) + }) + + it("createMessage should pass through unmappable allowed_tools entries unchanged", async () => { + const testTools = [ + { + type: "function" as const, + function: { + name: "test_tool", + description: "A test tool", + parameters: { type: "object", properties: { arg1: { type: "string" } }, required: ["arg1"] }, + }, + }, + ] + + // Entries that fail the function-name guard must pass through unchanged: a + // non-function type carrying a function ref, a null ref, a function value + // (typeof "function", not "object", whose .name is still a string), and an + // object whose name is not a string. + const probeFunction = () => 1 + const entries = [ + { type: "mcp", function: { name: "leaky" } }, + { type: "function", function: null }, + { type: "function", function: probeFunction }, + { type: "function", function: { name: 123 } }, + ] + + mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) + + const stream = handler.createMessage("test prompt", [], { + taskId: "test-task-id", + tools: testTools, + tool_choice: { type: "allowed_tools", allowed_tools: { mode: "auto", tools: entries } }, + }) + await stream.next() + + expect(mockResponsesCreate).toHaveBeenCalledWith( + expect.objectContaining({ + tool_choice: { + type: "allowed_tools", + mode: "auto", + tools: entries, + }, + }), + undefined, + ) + }) + + it("createMessage should send max_output_tokens and temperature for the default model", async () => { + mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) + + const stream = handler.createMessage("test prompt", []) + await stream.next() + + expect(mockResponsesCreate).toHaveBeenCalledWith( + expect.objectContaining({ + max_output_tokens: 65_536, + temperature: 0, + }), + undefined, + ) + }) + + it("createMessage should omit tool parameters when no tools are provided", async () => { + mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) + + const stream = handler.createMessage("test prompt", []) + await stream.next() + + const callArgs = mockResponsesCreate.mock.calls[mockResponsesCreate.mock.calls.length - 1][0] + expect(callArgs).not.toHaveProperty("tools") + expect(callArgs).not.toHaveProperty("tool_choice") + expect(callArgs).not.toHaveProperty("parallel_tool_calls") + }) + it("completePrompt should return text from Responses API", async () => { const expectedResponse = "This is a test response" mockResponsesCreate.mockResolvedValueOnce({ @@ -327,6 +428,17 @@ describe("XAIHandler", () => { ) }) + it("completePrompt should surface a native AbortError unmodified on abort", async () => { + const controller = new AbortController() + controller.abort() + const abortError = new Error("This operation was aborted") + abortError.name = "AbortError" + mockResponsesCreate.mockRejectedValueOnce(abortError) + + // The error must surface as the same instance, not wrapped by handleOpenAIError + await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toBe(abortError) + }) + it("completePrompt should pass abort signal through to client", async () => { const controller = new AbortController() mockResponsesCreate.mockResolvedValueOnce({ output_text: "response" }) @@ -348,6 +460,21 @@ describe("XAIHandler", () => { ) }) + it("completePrompt should send the full Responses API request body", async () => { + mockResponsesCreate.mockResolvedValueOnce({ output_text: "response" }) + + await handler.completePrompt("test prompt") + + expect(mockResponsesCreate).toHaveBeenCalledWith( + { + model: xaiDefaultModelId, + input: [{ role: "user", content: [{ type: "input_text", text: "test prompt" }] }], + store: false, + }, + undefined, + ) + }) + it("completePrompt should pass timeout through to client", async () => { const controller = new AbortController() mockResponsesCreate.mockResolvedValueOnce({ output_text: "response" }) @@ -465,6 +592,22 @@ describe("XAIHandler", () => { await expect(stream.next()).rejects.toThrow(`xAI completion error: ${errorMessage}`) }) + it("should capture createMessage failures in telemetry with operation context", async () => { + mockResponsesCreate.mockRejectedValueOnce(new Error("Stream error")) + + const stream = handler.createMessage("test prompt", []) + await expect(stream.next()).rejects.toThrow("xAI completion error: Stream error") + + expect(mockCaptureException).toHaveBeenCalledWith( + expect.objectContaining({ + name: "ApiProviderError", + provider: "xAI", + modelId: xaiDefaultModelId, + operation: "createMessage", + }), + ) + }) + it("createMessage should surface the SDK APIUserAbortError unmodified on abort", async () => { const abortedController = new AbortController() abortedController.abort() @@ -570,6 +713,8 @@ describe("XAIHandler", () => { expect(addEventListenerSpy).toHaveBeenCalledTimes(1) const [event, listener] = addEventListenerSpy.mock.calls[0] expect(event).toBe("abort") + // The listener is registered once so it detaches itself when the signal aborts. + expect(addEventListenerSpy.mock.calls[0][2]).toEqual({ once: true }) // The same retained callback must be detached once the stream is done. expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) From 262ec50a8ac1d04c356c084e9d666697c073c407 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 13:55:22 +0800 Subject: [PATCH 09/18] fix(api): remove dead code flagged by the mutation gate in anthropic, xai, and anthropic-vertex - anthropic.ts: collapse the inner prompt-caching switch, whose 19 case branches all executed an identical body plus an unreachable default, into that body: push the prompt-caching beta, build the anthropic-beta header, and attach the abort signal once before messages.create. Pinned by anthropic.spec.ts (comma-joined beta header and pending-abort-signal assertions). Deviation: tsconfig noImplicitReturns forbids a string switch without a default, so the identical branches were collapsed instead of deleting the default alone. - xai.ts: drop the call-site `stream: true` spread override from responses.create (the ResponseCreateParamsStreaming type requires the flag in the request body, so the body literal keeps it and is what is sent); collapse the model.maxTokens and model.temperature guards into unconditional assignments; bind metadata?.tool_choice and metadata?.parallelToolCalls once at the top of createMessage instead of re-reading the metadata at the use sites. Pinned by xai.spec.ts exact body assertions (stream: true, max_output_tokens 65_536, temperature 0, tool_choice "auto", parallel_tool_calls true). - anthropic-vertex.ts: send the single beta label directly (betas[0]) instead of building an array and joining it, and delete the no-op content_block_stop case. Pinned by anthropic-vertex.spec.ts exact anthropic-beta header assertions. - xai.spec.ts: add a createMessage test for parallelToolCalls: false so the defaulting of parallel_tool_calls stays pinned to an exact value. --- src/api/providers/__tests__/xai.spec.ts | 29 +++++++++++++++ src/api/providers/anthropic-vertex.ts | 10 ++---- src/api/providers/anthropic.ts | 48 +++++++------------------ src/api/providers/xai.ts | 21 +++++------ 4 files changed, 52 insertions(+), 56 deletions(-) diff --git a/src/api/providers/__tests__/xai.spec.ts b/src/api/providers/__tests__/xai.spec.ts index ea6a846bbd..fa2667b619 100644 --- a/src/api/providers/__tests__/xai.spec.ts +++ b/src/api/providers/__tests__/xai.spec.ts @@ -224,6 +224,35 @@ describe("XAIHandler", () => { ) }) + it("createMessage should honor an explicit parallelToolCalls false", async () => { + const testTools = [ + { + type: "function" as const, + function: { + name: "test_tool", + description: "A test Tool", + parameters: { type: "object", properties: { arg1: { type: "string" } }, required: ["arg1"] }, + }, + }, + ] + + mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) + + const stream = handler.createMessage("test prompt", [], { + taskId: "test-task-id", + tools: testTools, + parallelToolCalls: false, + }) + await stream.next() + + expect(mockResponsesCreate).toHaveBeenCalledWith( + expect.objectContaining({ + parallel_tool_calls: false, + }), + undefined, + ) + }) + it("createMessage should map a forced tool_choice to the Responses API shape", async () => { const testTools = [ { diff --git a/src/api/providers/anthropic-vertex.ts b/src/api/providers/anthropic-vertex.ts index 3f36720055..0c925989ef 100644 --- a/src/api/providers/anthropic-vertex.ts +++ b/src/api/providers/anthropic-vertex.ts @@ -139,7 +139,8 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple // and prompt caching const requestOptions: Anthropic.RequestOptions = {} if (betas?.length) { - requestOptions.headers = { "anthropic-beta": betas.join(",") } + // Model definitions carry at most one beta label, so use it directly. + requestOptions.headers = { "anthropic-beta": betas[0] } } if (abortSignal) { requestOptions.signal = abortSignal @@ -233,13 +234,6 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple break } - case "content_block_stop": { - // Block complete - no action needed for now. - // NativeToolCallParser handles tool call completion - // Note: Signature for multi-turn thinking would require using stream.finalMessage() - // after iteration completes, which requires restructuring the streaming approach. - break - } } } } finally { diff --git a/src/api/providers/anthropic.ts b/src/api/providers/anthropic.ts index 07a1c0e8f8..7e53b47a32 100644 --- a/src/api/providers/anthropic.ts +++ b/src/api/providers/anthropic.ts @@ -195,43 +195,21 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa stream: true, ...nativeToolParams, } + // prompt caching: https://x.com/alexalbert__/status/1823751995901272068 + // https://github.com/anthropics/anthropic-sdk-typescript?tab=readme-ov-file#default-headers + // https://github.com/anthropics/anthropic-sdk-typescript/commit/c920b77fc67bd839bfeb6716ceab9d7c9bbe7393 + // Every model that reaches this branch supports prompt caching, so the + // beta applies to all of them. + betas.push("prompt-caching-2024-07-31") + const requestOptions: Anthropic.RequestOptions = { + headers: { "anthropic-beta": betas.join(",") }, + } + if (abortSignal) { + requestOptions.signal = abortSignal + } stream = await this.client.messages.create( requestParams as Anthropic.Messages.MessageCreateParamsStreaming, - (() => { - // prompt caching: https://x.com/alexalbert__/status/1823751995901272068 - // https://github.com/anthropics/anthropic-sdk-typescript?tab=readme-ov-file#default-headers - // https://github.com/anthropics/anthropic-sdk-typescript/commit/c920b77fc67bd839bfeb6716ceab9d7c9bbe7393 - - // Then check for models that support prompt caching - switch (modelId) { - case "claude-sonnet-5": - case "claude-sonnet-4-6": - case "claude-sonnet-4-5": - case "claude-sonnet-4-20250514": - case "claude-opus-4-6": - case "claude-opus-4-7": - case "claude-opus-4-8": - case "claude-opus-5": - case "claude-fable-5-1": - case "claude-fable-5": - case "claude-opus-4-5-20251101": - case "claude-opus-4-1-20250805": - case "claude-opus-4-20250514": - case "claude-3-7-sonnet-20250219": - case "claude-3-5-sonnet-20241022": - case "claude-3-5-haiku-20241022": - case "claude-3-opus-20240229": - case "claude-haiku-4-5-20251001": - case "claude-3-haiku-20240307": - betas.push("prompt-caching-2024-07-31") - return { - headers: { "anthropic-beta": betas.join(",") }, - ...(abortSignal && { signal: abortSignal }), - } - default: - return abortSignal ? { signal: abortSignal } : undefined - } - })(), + requestOptions, ) } catch (error) { TelemetryService.instance.captureException( diff --git a/src/api/providers/xai.ts b/src/api/providers/xai.ts index edbc89a10c..50f34a358f 100644 --- a/src/api/providers/xai.ts +++ b/src/api/providers/xai.ts @@ -139,6 +139,8 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler // Convert directly from Anthropic format to Responses API input format const input = convertToResponsesApiInput(messages) const responseTools = this.mapResponseTools(metadata?.tools) + const toolChoice = metadata?.tool_choice + const parallelToolCalls = metadata?.parallelToolCalls // Bridge the external abort signal from request metadata into a per-request // controller so the SDK call is cancelled when the owning request is aborted @@ -174,20 +176,16 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler include: ["reasoning.encrypted_content"], } - if (model.maxTokens) { - requestBody.max_output_tokens = model.maxTokens - } - - if (model.temperature !== undefined) { - requestBody.temperature = model.temperature - } + // Model params are always resolved by getModel(); send them unconditionally. + requestBody.max_output_tokens = model.maxTokens + requestBody.temperature = model.temperature if (responseTools) { requestBody.tools = responseTools // Metadata carries a Chat Completions tool choice; the Responses API // uses its own shape, so map it explicitly instead of casting. - requestBody.tool_choice = this.mapToolChoice(metadata?.tool_choice ?? "auto") - requestBody.parallel_tool_calls = metadata?.parallelToolCalls ?? true + requestBody.tool_choice = this.mapToolChoice(toolChoice === undefined ? "auto" : toolChoice) + requestBody.parallel_tool_calls = parallelToolCalls ?? true } // Pass reasoning effort for models that support it (e.g., grok-4.5, grok-3-mini). @@ -200,10 +198,7 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler let stream: AsyncIterable try { stream = await this.client.responses.create( - { - ...requestBody, - stream: true, - }, + requestBody, abortSignal ? { signal: abortSignal } : undefined, ) } catch (error) { From 3d92d06561d7dfd49aa1017eb7b111371842a352 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sat, 5 Sep 2026 14:19:37 +0800 Subject: [PATCH 10/18] fix(api): attach anthropic abort signal via spread to drop the surviving guard mutant --- src/api/providers/anthropic.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/api/providers/anthropic.ts b/src/api/providers/anthropic.ts index 7e53b47a32..f59a6b20ab 100644 --- a/src/api/providers/anthropic.ts +++ b/src/api/providers/anthropic.ts @@ -203,9 +203,7 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa betas.push("prompt-caching-2024-07-31") const requestOptions: Anthropic.RequestOptions = { headers: { "anthropic-beta": betas.join(",") }, - } - if (abortSignal) { - requestOptions.signal = abortSignal + ...(abortSignal && { signal: abortSignal }), } stream = await this.client.messages.create( requestParams as Anthropic.Messages.MessageCreateParamsStreaming, From 895660b540aeb190ab84ef5394236f4ebc01ed3a Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Sun, 6 Sep 2026 00:10:35 +0800 Subject: [PATCH 11/18] fix(api): detach external abort listeners when anthropic/minimax requests end --- src/api/providers/__tests__/anthropic.spec.ts | 94 +++++++++++++++++++ src/api/providers/__tests__/minimax.spec.ts | 54 +++++++++++ .../__tests__/stream-cleanup.spec.ts | 52 ++++++++++ src/api/providers/anthropic.ts | 17 +++- src/api/providers/minimax.ts | 24 +++-- src/api/providers/stream-cleanup.ts | 19 ++++ 6 files changed, 252 insertions(+), 8 deletions(-) create mode 100644 src/api/providers/__tests__/stream-cleanup.spec.ts create mode 100644 src/api/providers/stream-cleanup.ts diff --git a/src/api/providers/__tests__/anthropic.spec.ts b/src/api/providers/__tests__/anthropic.spec.ts index f1ba8b5b0d..f6a47d31fe 100644 --- a/src/api/providers/__tests__/anthropic.spec.ts +++ b/src/api/providers/__tests__/anthropic.spec.ts @@ -599,6 +599,100 @@ describe("AnthropicHandler", () => { expect(addEventListenerSpy.mock.calls[0][2]).toEqual({ once: true }) }) + it("should remove the external abort listener when the stream completes", async () => { + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + mockCreate.mockImplementationOnce(async () => asyncStreamFrom([])) + + const stream = handler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await collectStream(stream) + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // The same retained callback must be detached once the stream is done. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) + + it("should remove the external abort listener when message creation fails", async () => { + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + mockCreate.mockRejectedValueOnce(new Error("Anthropic createMessage error")) + + const stream = handler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await expect(collectStream(stream)).rejects.toThrow("Anthropic createMessage error") + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // The same retained callback must be detached when creation fails. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) + + it("should propagate the creation error when no external abort signal is provided", async () => { + mockCreate.mockRejectedValueOnce(new Error("Anthropic createMessage error")) + + const stream = handler.createMessage(systemPrompt, [{ role: "user", content: "Hello" }]) + + await expect(collectStream(stream)).rejects.toThrow("Anthropic createMessage error") + }) + + it("should remove the external abort listener when message creation fails for a non-cached model", async () => { + const customHandler = new AnthropicHandler({ + apiKey: "test-api-key", + apiModelId: "claude-sonnet-5-bf", + }) + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + mockCreate.mockRejectedValueOnce(new Error("Anthropic createMessage error")) + + const stream = customHandler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await expect(collectStream(stream)).rejects.toThrow("Anthropic createMessage error") + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // The same retained callback must be detached when creation fails. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) + + it("should propagate the creation error for a non-cached model when no external abort signal is provided", async () => { + const customHandler = new AnthropicHandler({ + apiKey: "test-api-key", + apiModelId: "claude-sonnet-5-bf", + }) + + mockCreate.mockRejectedValueOnce(new Error("Anthropic createMessage error")) + + const stream = customHandler.createMessage(systemPrompt, [{ role: "user", content: "Hello" }]) + + await expect(collectStream(stream)).rejects.toThrow("Anthropic createMessage error") + }) + it("should join beta headers with a comma for the default model", async () => { mockCreate.mockImplementationOnce(async () => asyncStreamFrom([])) diff --git a/src/api/providers/__tests__/minimax.spec.ts b/src/api/providers/__tests__/minimax.spec.ts index 41dcd9bd0d..be2343f804 100644 --- a/src/api/providers/__tests__/minimax.spec.ts +++ b/src/api/providers/__tests__/minimax.spec.ts @@ -557,6 +557,60 @@ describe("MiniMaxHandler", () => { // The listener is registered once so it detaches itself when the signal aborts. expect(addEventListenerSpy.mock.calls[0][2]).toEqual({ once: true }) }) + + it("should remove the external abort listener when the stream completes", async () => { + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + mockCreate.mockResolvedValueOnce(asyncStreamFrom([])) + + const stream = handler.createMessage( + "system prompt", + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await expect(collectStream(stream)).resolves.toEqual([]) + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // The same retained callback must be detached once the stream is done. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) + + it("should remove the external abort listener when message creation fails", async () => { + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + mockCreate.mockRejectedValueOnce(new Error("MiniMax createMessage error")) + + const stream = handler.createMessage( + "system prompt", + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await expect(collectStream(stream)).rejects.toThrow("MiniMax createMessage error") + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // The same retained callback must be detached when creation fails. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) + + it("should propagate the creation error when no external abort signal is provided", async () => { + mockCreate.mockRejectedValueOnce(new Error("MiniMax createMessage error")) + + const stream = handler.createMessage("system prompt", [{ role: "user", content: "Hello" }]) + + await expect(collectStream(stream)).rejects.toThrow("MiniMax createMessage error") + }) }) describe("Model Configuration", () => { diff --git a/src/api/providers/__tests__/stream-cleanup.spec.ts b/src/api/providers/__tests__/stream-cleanup.spec.ts new file mode 100644 index 0000000000..677ef62c8f --- /dev/null +++ b/src/api/providers/__tests__/stream-cleanup.spec.ts @@ -0,0 +1,52 @@ +// npx vitest run src/api/providers/__tests__/stream-cleanup.spec.ts + +import { withFinallyCleanup } from "../stream-cleanup" +import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" + +describe("withFinallyCleanup", () => { + it("forwards every chunk and runs the cleanup after the stream completes", async () => { + const cleanup = vitest.fn() + + const chunks = await collectStream(withFinallyCleanup(asyncStreamFrom(["a", "b", "c"]), cleanup)) + + expect(chunks).toEqual(["a", "b", "c"]) + expect(cleanup).toHaveBeenCalledTimes(1) + }) + + it("runs the cleanup when the consumer stops early", async () => { + const cleanup = vitest.fn() + const stream = withFinallyCleanup(asyncStreamFrom(["a", "b", "c"]), cleanup) + const iterator = stream[Symbol.asyncIterator]() + + expect((await iterator.next()).value).toBe("a") + await iterator.return(undefined) + + expect(cleanup).toHaveBeenCalledTimes(1) + }) + + it("runs the cleanup when the underlying stream fails", async () => { + const cleanup = vitest.fn() + async function* failing() { + yield "a" + throw new Error("boom") + } + + await expect(collectStream(withFinallyCleanup(failing(), cleanup))).rejects.toThrow("boom") + expect(cleanup).toHaveBeenCalledTimes(1) + }) + + it("runs the cleanup at most once when the consumer stops after a failure", async () => { + const cleanup = vitest.fn() + async function* failing() { + yield* [] + throw new Error("boom") + } + const stream = withFinallyCleanup(failing(), cleanup) + const iterator = stream[Symbol.asyncIterator]() + + await expect(iterator.next()).rejects.toThrow("boom") + await iterator.return(undefined) + + expect(cleanup).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/api/providers/anthropic.ts b/src/api/providers/anthropic.ts index f59a6b20ab..7e594fc306 100644 --- a/src/api/providers/anthropic.ts +++ b/src/api/providers/anthropic.ts @@ -1,6 +1,7 @@ import { Anthropic } from "@anthropic-ai/sdk" import { Stream as AnthropicStream } from "@anthropic-ai/sdk/streaming" import { CacheControlEphemeral } from "@anthropic-ai/sdk/resources" +import { withFinallyCleanup } from "./stream-cleanup" import OpenAI from "openai" import { @@ -120,12 +121,18 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa // cancellation mechanism, preserving the existing behavior. const externalAbortSignal = metadata?.abortSignal let abortSignal: AbortSignal | undefined + let removeExternalAbortListener: (() => void) | undefined if (externalAbortSignal) { const controller = new AbortController() if (externalAbortSignal.aborted) { controller.abort() } else { - externalAbortSignal.addEventListener("abort", () => controller.abort(), { once: true }) + // Retain the callback so the listener can be detached once this request + // ends: an anonymous listener on a long-lived external signal would + // outlive the request and retain the per-request controller. + const onExternalAbort = () => controller.abort() + externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) + removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) } abortSignal = controller.signal } @@ -218,6 +225,9 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa "createMessage", ), ) + // The request failed before streaming: detach the bridged listener so + // it cannot outlive this request. + removeExternalAbortListener?.() throw error } break @@ -247,6 +257,9 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa "createMessage", ), ) + // The request failed before streaming: detach the bridged listener so + // it cannot outlive this request. + removeExternalAbortListener?.() throw error } break @@ -258,7 +271,7 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa let cacheWriteTokens = 0 let cacheReadTokens = 0 - for await (const chunk of stream) { + for await (const chunk of withFinallyCleanup(stream, removeExternalAbortListener)) { switch (chunk.type) { case "message_start": { // Tells us cache reads/writes/input/output. diff --git a/src/api/providers/minimax.ts b/src/api/providers/minimax.ts index 9e9c219a25..c5a4c3f5c5 100644 --- a/src/api/providers/minimax.ts +++ b/src/api/providers/minimax.ts @@ -1,6 +1,7 @@ import { Anthropic } from "@anthropic-ai/sdk" import { Stream as AnthropicStream } from "@anthropic-ai/sdk/streaming" import { CacheControlEphemeral } from "@anthropic-ai/sdk/resources" +import { withFinallyCleanup } from "./stream-cleanup" import OpenAI from "openai" import { type MinimaxModelId, minimaxDefaultModelId, minimaxModels } from "@roo-code/types" @@ -92,12 +93,18 @@ export class MiniMaxHandler extends BaseProvider implements SingleCompletionHand // cancellation mechanism, preserving the existing behavior. const externalAbortSignal = metadata?.abortSignal let abortSignal: AbortSignal | undefined + let removeExternalAbortListener: (() => void) | undefined if (externalAbortSignal) { const controller = new AbortController() if (externalAbortSignal.aborted) { controller.abort() } else { - externalAbortSignal.addEventListener("abort", () => controller.abort(), { once: true }) + // Retain the callback so the listener can be detached once this request + // ends: an anonymous listener on a long-lived external signal would + // outlive the request and retain the per-request controller. + const onExternalAbort = () => controller.abort() + externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) + removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) } abortSignal = controller.signal } @@ -130,17 +137,22 @@ export class MiniMaxHandler extends BaseProvider implements SingleCompletionHand tool_choice: convertOpenAIToolChoice(metadata?.tool_choice), } - const stream = await this.client.messages.create( - requestParams, - abortSignal ? { signal: abortSignal } : undefined, - ) + let stream: AnthropicStream + try { + stream = await this.client.messages.create(requestParams, abortSignal ? { signal: abortSignal } : undefined) + } catch (error) { + // Creation failed before streaming: detach the bridged listener so it + // cannot outlive this request. + removeExternalAbortListener?.() + throw error + } let inputTokens = 0 let outputTokens = 0 let cacheWriteTokens = 0 let cacheReadTokens = 0 - for await (const chunk of stream) { + for await (const chunk of withFinallyCleanup(stream, removeExternalAbortListener)) { switch (chunk.type) { case "message_start": { // Tells us cache reads/writes/input/output. diff --git a/src/api/providers/stream-cleanup.ts b/src/api/providers/stream-cleanup.ts new file mode 100644 index 0000000000..d20168c954 --- /dev/null +++ b/src/api/providers/stream-cleanup.ts @@ -0,0 +1,19 @@ +/** + * Forward chunks from an async iterable and run the cleanup callback once when + * iteration ends: normal completion, a failure from the underlying stream, or + * an early stop by the consumer. Providers use this to detach a bridged + * external abort listener as soon as the request's stream has finished, + * keeping the cleanup out of the (existing) stream loop body. + */ +export async function* withFinallyCleanup( + stream: AsyncIterable, + cleanup: (() => void) | undefined, +): AsyncGenerator { + try { + for await (const chunk of stream) { + yield chunk + } + } finally { + cleanup?.() + } +} From ac0e15110361691902a0af60ef34e99dbc1ed396 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Mon, 28 Sep 2026 03:05:45 +0800 Subject: [PATCH 12/18] test: assert non-cached Anthropic path rejects with AbortError on external abort --- src/api/providers/__tests__/anthropic.spec.ts | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/src/api/providers/__tests__/anthropic.spec.ts b/src/api/providers/__tests__/anthropic.spec.ts index ea2c4b07c2..f38b6a4552 100644 --- a/src/api/providers/__tests__/anthropic.spec.ts +++ b/src/api/providers/__tests__/anthropic.spec.ts @@ -705,6 +705,53 @@ describe("AnthropicHandler", () => { expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) }) + it("should reject with AbortError when the external signal aborts a non-cached model mid-flight", async () => { + const customHandler = new AnthropicHandler({ + apiKey: "test-api-key", + apiModelId: "claude-sonnet-5-bf", + }) + const controller = new AbortController() + + mockCreate.mockImplementation((_params: unknown, options?: { signal?: AbortSignal }) => { + return new Promise((_resolve, reject) => { + const signal = options?.signal + if (!signal) { + return + } + if (signal.aborted) { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + return + } + signal.addEventListener( + "abort", + () => { + const error = new Error("The operation was aborted") + error.name = "AbortError" + reject(error) + }, + { once: true }, + ) + }) + }) + + const stream = customHandler.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + const promise = stream.next() + controller.abort() + await expect(promise).rejects.toMatchObject({ name: "AbortError" }) + + // The non-cached path must hand the bridged signal to the SDK. + const requestOptions = mockCreate.mock.calls[mockCreate.mock.calls.length - 1]?.[1] + expect(requestOptions?.signal).toBeDefined() + expect(requestOptions?.signal?.aborted).toBe(true) + }) + it("should propagate the creation error for a non-cached model when no external abort signal is provided", async () => { const customHandler = new AnthropicHandler({ apiKey: "test-api-key", From 8521fa8c94426f5e5dfd5c6e6e060141d860ceb6 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Tue, 13 Oct 2026 10:43:35 +0800 Subject: [PATCH 13/18] Address review: shared abort bridge, canonical xAI aborts, beta join, cleanup coverage - Extract the four copies of the external-abort bridge (anthropic, anthropic-vertex, minimax, xai) into bridgeExternalAbortSignal in utils/abort-signal.ts (single canonical comment, unit-tested incl. a negative control that dispose detaches the exact retained callback). - xai: normalize cancellations through isRequestAborted + createAbortError in both createMessage and completePrompt; the raw OpenAI SDK APIUserAbortError (name "Error", message "Request was aborted.") no longer escapes to callers' AbortError guards. - anthropic-vertex: send anthropic-beta as betas.join(",") again so a second beta added to the builder later is not silently dropped. - Move the xAI tool-choice mapping (mapToolChoice + typed request body + unconditional max_output_tokens/temperature) and the vertex content_block_stop removal out of this abort-wiring unit per review: the hunks return to a separate PR; xai.ts keeps its main-side lint suppression count (7) accordingly. - Pin listener cleanup beyond normal completion: vertex and xai specs now cover create-rejects and consumer-early-return paths (vertex uses an inline finally, so stream-cleanup coverage does not reach it). Verification at this head: 273/273 green across the six touched specs; tsc --noEmit 0 errors; eslint --max-warnings=0 clean on touched files; prettier clean. --- .../__tests__/anthropic-vertex.spec.ts | 69 ++++++ src/api/providers/__tests__/xai.spec.ts | 234 ++++++------------ src/api/providers/anthropic-vertex.ts | 31 +-- src/api/providers/anthropic.ts | 19 +- src/api/providers/minimax.ts | 19 +- .../utils/__tests__/abort-signal.spec.ts | 47 ++++ src/api/providers/utils/abort-signal.ts | 39 +++ src/api/providers/xai.ts | 153 ++++-------- src/eslint-suppressions.json | 2 +- 9 files changed, 297 insertions(+), 316 deletions(-) diff --git a/src/api/providers/__tests__/anthropic-vertex.spec.ts b/src/api/providers/__tests__/anthropic-vertex.spec.ts index b3227f3e70..917527cd51 100644 --- a/src/api/providers/__tests__/anthropic-vertex.spec.ts +++ b/src/api/providers/__tests__/anthropic-vertex.spec.ts @@ -838,6 +838,75 @@ describe("VertexHandler", () => { expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) }) + it("should remove the external abort listener when message creation fails", async () => { + const handlerWithSignal = new AnthropicVertexHandler({ + apiModelId: "claude-3-5-sonnet-v2@20241022", + vertexProjectId: "test-project", + vertexRegion: "us-central1", + }) + + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + handlerWithSignal["client"].messages.create = vitest + .fn() + .mockRejectedValue(new Error("vertex create error")) + + const stream = handlerWithSignal.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await expect(collectStream(stream)).rejects.toThrow("vertex create error") + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // The inline finally must detach the listener on the failure path too, + // not only on normal completion. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) + + it("should remove the external abort listener when the consumer returns early", async () => { + const handlerWithSignal = new AnthropicVertexHandler({ + apiModelId: "claude-3-5-sonnet-v2@20241022", + vertexProjectId: "test-project", + vertexRegion: "us-central1", + }) + + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + handlerWithSignal["client"].messages.create = vitest.fn().mockImplementation(async () => + asyncStreamFrom([ + { type: "message_start", message: { usage: { input_tokens: 1, output_tokens: 0 } } }, + { type: "content_block_delta", delta: { type: "text_delta", text: "first" } }, + { type: "content_block_delta", delta: { type: "text_delta", text: "second" } }, + ]), + ) + + const stream = handlerWithSignal.createMessage( + systemPrompt, + [{ role: "user", content: "Hello" }], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + const iterator = stream[Symbol.asyncIterator]() + + await iterator.next() + await iterator.return(undefined) + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // Abandoning the generator mid-stream runs the finally block too. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) + it("should default message_start outputTokens to zero when output_tokens is omitted", async () => { handler = new AnthropicVertexHandler({ apiModelId: "claude-3-5-sonnet-v2@20241022", diff --git a/src/api/providers/__tests__/xai.spec.ts b/src/api/providers/__tests__/xai.spec.ts index 1fcca4e700..4686285920 100644 --- a/src/api/providers/__tests__/xai.spec.ts +++ b/src/api/providers/__tests__/xai.spec.ts @@ -268,154 +268,6 @@ describe("XAIHandler", () => { ) }) - it("createMessage should map a forced tool_choice to the Responses API shape", async () => { - const testTools = [ - { - type: "function" as const, - function: { - name: "test_tool", - description: "A test tool", - parameters: { type: "object", properties: { arg1: { type: "string" } }, required: ["arg1"] }, - }, - }, - ] - - mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) - - const stream = handler.createMessage("test prompt", [], { - taskId: "test-task-id", - tools: testTools, - tool_choice: { type: "function", function: { name: "test_tool" } }, - }) - await stream.next() - - expect(mockResponsesCreate).toHaveBeenCalledWith( - expect.objectContaining({ - tool_choice: { type: "function", name: "test_tool" }, - }), - undefined, - ) - }) - - it("createMessage should flatten allowed_tools entries to the Responses API shape", async () => { - const testTools = [ - { - type: "function" as const, - function: { - name: "test_tool", - description: "A test tool", - parameters: { type: "object", properties: { arg1: { type: "string" } }, required: ["arg1"] }, - }, - }, - ] - - mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) - - const stream = handler.createMessage("test prompt", [], { - taskId: "test-task-id", - tools: testTools, - tool_choice: { - type: "allowed_tools", - allowed_tools: { - mode: "required", - tools: [ - { type: "function", function: { name: "test_tool" } }, - { type: "mcp", server_label: "deepwiki" }, - ], - }, - }, - }) - await stream.next() - - expect(mockResponsesCreate).toHaveBeenCalledWith( - expect.objectContaining({ - tool_choice: { - type: "allowed_tools", - mode: "required", - tools: [ - { type: "function", name: "test_tool" }, - { type: "mcp", server_label: "deepwiki" }, - ], - }, - }), - undefined, - ) - }) - - it("createMessage should map a custom tool_choice to the Responses API shape", async () => { - const testTools = [ - { - type: "function" as const, - function: { - name: "test_tool", - description: "A test tool", - parameters: { type: "object", properties: { arg1: { type: "string" } }, required: ["arg1"] }, - }, - }, - ] - - mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) - - const stream = handler.createMessage("test prompt", [], { - taskId: "test-task-id", - tools: testTools, - tool_choice: { type: "custom", custom: { name: "custom_tool" } }, - }) - await stream.next() - - expect(mockResponsesCreate).toHaveBeenCalledWith( - expect.objectContaining({ - tool_choice: { type: "custom", name: "custom_tool" }, - }), - undefined, - ) - }) - - it("createMessage should pass through unmappable allowed_tools entries unchanged", async () => { - const testTools = [ - { - type: "function" as const, - function: { - name: "test_tool", - description: "A test tool", - parameters: { type: "object", properties: { arg1: { type: "string" } }, required: ["arg1"] }, - }, - }, - ] - - // Entries that fail the function-name guard must pass through unchanged: a - // non-function type carrying a function ref, a null ref, a function value - // (typeof "function", not "object", whose .name is still a string), and an - // object whose name is not a string. - const probeFunction = () => 1 - const entries = [ - { type: "mcp", function: { name: "leaky" } }, - { type: "function", function: null }, - { type: "function", function: probeFunction }, - { type: "function", function: { name: 123 } }, - ] - - mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) - - const stream = handler.createMessage("test prompt", [], { - taskId: "test-task-id", - tools: testTools, - tool_choice: { type: "allowed_tools", allowed_tools: { mode: "auto", tools: entries } }, - }) - await stream.next() - - expect(mockResponsesCreate).toHaveBeenCalledWith( - expect.objectContaining({ - tool_choice: { - type: "allowed_tools", - mode: "auto", - tools: entries, - }, - }), - undefined, - ) - }) - it("createMessage should send max_output_tokens and temperature for the default model", async () => { mockResponsesCreate.mockResolvedValueOnce(asyncStreamFrom([])) @@ -460,27 +312,33 @@ describe("XAIHandler", () => { await expect(handler.completePrompt("test prompt")).rejects.toThrow(`xAI completion error: ${errorMessage}`) }) - it("completePrompt should surface the SDK APIUserAbortError unmodified on abort", async () => { + it("completePrompt should normalize the SDK APIUserAbortError to the contract AbortError on abort", async () => { const controller = new AbortController() controller.abort() const sdkAbortError = new APIUserAbortError() mockResponsesCreate.mockRejectedValueOnce(sdkAbortError) - // The error must surface as the same SDK instance, not wrapped by handleOpenAIError - await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toBe( - sdkAbortError, - ) + // The raw SDK abort error (name "Error", message "Request was aborted.") does not + // satisfy the caller's AbortError contract; it must be normalized, not wrapped by + // handleOpenAIError. + await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toMatchObject({ + name: "AbortError", + message: "The xAI request was aborted", + }) }) - it("completePrompt should surface a native AbortError unmodified on abort", async () => { + it("completePrompt should normalize a native AbortError to the contract error on abort", async () => { const controller = new AbortController() controller.abort() const abortError = new Error("This operation was aborted") abortError.name = "AbortError" mockResponsesCreate.mockRejectedValueOnce(abortError) - // The error must surface as the same instance, not wrapped by handleOpenAIError - await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toBe(abortError) + // Normalized to the branded contract message while keeping the contract name. + await expect(handler.completePrompt("test prompt", { abortSignal: controller.signal })).rejects.toMatchObject({ + name: "AbortError", + message: "The xAI request was aborted", + }) }) it("completePrompt should pass abort signal through to client", async () => { @@ -652,7 +510,7 @@ describe("XAIHandler", () => { ) }) - it("createMessage should surface the SDK APIUserAbortError unmodified on abort", async () => { + it("createMessage should normalize the SDK APIUserAbortError to the contract AbortError on abort", async () => { const abortedController = new AbortController() abortedController.abort() const sdkAbortError = new APIUserAbortError() @@ -672,8 +530,12 @@ describe("XAIHandler", () => { makeCreateMessageMetadata({ abortSignal: abortedController.signal }), ) - // The SDK error must surface as the same instance, not wrapped by handleOpenAIError - await expect(stream.next()).rejects.toBe(sdkAbortError) + // The raw SDK abort error does not satisfy the caller's AbortError contract: it + // must be normalized, not wrapped by handleOpenAIError. + await expect(stream.next()).rejects.toMatchObject({ + name: "AbortError", + message: "The xAI request was aborted", + }) }) it("should reject with AbortError when createMessage is called with an already-aborted signal", async () => { @@ -763,4 +625,58 @@ describe("XAIHandler", () => { expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) }) + + it("should remove the external abort listener when the SDK call rejects", async () => { + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + mockResponsesCreate.mockRejectedValueOnce(new Error("xai createMessage error")) + + const stream = handler.createMessage( + "test prompt", + [], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + await expect(stream.next()).rejects.toThrow() + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // The finally block must detach the listener when the request fails, not + // only on normal completion. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) + + it("should remove the external abort listener when the consumer returns early", async () => { + const controller = new AbortController() + const addEventListenerSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeEventListenerSpy = vitest.spyOn(controller.signal, "removeEventListener") + + mockResponsesCreate.mockResolvedValueOnce( + asyncStreamFrom([ + { type: "response.output_text.delta", delta: "first" }, + { type: "response.output_text.delta", delta: "second" }, + ]), + ) + + const stream = handler.createMessage( + "test prompt", + [], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + const iterator = stream[Symbol.asyncIterator]() + + expect((await iterator.next()).value).toEqual({ type: "text", text: "first" }) + await iterator.return(undefined) + + expect(addEventListenerSpy).toHaveBeenCalledTimes(1) + const [event, listener] = addEventListenerSpy.mock.calls[0] + expect(event).toBe("abort") + // Abandoning the generator mid-stream runs the finally block too. + expect(removeEventListenerSpy).toHaveBeenCalledTimes(1) + expect(removeEventListenerSpy).toHaveBeenCalledWith("abort", listener) + }) }) diff --git a/src/api/providers/anthropic-vertex.ts b/src/api/providers/anthropic-vertex.ts index 4098e3920c..bdc473f69a 100644 --- a/src/api/providers/anthropic-vertex.ts +++ b/src/api/providers/anthropic-vertex.ts @@ -24,6 +24,7 @@ import { } from "../../core/prompts/tools/native-tools/converters" import { BaseProvider } from "./base-provider" +import { bridgeExternalAbortSignal } from "./utils/abort-signal" import { NOT_PROVIDED } from "./constants" import { parseVertexJsonCredentials } from "./utils/vertex-credentials" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" @@ -92,22 +93,8 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple // client-level timeout configured in the constructor remains the only // cancellation mechanism, preserving the existing behavior. const externalAbortSignal = metadata?.abortSignal - let abortSignal: AbortSignal | undefined - let removeExternalAbortListener: (() => void) | undefined - if (externalAbortSignal) { - const controller = new AbortController() - if (externalAbortSignal.aborted) { - controller.abort() - } else { - // Retain the listener so it can be removed again once streaming - // finishes; otherwise a long-lived external signal would keep one - // listener (and its closed-over controller) per completed request. - const onExternalAbort = () => controller.abort() - externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) - removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) - } - abortSignal = controller.signal - } + const { signal: abortSignal, dispose: removeExternalAbortListener } = + bridgeExternalAbortSignal(externalAbortSignal) /** * Vertex API has specific limitations for prompt caching: @@ -139,8 +126,9 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple // and prompt caching const requestOptions: Anthropic.RequestOptions = {} if (betas?.length) { - // Model definitions carry at most one beta label, so use it directly. - requestOptions.headers = { "anthropic-beta": betas[0] } + // Join rather than take the first entry: the header accepts a comma list, so a + // second beta added to the builder later is not silently dropped. + requestOptions.headers = { "anthropic-beta": betas.join(",") } } if (abortSignal) { requestOptions.signal = abortSignal @@ -245,6 +233,13 @@ export class AnthropicVertexHandler extends BaseProvider implements SingleComple break } + case "content_block_stop": { + // Block complete - no action needed for now. + // NativeToolCallParser handles tool call completion + // Note: Signature for multi-turn thinking would require using stream.finalMessage() + // after iteration completes, which requires restructuring the streaming approach. + break + } } } } finally { diff --git a/src/api/providers/anthropic.ts b/src/api/providers/anthropic.ts index fb92dc7a4f..0efc2afc62 100644 --- a/src/api/providers/anthropic.ts +++ b/src/api/providers/anthropic.ts @@ -23,6 +23,7 @@ import { getAnthropicProviderReasoning } from "../transform/reasoning" import { handleProviderError } from "./utils/error-handler" import { BaseProvider } from "./base-provider" +import { bridgeExternalAbortSignal } from "./utils/abort-signal" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { calculateApiCostAnthropic } from "../../shared/cost" import { @@ -120,22 +121,8 @@ export class AnthropicHandler extends BaseProvider implements SingleCompletionHa // client-level timeout configured in the constructor remains the only // cancellation mechanism, preserving the existing behavior. const externalAbortSignal = metadata?.abortSignal - let abortSignal: AbortSignal | undefined - let removeExternalAbortListener: (() => void) | undefined - if (externalAbortSignal) { - const controller = new AbortController() - if (externalAbortSignal.aborted) { - controller.abort() - } else { - // Retain the callback so the listener can be detached once this request - // ends: an anonymous listener on a long-lived external signal would - // outlive the request and retain the per-request controller. - const onExternalAbort = () => controller.abort() - externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) - removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) - } - abortSignal = controller.signal - } + const { signal: abortSignal, dispose: removeExternalAbortListener } = + bridgeExternalAbortSignal(externalAbortSignal) switch (modelId) { case "claude-sonnet-5": diff --git a/src/api/providers/minimax.ts b/src/api/providers/minimax.ts index c5a4c3f5c5..f10bca8a7a 100644 --- a/src/api/providers/minimax.ts +++ b/src/api/providers/minimax.ts @@ -13,6 +13,7 @@ import { getModelParams } from "../transform/model-params" import { mergeEnvironmentDetailsForMiniMax } from "../transform/minimax-format" import { BaseProvider } from "./base-provider" +import { bridgeExternalAbortSignal } from "./utils/abort-signal" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { calculateApiCostAnthropic } from "../../shared/cost" import { convertOpenAIToolsToAnthropic } from "../../core/prompts/tools/native-tools/converters" @@ -92,22 +93,8 @@ export class MiniMaxHandler extends BaseProvider implements SingleCompletionHand // client-level timeout configured in the constructor remains the only // cancellation mechanism, preserving the existing behavior. const externalAbortSignal = metadata?.abortSignal - let abortSignal: AbortSignal | undefined - let removeExternalAbortListener: (() => void) | undefined - if (externalAbortSignal) { - const controller = new AbortController() - if (externalAbortSignal.aborted) { - controller.abort() - } else { - // Retain the callback so the listener can be detached once this request - // ends: an anonymous listener on a long-lived external signal would - // outlive the request and retain the per-request controller. - const onExternalAbort = () => controller.abort() - externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) - removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) - } - abortSignal = controller.signal - } + const { signal: abortSignal, dispose: removeExternalAbortListener } = + bridgeExternalAbortSignal(externalAbortSignal) // MiniMax M2 models support prompt caching const supportsPromptCache = info.supportsPromptCache ?? false diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index f779d48e1f..9a3bd96c8d 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -1,4 +1,5 @@ import { + bridgeExternalAbortSignal, createAbortError, isRequestAborted, mergeAbortSignalAndTimeout, @@ -416,3 +417,49 @@ describe("abort-signal utilities", () => { }) }) }) + +describe("bridgeExternalAbortSignal", () => { + it("returns an undefined signal and a noop dispose without an external signal", () => { + const { signal, dispose } = bridgeExternalAbortSignal() + expect(signal).toBeUndefined() + expect(() => dispose()).not.toThrow() + }) + + it("returns an already-aborted signal when the external signal is aborted", () => { + const controller = new AbortController() + controller.abort() + const { signal, dispose } = bridgeExternalAbortSignal(controller.signal) + expect(signal?.aborted).toBe(true) + expect(() => dispose()).not.toThrow() + }) + + it("aborts the per-request signal when the external signal aborts", () => { + const controller = new AbortController() + const { signal } = bridgeExternalAbortSignal(controller.signal) + expect(signal?.aborted).toBe(false) + controller.abort() + expect(signal?.aborted).toBe(true) + }) + + it("detaches the retained listener on dispose", () => { + const controller = new AbortController() + const addSpy = vitest.spyOn(controller.signal, "addEventListener") + const removeSpy = vitest.spyOn(controller.signal, "removeEventListener") + + const { signal, dispose } = bridgeExternalAbortSignal(controller.signal) + + expect(addSpy).toHaveBeenCalledTimes(1) + const [event, listener, opts] = addSpy.mock.calls[0] + expect(event).toBe("abort") + expect(opts).toEqual({ once: true }) + + dispose() + expect(removeSpy).toHaveBeenCalledTimes(1) + expect(removeSpy).toHaveBeenCalledWith("abort", listener) + + // After dispose the bridge no longer forwards aborts: the per-request + // signal stays live, proving the exact retained callback was detached. + controller.abort() + expect(signal?.aborted).toBe(false) + }) +}) diff --git a/src/api/providers/utils/abort-signal.ts b/src/api/providers/utils/abort-signal.ts index cbb7a13aaf..9f6d2aade8 100644 --- a/src/api/providers/utils/abort-signal.ts +++ b/src/api/providers/utils/abort-signal.ts @@ -36,6 +36,45 @@ export function mergeAbortSignals(primarySignal: AbortSignal, secondarySignal?: return AbortSignal.any([primarySignal, secondarySignal]) } +/** + * Bridge an optional caller-owned (external) abort signal to a per-request + * signal, with an explicit detach handle. + * + * The returned `signal` aborts together with the external signal (immediately + * if it is already aborted). Callers must invoke `dispose()` once the request + * ends - on completion, failure, and early generator exit alike: the listener + * is registered on the caller's long-lived signal, and an anonymous listener + * left attached would outlive the request and retain the per-request + * controller for every completed request. + * + * With no external signal, returns `{ signal: undefined, dispose: noop }` so + * the client-level timeout remains the only cancellation mechanism. + */ +export function bridgeExternalAbortSignal(externalSignal?: AbortSignal): { + signal?: AbortSignal + dispose: () => void +} { + if (!externalSignal) { + return { signal: undefined, dispose: () => {} } + } + + const controller = new AbortController() + if (externalSignal.aborted) { + controller.abort() + return { signal: controller.signal, dispose: () => {} } + } + + // Retain the callback so the listener can be detached once this request + // ends: an anonymous listener on a long-lived external signal would + // outlive the request and retain the per-request controller. + const onExternalAbort = () => controller.abort() + externalSignal.addEventListener("abort", onExternalAbort, { once: true }) + return { + signal: controller.signal, + dispose: () => externalSignal.removeEventListener("abort", onExternalAbort), + } +} + /** * Throw an AbortError if the given signal is already aborted. * diff --git a/src/api/providers/xai.ts b/src/api/providers/xai.ts index 50f34a358f..ec609c8456 100644 --- a/src/api/providers/xai.ts +++ b/src/api/providers/xai.ts @@ -1,5 +1,5 @@ import { Anthropic } from "@anthropic-ai/sdk" -import OpenAI, { APIUserAbortError } from "openai" +import OpenAI from "openai" import { type XAIModelId, xaiDefaultModelId, xaiModels, ApiProviderError } from "@roo-code/types" import { TelemetryService } from "@roo-code/telemetry" @@ -13,6 +13,7 @@ import { getModelParams } from "../transform/model-params" import { DEFAULT_HEADERS, NOT_PROVIDED } from "./constants" import { BaseProvider } from "./base-provider" +import { bridgeExternalAbortSignal, createAbortError, isRequestAborted } from "./utils/abort-signal" import type { SingleCompletionHandler, ApiHandlerCreateMessageMetadata, CompletePromptOptions } from "../index" import { handleOpenAIError } from "./utils/error-handler" import { isMcpTool } from "../../utils/mcp-name" @@ -63,51 +64,6 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler * Uses base provider's convertToolSchemaForOpenAI() for schema hardening * (additionalProperties: false, ensureAllRequired) and handles MCP tools. */ - /** - * Map a Chat Completions tool choice to the Responses API shape so TypeScript - * validates the provider payload (the APIs use different object forms for the - * named-tool choices: Chat Completions nests the name under `function`/`custom`, - * Responses API puts it at the top level). String options (auto/required/none) - * are identical in both APIs and pass through unchanged. `allowed_tools` - * entries keep their allowlist mode, but Chat Completions function references - * ({ type: "function", function: { name } }) are flattened to the Responses - * API shape ({ type: "function", name }); other entry types pass through - * unchanged. - */ - private mapToolChoice( - toolChoice: NonNullable, - ): OpenAI.Responses.ResponseCreateParamsStreaming["tool_choice"] { - if (typeof toolChoice === "string") { - return toolChoice - } - switch (toolChoice.type) { - case "function": - return { type: "function", name: toolChoice.function.name } - case "custom": - return { type: "custom", name: toolChoice.custom.name } - case "allowed_tools": - return { - type: "allowed_tools", - mode: toolChoice.allowed_tools.mode, - tools: toolChoice.allowed_tools.tools.map((entry) => { - // Chat Completions allowlist entries nest the function reference - // ({ type: "function", function: { name } }); the Responses API - // expects the name at the top level ({ type: "function", name }). - const functionRef = entry["function"] - if ( - entry["type"] === "function" && - functionRef != null && - typeof functionRef === "object" && - typeof (functionRef as Record)["name"] === "string" - ) { - return { type: "function", name: (functionRef as Record)["name"] } - } - return entry - }), - } - } - } - private mapResponseTools(tools?: any[]): any[] | undefined { const converted = this.convertToolsForOpenAI(tools) if (!converted?.length) { @@ -139,8 +95,6 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler // Convert directly from Anthropic format to Responses API input format const input = convertToResponsesApiInput(messages) const responseTools = this.mapResponseTools(metadata?.tools) - const toolChoice = metadata?.tool_choice - const parallelToolCalls = metadata?.parallelToolCalls // Bridge the external abort signal from request metadata into a per-request // controller so the SDK call is cancelled when the owning request is aborted @@ -148,67 +102,55 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler // client-level timeout configured in the constructor remains the only // cancellation mechanism, preserving the existing behavior. const externalAbortSignal = metadata?.abortSignal - let abortSignal: AbortSignal | undefined - let removeExternalAbortListener: (() => void) | undefined - if (externalAbortSignal) { - const controller = new AbortController() - if (externalAbortSignal.aborted) { - controller.abort() - } else { - // Retain the listener so it can be removed again once streaming - // finishes; otherwise a long-lived external signal would keep one - // listener (and its closed-over controller) per completed request. - const onExternalAbort = () => controller.abort() - externalAbortSignal.addEventListener("abort", onExternalAbort, { once: true }) - removeExternalAbortListener = () => externalAbortSignal.removeEventListener("abort", onExternalAbort) - } - abortSignal = controller.signal + const { signal: abortSignal, dispose: removeExternalAbortListener } = + bridgeExternalAbortSignal(externalAbortSignal) + + // Build request options + const requestBody: Record = { + model: model.id, + instructions: systemPrompt, + input: input, + stream: true, + store: false, // Don't store responses server-side for privacy + include: ["reasoning.encrypted_content"], } - try { - // Build request options - const requestBody: OpenAI.Responses.ResponseCreateParamsStreaming = { - model: model.id, - instructions: systemPrompt, - input: input, - stream: true, - store: false, // Don't store responses server-side for privacy - include: ["reasoning.encrypted_content"], - } - - // Model params are always resolved by getModel(); send them unconditionally. + if (model.maxTokens) { requestBody.max_output_tokens = model.maxTokens + } + + if (model.temperature !== undefined) { requestBody.temperature = model.temperature + } - if (responseTools) { - requestBody.tools = responseTools - // Metadata carries a Chat Completions tool choice; the Responses API - // uses its own shape, so map it explicitly instead of casting. - requestBody.tool_choice = this.mapToolChoice(toolChoice === undefined ? "auto" : toolChoice) - requestBody.parallel_tool_calls = parallelToolCalls ?? true - } + if (responseTools) { + requestBody.tools = responseTools + // Cast tool_choice since metadata uses Chat Completions types but Responses API has its own type + requestBody.tool_choice = (metadata?.tool_choice ?? "auto") as any + requestBody.parallel_tool_calls = metadata?.parallelToolCalls ?? true + } - // Pass reasoning effort for models that support it (e.g., grok-4.5, grok-3-mini). - // The xAI Responses API uses `reasoning: { effort }` format (not `reasoning_effort` - // which is the Chat Completions format), so we convert from the OpenAI params shape. - if (model.reasoning) { - requestBody.reasoning = { effort: model.reasoning.reasoning_effort } - } + // Pass reasoning effort for models that support it (e.g., grok-4.5, grok-3-mini). + // The xAI Responses API uses `reasoning: { effort }` format (not `reasoning_effort` + // which is the Chat Completions format), so we convert from the OpenAI params shape. + if (model.reasoning) { + requestBody.reasoning = { effort: model.reasoning.reasoning_effort } + } - let stream: AsyncIterable + try { + let stream: AsyncIterable try { - stream = await this.client.responses.create( - requestBody, + stream = (await this.client.responses.create( + { ...requestBody, stream: true } as any, abortSignal ? { signal: abortSignal } : undefined, - ) + )) as unknown as AsyncIterable } catch (error) { - // Let abort errors propagate unmodified so callers can recognize them: - // native AbortError (error.name === "AbortError") and the OpenAI SDK's - // APIUserAbortError, which the SDK throws when the request signal aborts - // (the SDK class does not set a distinctive error.name in v5, so use - // instanceof). - if ((error instanceof Error && error.name === "AbortError") || error instanceof APIUserAbortError) { - throw error + // Normalize cancellations to the contract error: the OpenAI SDK's + // APIUserAbortError carries name "Error" and message "Request was + // aborted.", so rethrowing it raw would leak a non-conforming shape to + // the caller's AbortError guard. + if (isRequestAborted(error, abortSignal)) { + throw createAbortError(this.providerName) } const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, model.id, "createMessage") @@ -250,13 +192,12 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler // output_text is a convenience field on the Responses API response return response.output_text || "" } catch (error) { - // Let abort errors propagate unmodified so callers can recognize them: - // native AbortError (error.name === "AbortError") and the OpenAI SDK's - // APIUserAbortError, which the SDK throws when the request signal aborts - // (the SDK class does not set a distinctive error.name in v5, so use - // instanceof). - if ((error instanceof Error && error.name === "AbortError") || error instanceof APIUserAbortError) { - throw error + // Normalize cancellations to the contract error: the OpenAI SDK's + // APIUserAbortError carries name "Error" and message "Request was + // aborted.", so rethrowing it raw would leak a non-conforming shape to + // the caller's AbortError guard. + if (isRequestAborted(error, options?.abortSignal)) { + throw createAbortError(this.providerName) } const errorMessage = error instanceof Error ? error.message : String(error) const apiError = new ApiProviderError(errorMessage, this.providerName, model.id, "completePrompt") diff --git a/src/eslint-suppressions.json b/src/eslint-suppressions.json index 45b1a32d72..10e1c94931 100644 --- a/src/eslint-suppressions.json +++ b/src/eslint-suppressions.json @@ -451,7 +451,7 @@ }, "api/providers/xai.ts": { "@typescript-eslint/no-explicit-any": { - "count": 2 + "count": 7 } }, "api/transform/__tests__/ai-sdk.spec.ts": { From 67b5966a525e0f22f0a28bb6cdf9083ae6e9fadb Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 14 Oct 2026 17:43:18 +0800 Subject: [PATCH 14/18] chore(api): remove Stryker suppression comments from touched files Per repo policy (AGENTS.md: Stryker suppression comments are banned; the mutation gate rejects them in touched files). Equivalence arguments for the resulting mutants are recorded in the PR body. --- src/api/providers/utils/abort-signal.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/src/api/providers/utils/abort-signal.ts b/src/api/providers/utils/abort-signal.ts index 9f6d2aade8..f1b36abde8 100644 --- a/src/api/providers/utils/abort-signal.ts +++ b/src/api/providers/utils/abort-signal.ts @@ -154,7 +154,6 @@ export function rejectOnAbort(pending: Promise, signal: AbortSignal, provi } return new Promise((resolve, reject) => { const onAbort = () => reject(createAbortError(providerName)) - // Stryker disable next-line ObjectLiteral,BooleanLiteral: a signal fires its abort event exactly once and the settle handler removes this listener, so the once flag is unobservable signal.addEventListener("abort", onAbort, { once: true }) void pending.then( (value) => { From 888cefe13f2f30a14ce8850bc354a06caf8bc619 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 14 Oct 2026 18:29:10 +0800 Subject: [PATCH 15/18] fix(api): normalize stream-iteration aborts in the xAI responses path CodeRabbit: an abort landing after responses.create resolved escaped as the OpenAI SDK's APIUserAbortError (name "Error") from the yield* over processResponsesApiStream, outside the create catch, so callers classified the cancellation as a provider failure. Wrap the stream iteration and normalize through isRequestAborted/createAbortError to the series contract (name AbortError, "The xAI request was aborted"). Regression test pins the path; negative control: reverting the wrap fails exactly that test (1 red / 37 green). --- src/api/providers/__tests__/xai.spec.ts | 29 +++++++++++++++++++++++++ src/api/providers/xai.ts | 14 +++++++++++- 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/src/api/providers/__tests__/xai.spec.ts b/src/api/providers/__tests__/xai.spec.ts index c209001665..cbff98e0dd 100644 --- a/src/api/providers/__tests__/xai.spec.ts +++ b/src/api/providers/__tests__/xai.spec.ts @@ -537,6 +537,35 @@ describe("XAIHandler", () => { }) }) + it("createMessage should normalize an SDK abort raised while iterating the response stream", async () => { + const controller = new AbortController() + const sdkAbortError = new APIUserAbortError() + + // responses.create resolves; the cancellation lands while the stream is + // being consumed, so the SDK abort error escapes from the iterator. + mockResponsesCreate.mockResolvedValueOnce({ + [Symbol.asyncIterator]: () => ({ + next: async () => { + controller.abort() + throw sdkAbortError + }, + }), + }) + + const stream = handler.createMessage( + "test prompt", + [], + makeCreateMessageMetadata({ abortSignal: controller.signal }), + ) + + // The raw SDK abort error (name "Error") must be normalized to the + // contract AbortError even though it escapes outside the create call. + await expect(stream.next()).rejects.toMatchObject({ + name: "AbortError", + message: "The xAI request was aborted", + }) + }) + it("should reject with AbortError when createMessage is called with an already-aborted signal", async () => { const abortedController = new AbortController() abortedController.abort() diff --git a/src/api/providers/xai.ts b/src/api/providers/xai.ts index ec609c8456..bcf401284c 100644 --- a/src/api/providers/xai.ts +++ b/src/api/providers/xai.ts @@ -159,7 +159,19 @@ export class XAIHandler extends BaseProvider implements SingleCompletionHandler } const normalizeUsage = createUsageNormalizer() - yield* processResponsesApiStream(stream, normalizeUsage) + try { + yield* processResponsesApiStream(stream, normalizeUsage) + } catch (error) { + // An abort landing after responses.create resolved surfaces here as + // the SDK's APIUserAbortError thrown from stream iteration (name + // "Error", message "Request was aborted."), outside the create + // catch above. Normalize it to the contract AbortError too, so + // callers never classify a cancellation as a provider failure. + if (isRequestAborted(error, abortSignal)) { + throw createAbortError(this.providerName) + } + throw error + } } finally { // Release the listener once the stream is consumed, whether the // request completed, failed, or the generator was closed early. From 0f23d9875890bc1d118126da764bbc68f2cf1a31 Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 14 Oct 2026 20:27:52 +0800 Subject: [PATCH 16/18] fix(api): reject non-finite and cap overflowing mergeAbortSignalAndTimeout delays Port of the #1300 timeout guard finding (CodeRabbit): Node's AbortSignal.timeout throws ERR_OUT_OF_RANGE for non-integer delays (Infinity) and clamps delays above 2^31-1 ms to 1 ms, so a non-finite timeoutMs threw a RangeError instead of running the request, and an overflowing one aborted almost immediately. The helper now treats non-finite values as "no timeout" and caps finite delays at 2^31-1 ms. Regression tests added; negative controls (reverting the guard) fail exactly the two new tests. --- .../utils/__tests__/abort-signal.spec.ts | 23 +++++++++++++++++++ src/api/providers/utils/abort-signal.ts | 11 ++++++--- 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index 9a3bd96c8d..cb29c09d22 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -276,6 +276,29 @@ describe("abort-signal utilities", () => { await vi.waitFor(() => expect(result?.aborted).toBe(true)) }) + + it("treats a non-finite timeoutMs as no timeout", () => { + const controller = new AbortController() + + // Infinity must not reach AbortSignal.timeout(), which throws + // ERR_OUT_OF_RANGE for non-integer delays. + expect(() => mergeAbortSignalAndTimeout(undefined, Infinity)).not.toThrow() + expect(mergeAbortSignalAndTimeout(undefined, Infinity)).toBeUndefined() + expect(mergeAbortSignalAndTimeout(controller.signal, Infinity)).toBe(controller.signal) + }) + + it("caps an overflowing timeoutMs instead of aborting immediately", async () => { + // Node clamps AbortSignal.timeout delays above 2^31-1 ms to 1 ms; + // the helper caps the delay so an overflowing timeout does not abort + // the request within the first millisecond. + const result = mergeAbortSignalAndTimeout(undefined, 2_147_483_648) + + expect(result).toBeInstanceOf(AbortSignal) + expect(result?.aborted).toBe(false) + + await new Promise((resolve) => setTimeout(resolve, 20)) + expect(result?.aborted).toBe(false) + }) }) describe("mergeAbortSignals", () => { diff --git a/src/api/providers/utils/abort-signal.ts b/src/api/providers/utils/abort-signal.ts index f1b36abde8..0fcd961b3b 100644 --- a/src/api/providers/utils/abort-signal.ts +++ b/src/api/providers/utils/abort-signal.ts @@ -1,18 +1,23 @@ /** * Merge an optional external abort signal with an optional timeout. * - * Timeout values <= 0 are treated as disabled. The timeout is created via the + * Timeout values <= 0 and non-finite values are treated as disabled; finite + * delays above 2^31-1 ms are capped (Node clamps them to 1 ms otherwise). The timeout is created via the * native AbortSignal.timeout() API, which self-manages its timer — callers do * not need to (and cannot) clear it manually. */ export function mergeAbortSignalAndTimeout(externalSignal?: AbortSignal, timeoutMs?: number): AbortSignal | undefined { - const hasTimeout = typeof timeoutMs === "number" && timeoutMs > 0 + // Node's AbortSignal.timeout() throws ERR_OUT_OF_RANGE for non-integer + // delays (Infinity) and clamps delays above 2^31-1 ms to 1 ms, which would + // abort almost immediately; treat non-finite values as "no timeout" and + // cap the rest. + const hasTimeout = typeof timeoutMs === "number" && Number.isFinite(timeoutMs) && timeoutMs > 0 if (!hasTimeout) { return externalSignal } - const timeoutSignal = AbortSignal.timeout(timeoutMs) + const timeoutSignal = AbortSignal.timeout(Math.min(timeoutMs, 2_147_483_647)) if (!externalSignal) { return timeoutSignal From f160a99d705b6c831fa6b141435e6fd4a55cc73e Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 14 Oct 2026 21:42:50 +0800 Subject: [PATCH 17/18] fix(api): round fractional timeouts before AbortSignal.timeout CodeRabbit follow-up on the timeout guard: Node validates AbortSignal.timeout delays as uint32, so finite fractions (1500.5, 0.5) threw ERR_OUT_OF_RANGE just like Infinity did. The helper now rounds delays up (a positive fraction never becomes 0 ms) before capping at 2^31-1 ms. Regression test mergeAbortSignalAndTimeout (undefined, 1.5); negative control (reverting the rounding) fails exactly that test. --- .../providers/utils/__tests__/abort-signal.spec.ts | 14 ++++++++++++++ src/api/providers/utils/abort-signal.ts | 7 ++++--- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index cb29c09d22..fb8b725485 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -287,6 +287,20 @@ describe("abort-signal utilities", () => { expect(mergeAbortSignalAndTimeout(controller.signal, Infinity)).toBe(controller.signal) }) + it("rounds a fractional timeoutMs up before scheduling", async () => { + // Node validates AbortSignal.timeout delays as uint32: 1.5 or 0.5 + // would throw ERR_OUT_OF_RANGE. The helper rounds up, so a positive + // fraction never becomes 0 ms either. + let result: AbortSignal | undefined + expect(() => { + result = mergeAbortSignalAndTimeout(undefined, 1.5) + }).not.toThrow() + expect(result).toBeInstanceOf(AbortSignal) + expect(result?.aborted).toBe(false) + await vi.waitFor(() => expect(result?.aborted).toBe(true)) + expect(() => mergeAbortSignalAndTimeout(undefined, 0.5)).not.toThrow() + }) + it("caps an overflowing timeoutMs instead of aborting immediately", async () => { // Node clamps AbortSignal.timeout delays above 2^31-1 ms to 1 ms; // the helper caps the delay so an overflowing timeout does not abort diff --git a/src/api/providers/utils/abort-signal.ts b/src/api/providers/utils/abort-signal.ts index 0fcd961b3b..3f19335158 100644 --- a/src/api/providers/utils/abort-signal.ts +++ b/src/api/providers/utils/abort-signal.ts @@ -9,15 +9,16 @@ export function mergeAbortSignalAndTimeout(externalSignal?: AbortSignal, timeoutMs?: number): AbortSignal | undefined { // Node's AbortSignal.timeout() throws ERR_OUT_OF_RANGE for non-integer // delays (Infinity) and clamps delays above 2^31-1 ms to 1 ms, which would - // abort almost immediately; treat non-finite values as "no timeout" and - // cap the rest. + // abort almost immediately; treat non-finite values as "no timeout", + // round fractional delays up to whole milliseconds (uint32 validation + // throws for 1500.5), and cap the rest. const hasTimeout = typeof timeoutMs === "number" && Number.isFinite(timeoutMs) && timeoutMs > 0 if (!hasTimeout) { return externalSignal } - const timeoutSignal = AbortSignal.timeout(Math.min(timeoutMs, 2_147_483_647)) + const timeoutSignal = AbortSignal.timeout(Math.min(Math.max(1, Math.ceil(timeoutMs)), 2_147_483_647)) if (!externalSignal) { return timeoutSignal From efb3b4f8a97feda909911e5fcf25d61d0c7e948d Mon Sep 17 00:00:00 2001 From: Eason Liang Date: Wed, 14 Oct 2026 23:20:01 +0800 Subject: [PATCH 18/18] test(api): assert the rounded delay reaching AbortSignal.timeout CodeRabbit: the fractional-timeout test only proved the call did not throw and eventually aborted - a round-down mutation (1.5 -> 1) would still pass. The test now spies AbortSignal.timeout and pins the received delay: 2 for 1.5 and 1 for 0.5 (never 0). Negative control: mutating Math.ceil to Math.floor fails exactly this test. --- .../utils/__tests__/abort-signal.spec.ts | 25 ++++++++++++++----- 1 file changed, 19 insertions(+), 6 deletions(-) diff --git a/src/api/providers/utils/__tests__/abort-signal.spec.ts b/src/api/providers/utils/__tests__/abort-signal.spec.ts index fb8b725485..1d0bbff9d4 100644 --- a/src/api/providers/utils/__tests__/abort-signal.spec.ts +++ b/src/api/providers/utils/__tests__/abort-signal.spec.ts @@ -289,16 +289,29 @@ describe("abort-signal utilities", () => { it("rounds a fractional timeoutMs up before scheduling", async () => { // Node validates AbortSignal.timeout delays as uint32: 1.5 or 0.5 - // would throw ERR_OUT_OF_RANGE. The helper rounds up, so a positive - // fraction never becomes 0 ms either. + // would throw ERR_OUT_OF_RANGE. Pin the rounded delay that actually + // reaches AbortSignal.timeout (1.5 -> 2, 0.5 -> 1, never 0) so a + // round-down mutation reddens this test instead of still aborting + // eventually. + const timeoutSpy = vi.spyOn(AbortSignal, "timeout") let result: AbortSignal | undefined - expect(() => { - result = mergeAbortSignalAndTimeout(undefined, 1.5) - }).not.toThrow() + let tiny: AbortSignal | undefined + try { + expect(() => { + result = mergeAbortSignalAndTimeout(undefined, 1.5) + }).not.toThrow() + expect(timeoutSpy).toHaveBeenCalledWith(2) + expect(() => { + tiny = mergeAbortSignalAndTimeout(undefined, 0.5) + }).not.toThrow() + expect(timeoutSpy).toHaveBeenCalledWith(1) + } finally { + timeoutSpy.mockRestore() + } expect(result).toBeInstanceOf(AbortSignal) expect(result?.aborted).toBe(false) + expect(tiny).toBeInstanceOf(AbortSignal) await vi.waitFor(() => expect(result?.aborted).toBe(true)) - expect(() => mergeAbortSignalAndTimeout(undefined, 0.5)).not.toThrow() }) it("caps an overflowing timeoutMs instead of aborting immediately", async () => {