Skip to content

Commit 299c474

Browse files
committed
fix(api): pass abortSignal to streamText() for OpenAI Compatible provider (#404)
When user clicks stop during streaming, the HTTP request to the provider continues running because Vercel AI SDK's streamText() doesn't receive an abort signal. This wastes API tokens/compute on the provider side. Changes: - Add abortSignal?: AbortSignal to ApiHandlerCreateMessageMetadata interface - Pass Task.ts's AbortController.signal through metadata to createMessage() - Use signal in openai-compatible.ts streamText() options This fix affects all OpenAICompatibleHandler subclasses (LiteLLM, OpenRouter, Vercel AI Gateway, etc.) that use the Vercel AI SDK. Fixes #404
1 parent f1f7cb4 commit 299c474

18 files changed

Lines changed: 656 additions & 24 deletions

src/api/index.ts

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,12 @@ export interface ApiHandlerCreateMessageMetadata {
8989
* Only applies to providers that support function calling restrictions (e.g., Gemini).
9090
*/
9191
allowedFunctionNames?: string[]
92+
/**
93+
* Abort signal for cancelling the HTTP request mid-stream.
94+
* Passed through to AI SDK's streamText() so the underlying HTTP request is aborted
95+
* when the user clicks stop, preventing wasted API tokens/compute on the provider side.
96+
*/
97+
abortSignal?: AbortSignal
9298
}
9399

94100
export interface ApiHandler {
Lines changed: 225 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,225 @@
1+
// Tests for OpenAICompatibleHandler's abortSignal passing to streamText()
2+
// Verifies that when createMessage() is called with metadata containing an abortSignal,
3+
// the signal is passed through to AI SDK's streamText() so HTTP requests can be aborted.
4+
5+
const { mockStreamText } = vi.hoisted(() => ({
6+
mockStreamText: vi.fn(),
7+
}))
8+
9+
vi.mock("ai", async (importOriginal) => {
10+
const actual = await importOriginal<typeof import("ai")>()
11+
return {
12+
...actual,
13+
streamText: mockStreamText,
14+
}
15+
})
16+
17+
vi.mock("@ai-sdk/openai-compatible", () => ({
18+
createOpenAICompatible: vi.fn(() => {
19+
return vi.fn(() => ({
20+
modelId: "test-model",
21+
provider: "test-provider",
22+
}))
23+
}),
24+
}))
25+
26+
import type { Anthropic } from "@anthropic-ai/sdk"
27+
28+
import { OpenAICompatibleHandler, OpenAICompatibleConfig } from "../openai-compatible"
29+
import type { ApiHandlerOptions } from "../../../shared/api"
30+
import type { ModelInfo } from "@roo-code/types"
31+
32+
// Concrete test subclass of the abstract OpenAICompatibleHandler
33+
class TestOpenAiCompatibleHandler extends OpenAICompatibleHandler {
34+
constructor(options: ApiHandlerOptions, config: OpenAICompatibleConfig) {
35+
super(options, config)
36+
}
37+
38+
override getModel(): { id: string; info: ModelInfo } {
39+
return { id: this.config.modelId, info: this.config.modelInfo }
40+
}
41+
}
42+
43+
describe("OpenAICompatibleHandler abort signal", () => {
44+
let handler: TestOpenAiCompatibleHandler
45+
const mockOptions: ApiHandlerOptions = {}
46+
const config: OpenAICompatibleConfig = {
47+
providerName: "test-provider",
48+
baseURL: "https://api.test.com/v1",
49+
apiKey: "test-key",
50+
modelId: "test-model",
51+
modelInfo: { maxTokens: 8192, contextWindow: 128000, supportsImages: false, supportsPromptCache: true },
52+
}
53+
54+
beforeEach(() => {
55+
vi.clearAllMocks()
56+
handler = new TestOpenAiCompatibleHandler(mockOptions, config)
57+
})
58+
59+
describe("createMessage abortSignal passing", () => {
60+
const systemPrompt = "You are a helpful assistant."
61+
const messages: Anthropic.Messages.MessageParam[] = [
62+
{ role: "user", content: [{ type: "text" as const, text: "Hello!" }] },
63+
]
64+
65+
it("should pass abortSignal to streamText when provided in metadata", async () => {
66+
const controller = new AbortController()
67+
const mockAbortSignal = controller.signal
68+
69+
async function* mockFullStream() {
70+
yield { type: "text-delta", text: "Test response" }
71+
}
72+
73+
function createMockStream(yieldValue: any) {
74+
return {
75+
fullStream: (async function* () {
76+
yield yieldValue
77+
})(),
78+
usage: Promise.resolve({ inputTokens: 10, outputTokens: 5 }),
79+
}
80+
}
81+
82+
const mockUsage = Promise.resolve({ inputTokens: 10, outputTokens: 5 })
83+
84+
mockStreamText.mockReturnValue({
85+
fullStream: mockFullStream(),
86+
usage: mockUsage,
87+
})
88+
89+
await handler
90+
.createMessage(systemPrompt, messages, {
91+
taskId: "test-task",
92+
abortSignal: mockAbortSignal,
93+
})
94+
.next()
95+
96+
expect(mockStreamText).toHaveBeenCalledWith(
97+
expect.objectContaining({
98+
signal: mockAbortSignal,
99+
}),
100+
)
101+
})
102+
103+
it("should pass undefined signal to streamText when abortSignal is not provided", async () => {
104+
async function* mockFullStream() {
105+
yield { type: "text-delta", text: "Test response" }
106+
}
107+
108+
function createMockStream(yieldValue: any) {
109+
return {
110+
fullStream: (async function* () {
111+
yield yieldValue
112+
})(),
113+
usage: Promise.resolve({ inputTokens: 10, outputTokens: 5 }),
114+
}
115+
}
116+
117+
const mockUsage = Promise.resolve({ inputTokens: 10, outputTokens: 5 })
118+
119+
mockStreamText.mockReturnValue({
120+
fullStream: mockFullStream(),
121+
usage: mockUsage,
122+
})
123+
124+
await handler
125+
.createMessage(systemPrompt, messages, {
126+
taskId: "test-task",
127+
})
128+
.next()
129+
130+
expect(mockStreamText).toHaveBeenCalledWith(
131+
expect.objectContaining({
132+
signal: undefined,
133+
}),
134+
)
135+
})
136+
137+
it("should pass signal to streamText when metadata is undefined", async () => {
138+
async function* mockFullStream() {
139+
yield { type: "text-delta", text: "Test response" }
140+
}
141+
142+
function createMockStream(yieldValue: any) {
143+
return {
144+
fullStream: (async function* () {
145+
yield yieldValue
146+
})(),
147+
usage: Promise.resolve({ inputTokens: 10, outputTokens: 5 }),
148+
}
149+
}
150+
151+
const mockUsage = Promise.resolve({ inputTokens: 10, outputTokens: 5 })
152+
153+
mockStreamText.mockReturnValue({
154+
fullStream: mockFullStream(),
155+
usage: mockUsage,
156+
})
157+
158+
await handler.createMessage(systemPrompt, messages).next()
159+
160+
expect(mockStreamText).toHaveBeenCalledWith(
161+
expect.objectContaining({
162+
signal: undefined,
163+
}),
164+
)
165+
})
166+
167+
it("should pass the correct signal when it fires during streaming", async () => {
168+
const controller = new AbortController()
169+
const mockAbortSignal = controller.signal
170+
171+
let capturedOptions: any = null
172+
173+
mockStreamText.mockImplementation((options) => {
174+
capturedOptions = options
175+
return {
176+
fullStream: (async function* () {
177+
yield { type: "text-delta", text: "Partial" }
178+
})(),
179+
usage: Promise.resolve({ inputTokens: 5, outputTokens: 3 }),
180+
}
181+
})
182+
183+
const stream = handler.createMessage(systemPrompt, messages, {
184+
taskId: "test-task",
185+
abortSignal: mockAbortSignal,
186+
})
187+
188+
// Verify the signal was captured before aborting
189+
expect(capturedOptions).toBeDefined()
190+
expect(capturedOptions.signal).toBe(mockAbortSignal)
191+
192+
// Now abort - this should cause streamText to receive an aborted signal
193+
controller.abort()
194+
expect(controller.signal.aborted).toBe(true)
195+
})
196+
197+
it("should pass all other request options alongside the signal", async () => {
198+
const controller = new AbortController()
199+
200+
let capturedOptions: any = null
201+
202+
mockStreamText.mockImplementation((options) => {
203+
capturedOptions = options
204+
return {
205+
fullStream: (async function* () {
206+
yield { type: "text-delta", text: "Test" }
207+
})(),
208+
usage: Promise.resolve({ inputTokens: 10, outputTokens: 5 }),
209+
}
210+
})
211+
212+
await handler
213+
.createMessage(systemPrompt, messages, {
214+
taskId: "test-task",
215+
abortSignal: controller.signal,
216+
})
217+
.next()
218+
219+
expect(capturedOptions).toHaveProperty("model")
220+
expect(capturedOptions).toHaveProperty("system", systemPrompt)
221+
expect(capturedOptions).toHaveProperty("messages")
222+
expect(capturedOptions).toHaveProperty("signal", controller.signal)
223+
})
224+
})
225+
})

src/api/providers/base-openai-compatible-provider.ts

100644100755
Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,7 @@ export abstract class BaseOpenAiCompatibleProvider<ModelName extends string>
104104
}
105105

106106
try {
107-
return this.client.chat.completions.create(params, requestOptions)
107+
return this.client.chat.completions.create(params, { ...requestOptions, signal: metadata?.abortSignal })
108108
} catch (error) {
109109
throw handleOpenAIError(error, this.providerName)
110110
}

src/api/providers/deepseek.ts

100644100755
Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -133,7 +133,10 @@ export class DeepSeekHandler extends OpenAiHandler {
133133
try {
134134
stream = await this.client.chat.completions.create(
135135
requestOptions as OpenAI.Chat.Completions.ChatCompletionCreateParamsStreaming,
136-
isAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {},
136+
{
137+
...(isAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {}),
138+
signal: metadata?.abortSignal,
139+
},
137140
)
138141
} catch (error) {
139142
const { handleOpenAIError } = await import("./utils/openai-error-handler")

src/api/providers/lite-llm.ts

100644100755
Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -223,7 +223,9 @@ export class LiteLLMHandler extends RouterProvider implements SingleCompletionHa
223223
}
224224

225225
try {
226-
const { data: completion } = await this.client.chat.completions.create(requestOptions).withResponse()
226+
const { data: completion } = await this.client.chat.completions
227+
.create(requestOptions, { signal: metadata?.abortSignal })
228+
.withResponse()
227229

228230
let lastUsage
229231

src/api/providers/lm-studio.ts

100644100755
Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan
9999

100100
let results
101101
try {
102-
results = await this.client.chat.completions.create(params)
102+
results = await this.client.chat.completions.create(params, { signal: metadata?.abortSignal })
103103
} catch (error) {
104104
throw handleOpenAIError(error, this.providerName)
105105
}

src/api/providers/mimo.ts

100644100755
Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,10 @@ export class MimoHandler extends OpenAiHandler {
9999

100100
let stream: AsyncIterable<OpenAI.Chat.Completions.ChatCompletionChunk>
101101
try {
102-
stream = (await this.client.chat.completions.create(params as any)) as any
102+
stream = (await this.client.chat.completions.create(
103+
params as any,
104+
{ signal: metadata?.abortSignal } as any,
105+
)) as any
103106
} catch (error) {
104107
throw handleProviderError(error, "MiMo")
105108
}

src/api/providers/openai-compatible.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,7 @@ export abstract class OpenAICompatibleHandler extends BaseProvider implements Si
174174
maxOutputTokens: this.getMaxOutputTokens(),
175175
tools: aiSdkTools,
176176
toolChoice: this.mapToolChoice(metadata?.tool_choice),
177+
abortSignal: metadata?.abortSignal,
177178
}
178179

179180
// Use streamText for streaming responses

src/api/providers/openai.ts

100644100755
Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -176,10 +176,10 @@ export class OpenAiHandler extends BaseProvider implements SingleCompletionHandl
176176

177177
let stream
178178
try {
179-
stream = await this.client.chat.completions.create(
180-
requestOptions,
181-
isAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {},
182-
)
179+
stream = await this.client.chat.completions.create(requestOptions, {
180+
...(isAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {}),
181+
signal: metadata?.abortSignal,
182+
})
183183
} catch (error) {
184184
throw handleOpenAIError(error, this.providerName)
185185
}
@@ -371,10 +371,10 @@ export class OpenAiHandler extends BaseProvider implements SingleCompletionHandl
371371

372372
let stream
373373
try {
374-
stream = await this.client.chat.completions.create(
375-
requestOptions,
376-
methodIsAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {},
377-
)
374+
stream = await this.client.chat.completions.create(requestOptions, {
375+
...(methodIsAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {}),
376+
signal: metadata?.abortSignal,
377+
})
378378
} catch (error) {
379379
throw handleOpenAIError(error, this.providerName)
380380
}
@@ -405,10 +405,10 @@ export class OpenAiHandler extends BaseProvider implements SingleCompletionHandl
405405

406406
let response
407407
try {
408-
response = await this.client.chat.completions.create(
409-
requestOptions,
410-
methodIsAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {},
411-
)
408+
response = await this.client.chat.completions.create(requestOptions, {
409+
...(methodIsAzureAiInference ? { path: OPENAI_AZURE_AI_INFERENCE_PATH } : {}),
410+
signal: metadata?.abortSignal,
411+
})
412412
} catch (error) {
413413
throw handleOpenAIError(error, this.providerName)
414414
}

src/api/providers/opencode-go.ts

100644100755
Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ export class OpencodeGoHandler extends RouterProvider implements SingleCompletio
7070
parallel_tool_calls: metadata?.parallelToolCalls ?? true,
7171
}
7272

73-
const completion = await this.client.chat.completions.create(body)
73+
const completion = await this.client.chat.completions.create(body, { signal: metadata?.abortSignal })
7474

7575
for await (const chunk of completion) {
7676
const delta = chunk.choices[0]?.delta

0 commit comments

Comments
 (0)