feat(api): abort signal support for unbound, vercel-ai-gateway, zoo-gateway

src/api/providers/unbound.ts

@@ -1,5 +1,5 @@
 import { Anthropic } from "@anthropic-ai/sdk"
-import OpenAI from "openai"
+import OpenAI, { APIConnectionTimeoutError } from "openai"
 
 import {
 	type ModelInfo,
@@ -29,6 +29,13 @@
 } from "../index"
 import { handleOpenAIError } from "./utils/error-handler"
 import { applyRouterToolPreferences } from "./utils/router-tool-preferences"
+import {
+	createAbortError,
+	hasUsableTimeout,
+	isRequestAborted,
+	resolveModelWithAbort,
+	toSdkTimeoutMs,
+} from "./utils/abort-signal"
 import { extractReasoningFromDelta } from "./utils/extract-reasoning"
 
 // Unbound usage includes extra fields for Anthropic cache tokens.
@@ -143,14 +150,21 @@
 		messages: Anthropic.Messages.MessageParam[],
 		metadata?: ApiHandlerCreateMessageMetadata,
 	): ApiStream {
+		// Establish the cancellation scope around model resolution: a
+		// pre-aborted signal rejects before the lookup starts, and a signal
+		// that fires while model metadata is loading settles on the
+		// standardized AbortError; any other resolution failure propagates
+		// unchanged.
+		const externalAbortSignal = metadata?.abortSignal
+		const resolved = await resolveModelWithAbort(() => this.fetchModel(), externalAbortSignal, "Unbound")
 		const {
 			id: model,
 			info,
 			maxTokens: max_tokens,
 			temperature,
 			reasoningEffort: reasoning_effort,
 			reasoning: thinking,
-		} = await this.fetchModel()
+		} = resolved
 
 		const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [
 			{ role: "system", content: systemPrompt },
@@ -176,51 +190,96 @@
 			tool_choice: metadata?.tool_choice,
 		}
 
-		let stream
-		try {
-			stream = await this.client.chat.completions.create(completionParams)
-		} catch (error) {
-			throw handleOpenAIError(error, this.providerName)
+		// Per-request controller so an external abort signal (e.g. task
+		// cancellation) can interrupt the in-flight streaming request.
+		// Bridge it to our controller using the Bedrock pattern:
+		// - pre-aborted guard: check if already aborted before adding listener
+		// - { once: true }: remove listener after first abort to avoid leaks
+		// The listener is stored so it can be detached when the request ends:
+		// { once: true } only removes it on abort, so a task-scoped signal
+		// would otherwise accumulate one listener per request.
+		const controller = new AbortController()
+		const abortListener = () => controller.abort()
+		if (externalAbortSignal) {
+			if (externalAbortSignal.aborted) {
+				controller.abort()
+			} else {
+				externalAbortSignal.addEventListener("abort", abortListener, { once: true })
+			}
 		}
-		let lastUsage: any = undefined
-
-		for await (const chunk of stream) {
-			const delta = chunk.choices[0]?.delta
 
-			const reasoningText = extractReasoningFromDelta(delta)
-			if (reasoningText) {
-				yield { type: "reasoning", text: reasoningText }
+		try {
+			let stream
+			try {
+				stream = await this.client.chat.completions.create(completionParams, { signal: controller.signal })
+			} catch (error) {
+				// Preserve abort identity (series standard): a cancelled request
+				// must surface as a DOM-standard AbortError, not a wrapped
+				// completion error.
+				if (isRequestAborted(error, externalAbortSignal)) {
+					throw createAbortError("Unbound")
+				}
+				throw handleOpenAIError(error, this.providerName)
 			}
+			let lastUsage: any = undefined
 
-			if (delta?.content) {
-				yield { type: "text", text: delta.content }
-			}
+			try {
+				for await (const chunk of stream) {
+					const delta = chunk.choices[0]?.delta
+
+					const reasoningText = extractReasoningFromDelta(delta)
+					if (reasoningText) {
+						yield { type: "reasoning", text: reasoningText }
+					}
 
-			// Handle native tool calls
-			if (delta && "tool_calls" in delta && Array.isArray(delta.tool_calls)) {
-				for (const toolCall of delta.tool_calls) {
-					yield {
-						type: "tool_call_partial",
-						index: toolCall.index,
-						id: toolCall.id,
-						name: toolCall.function?.name,
-						arguments: toolCall.function?.arguments,
+					if (delta?.content) {
+						yield { type: "text", text: delta.content }
+					}
+
+					// Handle native tool calls
+					if (delta && "tool_calls" in delta && Array.isArray(delta.tool_calls)) {
+						for (const toolCall of delta.tool_calls) {
+							yield {
+								type: "tool_call_partial",
+								index: toolCall.index,
+								id: toolCall.id,
+								name: toolCall.function?.name,
+								arguments: toolCall.function?.arguments,
+							}
+						}
+					}
+
+					if (chunk.usage) {
+						lastUsage = chunk.usage
 					}
 				}
-			}
 
-			if (chunk.usage) {
-				lastUsage = chunk.usage
+				if (lastUsage) {
+					yield this.processUsageMetrics(lastUsage, info)
+				}
+			} catch (error) {
+				// Preserve abort identity (series standard): a cancellation that
+				// surfaces after the stream has started must also normalize to
+				// the standardized AbortError, not the raw SDK rejection.
+				if (isRequestAborted(error, externalAbortSignal)) {
+					throw createAbortError("Unbound")
+				}
+				throw error
 			}
-		}
-
-		if (lastUsage) {
-			yield this.processUsageMetrics(lastUsage, info)
+		} finally {
+			externalAbortSignal?.removeEventListener("abort", abortListener)
 		}
 	}
 
-	async completePrompt(prompt: string, _options?: CompletePromptOptions): Promise<string> {
-		const { id: model, maxTokens: max_tokens, temperature } = await this.fetchModel()
+	async completePrompt(prompt: string, options?: CompletePromptOptions): Promise<string> {
+		// Model resolution must honour the caller's cancellation scope the same
+		// way createMessage does: a signal that fires during the lookup settles on
+		// the standardized AbortError instead of waiting for the catalog to finish.
+		const {
+			id: model,
+			maxTokens: max_tokens,
+			temperature,
+		} = await resolveModelWithAbort(() => this.fetchModel(), options?.abortSignal, "Unbound")
 
 		const openAiMessages: OpenAI.Chat.ChatCompletionMessageParam[] = [{ role: "system", content: prompt }]
 
@@ -230,11 +289,34 @@
 			messages: openAiMessages,
 			temperature: temperature,
 		}
+		// Build request options with abortSignal and/or timeout.
+		// timeoutMs <= 0 means "no explicit timeout": omit the SDK timeout
+		// option entirely — the OpenAI SDK treats timeout: 0 as an immediate
+		// abort, which would cancel the request right away.
+		const createOptions: OpenAI.RequestOptions = {}
+		if (options?.abortSignal) {
+			createOptions.signal = options.abortSignal
+		}
+		const timeoutMs = options?.timeoutMs
+		if (hasUsableTimeout(timeoutMs)) {
+			createOptions.timeout = toSdkTimeoutMs(timeoutMs)
+		}
 
 		let response: OpenAI.Chat.ChatCompletion
 		try {
-			response = await this.client.chat.completions.create(completionParams)
+			response = await this.client.chat.completions.create(completionParams, createOptions)
 		} catch (error) {
+			// Preserve abort identity (series standard): caller-initiated
+			// cancellations and request timeouts must surface as a
+			// DOM-standard AbortError, not a wrapped completion error. The
+			// OpenAI SDK reports both with messages ending in a period
+			// ("Request was aborted.", "Request timed out."), which would not
+			// match task-level abort detection (message ending in "aborted").
+			// SDK request timeouts are not aborts, but the series standard maps
+			// them to the same AbortError identity as caller cancellations.
+			if (isRequestAborted(error, options?.abortSignal) || error instanceof APIConnectionTimeoutError) {
+				throw createAbortError("Unbound")
+			}
 			throw handleOpenAIError(error, this.providerName)
 		}
 		return response.choices[0]?.message.content || ""