diff --git a/.changeset/abort-signal-ai-bedrock.md b/.changeset/abort-signal-ai-bedrock.md new file mode 100644 index 0000000000..92c9e38fdc --- /dev/null +++ b/.changeset/abort-signal-ai-bedrock.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-bedrock': patch +--- + +Aborting a chat now cancels the Converse request to Bedrock, for streaming and for structured output. Before, the request kept running. diff --git a/.changeset/abort-signal-ai-byteplus.md b/.changeset/abort-signal-ai-byteplus.md new file mode 100644 index 0000000000..39d3c3a01f --- /dev/null +++ b/.changeset/abort-signal-ai-byteplus.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-byteplus': patch +--- + +Aborting an image, video, speech, or transcription call now cancels the BytePlus request. Before, the request kept running and was billed. The image and video requests still stop at the configured `timeout`. diff --git a/.changeset/abort-signal-ai-elevenlabs.md b/.changeset/abort-signal-ai-elevenlabs.md new file mode 100644 index 0000000000..1336ec05e3 --- /dev/null +++ b/.changeset/abort-signal-ai-elevenlabs.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-elevenlabs': patch +--- + +An abort now cancels the ElevenLabs request for speech, dialogue, music, sound effects, and transcription. Before, the adapters did not give the abort signal to the SDK, so the request ran to the end and was billed. diff --git a/.changeset/abort-signal-ai-fal.md b/.changeset/abort-signal-ai-fal.md new file mode 100644 index 0000000000..8f305dca79 --- /dev/null +++ b/.changeset/abort-signal-ai-fal.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-fal': patch +--- + +An abort now asks fal to cancel the job for image, audio, speech, and transcription. fal stops a queued job. For a job that already runs, the model app decides whether it stops. The speech audio download also stops on abort. diff --git a/.changeset/abort-signal-ai-gemini.md b/.changeset/abort-signal-ai-gemini.md new file mode 100644 index 0000000000..1a2567b5cc --- /dev/null +++ b/.changeset/abort-signal-ai-gemini.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-gemini': patch +--- + +Aborting a call now cancels the Gemini request for Lyria audio, speech, Veo and Omni video creation, the Veo input image download, and chat with agentic video. Before, the request kept running and was billed. diff --git a/.changeset/abort-signal-ai-grok.md b/.changeset/abort-signal-ai-grok.md new file mode 100644 index 0000000000..8cf0042328 --- /dev/null +++ b/.changeset/abort-signal-ai-grok.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-grok': patch +--- + +Aborting `generateImage()`, `generateSpeech()`, `generateTranscription()` or `generateVideo()` now stops the xAI request. Before, the call rejected but the request kept running and was billed. diff --git a/.changeset/abort-signal-ai-groq.md b/.changeset/abort-signal-ai-groq.md new file mode 100644 index 0000000000..4dbb0fb332 --- /dev/null +++ b/.changeset/abort-signal-ai-groq.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-groq': patch +--- + +Aborting `generateSpeech()` or `generateTranscription()` now stops the Groq request. Before, the call rejected but the request kept running and was billed. diff --git a/.changeset/abort-signal-ai-mistral.md b/.changeset/abort-signal-ai-mistral.md new file mode 100644 index 0000000000..04ab318e97 --- /dev/null +++ b/.changeset/abort-signal-ai-mistral.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-mistral': patch +--- + +Aborting a chat now cancels the request to Mistral, for streaming and for structured output. Structured output keeps the `timeoutMs` limit when you pass a signal. diff --git a/.changeset/abort-signal-ai-ollama.md b/.changeset/abort-signal-ai-ollama.md new file mode 100644 index 0000000000..90ef7ed195 --- /dev/null +++ b/.changeset/abort-signal-ai-ollama.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-ollama': patch +--- + +Aborting a chat now stops the Ollama stream when you pass your own `Ollama` client. Structured output with your own client still runs to the end, because `ollama` has no per-call signal. diff --git a/.changeset/abort-signal-ai-openai.md b/.changeset/abort-signal-ai-openai.md new file mode 100644 index 0000000000..c546c8097e --- /dev/null +++ b/.changeset/abort-signal-ai-openai.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-openai': patch +--- + +Aborting `generateImage()`, `generateSpeech()`, `generateTranscription()` or `generateVideo()` now stops the OpenAI request. Before, the call rejected but the request kept running and was billed. This also covers the image fetch for an `allowUrlFetch` image input. diff --git a/.changeset/abort-signal-ai-openrouter.md b/.changeset/abort-signal-ai-openrouter.md new file mode 100644 index 0000000000..d2e04b01c3 --- /dev/null +++ b/.changeset/abort-signal-ai-openrouter.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-openrouter': patch +--- + +Aborting an image or video call now cancels the OpenRouter request. Before, the request kept running and was billed. diff --git a/.changeset/abort-signal-ai-vercel-gateway.md b/.changeset/abort-signal-ai-vercel-gateway.md new file mode 100644 index 0000000000..c8a16447f1 --- /dev/null +++ b/.changeset/abort-signal-ai-vercel-gateway.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-vercel-gateway': patch +--- + +Aborting `generateImage()` now stops the Vercel AI Gateway request. Before, the call rejected but the request kept running and was billed. diff --git a/packages/ai-bedrock/src/adapters/converse-text.ts b/packages/ai-bedrock/src/adapters/converse-text.ts index c4400e9741..331f2ef2a4 100644 --- a/packages/ai-bedrock/src/adapters/converse-text.ts +++ b/packages/ai-bedrock/src/adapters/converse-text.ts @@ -241,12 +241,13 @@ export class BedrockConverseTextAdapter< protected async sendStream( input: ConverseStreamCommandInput, + signal?: AbortSignal | null, ): Promise> { const { ConverseStreamCommand } = await this.importBedrockRuntime() const client = await this.getClient() const command = new ConverseStreamCommand(input) addRestoreDocuments(command, input) - const res = await client.send(command) + const res = await client.send(command, { abortSignal: signal ?? undefined }) if (!res.stream) { throw new Error('Bedrock Converse: empty stream response') } @@ -255,12 +256,13 @@ export class BedrockConverseTextAdapter< protected async send( input: ConverseCommandInput, + signal?: AbortSignal | null, ): Promise { const { ConverseCommand } = await this.importBedrockRuntime() const client = await this.getClient() const command = new ConverseCommand(input) addRestoreDocuments(command, input) - return client.send(command) + return client.send(command, { abortSignal: signal ?? undefined }) } // --------------------------------------------------------------------------- @@ -281,7 +283,7 @@ export class BedrockConverseTextAdapter< if (!input.toolConfig && input.messages) { input.messages = toolBlocksToText(input.messages) } - const stream = await this.sendStream(input) + const stream = await this.sendStream(input, options.request?.signal) for await (const chunk of processConverseStream( stream, () => this.generateId(), @@ -344,7 +346,7 @@ export class BedrockConverseTextAdapter< ...this.buildInput(requestOptions), toolConfig: buildStructuredToolConfig(outputSchema), } - const res = await this.send(input) + const res = await this.send(input, chatOptions.request?.signal) // A forced tool call cut off at the output cap carries partial input, // which would read like a schema failure (issue #1426). if (res.stopReason === 'max_tokens') { @@ -408,7 +410,7 @@ export class BedrockConverseTextAdapter< ...this.buildInput(requestOptions), toolConfig: buildStructuredToolConfig(outputSchema), } - const stream = await this.sendStream(input) + const stream = await this.sendStream(input, chatOptions.request?.signal) // The forced tool streams its `input` as partial-JSON fragments inside // `contentBlockDelta.delta.toolUse.input`. We surface them as diff --git a/packages/ai-bedrock/tests/abort-signal-wire.test.ts b/packages/ai-bedrock/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..9fdeaed605 --- /dev/null +++ b/packages/ai-bedrock/tests/abort-signal-wire.test.ts @@ -0,0 +1,177 @@ +import { describe, expect, it, vi } from 'vitest' +import { BedrockRuntimeClient } from '@aws-sdk/client-bedrock-runtime' +import { EventType } from '@tanstack/ai' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { BedrockConverseTextAdapter } from '../src/adapters/converse-text' +import { createBedrockResponsesText } from '../src/adapters/responses-text' +import { createBedrockChat } from '../src/adapters/text' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +/** Drains a stream. A RUN_ERROR chunk counts as a rejection. */ +async function drain(stream: AsyncIterable<{ type: string }>) { + for await (const chunk of stream) { + if (chunk.type === EventType.RUN_ERROR) throw new Error('RUN_ERROR') + } +} + +const messages = [{ role: 'user' as const, content: 'Hi' }] +const outputSchema = { type: 'object', properties: {} } +const converseModel = 'us.amazon.nova-pro-v1:0' +const mantleModel = 'openai.gpt-oss-120b-1:0' + +/** A Converse adapter whose AWS SDK client sends through the fake fetch. */ +function converseAdapter(wire: ReturnType) { + const client = new BedrockRuntimeClient({ + region: 'us-east-1', + maxAttempts: 1, + credentials: { accessKeyId: 'test', secretAccessKey: 'test' }, + requestHandler: { + handle: (_request: unknown, options?: { abortSignal?: AbortSignal }) => + wire.fetch('https://bedrock.test', { + ...(options?.abortSignal && { signal: options.abortSignal }), + }), + }, + }) + class WireAdapter extends BedrockConverseTextAdapter { + protected override async getClient(): Promise { + return client + } + } + return new WireAdapter({ apiKey: 'test-key' }, converseModel) +} + +describe('Bedrock Converse aborts the provider request', () => { + it('chatStream', async () => { + const wire = hangingFetch() + const adapter = converseAdapter(wire) + await expectAbortReachesFetch(wire, (signal) => + drain( + adapter.chatStream({ + threadId: 'thread-1', + logger, + model: converseModel, + messages, + request: { signal }, + }), + ), + ) + }) + + it('structuredOutput', async () => { + const wire = hangingFetch() + const adapter = converseAdapter(wire) + await expectAbortReachesFetch(wire, (signal) => + adapter.structuredOutput({ + chatOptions: { + threadId: 'thread-1', + logger, + model: converseModel, + messages, + request: { signal }, + }, + outputSchema, + }), + ) + }) + + it('structuredOutputStream', async () => { + const wire = hangingFetch() + const adapter = converseAdapter(wire) + await expectAbortReachesFetch(wire, (signal) => + drain( + adapter.structuredOutputStream({ + chatOptions: { + threadId: 'thread-1', + logger, + model: converseModel, + messages, + request: { signal }, + }, + outputSchema, + }), + ), + ) + }) +}) + +describe.each([ + ['Mantle chat', createBedrockChat], + ['Mantle responses', createBedrockResponsesText], +] as const)('Bedrock %s aborts the provider request', (_name, create) => { + function adapter(wire: ReturnType) { + return create(mantleModel, 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + } + + it('chatStream', async () => { + const wire = hangingFetch() + const a = adapter(wire) + await expectAbortReachesFetch(wire, (signal) => + drain( + a.chatStream({ + threadId: 'thread-1', + logger, + model: mantleModel, + messages, + request: { signal }, + }), + ), + ) + }) + + it('structuredOutput', async () => { + const wire = hangingFetch() + const a = adapter(wire) + await expectAbortReachesFetch(wire, (signal) => + a.structuredOutput({ + chatOptions: { + threadId: 'thread-1', + logger, + model: mantleModel, + messages, + request: { signal }, + }, + outputSchema, + }), + ) + }) +}) diff --git a/packages/ai-byteplus/src/adapters/image.ts b/packages/ai-byteplus/src/adapters/image.ts index b68cddf7e0..00ad808b93 100644 --- a/packages/ai-byteplus/src/adapters/image.ts +++ b/packages/ai-byteplus/src/adapters/image.ts @@ -218,7 +218,10 @@ export class BytePlusImageAdapter< ) const fetchImpl = this.clientConfig.fetch ?? fetch - const signal = bytePlusTimeoutSignal(this.clientConfig.timeout) + const signal = bytePlusTimeoutSignal( + this.clientConfig.timeout, + options.abortSignal, + ) const response = await fetchImpl( `${this.clientConfig.baseURL}/images/generations`, { diff --git a/packages/ai-byteplus/src/adapters/transcription.ts b/packages/ai-byteplus/src/adapters/transcription.ts index feecdebca4..ea413eb757 100644 --- a/packages/ai-byteplus/src/adapters/transcription.ts +++ b/packages/ai-byteplus/src/adapters/transcription.ts @@ -160,6 +160,7 @@ export class BytePlusTranscriptionAdapter< [BYTEPLUS_ASR_RESOURCE_HEADER]: BYTEPLUS_ASR_RESOURCE_ID, }), body: JSON.stringify(body), + ...(options.abortSignal && { signal: options.abortSignal }), }, ) diff --git a/packages/ai-byteplus/src/adapters/tts.ts b/packages/ai-byteplus/src/adapters/tts.ts index d8d8de9ef0..e3c3725cfc 100644 --- a/packages/ai-byteplus/src/adapters/tts.ts +++ b/packages/ai-byteplus/src/adapters/tts.ts @@ -189,6 +189,7 @@ export class BytePlusTTSAdapter< 'X-Api-Request-Id': newRequestId(), }), body: JSON.stringify(body), + ...(options.abortSignal && { signal: options.abortSignal }), }, ) diff --git a/packages/ai-byteplus/src/adapters/video.ts b/packages/ai-byteplus/src/adapters/video.ts index 777991f988..8fc66966be 100644 --- a/packages/ai-byteplus/src/adapters/video.ts +++ b/packages/ai-byteplus/src/adapters/video.ts @@ -211,9 +211,10 @@ export class BytePlusVideoAdapter< private async request( path: string, init?: Omit, + abortSignal?: AbortSignal, ): Promise<{ response: Response; body: unknown }> { const fetchImpl = this.clientConfig.fetch ?? fetch - const signal = bytePlusTimeoutSignal(this.clientConfig.timeout) + const signal = bytePlusTimeoutSignal(this.clientConfig.timeout, abortSignal) const response = await fetchImpl(`${this.clientConfig.baseURL}${path}`, { ...init, ...(signal && { signal }), @@ -462,10 +463,11 @@ export class BytePlusVideoAdapter< { provider: this.name, model }, ) - const { response, body } = await this.request(TASKS_PATH, { - method: 'POST', - body: JSON.stringify(request), - }) + const { response, body } = await this.request( + TASKS_PATH, + { method: 'POST', body: JSON.stringify(request) }, + options.abortSignal, + ) if (!response.ok) { throw bytePlusArkError(response.status, body, 'video task creation') } diff --git a/packages/ai-byteplus/src/utils/client.ts b/packages/ai-byteplus/src/utils/client.ts index 164756f930..b1e6b0d80d 100644 --- a/packages/ai-byteplus/src/utils/client.ts +++ b/packages/ai-byteplus/src/utils/client.ts @@ -232,13 +232,21 @@ function applyReservedHeaders( * * `maxRetries` has no equivalent here and stays SDK-path-only; it is * documented as such on {@link BytePlusArkConfig}. + * + * The caller's `abortSignal`, when given, is combined with the timeout, so an + * abort cancels the request too. */ export function bytePlusTimeoutSignal( timeout: number | undefined, + abortSignal?: AbortSignal, ): AbortSignal | undefined { - return typeof timeout === 'number' && timeout > 0 - ? AbortSignal.timeout(timeout) - : undefined + const timeoutSignal = + typeof timeout === 'number' && timeout > 0 + ? AbortSignal.timeout(timeout) + : undefined + return timeoutSignal && abortSignal + ? AbortSignal.any([timeoutSignal, abortSignal]) + : (timeoutSignal ?? abortSignal) } /** diff --git a/packages/ai-byteplus/tests/abort-signal-wire.test.ts b/packages/ai-byteplus/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..de02b6ff1a --- /dev/null +++ b/packages/ai-byteplus/tests/abort-signal-wire.test.ts @@ -0,0 +1,139 @@ +import { describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createBytePlusImage } from '../src/adapters/image' +import { createBytePlusText } from '../src/adapters/text' +import { createBytePlusTranscription } from '../src/adapters/transcription' +import { createBytePlusSpeech } from '../src/adapters/tts' +import { createBytePlusVideo } from '../src/adapters/video' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +describe('BytePlus adapters abort the provider request', () => { + it.each(['stream', 'structured'] as const)('text %s', async (mode) => { + const wire = hangingFetch() + const adapter = createBytePlusText('seed-1-6-250915', 'ark-test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, async (signal) => { + const chatOptions = { + threadId: 'thread-1', + logger, + model: 'seed-1-6-250915' as const, + messages: [{ role: 'user' as const, content: 'Hi' }], + request: { signal }, + } + if (mode === 'structured') { + await adapter.structuredOutput({ + chatOptions, + outputSchema: { type: 'object', properties: {} }, + }) + return + } + // The stream reports an abort as a RUN_ERROR chunk, not a throw. + for await (const chunk of adapter.chatStream(chatOptions)) { + if (chunk.type === 'RUN_ERROR') throw new Error(chunk.message) + } + }) + }) + + // The config timeout must not replace the caller signal. + it.each([undefined, 60_000])('image (timeout %s)', async (timeout) => { + const wire = hangingFetch() + const adapter = createBytePlusImage('seedream-4-0-250828', 'ark-test-key', { + fetch: wire.fetch, + ...(timeout && { timeout }), + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: 'seedream-4-0-250828', + prompt: 'a guitar', + abortSignal, + logger, + }), + ) + }) + + it.each([undefined, 60_000])('video create (timeout %s)', async (timeout) => { + const wire = hangingFetch() + const adapter = createBytePlusVideo( + 'seedance-1-0-pro-fast-251015', + 'ark-test-key', + { fetch: wire.fetch, ...(timeout && { timeout }) }, + ) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'seedance-1-0-pro-fast-251015', + prompt: 'a guitar', + abortSignal, + logger, + }), + ) + }) + + it('transcription', async () => { + const wire = hangingFetch() + const adapter = createBytePlusTranscription('seed-asr', 'voice-key', { + fetch: wire.fetch, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.transcribe({ + model: 'seed-asr', + audio: 'https://example.com/guitar.mp3', + abortSignal, + logger, + }), + ) + }) + + it('speech', async () => { + const wire = hangingFetch() + const adapter = createBytePlusSpeech('seed-audio-1.0', 'voice-key', { + fetch: wire.fetch, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateSpeech({ + model: 'seed-audio-1.0', + text: 'hello', + abortSignal, + logger, + }), + ) + }) +}) diff --git a/packages/ai-cloudflare/tests/abort-signal-wire.test.ts b/packages/ai-cloudflare/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..3ed376d96e --- /dev/null +++ b/packages/ai-cloudflare/tests/abort-signal-wire.test.ts @@ -0,0 +1,153 @@ +import { describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createCloudflareDecider } from '../src/adapters/evaluate' +import { createCloudflareImage } from '../src/adapters/image' +import { createCloudflareText } from '../src/adapters/text' +import { createCloudflareTranscription } from '../src/adapters/transcription' +import { createCloudflareTTS } from '../src/adapters/tts' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +// Binding mode (`env.AI.run`) is not covered: the binding takes no abort +// signal, so there is no request to cancel. Only REST mode can abort. +const rest = (fetch: typeof globalThis.fetch) => ({ + accountId: 'account-1', + apiKey: 'test-key', + fetch, +}) +const audio = new Uint8Array([1, 2, 3]).buffer + +describe('Cloudflare REST adapters abort the provider request', () => { + it.each(['stream', 'structured'] as const)('text %s', async (mode) => { + const wire = hangingFetch() + const adapter = createCloudflareText('@cf/test', { + ...rest(wire.fetch), + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, async (signal) => { + const chatOptions = { + threadId: 'thread-1', + logger, + model: '@cf/test', + messages: [{ role: 'user' as const, content: 'Hi' }], + request: { signal }, + } + if (mode === 'structured') { + await adapter.structuredOutput({ + chatOptions, + outputSchema: { type: 'object', properties: {} }, + }) + return + } + // The stream reports an abort as a RUN_ERROR chunk, not a throw. + for await (const chunk of adapter.chatStream(chatOptions)) { + if (chunk.type === 'RUN_ERROR') throw new Error(chunk.message) + } + }) + }) + + it('image', async () => { + const wire = hangingFetch() + const adapter = createCloudflareImage('@cf/test', rest(wire.fetch)) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: '@cf/test', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('speech', async () => { + const wire = hangingFetch() + const adapter = createCloudflareTTS('@cf/test', rest(wire.fetch)) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateSpeech({ + model: '@cf/test', + text: 'hello', + abortSignal, + logger, + }), + ) + }) + + it.each(['@cf/openai/whisper-large-v3-turbo', '@cf/deepgram/nova-3'])( + 'transcription (%s)', + async (model) => { + const wire = hangingFetch() + const adapter = createCloudflareTranscription(model, rest(wire.fetch)) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.transcribe({ model, audio, abortSignal, logger }), + ) + }, + ) + + it('transcription: audio URL fetch', async () => { + const wire = hangingFetch() + const adapter = createCloudflareTranscription( + '@cf/openai/whisper-large-v3-turbo', + rest(wire.fetch), + ) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.transcribe({ + model: '@cf/openai/whisper-large-v3-turbo', + audio: 'https://example.test/a.mp3', + abortSignal, + logger, + }), + ) + }) + + it('evaluate', async () => { + const wire = hangingFetch() + const adapter = createCloudflareDecider('typesafe/jev', rest(wire.fetch)) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.evaluate({ + model: 'typesafe/jev', + state: { subject: 'Invoice 1842' }, + questions: { + refund: { type: 'noul', instructions: 'Refund asked?' }, + }, + abortSignal, + logger, + }), + ) + }) +}) diff --git a/packages/ai-elevenlabs/src/adapters/audio.ts b/packages/ai-elevenlabs/src/adapters/audio.ts index bf62b7249c..49c4573ddf 100644 --- a/packages/ai-elevenlabs/src/adapters/audio.ts +++ b/packages/ai-elevenlabs/src/adapters/audio.ts @@ -149,26 +149,29 @@ export class ElevenLabsAudioAdapter< const music = (options.modelOptions ?? {}) as ElevenLabsMusicProviderOptions const outputFormat = music.outputFormat - const stream = await this.client.music.compose({ - modelId, - ...(options.prompt && !music.compositionPlan - ? { prompt: options.prompt } - : {}), - ...(music.compositionPlan - ? { compositionPlan: toMusicPrompt(music.compositionPlan) } - : {}), - ...(options.duration != null && !music.compositionPlan - ? { musicLengthMs: Math.round(options.duration * 1000) } - : {}), - ...(outputFormat ? { outputFormat } : {}), - ...(music.seed != null ? { seed: music.seed } : {}), - ...(music.forceInstrumental != null - ? { forceInstrumental: music.forceInstrumental } - : {}), - ...(music.respectSectionsDurations != null - ? { respectSectionsDurations: music.respectSectionsDurations } - : {}), - }) + const stream = await this.client.music.compose( + { + modelId, + ...(options.prompt && !music.compositionPlan + ? { prompt: options.prompt } + : {}), + ...(music.compositionPlan + ? { compositionPlan: toMusicPrompt(music.compositionPlan) } + : {}), + ...(options.duration != null && !music.compositionPlan + ? { musicLengthMs: Math.round(options.duration * 1000) } + : {}), + ...(outputFormat ? { outputFormat } : {}), + ...(music.seed != null ? { seed: music.seed } : {}), + ...(music.forceInstrumental != null + ? { forceInstrumental: music.forceInstrumental } + : {}), + ...(music.respectSectionsDurations != null + ? { respectSectionsDurations: music.respectSectionsDurations } + : {}), + }, + options.abortSignal ? { abortSignal: options.abortSignal } : {}, + ) return this.finalize(stream, outputFormat, options.duration) } @@ -182,19 +185,22 @@ export class ElevenLabsAudioAdapter< {}) as ElevenLabsSoundEffectsProviderOptions const outputFormat = sfx.outputFormat - const stream = await this.client.textToSoundEffects.convert({ - text: options.prompt, - // The SDK types list only the sound models it knew at release. - modelId: modelId as ElevenLabs.SfxModelId, - ...(options.duration != null - ? { durationSeconds: options.duration } - : {}), - ...(outputFormat ? { outputFormat } : {}), - ...(sfx.promptInfluence != null - ? { promptInfluence: sfx.promptInfluence } - : {}), - ...(sfx.loop != null ? { loop: sfx.loop } : {}), - }) + const stream = await this.client.textToSoundEffects.convert( + { + text: options.prompt, + // The SDK types list only the sound models it knew at release. + modelId: modelId as ElevenLabs.SfxModelId, + ...(options.duration != null + ? { durationSeconds: options.duration } + : {}), + ...(outputFormat ? { outputFormat } : {}), + ...(sfx.promptInfluence != null + ? { promptInfluence: sfx.promptInfluence } + : {}), + ...(sfx.loop != null ? { loop: sfx.loop } : {}), + }, + options.abortSignal ? { abortSignal: options.abortSignal } : {}, + ) return this.finalize(stream, outputFormat, options.duration) } diff --git a/packages/ai-elevenlabs/src/adapters/speech.ts b/packages/ai-elevenlabs/src/adapters/speech.ts index ad022ea31d..ba2a94da17 100644 --- a/packages/ai-elevenlabs/src/adapters/speech.ts +++ b/packages/ai-elevenlabs/src/adapters/speech.ts @@ -137,6 +137,9 @@ export class ElevenLabsSpeechAdapter< const { format, contentType } = wrapAsWav ? { format: 'wav', contentType: 'audio/wav' } : parseOutputFormat(effectiveOutputFormat) + const requestOptions = options.abortSignal + ? { abortSignal: options.abortSignal } + : {} // `format: 'wav'` asks ElevenLabs for pcm_44100 and wraps it here, on // every endpoint: the byte-stream pair hands back raw bytes, the // timestamped pair hands back the same bytes as base64. @@ -172,7 +175,10 @@ export class ElevenLabsSpeechAdapter< if (options.timestamps) { const response = - await this.client.textToDialogue.convertWithTimestamps(dialogue) + await this.client.textToDialogue.convertWithTimestamps( + dialogue, + requestOptions, + ) return { id: generateId(this.name), model: this.model, @@ -183,7 +189,10 @@ export class ElevenLabsSpeechAdapter< } } - const stream = await this.client.textToDialogue.convert(dialogue) + const stream = await this.client.textToDialogue.convert( + dialogue, + requestOptions, + ) return { id: generateId(this.name), model: this.model, @@ -221,6 +230,7 @@ export class ElevenLabsSpeechAdapter< const response = await this.client.textToSpeech.convertWithTimestamps( voiceId, single, + requestOptions, ) return { id: generateId(this.name), @@ -232,12 +242,16 @@ export class ElevenLabsSpeechAdapter< } } - const stream = await this.client.textToSpeech.convert(voiceId, { - ...single, - ...(optimizeStreamingLatency != null - ? { optimizeStreamingLatency } - : {}), - }) + const stream = await this.client.textToSpeech.convert( + voiceId, + { + ...single, + ...(optimizeStreamingLatency != null + ? { optimizeStreamingLatency } + : {}), + }, + requestOptions, + ) return { id: generateId(this.name), diff --git a/packages/ai-elevenlabs/src/adapters/transcription.ts b/packages/ai-elevenlabs/src/adapters/transcription.ts index fa459b9928..0145425b62 100644 --- a/packages/ai-elevenlabs/src/adapters/transcription.ts +++ b/packages/ai-elevenlabs/src/adapters/transcription.ts @@ -101,53 +101,56 @@ export class ElevenLabsTranscriptionAdapter< const modelOpts = options.modelOptions ?? {} const audioInput = normalizeAudioInput(options.audio) - const response = await this.client.speechToText.convert({ - modelId: this.model, - ...(audioInput.kind === 'file' - ? { file: audioInput.value } - : { cloudStorageUrl: audioInput.value }), - ...(options.language ? { languageCode: options.language } : {}), - ...(modelOpts.tagAudioEvents != null - ? { tagAudioEvents: modelOpts.tagAudioEvents } - : {}), - ...(modelOpts.numSpeakers != null - ? { numSpeakers: modelOpts.numSpeakers } - : {}), - ...(modelOpts.timestampsGranularity - ? { timestampsGranularity: modelOpts.timestampsGranularity } - : {}), - ...(modelOpts.diarize != null ? { diarize: modelOpts.diarize } : {}), - ...(modelOpts.diarizationThreshold != null - ? { diarizationThreshold: modelOpts.diarizationThreshold } - : {}), - ...(modelOpts.detectSpeakerRoles != null - ? { detectSpeakerRoles: modelOpts.detectSpeakerRoles } - : {}), - ...(modelOpts.keyterms ? { keyterms: modelOpts.keyterms } : {}), - ...(modelOpts.entityDetection - ? { entityDetection: modelOpts.entityDetection } - : {}), - ...(modelOpts.entityRedaction - ? { entityRedaction: modelOpts.entityRedaction } - : {}), - ...(modelOpts.entityRedactionMode - ? { entityRedactionMode: modelOpts.entityRedactionMode } - : {}), - ...(modelOpts.noVerbatim != null - ? { noVerbatim: modelOpts.noVerbatim } - : {}), - ...(modelOpts.temperature != null - ? { temperature: modelOpts.temperature } - : {}), - ...(modelOpts.seed != null ? { seed: modelOpts.seed } : {}), - ...(modelOpts.enableLogging != null - ? { enableLogging: modelOpts.enableLogging } - : {}), - ...(modelOpts.useMultiChannel != null - ? { useMultiChannel: modelOpts.useMultiChannel } - : {}), - ...(modelOpts.fileFormat ? { fileFormat: modelOpts.fileFormat } : {}), - } as Parameters[0]) + const response = await this.client.speechToText.convert( + { + modelId: this.model, + ...(audioInput.kind === 'file' + ? { file: audioInput.value } + : { cloudStorageUrl: audioInput.value }), + ...(options.language ? { languageCode: options.language } : {}), + ...(modelOpts.tagAudioEvents != null + ? { tagAudioEvents: modelOpts.tagAudioEvents } + : {}), + ...(modelOpts.numSpeakers != null + ? { numSpeakers: modelOpts.numSpeakers } + : {}), + ...(modelOpts.timestampsGranularity + ? { timestampsGranularity: modelOpts.timestampsGranularity } + : {}), + ...(modelOpts.diarize != null ? { diarize: modelOpts.diarize } : {}), + ...(modelOpts.diarizationThreshold != null + ? { diarizationThreshold: modelOpts.diarizationThreshold } + : {}), + ...(modelOpts.detectSpeakerRoles != null + ? { detectSpeakerRoles: modelOpts.detectSpeakerRoles } + : {}), + ...(modelOpts.keyterms ? { keyterms: modelOpts.keyterms } : {}), + ...(modelOpts.entityDetection + ? { entityDetection: modelOpts.entityDetection } + : {}), + ...(modelOpts.entityRedaction + ? { entityRedaction: modelOpts.entityRedaction } + : {}), + ...(modelOpts.entityRedactionMode + ? { entityRedactionMode: modelOpts.entityRedactionMode } + : {}), + ...(modelOpts.noVerbatim != null + ? { noVerbatim: modelOpts.noVerbatim } + : {}), + ...(modelOpts.temperature != null + ? { temperature: modelOpts.temperature } + : {}), + ...(modelOpts.seed != null ? { seed: modelOpts.seed } : {}), + ...(modelOpts.enableLogging != null + ? { enableLogging: modelOpts.enableLogging } + : {}), + ...(modelOpts.useMultiChannel != null + ? { useMultiChannel: modelOpts.useMultiChannel } + : {}), + ...(modelOpts.fileFormat ? { fileFormat: modelOpts.fileFormat } : {}), + } as Parameters[0], + options.abortSignal ? { abortSignal: options.abortSignal } : {}, + ) return this.transformResponse(response) } catch (error) { diff --git a/packages/ai-elevenlabs/tests/abort-signal-wire.test.ts b/packages/ai-elevenlabs/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..869311b995 --- /dev/null +++ b/packages/ai-elevenlabs/tests/abort-signal-wire.test.ts @@ -0,0 +1,143 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createElevenLabsAudio } from '../src/adapters/audio' +import { createElevenLabsSpeech } from '../src/adapters/speech' +import { createElevenLabsTranscription } from '../src/adapters/transcription' +import { createElevenLabsVoiceDesign } from '../src/adapters/voice' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +// The adapter config has no `fetch` option, so stub the global one. +function stubFetch() { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + return wire +} + +const config = { maxRetries: 0 } + +afterEach(() => { + vi.unstubAllGlobals() +}) + +describe('ElevenLabs adapters abort the provider request', () => { + it.each([ + ['speech', false, false], + ['speech with timestamps', false, true], + ['dialogue', true, false], + ['dialogue with timestamps', true, true], + ] as const)('%s', async (_name, dialogue, timestamps) => { + const wire = stubFetch() + const adapter = createElevenLabsSpeech('eleven_v3', 'test-key', config) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateSpeech({ + model: 'eleven_v3', + text: 'hello', + voice: 'voice-1', + ...(dialogue && { + turns: [ + { text: 'hi', voice: 'voice-1' }, + { text: 'hey', voice: 'voice-2' }, + ], + }), + timestamps, + abortSignal, + logger, + }), + ) + }) + + it('listVoices', async () => { + const wire = stubFetch() + const adapter = createElevenLabsSpeech('eleven_v3', 'test-key', config) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.listVoices({ abortSignal }), + ) + }) + + it.each(['music_v1', 'eleven_text_to_sound_v2'] as const)( + 'audio (%s)', + async (model) => { + const wire = stubFetch() + const adapter = createElevenLabsAudio(model, 'test-key', config) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateAudio({ + model, + prompt: 'rain on a roof', + abortSignal, + logger, + }), + ) + }, + ) + + it('transcription', async () => { + const wire = stubFetch() + const adapter = createElevenLabsTranscription( + 'scribe_v2', + 'test-key', + config, + ) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.transcribe({ + model: 'scribe_v2', + audio: new Blob([new Uint8Array([1, 2, 3])], { type: 'audio/mpeg' }), + abortSignal, + logger, + }), + ) + }) + + it('voice design', async () => { + const wire = stubFetch() + const adapter = createElevenLabsVoiceDesign( + 'eleven_ttv_v3', + 'test-key', + config, + ) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateVoice({ + model: 'eleven_ttv_v3', + prompt: 'a calm narrator', + abortSignal, + logger, + }), + ) + }) +}) diff --git a/packages/ai-fal/src/adapters/audio.ts b/packages/ai-fal/src/adapters/audio.ts index 26c067e93f..6373d9ba1f 100644 --- a/packages/ai-fal/src/adapters/audio.ts +++ b/packages/ai-fal/src/adapters/audio.ts @@ -82,15 +82,24 @@ export class FalAudioAdapter extends BaseAudioAdapter< provider: 'fal', model: this.model, }) + let requestId: string | undefined try { const input = this.buildInput(options) // Request-specific abortSignal only — not fal.config() (global). const result = await fal.subscribe(this.model, { input, ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}), + onEnqueue: (id) => { + requestId = id + }, }) return this.transformResponse(result) } catch (error) { + // The SDK stops waiting on abort but leaves the queued job running. + // Cancel it so it stops billing. + if (requestId && options.abortSignal?.aborted) { + fal.queue.cancel(this.model, { requestId }).catch(() => {}) + } logger.errors('fal.generateAudio fatal', { error, source: 'fal.generateAudio', diff --git a/packages/ai-fal/src/adapters/image.ts b/packages/ai-fal/src/adapters/image.ts index bf599784e5..ef04466d7c 100644 --- a/packages/ai-fal/src/adapters/image.ts +++ b/packages/ai-fal/src/adapters/image.ts @@ -87,6 +87,7 @@ export class FalImageAdapter extends BaseImageAdapter< ) } + let requestId: string | undefined try { const input = this.buildInput(options, resolved) // Pass request-specific abortSignal only — never via fal.config(), which @@ -94,9 +95,17 @@ export class FalImageAdapter extends BaseImageAdapter< const result = await fal.subscribe(this.model, { input, ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}), + onEnqueue: (id) => { + requestId = id + }, }) return this.transformResponse(result) } catch (error) { + // The SDK stops waiting on abort but leaves the queued job running. + // Cancel it so it stops billing. + if (requestId && options.abortSignal?.aborted) { + fal.queue.cancel(this.model, { requestId }).catch(() => {}) + } logger.errors('fal.generateImage fatal', { error, source: 'fal.generateImage', diff --git a/packages/ai-fal/src/adapters/speech.ts b/packages/ai-fal/src/adapters/speech.ts index 44f9a05e45..420b8a5f04 100644 --- a/packages/ai-fal/src/adapters/speech.ts +++ b/packages/ai-fal/src/adapters/speech.ts @@ -54,15 +54,24 @@ export class FalSpeechAdapter extends BaseTTSAdapter< provider: 'fal', model: this.model, }) + let requestId: string | undefined try { const input = this.buildInput(options) // Request-specific abortSignal only — not fal.config() (global). const result = await fal.subscribe(this.model, { input, ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}), + onEnqueue: (id) => { + requestId = id + }, }) - return await this.transformResponse(result) + return await this.transformResponse(result, options.abortSignal) } catch (error) { + // The SDK stops waiting on abort but leaves the queued job running. + // Cancel it so it stops billing. + if (requestId && options.abortSignal?.aborted) { + fal.queue.cancel(this.model, { requestId }).catch(() => {}) + } logger.errors('fal.generateSpeech fatal', { error, source: 'fal.generateSpeech', @@ -91,6 +100,7 @@ export class FalSpeechAdapter extends BaseTTSAdapter< private async transformResponse( response: Result>, + abortSignal: AbortSignal | undefined, ): Promise { const data = response.data as Record @@ -117,7 +127,10 @@ export class FalSpeechAdapter extends BaseTTSAdapter< // Fetch the audio and convert to base64 to match TTSResult contract. // Using a chunked helper here — spreading Uint8Array into btoa exceeds // V8's argument limit (~65k) for any realistic TTS clip. - const audioResponse = await fetch(audioUrl) + const audioResponse = await fetch( + audioUrl, + abortSignal ? { signal: abortSignal } : undefined, + ) if (!audioResponse.ok) { throw new Error( `Failed to fetch audio from ${audioUrl}: ${audioResponse.status} ${audioResponse.statusText}`, diff --git a/packages/ai-fal/src/adapters/transcription.ts b/packages/ai-fal/src/adapters/transcription.ts index 4a0a24d096..54ca7f5e95 100644 --- a/packages/ai-fal/src/adapters/transcription.ts +++ b/packages/ai-fal/src/adapters/transcription.ts @@ -68,15 +68,24 @@ export class FalTranscriptionAdapter< model: this.model, }, ) + let requestId: string | undefined try { const input = this.buildInput(options) // Request-specific abortSignal only — not fal.config() (global). const result = await fal.subscribe(this.model, { input, ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}), + onEnqueue: (id) => { + requestId = id + }, }) return this.transformResponse(result) } catch (error) { + // The SDK stops waiting on abort but leaves the queued job running. + // Cancel it so it stops billing. + if (requestId && options.abortSignal?.aborted) { + fal.queue.cancel(this.model, { requestId }).catch(() => {}) + } logger.errors('fal.generateTranscription fatal', { error, source: 'fal.generateTranscription', diff --git a/packages/ai-fal/tests/abort-signal-wire.test.ts b/packages/ai-fal/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..73c43b73b8 --- /dev/null +++ b/packages/ai-fal/tests/abort-signal-wire.test.ts @@ -0,0 +1,206 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { falAudio } from '../src/adapters/audio' +import { falImage } from '../src/adapters/image' +import { falSpeech } from '../src/adapters/speech' +import { falTranscription } from '../src/adapters/transcription' +import { falVideo } from '../src/adapters/video' + +const logger = resolveDebugOption(false) + +function json(body: unknown) { + return new Response(JSON.stringify(body), { + headers: { 'content-type': 'application/json' }, + }) +} + +/** Rejects at once without a signal, else rejects only when it aborts. */ +function hang(signals: Array, init?: RequestInit) { + const signal = init?.signal + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) +} + +/** + * A fake fal queue. Submit answers with request `req-1`, cancel answers at + * once, and every status or result poll hangs until its signal aborts. With + * `completed`, polls answer with a finished job whose result is `result`. + */ +function falQueue(completed?: { result: unknown }) { + const signals: Array = [] + const cancels: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const url = input instanceof Request ? input.url : String(input) + if (url.endsWith('/cancel')) { + cancels.push(url) + return Promise.resolve(json({ status: 'CANCELLATION_REQUESTED' })) + } + if (init?.method?.toLowerCase() === 'post') { + return Promise.resolve(json({ request_id: 'req-1' })) + } + if (completed) { + return Promise.resolve( + json( + url.includes('/status') ? { status: 'COMPLETED' } : completed.result, + ), + ) + } + return hang(signals, init) + }) + return { fetch, signals, cancels } +} + +/** + * Starts the call, aborts the caller, and expects the request to abort, the + * queued job to be cancelled, and the call to reject. + */ +async function expectAbortCancelsJob( + queue: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(queue.signals.length).toBeGreaterThan(0)) + controller.abort() + await vi.waitFor( + () => expect(queue.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() + await vi.waitFor( + () => + expect(queue.cancels).toEqual([expect.stringContaining('req-1/cancel')]), + { timeout: 500 }, + ) +} + +afterEach(() => { + vi.unstubAllGlobals() +}) + +describe('fal adapters abort the provider request', () => { + it('image', async () => { + const queue = falQueue() + const adapter = falImage('fal-ai/flux/dev', { + apiKey: 'test-key', + fetch: queue.fetch, + }) + await expectAbortCancelsJob(queue, (abortSignal) => + adapter.generateImages({ + model: 'fal-ai/flux/dev', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('audio', async () => { + const queue = falQueue() + const adapter = falAudio('fal-ai/minimax-music/v2', { + apiKey: 'test-key', + fetch: queue.fetch, + }) + await expectAbortCancelsJob(queue, (abortSignal) => + adapter.generateAudio({ + model: 'fal-ai/minimax-music/v2', + prompt: 'calm piano', + abortSignal, + logger, + }), + ) + }) + + it('speech', async () => { + const queue = falQueue() + const adapter = falSpeech('fal-ai/index-tts-2/text-to-speech', { + apiKey: 'test-key', + fetch: queue.fetch, + }) + await expectAbortCancelsJob(queue, (abortSignal) => + adapter.generateSpeech({ + model: 'fal-ai/index-tts-2/text-to-speech', + text: 'hello', + abortSignal, + logger, + }), + ) + }) + + it('transcription', async () => { + const queue = falQueue() + const adapter = falTranscription('fal-ai/whisper', { + apiKey: 'test-key', + fetch: queue.fetch, + }) + await expectAbortCancelsJob(queue, (abortSignal) => + adapter.transcribe({ + model: 'fal-ai/whisper', + audio: 'https://example.test/a.mp3', + abortSignal, + logger, + }), + ) + }) + + it('speech: audio download', async () => { + const queue = falQueue({ + result: { audio: { url: 'https://fal.media/files/speech.mp3' } }, + }) + const download: Array = [] + const downloadFetch = vi.fn( + (_input: RequestInfo | URL, init?: RequestInit) => hang(download, init), + ) + vi.stubGlobal('fetch', downloadFetch) + const adapter = falSpeech('fal-ai/index-tts-2/text-to-speech', { + apiKey: 'test-key', + fetch: queue.fetch, + }) + const controller = new AbortController() + const run = adapter.generateSpeech({ + model: 'fal-ai/index-tts-2/text-to-speech', + text: 'hello', + abortSignal: controller.signal, + logger, + }) + run.catch(() => {}) + await vi.waitFor(() => expect(downloadFetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(download.at(-1)?.aborted, 'download aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() + }) + + it('video submit', async () => { + const signals: Array = [] + const fetch = vi.fn((_input: RequestInfo | URL, init?: RequestInit) => + hang(signals, init), + ) + const adapter = falVideo('fal-ai/veo3/image-to-video', { + apiKey: 'test-key', + fetch, + }) + const controller = new AbortController() + const run = adapter.createVideoJob({ + model: 'fal-ai/veo3/image-to-video', + prompt: 'a cat', + abortSignal: controller.signal, + logger, + }) + run.catch(() => {}) + await vi.waitFor(() => expect(fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() + }) +}) diff --git a/packages/ai-fal/tests/speech-adapter.test.ts b/packages/ai-fal/tests/speech-adapter.test.ts index cb3d531c8b..ca624a8112 100644 --- a/packages/ai-fal/tests/speech-adapter.test.ts +++ b/packages/ai-fal/tests/speech-adapter.test.ts @@ -190,7 +190,10 @@ describe('Fal Speech Adapter', () => { }) expect(result.format).toBe('mp3') - expect(mockFetch).toHaveBeenCalledWith('https://fal.media/files/speech.mp3') + expect(mockFetch).toHaveBeenCalledWith( + 'https://fal.media/files/speech.mp3', + undefined, + ) }) it('handles large audio payloads without blowing the stack', async () => { diff --git a/packages/ai-gemini/src/adapters/audio.ts b/packages/ai-gemini/src/adapters/audio.ts index fa04985cb5..11926e9071 100644 --- a/packages/ai-gemini/src/adapters/audio.ts +++ b/packages/ai-gemini/src/adapters/audio.ts @@ -98,6 +98,7 @@ export class GeminiAudioAdapter< config: { responseModalities: ['AUDIO', 'TEXT'], ...(modelOptions?.seed != null ? { seed: modelOptions.seed } : {}), + ...(options.abortSignal && { abortSignal: options.abortSignal }), }, }) diff --git a/packages/ai-gemini/src/adapters/text.ts b/packages/ai-gemini/src/adapters/text.ts index a5d6acf2f6..cdce069476 100644 --- a/packages/ai-gemini/src/adapters/text.ts +++ b/packages/ai-gemini/src/adapters/text.ts @@ -23,6 +23,7 @@ import { generateId, getGeminiApiKeyFromEnv, isGeminiContextOverflow, + withHttpTimeout, } from '../utils' import { GEMINI_COMBINED_TOOLS_AND_SCHEMA_MODELS, @@ -452,6 +453,7 @@ export class GeminiTextAdapter< override readonly inputModalities = GEMINI_MODEL_INPUT_MODALITIES[this.model] private readonly client: GoogleGenAI + private readonly timeoutMs: number | undefined constructor(config: GeminiTextConfig, model: TModel) { super({}, model) @@ -464,6 +466,7 @@ export class GeminiTextAdapter< ? 'google-vertex' : 'google-generative-ai' this.client = createGeminiClient(config) + this.timeoutMs = config.httpOptions?.timeout } async *chatStream( @@ -570,13 +573,16 @@ export class GeminiTextAdapter< // The installed @google/genai (2.10.0) Interactions `VideoContent` type // predates the `processing` field, so the structurally-built input is // cast at the call boundary. The SDK forwards it to the wire unchanged. - const interaction = await this.client.interactions.create({ - model, - ...(systemInstruction !== undefined && { - system_instruction: systemInstruction, - }), - input: input as never, - }) + const interaction = await this.client.interactions.create( + { + model, + ...(systemInstruction !== undefined && { + system_instruction: systemInstruction, + }), + input: input as never, + }, + { signal: withHttpTimeout(options.request?.signal, this.timeoutMs) }, + ) const text = interaction.output_text ?? '' const responseMetadata = { diff --git a/packages/ai-gemini/src/adapters/tts.ts b/packages/ai-gemini/src/adapters/tts.ts index 856634b4cf..b8e9c89d4b 100644 --- a/packages/ai-gemini/src/adapters/tts.ts +++ b/packages/ai-gemini/src/adapters/tts.ts @@ -215,6 +215,7 @@ export class GeminiTTSAdapter< config: { responseModalities: ['AUDIO'], speechConfig, + ...(options.abortSignal && { abortSignal: options.abortSignal }), // systemInstruction belongs inside `config` per the @google/genai // contract — matches sibling Gemini adapters (summarize, text). ...(modelOptions?.systemInstruction && { diff --git a/packages/ai-gemini/src/adapters/video.ts b/packages/ai-gemini/src/adapters/video.ts index 4593a8e72f..538f2fd5bb 100644 --- a/packages/ai-gemini/src/adapters/video.ts +++ b/packages/ai-gemini/src/adapters/video.ts @@ -10,7 +10,11 @@ import { } from '@tanstack/ai' import { BaseVideoAdapter } from '@tanstack/ai/adapters' import { arrayBufferToBase64 } from '@tanstack/ai-utils' -import { createGeminiClient, getGeminiApiKeyFromEnv } from '../utils' +import { + createGeminiClient, + getGeminiApiKeyFromEnv, + withHttpTimeout, +} from '../utils' import { getGeminiVideoDurationOptions, isInteractionsVideoModel, @@ -89,6 +93,7 @@ function operationErrorMessage(error: Record): string { async function imagePartToVeoImage( part: ImagePart, allowUrlFetch: boolean, + abortSignal: AbortSignal | undefined, ): Promise { if (part.source.type === 'data') { return { @@ -132,7 +137,9 @@ async function imagePartToVeoImage( `adapter config to opt into fetching. URL: ${url}`, ) } - const response = await fetch(url) + const response = abortSignal + ? await fetch(url, { signal: abortSignal }) + : await fetch(url) if (!response.ok) { throw new Error( `Failed to fetch image input (${response.status} ${response.statusText}): ${url}`, @@ -275,11 +282,13 @@ export class GeminiVideoAdapter< protected client: GoogleGenAI private readonly allowUrlFetch: boolean + private readonly timeoutMs: number | undefined constructor(config: GeminiVideoConfig, model: TModel) { super({}, model) this.client = createGeminiClient(config) this.allowUrlFetch = config.allowUrlFetch ?? false + this.timeoutMs = config.httpOptions?.timeout } async createVideoJob( @@ -319,6 +328,7 @@ export class GeminiVideoAdapter< const { image, lastFrame, referenceImages } = await this.routeImageParts( resolved.images, + options.abortSignal, ) const config: GenerateVideosConfig = { @@ -327,6 +337,7 @@ export class GeminiVideoAdapter< ...(duration !== undefined && { durationSeconds: duration }), ...(lastFrame && { lastFrame }), ...(referenceImages.length > 0 && { referenceImages }), + ...(options.abortSignal && { abortSignal: options.abortSignal }), } const operation = await this.client.models.generateVideos({ @@ -427,14 +438,18 @@ export class GeminiVideoAdapter< } : {} - const interaction = await this.client.interactions.create({ + const request = { ...modelOptions, model: this.model, - input: [{ type: 'user_input', content }], - response_modalities: ['video'], + input: [{ type: 'user_input' as const, content }], + response_modalities: ['video' as const], background: true, ...responseFormat, - }) + } + const signal = withHttpTimeout(options.abortSignal, this.timeoutMs) + const interaction = signal + ? await this.client.interactions.create(request, { signal }) + : await this.client.interactions.create(request) if (!interaction.id) { throw new Error( @@ -457,6 +472,7 @@ export class GeminiVideoAdapter< */ private async routeImageParts( parts: Array>, + abortSignal: AbortSignal | undefined, ): Promise<{ image: Image | undefined lastFrame: Image | undefined @@ -475,13 +491,21 @@ export class GeminiVideoAdapter< `${this.name}: Veo accepts at most one 'end_frame' image.`, ) } - lastFrame = await imagePartToVeoImage(part, this.allowUrlFetch) + lastFrame = await imagePartToVeoImage( + part, + this.allowUrlFetch, + abortSignal, + ) break } case 'reference': case 'character': { referenceImages.push({ - image: await imagePartToVeoImage(part, this.allowUrlFetch), + image: await imagePartToVeoImage( + part, + this.allowUrlFetch, + abortSignal, + ), referenceType: VideoGenerationReferenceType.ASSET, }) break @@ -493,7 +517,11 @@ export class GeminiVideoAdapter< `${this.name}: Veo accepts at most one starting image; received multiple 'start_frame'/un-roled images. Use metadata.role ('end_frame', 'reference') to disambiguate the others.`, ) } - image = await imagePartToVeoImage(part, this.allowUrlFetch) + image = await imagePartToVeoImage( + part, + this.allowUrlFetch, + abortSignal, + ) break } case 'mask': diff --git a/packages/ai-gemini/src/utils/client.ts b/packages/ai-gemini/src/utils/client.ts index 7076056fb8..aef12238a4 100644 --- a/packages/ai-gemini/src/utils/client.ts +++ b/packages/ai-gemini/src/utils/client.ts @@ -43,6 +43,21 @@ export function createGeminiClient(config: GeminiClientConfig): GoogleGenAI { return new GoogleGenAI(options) } +/** + * The signal for a `client.interactions.create` call. That SDK client uses + * `httpOptions.timeout` only when the request has no signal. So with a caller + * signal, add the timeout to the signal here. + */ +export function withHttpTimeout( + signal: AbortSignal | null | undefined, + timeoutMs: number | undefined, +): AbortSignal | undefined { + if (!signal || timeoutMs === undefined || timeoutMs <= 0) { + return signal ?? undefined + } + return AbortSignal.any([signal, AbortSignal.timeout(timeoutMs)]) +} + /** * Gets Google API key from environment variables * @throws Error if GOOGLE_API_KEY or GEMINI_API_KEY is not found diff --git a/packages/ai-gemini/src/utils/index.ts b/packages/ai-gemini/src/utils/index.ts index 8499be23a8..4100879182 100644 --- a/packages/ai-gemini/src/utils/index.ts +++ b/packages/ai-gemini/src/utils/index.ts @@ -3,5 +3,6 @@ export { generateId, getGeminiApiKeyFromEnv, isGeminiContextOverflow, + withHttpTimeout, type GeminiClientConfig, } from './client' diff --git a/packages/ai-gemini/tests/abort-signal-wire.test.ts b/packages/ai-gemini/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..22724e94d9 --- /dev/null +++ b/packages/ai-gemini/tests/abort-signal-wire.test.ts @@ -0,0 +1,336 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createGeminiAudio } from '../src/adapters/audio' +import { createGeminiImage } from '../src/adapters/image' +import { createGeminiChat } from '../src/adapters/text' +import { createGeminiSpeech } from '../src/adapters/tts' +import { createGeminiVideo } from '../src/adapters/video' +import { createGeminiTextInteractions } from '../src/experimental' +import type { ModelMessage } from '@tanstack/ai' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** + * Replaces `interactions.create` of the adapter's SDK client with a call that + * never answers. The Interactions client sends `request.clone()`, and after a + * GC the signal of the clone stops following the caller signal. So a fake + * `fetch` cannot show the abort reliably. This check reads the signal that + * the adapter gives to the SDK. + */ +function hangingInteractionsCreate(adapter: object) { + const { interactions } = (adapter as { client: { interactions: object } }) + .client + const signals: Array = [] + const fetch = vi + .spyOn(Object.getPrototypeOf(interactions), 'create') + .mockImplementation((...args: Array) => { + const signal = (args[1] as { signal?: AbortSignal } | undefined)?.signal + if (!signal) return Promise.reject(new Error('SDK got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: { fetch: object; signals: Array }, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +/** @google/genai has no `fetch` option, so the global one is stubbed. */ +function stubFetch() { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + return wire +} + +afterEach(() => { + vi.restoreAllMocks() + vi.unstubAllGlobals() +}) + +const text: Array = [{ role: 'user', content: 'Hi' }] +const agenticVideo: Array = [ + { + role: 'user', + content: [ + { type: 'text', content: 'What happens?' }, + { + type: 'video', + source: { type: 'url', value: 'files/abc', mimeType: 'video/mp4' }, + metadata: { processing: 'agentic' }, + }, + ], + }, +] +const urlImage = { + type: 'image' as const, + source: { type: 'url' as const, value: 'https://example.test/in.png' }, +} + +type TextAdapter = { + chatStream: (options: never) => AsyncIterable<{ type: string }> + structuredOutput: (options: never) => Promise +} + +/** Runs one text call with the caller signal in `request`. */ +function runText( + adapter: TextAdapter, + mode: 'stream' | 'structured', + messages: Array, + signal: AbortSignal, +) { + const chatOptions = { + threadId: 'thread-1', + logger, + model: 'gemini-3.8-flash', + messages, + request: { signal }, + } + if (mode === 'structured') { + return adapter.structuredOutput({ + chatOptions, + outputSchema: { type: 'object', properties: {} }, + } as never) + } + return (async () => { + // The stream reports an abort as a RUN_ERROR chunk, not a throw. + for await (const chunk of adapter.chatStream(chatOptions as never)) { + if (chunk.type === 'RUN_ERROR') throw new Error('run error') + } + })() +} + +describe('Gemini adapters abort the provider request', () => { + it.each(['stream', 'structured'] as const)('text %s', async (mode) => { + const wire = stubFetch() + const adapter = createGeminiChat('gemini-3.8-flash', 'test-key') + await expectAbortReachesFetch(wire, (signal) => + runText(adapter as unknown as TextAdapter, mode, text, signal), + ) + }) + + it('text stream, agentic video (Interactions API)', async () => { + const adapter = createGeminiChat('gemini-3.8-flash', 'test-key') + const wire = hangingInteractionsCreate(adapter) + await expectAbortReachesFetch(wire, (signal) => + runText( + adapter as unknown as TextAdapter, + 'stream', + agenticVideo, + signal, + ), + ) + }) + + it.each(['stream', 'structured'] as const)( + 'experimental text interactions %s', + async (mode) => { + const adapter = createGeminiTextInteractions( + 'gemini-3.8-flash', + 'test-key', + ) + const wire = hangingInteractionsCreate(adapter) + await expectAbortReachesFetch(wire, (signal) => + runText(adapter as unknown as TextAdapter, mode, text, signal), + ) + }, + ) + + it('image', async () => { + const adapter = createGeminiImage('gemini-3.1-flash-image', 'test-key') + const wire = hangingInteractionsCreate(adapter) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: 'gemini-3.1-flash-image', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('audio (Lyria)', async () => { + const wire = stubFetch() + const adapter = createGeminiAudio('lyria-3-clip-preview', 'test-key') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateAudio({ + model: 'lyria-3-clip-preview', + prompt: 'calm piano', + abortSignal, + logger, + }), + ) + }) + + it('speech', async () => { + const wire = stubFetch() + const adapter = createGeminiSpeech( + 'gemini-2.5-flash-preview-tts', + 'test-key', + ) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateSpeech({ + model: 'gemini-2.5-flash-preview-tts', + text: 'hello', + abortSignal, + logger, + }), + ) + }) + + it('video create (Veo)', async () => { + const wire = stubFetch() + const adapter = createGeminiVideo('veo-3.1-generate-preview', 'test-key') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'veo-3.1-generate-preview', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('video create (Veo): input image URL fetch', async () => { + const wire = stubFetch() + const adapter = createGeminiVideo('veo-3.1-generate-preview', 'test-key', { + allowUrlFetch: true, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'veo-3.1-generate-preview', + prompt: [{ type: 'text', content: 'a cat' }, urlImage], + abortSignal, + logger, + }), + ) + }) + + it('video create (Omni)', async () => { + const adapter = createGeminiVideo('gemini-omni-1.1-flash', 'test-key') + const wire = hangingInteractionsCreate(adapter) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'gemini-omni-1.1-flash', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) +}) + +// The Interactions client drops its own `httpOptions.timeout` when the +// request has a signal. So a caller signal must not remove that timeout. +describe('Gemini Interactions calls keep httpOptions.timeout', () => { + const httpOptions = { timeout: 50 } + + /** Starts the call with a caller signal that never aborts. */ + async function expectTimeoutAborts( + wire: { fetch: object; signals: Array }, + start: (signal: AbortSignal) => Promise, + ) { + const run = start(new AbortController().signal) + run.catch(() => {}) + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 1000 }, + ) + await expect(run).rejects.toBeDefined() + } + + it('text stream, agentic video: the timeout aborts', async () => { + const adapter = createGeminiChat('gemini-3.8-flash', 'test-key', { + httpOptions, + }) + const wire = hangingInteractionsCreate(adapter) + await expectTimeoutAborts(wire, (signal) => + runText( + adapter as unknown as TextAdapter, + 'stream', + agenticVideo, + signal, + ), + ) + }) + + it('text stream, agentic video: the caller still aborts', async () => { + const adapter = createGeminiChat('gemini-3.8-flash', 'test-key', { + httpOptions: { timeout: 60_000 }, + }) + const wire = hangingInteractionsCreate(adapter) + await expectAbortReachesFetch(wire, (signal) => + runText( + adapter as unknown as TextAdapter, + 'stream', + agenticVideo, + signal, + ), + ) + }) + + it('video create (Omni): the timeout aborts', async () => { + const adapter = createGeminiVideo('gemini-omni-1.1-flash', 'test-key', { + httpOptions, + }) + const wire = hangingInteractionsCreate(adapter) + await expectTimeoutAborts(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'gemini-omni-1.1-flash', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('video create (Omni): the caller still aborts', async () => { + const adapter = createGeminiVideo('gemini-omni-1.1-flash', 'test-key', { + httpOptions: { timeout: 60_000 }, + }) + const wire = hangingInteractionsCreate(adapter) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'gemini-omni-1.1-flash', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) +}) diff --git a/packages/ai-grok/src/adapters/image.ts b/packages/ai-grok/src/adapters/image.ts index 9a79787634..353e402a10 100644 --- a/packages/ai-grok/src/adapters/image.ts +++ b/packages/ai-grok/src/adapters/image.ts @@ -159,7 +159,9 @@ export class GrokImageAdapter< `activity=image provider=${this.name} model=${model} n=${request.n ?? 1} size=${request.size ?? 'default'}`, { provider: this.name, model }, ) - const response = await this.client.images.generate(request) + const response = await this.client.images.generate(request, { + signal: options.abortSignal, + }) const images: Array = (response.data ?? []).flatMap( (item): Array => { @@ -266,6 +268,7 @@ export class GrokImageAdapter< Authorization: `Bearer ${this.clientConfig.apiKey}`, }, body: JSON.stringify(request), + signal: options.abortSignal, }, ) if (!response.ok) { diff --git a/packages/ai-grok/src/adapters/transcription.ts b/packages/ai-grok/src/adapters/transcription.ts index 8037b7eef8..5a8606f2d6 100644 --- a/packages/ai-grok/src/adapters/transcription.ts +++ b/packages/ai-grok/src/adapters/transcription.ts @@ -106,6 +106,7 @@ export class GrokTranscriptionAdapter< Authorization: `Bearer ${this.apiKey}`, }, body: form, + signal: options.abortSignal, }) if (!response.ok) { diff --git a/packages/ai-grok/src/adapters/tts.ts b/packages/ai-grok/src/adapters/tts.ts index 2688564320..66fa3e1ee2 100644 --- a/packages/ai-grok/src/adapters/tts.ts +++ b/packages/ai-grok/src/adapters/tts.ts @@ -75,6 +75,7 @@ export class GrokSpeechAdapter< 'Content-Type': 'application/json', }, body: JSON.stringify(body), + signal: options.abortSignal, }) if (!response.ok) { diff --git a/packages/ai-grok/src/adapters/video.ts b/packages/ai-grok/src/adapters/video.ts index da3a997802..508956ae2e 100644 --- a/packages/ai-grok/src/adapters/video.ts +++ b/packages/ai-grok/src/adapters/video.ts @@ -269,6 +269,7 @@ export class GrokVideoAdapter< size, genericDuration: options.duration, logger, + signal: options.abortSignal, }) } @@ -410,6 +411,7 @@ export class GrokVideoAdapter< model, logger, logLine: `activity=video.create provider=${this.name} model=${model} mode=generate size=${size ?? 'default'} duration=${duration ?? 'default'}`, + signal: options.abortSignal, }) } @@ -430,6 +432,7 @@ export class GrokVideoAdapter< size: string | undefined genericDuration: number | undefined logger: VideoGenerationOptions['logger'] + signal: AbortSignal | undefined }): Promise { const { model, mode, sourceVideo, resolved, wireOptions, logger } = args const endpoint = mode === 'edit' ? '/videos/edits' : '/videos/extensions' @@ -493,6 +496,7 @@ export class GrokVideoAdapter< model, logger, logLine: `activity=video.create provider=${this.name} model=${model} mode=${mode} duration=${duration ?? 'default'}`, + signal: args.signal, }) } @@ -508,15 +512,17 @@ export class GrokVideoAdapter< model: string logger: VideoGenerationOptions['logger'] logLine: string + signal: AbortSignal | undefined }, ): Promise { - const { model, logger, logLine } = context + const { model, logger, logLine, signal } = context try { logger.request(logLine, { provider: this.name, model }) const response = await this.request(endpoint, { method: 'POST', body: JSON.stringify(request), + signal, }) if (!response.ok) { throw new Error( diff --git a/packages/ai-grok/tests/abort-signal-wire.test.ts b/packages/ai-grok/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..bade822325 --- /dev/null +++ b/packages/ai-grok/tests/abort-signal-wire.test.ts @@ -0,0 +1,171 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createGrokImage } from '../src/adapters/image' +import { createGrokText } from '../src/adapters/text' +import { createGrokTranscription } from '../src/adapters/transcription' +import { createGrokSpeech } from '../src/adapters/tts' +import { createGrokVideo } from '../src/adapters/video' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +afterEach(() => { + vi.unstubAllGlobals() +}) + +describe('Grok adapters abort the provider request', () => { + it('image generate', async () => { + const wire = hangingFetch() + const adapter = createGrokImage('grok-imagine-image', 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: 'grok-imagine-image', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('image edit', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createGrokImage('grok-imagine-image', 'test-key') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: 'grok-imagine-image', + prompt: [ + { type: 'text', content: 'make it blue' }, + { + type: 'image', + source: { type: 'url', value: 'https://example.test/in.png' }, + }, + ], + abortSignal, + logger, + }), + ) + }) + + it('transcription', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createGrokTranscription('grok-stt', 'test-key') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.transcribe({ + model: 'grok-stt', + audio: new Blob([new Uint8Array([1, 2, 3])], { type: 'audio/mpeg' }), + abortSignal, + logger, + }), + ) + }) + + it('speech', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createGrokSpeech('grok-tts', 'test-key') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateSpeech({ + model: 'grok-tts', + text: 'hello', + abortSignal, + logger, + }), + ) + }) + + it.each([undefined, 'edit', 'extend'] as const)( + 'video create (mode %s)', + async (mode) => { + const wire = hangingFetch() + const adapter = createGrokVideo('grok-imagine-video', 'test-key', { + fetch: wire.fetch, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'grok-imagine-video', + prompt: mode + ? [ + { type: 'text', content: 'make the sky stormy' }, + { + type: 'video', + source: { type: 'url', value: 'https://example.test/a.mp4' }, + }, + ] + : 'a cat', + ...(mode && { modelOptions: { mode } }), + abortSignal, + logger, + }), + ) + }, + ) + + it.each(['stream', 'structured'] as const)('text %s', async (mode) => { + const wire = hangingFetch() + const model = 'grok-4.3' + const adapter = createGrokText(model, 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, async (signal) => { + const chatOptions = { + threadId: 'thread-1', + logger, + model, + messages: [{ role: 'user' as const, content: 'Hi' }], + request: { signal }, + } + if (mode === 'structured') { + await adapter.structuredOutput({ + chatOptions, + outputSchema: { type: 'object', properties: {} }, + }) + return + } + // The stream reports the abort as a RUN_ERROR chunk. + for await (const chunk of adapter.chatStream(chatOptions)) { + if (chunk.type === 'RUN_ERROR') throw new Error(chunk.message) + } + }) + }) +}) diff --git a/packages/ai-grok/tests/grok-adapter.test.ts b/packages/ai-grok/tests/grok-adapter.test.ts index ee39007b77..08a6c38bbb 100644 --- a/packages/ai-grok/tests/grok-adapter.test.ts +++ b/packages/ai-grok/tests/grok-adapter.test.ts @@ -518,6 +518,7 @@ describe('Grok adapters', () => { expect(mockGenerate).toHaveBeenCalledWith( expect.objectContaining({ model: 'grok-imagine-image', prompt }), + expect.anything(), ) }) @@ -541,6 +542,7 @@ describe('Grok adapters', () => { aspect_ratio: '16:9', resolution: '2k', }), + expect.anything(), ) expect(mockGenerate.mock.calls[0]![0]).not.toHaveProperty('size') }) @@ -568,6 +570,7 @@ describe('Grok adapters', () => { aspect_ratio: '16:9', quality: 'low', }), + expect.anything(), ) }) }) diff --git a/packages/ai-groq/src/adapters/transcription.ts b/packages/ai-groq/src/adapters/transcription.ts index 2ac6999b2a..99ead2e983 100644 --- a/packages/ai-groq/src/adapters/transcription.ts +++ b/packages/ai-groq/src/adapters/transcription.ts @@ -157,6 +157,7 @@ export class GroqTranscriptionAdapter< Authorization: `Bearer ${this.apiKey}`, }, body: form, + signal: options.abortSignal, }) if (!response.ok) { diff --git a/packages/ai-groq/src/adapters/tts.ts b/packages/ai-groq/src/adapters/tts.ts index 64d7a55d88..241fcceb1e 100644 --- a/packages/ai-groq/src/adapters/tts.ts +++ b/packages/ai-groq/src/adapters/tts.ts @@ -70,7 +70,9 @@ export class GroqTTSAdapter extends BaseTTSAdapter< `activity=tts provider=${this.name} model=${model} format=${request.response_format ?? 'default'} voice=${request.voice}`, { provider: this.name, model }, ) - const response = await this.client.audio.speech.create(request) + const response = await this.client.audio.speech.create(request, { + signal: options.abortSignal, + }) const arrayBuffer = await response.arrayBuffer() const base64 = arrayBufferToBase64(arrayBuffer) diff --git a/packages/ai-groq/tests/abort-signal-wire.test.ts b/packages/ai-groq/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..4b8393ed1a --- /dev/null +++ b/packages/ai-groq/tests/abort-signal-wire.test.ts @@ -0,0 +1,112 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createGroqText } from '../src/adapters/text' +import { createGroqTranscription } from '../src/adapters/transcription' +import { createGroqSpeech } from '../src/adapters/tts' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +afterEach(() => { + vi.unstubAllGlobals() +}) + +describe('Groq adapters abort the provider request', () => { + it('speech', async () => { + const wire = hangingFetch() + const adapter = createGroqSpeech( + 'canopylabs/orpheus-v1-english', + 'test-key', + { + fetch: wire.fetch, + maxRetries: 0, + }, + ) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateSpeech({ + model: 'canopylabs/orpheus-v1-english', + text: 'hello', + abortSignal, + logger, + }), + ) + }) + + it('transcription', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createGroqTranscription('whisper-large-v3', 'test-key') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.transcribe({ + model: 'whisper-large-v3', + audio: new Blob([new Uint8Array([1, 2, 3])], { type: 'audio/mpeg' }), + abortSignal, + logger, + }), + ) + }) + + it.each(['stream', 'structured'] as const)('text %s', async (mode) => { + const wire = hangingFetch() + const model = 'openai/gpt-oss-120b' + const adapter = createGroqText(model, 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, async (signal) => { + const chatOptions = { + threadId: 'thread-1', + logger, + model, + messages: [{ role: 'user' as const, content: 'Hi' }], + request: { signal }, + } + if (mode === 'structured') { + await adapter.structuredOutput({ + chatOptions, + outputSchema: { type: 'object', properties: {} }, + }) + return + } + // The stream reports the abort as a RUN_ERROR chunk. + for await (const chunk of adapter.chatStream(chatOptions)) { + if (chunk.type === 'RUN_ERROR') throw new Error(chunk.message) + } + }) + }) +}) diff --git a/packages/ai-mistral/src/adapters/text.ts b/packages/ai-mistral/src/adapters/text.ts index 59d25e39f7..1b58b41ad8 100644 --- a/packages/ai-mistral/src/adapters/text.ts +++ b/packages/ai-mistral/src/adapters/text.ts @@ -356,6 +356,7 @@ export class MistralTextAdapter< body, this.rawConfig, options.wrapFetch, + options.request?.signal ?? undefined, ) for await (const chunk of this.processMistralStreamChunks( stream, @@ -456,17 +457,29 @@ export class MistralTextAdapter< : chatOptions.wrapFetch ? createMistralClient(this.rawConfig, next) : this.client - const response = await client.chat.complete({ + // The SDK drops its default 30 s timeout when the request has a signal + // (lib/sdks.js), so keep that timeout next to the caller signal. + const signal = chatOptions.request?.signal + const timeoutMs = this.rawConfig.timeoutMs || 30_000 + const params = { ...nonStreamParams, responseFormat: { - type: 'json_schema', + type: 'json_schema' as const, jsonSchema: { name: 'structured_output', schemaDefinition: jsonSchema, strict, }, }, - }) + } + const response = signal + ? await client.chat.complete(params, { + signal: + timeoutMs > 0 + ? AbortSignal.any([signal, AbortSignal.timeout(timeoutMs)]) + : signal, + }) + : await client.chat.complete(params) // A response cut off at the output cap is a truncated JSON document; // report it before the parse error (issue #1426). @@ -956,6 +969,7 @@ export class MistralTextAdapter< body: Record, config: MistralClientConfig, wrapFetch: FetchWrapper | undefined, + signal: AbortSignal | undefined, ): AsyncGenerator { const serverURL = ( config.baseURL ?? @@ -983,6 +997,7 @@ export class MistralTextAdapter< method: 'POST', headers, body: JSON.stringify(body), + signal, }) if (!response.ok) { diff --git a/packages/ai-mistral/tests/abort-signal-wire.test.ts b/packages/ai-mistral/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..36b606813d --- /dev/null +++ b/packages/ai-mistral/tests/abort-signal-wire.test.ts @@ -0,0 +1,113 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { EventType } from '@tanstack/ai' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createMistralText } from '../src/adapters/text' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +/** Drains a stream. A RUN_ERROR chunk counts as a rejection. */ +async function drain(stream: AsyncIterable<{ type: string }>) { + for await (const chunk of stream) { + if (chunk.type === EventType.RUN_ERROR) throw new Error('RUN_ERROR') + } +} + +const model = 'mistral-large-latest' +const messages = [{ role: 'user' as const, content: 'Hi' }] +const outputSchema = { type: 'object', properties: {} } + +afterEach(() => { + vi.unstubAllGlobals() +}) + +describe('Mistral text adapter aborts the provider request', () => { + it('chatStream', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createMistralText(model, 'test-key') + await expectAbortReachesFetch(wire, (signal) => + drain( + adapter.chatStream({ + threadId: 'thread-1', + logger, + model, + messages, + request: { signal }, + }), + ), + ) + }) + + it('structuredOutput', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createMistralText(model, 'test-key') + await expectAbortReachesFetch(wire, (signal) => + adapter.structuredOutput({ + chatOptions: { + threadId: 'thread-1', + logger, + model, + messages, + request: { signal }, + }, + outputSchema, + }), + ) + }) + + // A signal turns off the SDK default timeout, so the adapter keeps it. + it('structuredOutput keeps timeoutMs when the caller passes a signal', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createMistralText(model, 'test-key', { timeoutMs: 50 }) + const run = adapter.structuredOutput({ + chatOptions: { + threadId: 'thread-1', + logger, + model, + messages, + request: { signal: new AbortController().signal }, + }, + outputSchema, + }) + await expect(run).rejects.toBeDefined() + expect(wire.signals.at(-1)?.aborted).toBe(true) + }) +}) diff --git a/packages/ai-ollama/src/adapters/text.ts b/packages/ai-ollama/src/adapters/text.ts index 3fd3e764f6..fd32148253 100644 --- a/packages/ai-ollama/src/adapters/text.ts +++ b/packages/ai-ollama/src/adapters/text.ts @@ -194,17 +194,35 @@ export class OllamaTextAdapter extends BaseTextAdapter< ...mappedOptions, stream: true, }) - for await (const chunk of this.processOllamaStreamChunks( - response, - options, - logger, - )) { - yield { - ...chunk, - metadata: { - ...chunk.metadata, - tanstack: { ...tanstackMetadata(chunk), source }, - }, + // ponytail: an injected client has no per-call signal, so stop this + // stream with its own abort(). It cannot cancel before the headers. + const signal = this.clientConfig ? undefined : options.request?.signal + const onAbort = () => response.abort() + if (signal?.aborted) onAbort() + else signal?.addEventListener('abort', onAbort, { once: true }) + let ended = false + try { + for await (const chunk of this.processOllamaStreamChunks( + response, + options, + logger, + )) { + yield { + ...chunk, + metadata: { + ...chunk.metadata, + tanstack: { ...tanstackMetadata(chunk), source }, + }, + } + } + ended = true + } finally { + if (signal) { + // chat() reuses one signal for every agent-loop step. + signal.removeEventListener('abort', onAbort) + // A consumer that stops early does not abort the request in + // `ollama`, and the listener is gone now, so stop it here. + if (!ended) response.abort() } } } catch (error: unknown) { diff --git a/packages/ai-ollama/tests/abort-signal-wire.test.ts b/packages/ai-ollama/tests/abort-signal-wire.test.ts index 61f5db1f8e..58923af22a 100644 --- a/packages/ai-ollama/tests/abort-signal-wire.test.ts +++ b/packages/ai-ollama/tests/abort-signal-wire.test.ts @@ -1,7 +1,9 @@ +import { getEventListeners } from 'node:events' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { Ollama } from 'ollama' import { chat } from '@tanstack/ai' import { resolveDebugOption } from '@tanstack/ai/adapter-internals' -import { createOllamaChat } from '../src/adapters/text' +import { OllamaTextAdapter, createOllamaChat } from '../src/adapters/text' import type { FetchWrapper } from '@tanstack/ai' const logger = resolveDebugOption(false) @@ -99,3 +101,114 @@ describe('Ollama abort signal', () => { await run }) }) + +// `ollama` takes no per-call signal, so with an injected client only the +// stream's own `abort()` can end the request. It needs the response headers, +// so this fake answers at once with a body that never ends. +describe('Ollama abort signal with an injected client', () => { + it('aborts the stream request when the caller aborts', async () => { + const signals: Array = [] + const fetch = vi.fn((_input: RequestInfo | URL, init?: RequestInit) => { + const signal = init?.signal + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + const body = new ReadableStream({ + start(controller) { + signal.addEventListener('abort', () => + controller.error(signal.reason), + ) + }, + }) + return Promise.resolve(new Response(body)) + }) + const client = new Ollama({ host: 'http://ollama.test', fetch }) + const abortController = new AbortController() + const run = (async () => { + for await (const _ of new OllamaTextAdapter(client, model).chatStream({ + threadId: 'thread-1', + logger, + model, + messages, + request: { signal: abortController.signal }, + })) { + // drain + } + })() + + await vi.waitFor(() => expect(fetch).toHaveBeenCalled()) + abortController.abort() + + await vi.waitFor( + () => expect(signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await run + }) + + it('leaves no abort listener on the signal after the stream ends', async () => { + const line = `${JSON.stringify({ + model, + created_at: '2026-11-02T00:00:00Z', + message: { role: 'assistant', content: 'Hi' }, + done: true, + done_reason: 'stop', + })}\n` + const fetch = vi.fn(() => Promise.resolve(new Response(line))) + const client = new Ollama({ host: 'http://ollama.test', fetch }) + const adapter = new OllamaTextAdapter(client, model) + // chat() reuses one signal for every step of the agent loop. + const { signal } = new AbortController() + for (let step = 0; step < 2; step++) { + for await (const _ of adapter.chatStream({ + threadId: 'thread-1', + logger, + model, + messages, + request: { signal }, + })) { + // drain + } + } + + expect(fetch).toHaveBeenCalledTimes(2) + expect(getEventListeners(signal, 'abort')).toHaveLength(0) + }) + + it('aborts the request when the consumer stops before the stream ends', async () => { + const signals: Array = [] + const line = `${JSON.stringify({ + model, + created_at: '2026-11-02T00:00:00Z', + message: { role: 'assistant', content: 'Hi' }, + done: false, + })}\n` + const fetch = vi.fn((_input: RequestInfo | URL, init?: RequestInit) => { + const signal = init?.signal + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + // One chunk, then a body that never ends. + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(line)) + signal.addEventListener('abort', () => + controller.error(signal.reason), + ) + }, + }) + return Promise.resolve(new Response(body)) + }) + const client = new Ollama({ host: 'http://ollama.test', fetch }) + const { signal } = new AbortController() + for await (const chunk of new OllamaTextAdapter(client, model).chatStream({ + threadId: 'thread-1', + logger, + model, + messages, + request: { signal }, + })) { + if (chunk.type === 'TEXT_MESSAGE_CONTENT') break + } + + expect(signals.at(-1)?.aborted).toBe(true) + }) +}) diff --git a/packages/ai-openai/src/adapters/image.ts b/packages/ai-openai/src/adapters/image.ts index fecaf83d4e..e50528ced6 100644 --- a/packages/ai-openai/src/adapters/image.ts +++ b/packages/ai-openai/src/adapters/image.ts @@ -112,6 +112,7 @@ export class OpenAIImageAdapter< modelOptions, imageInputs: resolved.images, logger: options.logger, + abortSignal: options.abortSignal, }) } @@ -136,10 +137,10 @@ export class OpenAIImageAdapter< `activity=image provider=${this.name} model=${model} n=${request.n ?? 1} size=${request.size ?? 'default'}`, { provider: this.name, model }, ) - const response = await this.client.images.generate({ - ...request, - stream: false, - }) + const response = await this.client.images.generate( + { ...request, stream: false }, + { signal: options.abortSignal }, + ) const images: Array = (response.data ?? []).flatMap( (item): Array => { @@ -200,8 +201,17 @@ export class OpenAIImageAdapter< modelOptions?: OpenAIImageProviderOptions imageInputs: ReadonlyArray> logger: ImageGenerationOptions['logger'] + abortSignal?: AbortSignal }): Promise { - const { model, prompt, numberOfImages, size, modelOptions, logger } = args + const { + model, + prompt, + numberOfImages, + size, + modelOptions, + logger, + abortSignal, + } = args const maskParts = args.imageInputs.filter( (part) => part.metadata?.role === 'mask', ) @@ -227,12 +237,17 @@ export class OpenAIImageAdapter< const sourceFiles = await Promise.all( sourceParts.map((part, i) => - imagePartToFile(part, `source-${i}`, this.allowUrlFetch), + imagePartToFile(part, `source-${i}`, this.allowUrlFetch, abortSignal), ), ) const [firstSourceFile] = sourceFiles const maskFile = maskParts[0] - ? await imagePartToFile(maskParts[0], 'mask', this.allowUrlFetch) + ? await imagePartToFile( + maskParts[0], + 'mask', + this.allowUrlFetch, + abortSignal, + ) : undefined // The SDK's edit params do not list every gpt-image option (e.g. the @@ -261,7 +276,9 @@ export class OpenAIImageAdapter< `activity=imageEdit provider=${this.name} model=${model} n=${request.n ?? 1} size=${request.size ?? 'default'} sources=${sourceFiles.length}${maskFile ? ' mask' : ''}`, { provider: this.name, model }, ) - const response = await this.client.images.edit(request) + const response = await this.client.images.edit(request, { + signal: abortSignal, + }) const images: Array = (response.data ?? []).flatMap( (item): Array => { diff --git a/packages/ai-openai/src/adapters/transcription.ts b/packages/ai-openai/src/adapters/transcription.ts index 1d9ca3c176..e4e6dc6f49 100644 --- a/packages/ai-openai/src/adapters/transcription.ts +++ b/packages/ai-openai/src/adapters/transcription.ts @@ -173,6 +173,7 @@ export class OpenAITranscriptionAdapter< if (responseMode === 'diarized') { const response = (await this.client.audio.transcriptions.create( request, + { signal: options.abortSignal }, )) as OpenAI_SDK.Audio.TranscriptionDiarized // Guard the cast: a proxy/gateway or API change that returns a @@ -213,10 +214,10 @@ export class OpenAITranscriptionAdapter< } if (responseMode === 'verbose') { - const response = (await this.client.audio.transcriptions.create({ - ...request, - response_format: 'verbose_json', - })) as OpenAI_SDK.Audio.Transcriptions.TranscriptionVerbose + const response = (await this.client.audio.transcriptions.create( + { ...request, response_format: 'verbose_json' }, + { signal: options.abortSignal }, + )) as OpenAI_SDK.Audio.Transcriptions.TranscriptionVerbose // `TranscriptionResult` declares optional fields without `| undefined`, // so under exactOptionalPropertyTypes we must omit absent fields rather @@ -255,7 +256,9 @@ export class OpenAITranscriptionAdapter< } } - const response = await this.client.audio.transcriptions.create(request) + const response = await this.client.audio.transcriptions.create(request, { + signal: options.abortSignal, + }) const usage = typeof response === 'string' diff --git a/packages/ai-openai/src/adapters/tts.ts b/packages/ai-openai/src/adapters/tts.ts index cc366abf39..c42106f59b 100644 --- a/packages/ai-openai/src/adapters/tts.ts +++ b/packages/ai-openai/src/adapters/tts.ts @@ -76,7 +76,9 @@ export class OpenAITTSAdapter< `activity=tts provider=${this.name} model=${model} format=${request.response_format ?? 'default'} voice=${request.voice}`, { provider: this.name, model }, ) - const response = await this.client.audio.speech.create(request) + const response = await this.client.audio.speech.create(request, { + signal: options.abortSignal, + }) // Convert response to base64. Buffer is Node-only; use atob fallback in // browser/edge runtimes where the SDK can run. diff --git a/packages/ai-openai/src/adapters/video.ts b/packages/ai-openai/src/adapters/video.ts index f68eec3d37..7c83365a69 100644 --- a/packages/ai-openai/src/adapters/video.ts +++ b/packages/ai-openai/src/adapters/video.ts @@ -127,6 +127,7 @@ export class OpenAIVideoAdapter< inputReference, 'input-reference', this.clientConfig.allowUrlFetch ?? false, + options.abortSignal, ) } // `VideoCreateParams.size` is `size?: VideoSize` (no `| undefined`), so we @@ -147,7 +148,9 @@ export class OpenAIVideoAdapter< { provider: this.name, model }, ) const videosClient = this.getVideosClient() - const response = await videosClient.create(request) + const response = await videosClient.create(request, { + signal: options.abortSignal, + }) return { jobId: response.id, model } } catch (error: any) { options.logger.errors(`${this.name}.createVideoJob fatal`, { @@ -170,7 +173,10 @@ export class OpenAIVideoAdapter< * the path uniformly even when the SDK lacks first-class typings here. */ private getVideosClient(): { - create: (req: Record) => Promise<{ id: string }> + create: ( + req: Record, + options?: { signal?: AbortSignal | undefined }, + ) => Promise<{ id: string }> retrieve: (id: string) => Promise<{ id: string status: string diff --git a/packages/ai-openai/src/image/image-input-to-file.ts b/packages/ai-openai/src/image/image-input-to-file.ts index 28b864cad6..9cb04a3fcc 100644 --- a/packages/ai-openai/src/image/image-input-to-file.ts +++ b/packages/ai-openai/src/image/image-input-to-file.ts @@ -42,6 +42,7 @@ export async function imagePartToFile( part: ImagePart, fallbackName: string, allowUrlFetch: boolean, + abortSignal?: AbortSignal, ): Promise { ensureFileSupport() @@ -76,7 +77,7 @@ export async function imagePartToFile( } // URL source — also handles data: URIs uniformly via fetch(). - const response = await fetch(part.source.value) + const response = await fetch(part.source.value, { signal: abortSignal }) if (!response.ok) { throw new Error( `Failed to fetch image input (${response.status} ${response.statusText}): ${part.source.value}`, diff --git a/packages/ai-openai/tests/abort-signal-wire.test.ts b/packages/ai-openai/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..4acc428e40 --- /dev/null +++ b/packages/ai-openai/tests/abort-signal-wire.test.ts @@ -0,0 +1,239 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createOpenaiDecider } from '../src/adapters/evaluate' +import { createOpenaiImage } from '../src/adapters/image' +import { createOpenaiChat } from '../src/adapters/text' +import { createOpenaiChatCompletions } from '../src/adapters/text-chat-completions' +import { createOpenaiTranscription } from '../src/adapters/transcription' +import { createOpenaiSpeech } from '../src/adapters/tts' +import { createOpenaiVideo } from '../src/adapters/video' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +const audio = new Blob([new Uint8Array([1, 2, 3])], { type: 'audio/mpeg' }) +const dataImage = { + type: 'image' as const, + source: { type: 'data' as const, value: 'aGk=', mimeType: 'image/png' }, +} + +afterEach(() => { + vi.unstubAllGlobals() +}) + +describe('OpenAI adapters abort the provider request', () => { + it('image generate', async () => { + const wire = hangingFetch() + const adapter = createOpenaiImage('gpt-image-2', 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: 'gpt-image-2', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('image edit', async () => { + const wire = hangingFetch() + const adapter = createOpenaiImage('gpt-image-2', 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: 'gpt-image-2', + prompt: [{ type: 'text', content: 'make it blue' }, dataImage], + abortSignal, + logger, + }), + ) + }) + + it('image edit: input image URL fetch', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createOpenaiImage('gpt-image-2', 'test-key', { + allowUrlFetch: true, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: 'gpt-image-2', + prompt: [ + { type: 'text', content: 'make it blue' }, + { + type: 'image', + source: { type: 'url', value: 'https://example.test/in.png' }, + }, + ], + abortSignal, + logger, + }), + ) + }) + + it.each([ + ['plain', 'gpt-4o-transcribe'], + ['verbose', 'whisper-1'], + ['diarized', 'gpt-4o-transcribe-diarize'], + ] as const)('transcription (%s)', async (_mode, model) => { + const wire = hangingFetch() + const adapter = createOpenaiTranscription(model, 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.transcribe({ model, audio, abortSignal, logger }), + ) + }) + + it('speech', async () => { + const wire = hangingFetch() + const adapter = createOpenaiSpeech('gpt-4o-mini-tts', 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateSpeech({ + model: 'gpt-4o-mini-tts', + text: 'hello', + abortSignal, + logger, + }), + ) + }) + + it('video create', async () => { + const wire = hangingFetch() + const adapter = createOpenaiVideo('sora-2', 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'sora-2', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('evaluate', async () => { + const wire = hangingFetch() + const adapter = createOpenaiDecider('gpt-6-luna', 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.evaluate({ + model: 'gpt-6-luna', + state: 'Charged twice', + questions: { + refund: { type: 'noul', instructions: 'Refund asked?' }, + }, + abortSignal, + logger, + }), + ) + }) + + it.each([ + ['Responses', createOpenaiChat], + ['Chat Completions', createOpenaiChatCompletions], + ] as const)( + 'text %s: chatStream + structuredOutput', + async (_api, create) => { + for (const mode of ['stream', 'structured'] as const) { + const wire = hangingFetch() + const adapter = create('gpt-5.5', 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, async (signal) => { + const chatOptions = { + threadId: 'thread-1', + logger, + model: 'gpt-5.5' as const, + messages: [{ role: 'user' as const, content: 'Hi' }], + request: { signal }, + } + if (mode === 'structured') { + await adapter.structuredOutput({ + chatOptions, + outputSchema: { type: 'object', properties: {} }, + }) + return + } + // The stream reports the abort as a RUN_ERROR chunk. + for await (const chunk of adapter.chatStream(chatOptions)) { + if (chunk.type === 'RUN_ERROR') throw new Error(chunk.message) + } + }) + } + }, + ) + + it('video create: input reference URL fetch', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createOpenaiVideo('sora-2', 'test-key', { + allowUrlFetch: true, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'sora-2', + prompt: [ + { type: 'text', content: 'a cat' }, + { + type: 'image', + source: { type: 'url', value: 'https://example.test/in.png' }, + }, + ], + abortSignal, + logger, + }), + ) + }) +}) diff --git a/packages/ai-openai/tests/image-adapter.test.ts b/packages/ai-openai/tests/image-adapter.test.ts index efb6df67dd..28018cba56 100644 --- a/packages/ai-openai/tests/image-adapter.test.ts +++ b/packages/ai-openai/tests/image-adapter.test.ts @@ -136,13 +136,16 @@ describe('OpenAI Image Adapter', () => { logger: testLogger, }) - expect(mockGenerate).toHaveBeenCalledWith({ - model: 'gpt-image-1', - prompt: 'A cat wearing a hat', - n: 1, - size: '1024x1024', - stream: false, - }) + expect(mockGenerate).toHaveBeenCalledWith( + { + model: 'gpt-image-1', + prompt: 'A cat wearing a hat', + n: 1, + size: '1024x1024', + stream: false, + }, + { signal: undefined }, + ) expect(result.model).toBe('gpt-image-1') expect(result.images).toHaveLength(1) @@ -176,14 +179,17 @@ describe('OpenAI Image Adapter', () => { logger: testLogger, }) - expect(mockGenerate).toHaveBeenCalledWith({ - model, - prompt: 'A cat wearing a hat', - n: 1, - size: '1536x1024', - quality: 'xhigh', - stream: false, - }) + expect(mockGenerate).toHaveBeenCalledWith( + { + model, + prompt: 'A cat wearing a hat', + n: 1, + size: '1536x1024', + quality: 'xhigh', + stream: false, + }, + { signal: undefined }, + ) expect(result.images[0]!.b64Json).toBe('img') }, ) @@ -327,7 +333,10 @@ describe('OpenAI Image Adapter', () => { ], logger: testLogger, }) - expect(fetchMock).toHaveBeenCalledWith('https://example.com/photo.jpg') + expect(fetchMock).toHaveBeenCalledWith( + 'https://example.com/photo.jpg', + { signal: undefined }, + ) expect(editSpy).toHaveBeenCalledTimes(1) expect(editSpy.mock.calls[0]![0].image).toBeInstanceOf(File) expect(result.images[0]!.b64Json).toBe('edited-base64') diff --git a/packages/ai-openai/tests/transcription-adapter.test.ts b/packages/ai-openai/tests/transcription-adapter.test.ts index 1ea8dae57c..c289d1f287 100644 --- a/packages/ai-openai/tests/transcription-adapter.test.ts +++ b/packages/ai-openai/tests/transcription-adapter.test.ts @@ -72,6 +72,7 @@ describe('OpenAI transcription adapter', () => { response_format: 'diarized_json', chunking_strategy: 'auto', }), + { signal: undefined }, ) expect(result.text).toBe('Agent: Hello\nCustomer: Hi') expect(result.segments).toEqual([ @@ -145,6 +146,7 @@ describe('OpenAI transcription adapter', () => { known_speaker_names: ['agent'], known_speaker_references: ['data:audio/wav;base64,AAA='], }), + { signal: undefined }, ) expect(result.segments?.[0]?.id).toBe(0) }) @@ -188,6 +190,7 @@ describe('OpenAI transcription adapter', () => { response_format: 'diarized_json', chunking_strategy: null, }), + { signal: undefined }, ) expect(result.segments?.[0]?.speaker).toBe('agent') }) @@ -220,6 +223,7 @@ describe('OpenAI transcription adapter', () => { expect.objectContaining({ chunking_strategy: null, }), + { signal: undefined }, ) // Diarized requests always report segments, even when empty — an empty // list must not look like a non-diarized result. @@ -250,6 +254,7 @@ describe('OpenAI transcription adapter', () => { response_format: 'json', chunking_strategy: 'auto', }), + { signal: undefined }, ) expect(result).toMatchObject({ model: 'gpt-4o-transcribe-diarize', @@ -355,6 +360,7 @@ describe('OpenAI transcription adapter', () => { model: 'gpt-4o-transcribe', chunking_strategy: 'auto', }), + { signal: undefined }, ) }) @@ -506,6 +512,7 @@ describe('OpenAI transcription adapter', () => { known_speaker_names: names, known_speaker_references: references, }), + { signal: undefined }, ) }) @@ -674,6 +681,7 @@ describe('OpenAI transcription adapter', () => { expect.objectContaining({ response_format: 'verbose_json', }), + { signal: undefined }, ) expect(result.segments).toEqual([ { id: 0, start: 0, end: 3, text: 'Hello world', confidence: 1 }, @@ -704,6 +712,7 @@ describe('OpenAI transcription adapter', () => { expect.objectContaining({ response_format: 'text', }), + { signal: undefined }, ) expect(result.text).toBe('Hello world') expect(result.segments).toBeUndefined() diff --git a/packages/ai-openai/tests/video-adapter.test.ts b/packages/ai-openai/tests/video-adapter.test.ts index be30d9bedd..06a25fb956 100644 --- a/packages/ai-openai/tests/video-adapter.test.ts +++ b/packages/ai-openai/tests/video-adapter.test.ts @@ -102,7 +102,9 @@ describe('OpenAI Video Adapter', () => { vi.unstubAllGlobals() } - expect(fetchMock).toHaveBeenCalledWith('https://example.com/ref.jpg') + expect(fetchMock).toHaveBeenCalledWith('https://example.com/ref.jpg', { + signal: undefined, + }) expect(mockCreate.mock.calls[0]![0].input_reference).toBeInstanceOf(File) }) diff --git a/packages/ai-openrouter/src/adapters/image.ts b/packages/ai-openrouter/src/adapters/image.ts index 9785b17b39..b8a0c8487d 100644 --- a/packages/ai-openrouter/src/adapters/image.ts +++ b/packages/ai-openrouter/src/adapters/image.ts @@ -157,40 +157,43 @@ export class OpenRouterImageAdapter< }, ) - const response = await this.client.chat.send({ - chatRequest: { - model, - messages: [ - { - role: 'user', - content, + const response = await this.client.chat.send( + { + chatRequest: { + model, + messages: [ + { + role: 'user', + content, + }, + ], + modalities: ['image'], + stream: false, + // The SDK serializes this record verbatim as `image_config`, so keys + // must match the HTTP API's documented snake_case fields — miskeyed + // entries are silently ignored by the gateway (verified live: + // `aspect_ratio` changes output dimensions, `aspectRatio` does not). + imageConfig: { + ...(aspectRatio + ? { + aspect_ratio: aspectRatio, + } + : {}), + ...(modelOptions?.image_size + ? { + image_size: modelOptions.image_size, + } + : {}), + ...(modelOptions?.strength !== undefined + ? { + strength: modelOptions.strength, + } + : {}), }, - ], - modalities: ['image'], - stream: false, - // The SDK serializes this record verbatim as `image_config`, so keys - // must match the HTTP API's documented snake_case fields — miskeyed - // entries are silently ignored by the gateway (verified live: - // `aspect_ratio` changes output dimensions, `aspectRatio` does not). - imageConfig: { - ...(aspectRatio - ? { - aspect_ratio: aspectRatio, - } - : {}), - ...(modelOptions?.image_size - ? { - image_size: modelOptions.image_size, - } - : {}), - ...(modelOptions?.strength !== undefined - ? { - strength: modelOptions.strength, - } - : {}), }, }, - }) + { signal: options.abortSignal }, + ) // Check for error in response if ('error' in response && response.error) { diff --git a/packages/ai-openrouter/src/adapters/video.ts b/packages/ai-openrouter/src/adapters/video.ts index 138dd1cc8d..68968bdc2e 100644 --- a/packages/ai-openrouter/src/adapters/video.ts +++ b/packages/ai-openrouter/src/adapters/video.ts @@ -292,9 +292,10 @@ export class OpenRouterVideoAdapter< `activity=video.create provider=${this.name} model=${this.model} size=${size ?? 'default'} duration=${duration ?? 'default'}`, { provider: this.name, model: this.model }, ) - const response = await this.client.videoGeneration.generate({ - videoGenerationRequest: request, - }) + const response = await this.client.videoGeneration.generate( + { videoGenerationRequest: request }, + { signal: options.abortSignal }, + ) return { jobId: response.id, model: this.model } } catch (error) { logger.errors(`${this.name}.createVideoJob fatal`, { diff --git a/packages/ai-openrouter/tests/abort-signal-wire.test.ts b/packages/ai-openrouter/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..092d024517 --- /dev/null +++ b/packages/ai-openrouter/tests/abort-signal-wire.test.ts @@ -0,0 +1,200 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createOpenRouterDecider } from '../src/adapters/evaluate' +import { createOpenRouterImage } from '../src/adapters/image' +import { createOpenRouterRerank } from '../src/adapters/rerank' +import { createOpenRouterResponsesText } from '../src/adapters/responses-text' +import { createOpenRouterText } from '../src/adapters/text' +import { createOpenRouterVideo } from '../src/adapters/video' + +const logger = resolveDebugOption(false) + +type Wire = { + fetch: { mock: { calls: Array } } + signals: Array +} + +/** Hangs until `signal` aborts. Rejects at once when there is no signal. */ +function hang(signals: Array, signal: AbortSignal | undefined) { + if (!signal) return Promise.reject(new Error('request got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) +} + +/** A fake `fetch` that never answers, for the adapters that call `fetch`. */ +function hangingFetch(): Wire { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => + hang( + signals, + init?.signal ?? (input instanceof Request ? input.signal : undefined), + ), + ) + return { fetch, signals } +} + +/** + * Replaces one OpenRouter SDK method with a call that never answers. The SDK + * sends `request.clone()`, and after a GC the signal of the clone stops + * following the caller signal. So a fake `fetch` cannot show the abort + * reliably. This check reads the signal that the adapter gives to the SDK. + */ +function hangingSdkCall(resource: object, method: string): Wire { + const signals: Array = [] + const fetch = vi + .spyOn(Object.getPrototypeOf(resource), method) + .mockImplementation((...args: Array) => { + const options = args[1] as + | { + signal?: AbortSignal | null + fetchOptions?: { signal?: AbortSignal | null } + } + | undefined + return hang( + signals, + options?.signal ?? options?.fetchOptions?.signal ?? undefined, + ) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: Wire, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +/** Reads the private SDK client of an adapter. */ +function sdkOf(adapter: object) { + const holder = adapter as { + client?: Record + orClient?: Record + } + const client = holder.client ?? holder.orClient + if (!client) throw new Error('adapter has no SDK client') + return client +} + +afterEach(() => { + vi.restoreAllMocks() + vi.unstubAllGlobals() +}) + +describe('OpenRouter adapters abort the provider request', () => { + it.each([ + ['Chat', 'stream'], + ['Chat', 'structured'], + ['Responses', 'stream'], + ['Responses', 'structured'], + ] as const)('text %s %s', async (api, mode) => { + const adapter = + api === 'Chat' + ? createOpenRouterText('openai/gpt-4o', 'sk-or-test') + : createOpenRouterResponsesText('openai/gpt-4o', 'sk-or-test') + const client = sdkOf(adapter) + const wire = + api === 'Chat' + ? hangingSdkCall(client.chat, 'send') + : hangingSdkCall(client.beta.responses, 'send') + await expectAbortReachesFetch(wire, async (signal) => { + const chatOptions = { + threadId: 'thread-1', + logger, + model: 'openai/gpt-4o' as const, + messages: [{ role: 'user' as const, content: 'Hi' }], + request: { signal }, + } + if (mode === 'structured') { + await adapter.structuredOutput({ + chatOptions, + outputSchema: { type: 'object', properties: {} }, + }) + return + } + // The stream reports an abort as a RUN_ERROR chunk, not a throw. + for await (const chunk of adapter.chatStream(chatOptions)) { + if (chunk.type === 'RUN_ERROR') throw new Error(chunk.message) + } + }) + }) + + it('image', async () => { + const adapter = createOpenRouterImage( + 'google/gemini-2.5-flash-image', + 'sk-or-test', + ) + const wire = hangingSdkCall(sdkOf(adapter).chat, 'send') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: 'google/gemini-2.5-flash-image', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('video create', async () => { + const adapter = createOpenRouterVideo( + 'bytedance/seedance-2.0', + 'sk-or-test', + ) + const wire = hangingSdkCall(sdkOf(adapter).videoGeneration, 'generate') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.createVideoJob({ + model: 'bytedance/seedance-2.0', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('rerank', async () => { + const adapter = createOpenRouterRerank('cohere/rerank-v3.5', 'sk-or-test') + const wire = hangingSdkCall(sdkOf(adapter).rerank, 'rerank') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.rerank({ + model: 'cohere/rerank-v3.5', + query: 'rain', + documents: ['a', 'b'], + abortSignal, + logger, + }), + ) + }) + + it('evaluate', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createOpenRouterDecider( + '~typesafe/jev-latest', + 'sk-or-test', + ) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.evaluate({ + model: '~typesafe/jev-latest', + state: { subject: 'Invoice 1842' }, + questions: { + refund: { type: 'noul', instructions: 'Refund asked?' }, + }, + abortSignal, + logger, + }), + ) + }) +}) diff --git a/packages/ai-vercel-gateway/src/adapters/image.ts b/packages/ai-vercel-gateway/src/adapters/image.ts index 576fda4ac8..1c00bba175 100644 --- a/packages/ai-vercel-gateway/src/adapters/image.ts +++ b/packages/ai-vercel-gateway/src/adapters/image.ts @@ -91,10 +91,10 @@ export class VercelGatewayImageAdapter< `activity=image provider=${this.name} model=${model} n=${request.n ?? 1} size=${request.size ?? 'default'}`, { provider: this.name, model }, ) - const response = await this.client.images.generate({ - ...request, - stream: false, - }) + const response = await this.client.images.generate( + { ...request, stream: false }, + { signal: options.abortSignal }, + ) const images: Array = (response.data ?? []).flatMap( (item): Array => { diff --git a/packages/ai-vercel-gateway/tests/abort-signal-wire.test.ts b/packages/ai-vercel-gateway/tests/abort-signal-wire.test.ts new file mode 100644 index 0000000000..b5049bee17 --- /dev/null +++ b/packages/ai-vercel-gateway/tests/abort-signal-wire.test.ts @@ -0,0 +1,131 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { resolveDebugOption } from '@tanstack/ai/adapter-internals' +import { createVercelGatewayDecider } from '../src/adapters/evaluate' +import { createVercelGatewayImage } from '../src/adapters/image' +import { createVercelGatewayResponsesText } from '../src/adapters/responses-text' +import { VercelGatewayTextAdapter } from '../src/adapters/text' + +const logger = resolveDebugOption(false) + +/** + * A fake `fetch` that never answers. It rejects at once when it gets no abort + * signal. Otherwise it rejects only when its signal aborts. + */ +function hangingFetch() { + const signals: Array = [] + const fetch = vi.fn((input: RequestInfo | URL, init?: RequestInit) => { + const signal = + init?.signal ?? (input instanceof Request ? input.signal : undefined) + if (!signal) return Promise.reject(new Error('fetch got no abort signal')) + signals.push(signal) + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason) + signal.addEventListener('abort', () => reject(signal.reason)) + }) + }) + return { fetch, signals } +} + +/** Starts the call, aborts the caller, and expects the request to abort. */ +async function expectAbortReachesFetch( + wire: ReturnType, + start: (signal: AbortSignal) => Promise, +) { + const controller = new AbortController() + const run = start(controller.signal) + run.catch(() => {}) + await vi.waitFor(() => expect(wire.fetch).toHaveBeenCalled()) + controller.abort() + await vi.waitFor( + () => expect(wire.signals.at(-1)?.aborted, 'request aborted').toBe(true), + { timeout: 500 }, + ) + await expect(run).rejects.toBeDefined() +} + +afterEach(() => { + vi.unstubAllGlobals() +}) + +describe('Vercel AI Gateway adapters abort the provider request', () => { + it('image generate', async () => { + const wire = hangingFetch() + const adapter = createVercelGatewayImage('openai/gpt-image-1', 'test-key', { + fetch: wire.fetch, + maxRetries: 0, + }) + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.generateImages({ + model: 'openai/gpt-image-1', + prompt: 'a cat', + abortSignal, + logger, + }), + ) + }) + + it('evaluate', async () => { + const wire = hangingFetch() + vi.stubGlobal('fetch', wire.fetch) + const adapter = createVercelGatewayDecider('typesafe-ai/jev', 'test-key') + await expectAbortReachesFetch(wire, (abortSignal) => + adapter.evaluate({ + model: 'typesafe-ai/jev', + state: 'Charged twice', + questions: { + refund: { type: 'noul', instructions: 'Refund asked?' }, + }, + abortSignal, + logger, + }), + ) + }) + + const model = 'openai/gpt-5.5' + it.each([ + [ + 'Chat Completions', + (fetch: typeof globalThis.fetch) => + new VercelGatewayTextAdapter( + { apiKey: 'test-key', fetch, maxRetries: 0 }, + model, + ), + ], + [ + 'Responses', + (fetch: typeof globalThis.fetch) => + createVercelGatewayResponsesText(model, 'test-key', { + fetch, + maxRetries: 0, + }), + ], + ] as const)( + 'text %s: chatStream + structuredOutput', + async (_api, create) => { + for (const mode of ['stream', 'structured'] as const) { + const wire = hangingFetch() + const adapter = create(wire.fetch) + await expectAbortReachesFetch(wire, async (signal) => { + const chatOptions = { + threadId: 'thread-1', + logger, + model, + messages: [{ role: 'user' as const, content: 'Hi' }], + request: { signal }, + } + if (mode === 'structured') { + await adapter.structuredOutput({ + chatOptions, + outputSchema: { type: 'object', properties: {} }, + }) + return + } + // The stream reports the abort as a RUN_ERROR chunk. + for await (const chunk of adapter.chatStream(chatOptions)) { + if (chunk.type === 'RUN_ERROR') throw new Error(chunk.message) + } + }) + } + }, + ) +})