fix(api): fix abort signal propagation across all providers

Author: easonliang28

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

@@ -637,48 +637,46 @@ describe("DeepSeekHandler", () => {
 			const toolCallChunks = chunks.filter((chunk) => chunk.type === "tool_call_partial")
 			expect(toolCallChunks.length).toBeGreaterThan(0)
 			expect(toolCallChunks[0].name).toBe("get_weather")
+		})
+	})
 
-			describe("abortSignal support", () => {
-				it("should pass abortSignal to chat.completions.create when provided in metadata", async () => {
-					const handler = new DeepSeekHandler({ ...mockOptions, apiKey: "test-key" })
-					const systemPrompt = "You are a helpful assistant."
-					const messages: Anthropic.Messages.MessageParam[] = [
-						{ role: "user", content: [{ type: "text" as const, text: "Hello!" }] },
-					]
-
-					const controller = new AbortController()
-					const mockAbortSignal = controller.signal
-
-					await handler.createMessage(systemPrompt, messages, {
-						taskId: "test",
-						abortSignal: mockAbortSignal,
-					})
-					for await (const _chunk of handler.createMessage(systemPrompt, messages)) {
-						break
-					}
-
-					expect(mockCreate).toHaveBeenCalled()
-					const callArgs = mockCreate.mock.calls[0][0]
-					expect(callArgs.signal).toBe(mockAbortSignal)
-				})
+	describe("abortSignal support", () => {
+		it("should pass abortSignal to chat.completions.create when provided in metadata", async () => {
+			const handler = new DeepSeekHandler({ ...mockOptions, apiKey: "test-key" })
+			const systemPrompt = "You are a helpful assistant."
+			const messages: Anthropic.Messages.MessageParam[] = [
+				{ role: "user", content: [{ type: "text" as const, text: "Hello!" }] },
+			]
 
-				it("should not include signal when abortSignal is not provided", async () => {
-					const handler = new DeepSeekHandler({ ...mockOptions, apiKey: "test-key" })
-					const systemPrompt = "You are a helpful assistant."
-					const messages: Anthropic.Messages.MessageParam[] = [
-						{ role: "user", content: [{ type: "text" as const, text: "Hello!" }] },
-					]
-
-					await handler.createMessage(systemPrompt, messages)
-					for await (const _chunk of handler.createMessage(systemPrompt, messages)) {
-						break
-					}
-
-					expect(mockCreate).toHaveBeenCalled()
-					const callArgs = mockCreate.mock.calls[0][0]
-					expect(callArgs.signal).toBeUndefined()
-				})
-			})
+			const controller = new AbortController()
+			const mockAbortSignal = controller.signal
+
+			for await (const _chunk of handler.createMessage(systemPrompt, messages, {
+				taskId: "test",
+				abortSignal: mockAbortSignal,
+			})) {
+				break
+			}
+
+			expect(mockCreate).toHaveBeenCalled()
+			const requestOptions = mockCreate.mock.calls[0][1]
+			expect(requestOptions?.signal).toBe(mockAbortSignal)
+		})
+
+		it("should not include signal when abortSignal is not provided", async () => {
+			const handler = new DeepSeekHandler({ ...mockOptions, apiKey: "test-key" })
+			const systemPrompt = "You are a helpful assistant."
+			const messages: Anthropic.Messages.MessageParam[] = [
+				{ role: "user", content: [{ type: "text" as const, text: "Hello!" }] },
+			]
+
+			for await (const _chunk of handler.createMessage(systemPrompt, messages)) {
+				break
+			}
+
+			expect(mockCreate).toHaveBeenCalled()
+			const requestOptions = mockCreate.mock.calls[0][1]
+			expect(requestOptions?.signal).toBeUndefined()
 		})
 	})
 })

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

@@ -416,7 +416,7 @@ describe("GeminiHandler", () => {
 
 			expect(mockGenerateContentStream).toHaveBeenCalled()
 			const callArgs = mockGenerateContentStream.mock.calls[0][0]
-			expect(callArgs.signal).toBe(mockAbortSignal)
+			expect(callArgs.config?.abortSignal).toBe(mockAbortSignal)
 		})
 
 		it("should pass undefined signal when abortSignal is not provided", async () => {
@@ -434,7 +434,7 @@ describe("GeminiHandler", () => {
 
 			expect(mockGenerateContentStream).toHaveBeenCalled()
 			const callArgs = mockGenerateContentStream.mock.calls[0][0]
-			expect(callArgs.signal).toBeUndefined()
+			expect(callArgs.config?.abortSignal).toBeUndefined()
 		})
 	})
 })

src/api/providers/__tests__/native-ollama.spec.ts

@@ -639,10 +639,8 @@ describe("NativeOllamaHandler", () => {
 	})
 
 	describe("abortSignal support", () => {
-		it("should pass abortSignal to chat when provided in metadata", async () => {
+		it("should wire abortSignal to per-request client's abort() method", async () => {
 			vitest.clearAllMocks()
-			const mockAbortController: any = { signal: Symbol("abort") }
-
 			;(mockGetOllamaModels as any).mockImplementationOnce(async () => ({
 				llama2: { contextWindow: 4096, maxTokens: 4096, supportsImages: false, supportsPromptCache: false },
 			}))
@@ -657,20 +655,23 @@ describe("NativeOllamaHandler", () => {
 				ollamaBaseUrl: "http://localhost:11434",
 			})
 
+			const controller = new AbortController()
+
 			for await (const _chunk of handlerWithSignal.createMessage(
 				"system",
 				[{ role: "user", content: "Hello!" }],
-				{ taskId: "test", abortSignal: mockAbortController.signal },
+				{ taskId: "test", abortSignal: controller.signal },
 			)) {
 				break
 			}
 
 			expect(mockedData.mockChat).toHaveBeenCalled()
+			// The chat call should NOT have signal in options (we use per-request client instead)
 			const callArgs = mockedData.mockChat.mock.calls[0][0]
-			expect(callArgs.signal).toBe(mockAbortController.signal)
+			expect(callArgs.signal).toBeUndefined()
 		})
 
-		it("should pass undefined signal when abortSignal is not provided", async () => {
+		it("should not pass signal in options when abortSignal is not provided", async () => {
 			vitest.clearAllMocks()
 			;(mockGetOllamaModels as any).mockImplementationOnce(async () => ({
 				llama2: { contextWindow: 4096, maxTokens: 4096, supportsImages: false, supportsPromptCache: false },

src/api/providers/__tests__/vercel-ai-gateway.spec.ts

@@ -586,13 +586,16 @@ describe("VercelAiGatewayHandler", () => {
 			const result = await handler.completePrompt(prompt)
 
 			expect(result).toBe("Test completion response")
-			expect(mockCreate).toHaveBeenCalledWith({
-				model: "anthropic/claude-sonnet-4",
-				messages: [{ role: "user", content: prompt }],
-				stream: false,
-				temperature: VERCEL_AI_GATEWAY_DEFAULT_TEMPERATURE,
-				max_completion_tokens: 64000,
-			})
+			expect(mockCreate).toHaveBeenCalledWith(
+				{
+					model: "anthropic/claude-sonnet-4",
+					messages: [{ role: "user", content: prompt }],
+					stream: false,
+					temperature: VERCEL_AI_GATEWAY_DEFAULT_TEMPERATURE,
+					max_completion_tokens: 64000,
+				},
+				{ signal: undefined },
+			)
 		})
 
 		it("uses custom temperature for completion", async () => {
@@ -608,6 +611,7 @@ describe("VercelAiGatewayHandler", () => {
 				expect.objectContaining({
 					temperature: customTemp,
 				}),
+				expect.objectContaining({ signal: undefined }),
 			)
 		})
 
@@ -656,6 +660,7 @@ describe("VercelAiGatewayHandler", () => {
 				expect.objectContaining({
 					temperature: 0.9,
 				}),
+				expect.objectContaining({ signal: undefined }),
 			)
 		})
 

src/api/providers/__tests__/zoo-gateway.spec.ts

@@ -464,6 +464,7 @@ describe("ZooGatewayHandler", () => {
 					temperature: ZOO_GATEWAY_DEFAULT_TEMPERATURE,
 					max_completion_tokens: 64000,
 				}),
+				expect.objectContaining({ signal: undefined }),
 			)
 		})
 

src/api/providers/bedrock.ts

@@ -534,12 +534,21 @@ export class AwsBedrockHandler extends BaseProvider implements SingleCompletionH
 		const controller = new AbortController()
 		let timeoutId: NodeJS.Timeout | undefined
 
-		// Listen for external abort signal from metadata and forward to internal controller
+		// Listen for external abort signal from metadata and forward to internal controller.
+		// Handle both pre-aborted signals and future abort events.
 		const externalAbortSignal = metadata?.abortSignal
 		if (externalAbortSignal) {
-			externalAbortSignal.addEventListener("abort", () => {
+			if (externalAbortSignal.aborted) {
 				controller.abort()
-			})
+			} else {
+				externalAbortSignal.addEventListener(
+					"abort",
+					() => {
+						controller.abort()
+					},
+					{ once: true },
+				)
+			}
 		}
 
 		try {

src/api/providers/gemini.ts

@@ -343,7 +343,11 @@ export class GeminiHandler extends BaseProvider implements SingleCompletionHandl
 			}
 		}
 
-		const params: any = { model, contents, config, signal: metadata?.abortSignal }
+		const params: any = {
+			model,
+			contents,
+			config: { ...config, abortSignal: metadata?.abortSignal },
+		}
 
 		try {
 			const result = await this.client.models.generateContentStream(params)

src/api/providers/native-ollama.ts

@@ -205,10 +205,37 @@ export class NativeOllamaHandler extends BaseProvider implements SingleCompletio
 		messages: Anthropic.Messages.MessageParam[],
 		metadata?: ApiHandlerCreateMessageMetadata,
 	): ApiStream {
-		const client = this.ensureClient()
 		const { id: modelId } = await this.fetchModel()
 		const useR1Format = modelId.toLowerCase().includes("deepseek-r1")
 
+		// Create a per-request Ollama client since the SDK doesn't support abortSignal in options.
+		// We listen to metadata.abortSignal and call client.abort() when it fires.
+		const requestClient = new Ollama({
+			host: this.options.ollamaBaseUrl || "http://localhost:11434",
+		})
+
+		// Add API key if provided
+		if (this.options.ollamaApiKey) {
+			;(requestClient as any).config = {
+				...((requestClient as any).config ?? {}),
+				headers: { Authorization: `Bearer ${this.options.ollamaApiKey}` },
+			}
+		}
+
+		// Wire external abort signal to per-request client's abort() method
+		const externalAbortSignal = metadata?.abortSignal
+		if (externalAbortSignal) {
+			if (externalAbortSignal.aborted) {
+				requestClient.abort()
+			} else {
+				const abortListener = () => {
+					requestClient.abort()
+					externalAbortSignal.removeEventListener("abort", abortListener)
+				}
+				externalAbortSignal.addEventListener("abort", abortListener, { once: true })
+			}
+		}
+
 		const ollamaMessages: Message[] = [
 			{ role: "system", content: systemPrompt },
 			...convertToOllamaMessages(messages),
@@ -234,14 +261,13 @@ export class NativeOllamaHandler extends BaseProvider implements SingleCompletio
 				chatOptions.num_ctx = this.options.ollamaNumCtx
 			}
 
-			// Create the actual API request promise
-			const stream = await client.chat({
+			// Create the actual API request promise (use per-request client, not signal in options)
+			const stream = await requestClient.chat({
 				model: modelId,
 				messages: ollamaMessages,
 				stream: true,
 				options: chatOptions,
 				tools: this.convertToolsToOllama(metadata?.tools),
-				signal: metadata?.abortSignal,
 			} as any)
 
 			let totalInputTokens = 0
@@ -346,27 +372,48 @@ export class NativeOllamaHandler extends BaseProvider implements SingleCompletio
 	}
 
 	async completePrompt(prompt: string, metadata?: ApiHandlerCreateMessageMetadata): Promise<string> {
+		// Create a per-request Ollama client since the SDK doesn't support abortSignal in options.
+		const requestClient = new Ollama({
+			host: this.options.ollamaBaseUrl || "http://localhost:11434",
+		})
+
+		if (this.options.ollamaApiKey) {
+			;(requestClient as any).config = {
+				headers: { Authorization: `Bearer ${this.options.ollamaApiKey}` },
+			}
+		}
+
+		// Wire external abort signal to per-request client's abort() method
+		const externalAbortSignal = metadata?.abortSignal
+		if (externalAbortSignal) {
+			if (externalAbortSignal.aborted) {
+				requestClient.abort()
+			} else {
+				const abortListener = () => {
+					requestClient.abort()
+					externalAbortSignal.removeEventListener("abort", abortListener)
+				}
+				externalAbortSignal.addEventListener("abort", abortListener, { once: true })
+			}
+		}
+
 		try {
-			const client = this.ensureClient()
 			const { id: modelId } = await this.fetchModel()
 			const useR1Format = modelId.toLowerCase().includes("deepseek-r1")
 
-			// Build options object conditionally
 			const chatOptions: OllamaChatOptions = {
 				temperature: this.options.modelTemperature ?? (useR1Format ? DEEP_SEEK_DEFAULT_TEMPERATURE : 0),
 			}
 
-			// Only include num_ctx if explicitly set via ollamaNumCtx
 			if (this.options.ollamaNumCtx !== undefined) {
 				chatOptions.num_ctx = this.options.ollamaNumCtx
 			}
 
-			const response = await client.chat({
+			const response = await requestClient.chat({
 				model: modelId,
 				messages: [{ role: "user", content: prompt }],
 				stream: false,
 				options: chatOptions,
-				signal: metadata?.abortSignal,
 			} as any)
 
 			return ((response as any).message?.content as string) || ""
@@ -375,6 +422,8 @@ export class NativeOllamaHandler extends BaseProvider implements SingleCompletio
 				throw new Error(`Ollama completion error: ${error.message}`)
 			}
 			throw error
+		} finally {
+			requestClient.abort()
 		}
 	}
 }

src/api/providers/openai-codex.ts

@@ -188,7 +188,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion
 		// Make the request with retry on auth failure
 		for (let attempt = 0; attempt < 2; attempt++) {
 			try {
-				yield* this.executeRequest(requestBody, model, accessToken, metadata?.taskId)
+				yield* this.executeRequest(requestBody, model, accessToken, metadata?.taskId, metadata)
 				return
 			} catch (error) {
 				const message = error instanceof Error ? error.message : String(error)