Skip to content

Commit ee47fbb

Browse files
committed
fix(api): pass abortSignal to streaming API calls for all providers (#404)
When user clicks stop during streaming, the HTTP request 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 all providers' streamText() options
1 parent 3df406e commit ee47fbb

24 files changed

Lines changed: 1476 additions & 57 deletions

packages/types/src/tool-params.ts

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,14 @@ export type ReadFileToolParams = ReadFileParams | LegacyReadFileParams
8989
* Type guard to check if params are in legacy format.
9090
*/
9191
export function isLegacyReadFileParams(params: ReadFileToolParams): params is LegacyReadFileParams {
92-
return "_legacyFormat" in params && params._legacyFormat === true
92+
// `NativeToolCallParser` always tags freshly parsed legacy calls with `_legacyFormat: true`.
93+
// The bare-`files` fallback only matters for chat history persisted before that flag was
94+
// introduced (commit cc86049f1) and re-hydrated on a later run. Note that params matched via
95+
// that fallback narrow to `LegacyReadFileParams` but leave `_legacyFormat` `undefined`, so
96+
// callers should branch on the presence of `files`, not on `_legacyFormat === true`.
97+
const hasLegacyFlag = "_legacyFormat" in params && params._legacyFormat === true
98+
const hasFilesArray = "files" in params && Array.isArray((params as unknown as Record<string, unknown>).files)
99+
return hasLegacyFlag || hasFilesArray
93100
}
94101

95102
export interface Coordinate {

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 {

src/api/providers/__tests__/fireworks.spec.ts

Lines changed: 44 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -95,25 +95,46 @@ describe("FireworksHandler", () => {
9595
})
9696

9797
it.each([
98-
{ modelId: "accounts/fireworks/models/glm-5p1" as const, contextWindow: 202752, inputPrice: 1.4, outputPrice: 4.4, cacheReadsPrice: 0.26 },
99-
{ modelId: "accounts/fireworks/models/kimi-k2p6" as const, contextWindow: 262144, inputPrice: 0.95, outputPrice: 4.0, cacheReadsPrice: 0.16 },
100-
{ modelId: "accounts/fireworks/models/deepseek-v4-pro" as const, contextWindow: 1048576, inputPrice: 1.74, outputPrice: 3.48, cacheReadsPrice: 0.14 },
101-
])("should expose newly added model $modelId", ({ modelId, contextWindow, inputPrice, outputPrice, cacheReadsPrice }) => {
102-
expect(fireworksModels[modelId]).toBeDefined()
103-
const info = fireworksModels[modelId]
104-
expect(info.maxTokens).toBeGreaterThan(0)
105-
expect(info.contextWindow).toBe(contextWindow)
106-
expect(info.inputPrice).toBe(inputPrice)
107-
expect(info.outputPrice).toBe(outputPrice)
108-
expect(info.cacheReadsPrice).toBe(cacheReadsPrice)
109-
expect(info.description).toBeTruthy()
110-
111-
const handlerWithModel = new FireworksHandler({
112-
apiModelId: modelId,
113-
fireworksApiKey: "test-fireworks-api-key",
114-
})
115-
expect(handlerWithModel.getModel().id).toBe(modelId)
116-
})
98+
{
99+
modelId: "accounts/fireworks/models/glm-5p1" as const,
100+
contextWindow: 202752,
101+
inputPrice: 1.4,
102+
outputPrice: 4.4,
103+
cacheReadsPrice: 0.26,
104+
},
105+
{
106+
modelId: "accounts/fireworks/models/kimi-k2p6" as const,
107+
contextWindow: 262144,
108+
inputPrice: 0.95,
109+
outputPrice: 4.0,
110+
cacheReadsPrice: 0.16,
111+
},
112+
{
113+
modelId: "accounts/fireworks/models/deepseek-v4-pro" as const,
114+
contextWindow: 1048576,
115+
inputPrice: 1.74,
116+
outputPrice: 3.48,
117+
cacheReadsPrice: 0.14,
118+
},
119+
])(
120+
"should expose newly added model $modelId",
121+
({ modelId, contextWindow, inputPrice, outputPrice, cacheReadsPrice }) => {
122+
expect(fireworksModels[modelId]).toBeDefined()
123+
const info = fireworksModels[modelId]
124+
expect(info.maxTokens).toBeGreaterThan(0)
125+
expect(info.contextWindow).toBe(contextWindow)
126+
expect(info.inputPrice).toBe(inputPrice)
127+
expect(info.outputPrice).toBe(outputPrice)
128+
expect(info.cacheReadsPrice).toBe(cacheReadsPrice)
129+
expect(info.description).toBeTruthy()
130+
131+
const handlerWithModel = new FireworksHandler({
132+
apiModelId: modelId,
133+
fireworksApiKey: "test-fireworks-api-key",
134+
})
135+
expect(handlerWithModel.getModel().id).toBe(modelId)
136+
},
137+
)
117138

118139
it("should return Kimi K2 Instruct model with correct configuration", () => {
119140
const testModelId: FireworksModelId = "accounts/fireworks/models/kimi-k2-instruct"
@@ -465,7 +486,7 @@ describe("FireworksHandler", () => {
465486
stream: true,
466487
stream_options: { include_usage: true },
467488
}),
468-
undefined,
489+
expect.any(Object),
469490
)
470491
})
471492

@@ -491,7 +512,7 @@ describe("FireworksHandler", () => {
491512
expect.objectContaining({
492513
temperature: 0.5,
493514
}),
494-
undefined,
515+
expect.any(Object),
495516
)
496517
})
497518

@@ -518,7 +539,7 @@ describe("FireworksHandler", () => {
518539
expect.objectContaining({
519540
temperature: 1.0,
520541
}),
521-
undefined,
542+
expect.any(Object),
522543
)
523544
})
524545

@@ -546,7 +567,7 @@ describe("FireworksHandler", () => {
546567
expect.objectContaining({
547568
temperature: 0.7,
548569
}),
549-
undefined,
570+
expect.any(Object),
550571
)
551572
})
552573

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/__tests__/sambanova.spec.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -146,7 +146,7 @@ describe("SambaNovaHandler", () => {
146146
stream: true,
147147
stream_options: { include_usage: true },
148148
}),
149-
undefined,
149+
expect.any(Object),
150150
)
151151
})
152152
})

src/api/providers/__tests__/zai.spec.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -469,7 +469,7 @@ describe("ZAiHandler", () => {
469469
stream: true,
470470
stream_options: { include_usage: true },
471471
}),
472-
undefined,
472+
expect.any(Object),
473473
)
474474
})
475475
})

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")

0 commit comments

Comments
 (0)