feat(ai): add abort signals and timeouts to media generation activities

.changeset/activity-abort-timeout.md

@@ -0,0 +1,10 @@
+---
+'@tanstack/ai': minor
+'@tanstack/ai-fal': minor
+---
+
+feat(ai): add `timeout` and `abortSignal` to media generation activities
+
+Media activities (`generateImage`, `generateAudio`, `generateVideo`, `generateSpeech`, `generateTranscription`, and `summarize`) now accept optional `timeout` and `abortSignal`. Core composes them into a request-specific effective signal, races the adapter call so hung providers reject, clears timeout resources on settle, and routes aborts to middleware `onAbort` (not `onError`).
+
+`@tanstack/ai-fal` forwards the signal to `fal.subscribe()` / `fal.queue.submit()` per request — never via global `fal.config()` — so concurrent generations stay isolated.

packages/ai-fal/src/adapters/audio.ts

@@ -84,7 +84,11 @@ export class FalAudioAdapter<TModel extends FalModel> extends BaseAudioAdapter<
     })
     try {
       const input = this.buildInput(options)
-      const result = await fal.subscribe(this.model, { input })
+      // Request-specific abortSignal only — not fal.config() (global).
+      const result = await fal.subscribe(this.model, {
+        input,
+        ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}),
+      })
       return this.transformResponse(result)
     } catch (error) {
       logger.errors('fal.generateAudio fatal', {

packages/ai-fal/src/adapters/image.ts

@@ -87,7 +87,12 @@ export class FalImageAdapter<TModel extends FalModel> extends BaseImageAdapter<
 
     try {
       const input = this.buildInput(options, resolved)
-      const result = await fal.subscribe(this.model, { input })
+      // Pass request-specific abortSignal only — never via fal.config(), which
+      // is global and would cancel concurrent generations from other calls.
+      const result = await fal.subscribe(this.model, {
+        input,
+        ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}),
+      })
       return this.transformResponse(result)
     } catch (error) {
       logger.errors('fal.generateImage fatal', {

packages/ai-fal/src/adapters/speech.ts

@@ -56,7 +56,11 @@ export class FalSpeechAdapter<TModel extends FalModel> extends BaseTTSAdapter<
     })
     try {
       const input = this.buildInput(options)
-      const result = await fal.subscribe(this.model, { input })
+      // Request-specific abortSignal only — not fal.config() (global).
+      const result = await fal.subscribe(this.model, {
+        input,
+        ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}),
+      })
       return await this.transformResponse(result)
     } catch (error) {
       logger.errors('fal.generateSpeech fatal', {

packages/ai-fal/src/adapters/transcription.ts

@@ -70,7 +70,11 @@ export class FalTranscriptionAdapter<
     )
     try {
       const input = this.buildInput(options)
-      const result = await fal.subscribe(this.model, { input })
+      // Request-specific abortSignal only — not fal.config() (global).
+      const result = await fal.subscribe(this.model, {
+        input,
+        ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}),
+      })
       return this.transformResponse(result)
     } catch (error) {
       logger.errors('fal.generateTranscription fatal', {

packages/ai-fal/src/adapters/video.ts

@@ -179,9 +179,11 @@ export class FalVideoAdapter<TModel extends FalModel> extends BaseVideoAdapter<
         ...(duration ? { duration } : {}),
       } as FalModelInput<TModel>
 
-      // Submit to queue and get request ID
+      // Submit to queue and get request ID. Request-specific abortSignal only —
+      // never via fal.config() (global; would cancel concurrent jobs).
       const { request_id } = await fal.queue.submit(this.model, {
         input,
+        ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}),
       })
 
       return {

packages/ai-fal/tests/image-adapter.test.ts

@@ -149,6 +149,50 @@ describe('Fal Image Adapter', () => {
     })
   })
 
+  it('forwards request-specific abortSignal to fal.subscribe()', async () => {
+    const mockResponse = createMockImageResponse([
+      { url: 'https://fal.media/files/image.png' },
+    ])
+    mockSubscribe.mockResolvedValueOnce(mockResponse)
+
+    const adapter = createAdapter()
+    const controller = new AbortController()
+
+    await generateImage({
+      adapter,
+      prompt: 'A landscape',
+      abortSignal: controller.signal,
+    })
+
+    expect(mockSubscribe).toHaveBeenCalledTimes(1)
+    const [, options] = mockSubscribe.mock.calls[0]!
+    expect(options.abortSignal).toBeInstanceOf(AbortSignal)
+    // Must be request-scoped options, not a side effect of fal.config().
+    expect(mockConfig).toHaveBeenCalled()
+    for (const call of mockConfig.mock.calls) {
+      expect(call[0]).not.toHaveProperty('abortSignal')
+    }
+  })
+
+  it('forwards activity timeout as abortSignal to fal.subscribe()', async () => {
+    const mockResponse = createMockImageResponse([
+      { url: 'https://fal.media/files/image.png' },
+    ])
+    mockSubscribe.mockResolvedValueOnce(mockResponse)
+
+    const adapter = createAdapter()
+
+    await generateImage({
+      adapter,
+      prompt: 'A landscape',
+      timeout: 60_000,
+    })
+
+    const [, options] = mockSubscribe.mock.calls[0]!
+    expect(options.abortSignal).toBeInstanceOf(AbortSignal)
+    expect(options.abortSignal.aborted).toBe(false)
+  })
+
   it('passes custom image_size through model options', async () => {
     const mockResponse = createMockImageResponse([
       { url: 'https://fal.media/files/image.png' },

packages/ai/src/activities/generateAudio/index.ts

@@ -11,11 +11,18 @@ import { resolveDebugOption } from '../../logger/resolve'
 import {
   applyGenerationResultTransforms,
   createGenerationContext,
+  runGenerationAbort,
   runGenerationError,
   runGenerationFinish,
   runGenerationStart,
   runGenerationUsage,
 } from '../middleware/run'
+import {
+  abortReasonMessage,
+  createActivityAbortControls,
+  isActivityAbortError,
+  raceWithAbort,
+} from '../../utilities/activity-abort'
 import type { InternalLogger } from '../../logger/internal-logger'
 import type { DebugOption } from '../../logger/types'
 import type { GenerationMiddleware } from '../middleware/types'
@@ -89,6 +96,18 @@ export interface AudioActivityOptions<
   threadId?: string
   /** Stable run id for correlating this run when persisted. */
   runId?: string
+  /**
+   * Maximum duration of this activity invocation in milliseconds.
+   * No SDK-wide default — choose a value suitable for the provider and job.
+   * Composed with {@link abortSignal}; the first abort wins.
+   */
+  timeout?: number
+  /**
+   * Caller cancellation signal (request disconnects, job/runtime cancellation).
+   * Composed with {@link timeout} into an effective signal forwarded to the
+   * adapter. Request-specific — not stored on global provider client config.
+   */
+  abortSignal?: AbortSignal
 }
 
 // ===========================
@@ -167,12 +186,18 @@ async function runGenerateAudio<
     middleware,
     threadId,
     runId,
+    timeout,
+    abortSignal: callerAbortSignal,
     ...rest
   } = options
   const model = adapter.model
   const requestId = createId('audio')
   const startTime = Date.now()
   const logger: InternalLogger = resolveDebugOption(options.debug)
+  const abortControls = createActivityAbortControls({
+    timeout,
+    abortSignal: callerAbortSignal,
+  })
   const providerName =
     (adapter as { name?: string; provider?: string }).provider ??
     (adapter as { name?: string }).name ??
@@ -208,7 +233,16 @@ async function runGenerateAudio<
   })
 
   try {
-    const rawResult = await adapter.generateAudio({ ...rest, model, logger })
+    const rawResult = await raceWithAbort(
+      adapter.generateAudio({
+        ...rest,
+        model,
+        logger,
+        ...(abortControls.signal ? { abortSignal: abortControls.signal } : {}),
+      }),
+      abortControls.signal,
+    )
+    abortControls.clear()
     const result = await applyGenerationResultTransforms(mwCtx, rawResult)
     const elapsedMs = Date.now() - startTime
 
@@ -245,6 +279,7 @@ async function runGenerateAudio<
 
     return result
   } catch (error) {
+    abortControls.clear()
     const elapsedMs = Date.now() - startTime
     const err = error as Error
     aiEventClient.emit('audio:request:error', {
@@ -256,10 +291,17 @@ async function runGenerateAudio<
       modelOptions: rest.modelOptions as Record<string, unknown> | undefined,
       timestamp: Date.now(),
     })
-    await runGenerationError(middleware, mwCtx, {
-      error,
-      duration: elapsedMs,
-    })
+    if (isActivityAbortError(error, abortControls.signal)) {
+      await runGenerationAbort(middleware, mwCtx, {
+        reason: abortReasonMessage(error, abortControls.signal),
+        duration: elapsedMs,
+      })
+    } else {
+      await runGenerationError(middleware, mwCtx, {
+        error,
+        duration: elapsedMs,
+      })
+    }
     logger.errors('generateAudio activity failed', {
       error,
       source: 'generateAudio',

packages/ai/src/activities/generateImage/index.ts

@@ -11,11 +11,18 @@ import { resolveDebugOption } from '../../logger/resolve'
 import {
   applyGenerationResultTransforms,
   createGenerationContext,
+  runGenerationAbort,
   runGenerationError,
   runGenerationFinish,
   runGenerationStart,
   runGenerationUsage,
 } from '../middleware/run'
+import {
+  abortReasonMessage,
+  createActivityAbortControls,
+  isActivityAbortError,
+  raceWithAbort,
+} from '../../utilities/activity-abort'
 import { resolveMediaPrompt } from '../../utilities/media-prompt'
 import type { InternalLogger } from '../../logger/internal-logger'
 import type { DebugOption } from '../../logger/types'
@@ -142,6 +149,18 @@ export type ImageActivityOptions<
   threadId?: string
   /** Stable run id for correlating this run when persisted. */
   runId?: string
+  /**
+   * Maximum duration of this activity invocation in milliseconds.
+   * No SDK-wide default — choose a value suitable for the provider and job.
+   * Composed with {@link abortSignal}; the first abort wins.
+   */
+  timeout?: number
+  /**
+   * Caller cancellation signal (request disconnects, job/runtime cancellation).
+   * Composed with {@link timeout} into an effective signal forwarded to the
+   * adapter. Request-specific — not stored on global provider client config.
+   */
+  abortSignal?: AbortSignal
 } & ({} extends ImageProviderOptionsForModel<TAdapter, TAdapter['model']>
   ? {
       /** Provider-specific options for image generation */ modelOptions?: ImageProviderOptionsForModel<
@@ -260,12 +279,18 @@ async function runGenerateImage<
     middleware,
     threadId,
     runId,
+    timeout,
+    abortSignal: callerAbortSignal,
     ...rest
   } = options
   const model = adapter.model
   const requestId = createId('image')
   const startTime = Date.now()
   const logger: InternalLogger = resolveDebugOption(options.debug)
+  const abortControls = createActivityAbortControls({
+    timeout,
+    abortSignal: callerAbortSignal,
+  })
 
   const mwCtx = createGenerationContext({
     requestId,
@@ -311,7 +336,16 @@ async function runGenerateImage<
   })
 
   try {
-    const rawResult = await adapter.generateImages({ ...rest, model, logger })
+    const rawResult = await raceWithAbort(
+      adapter.generateImages({
+        ...rest,
+        model,
+        logger,
+        ...(abortControls.signal ? { abortSignal: abortControls.signal } : {}),
+      }),
+      abortControls.signal,
+    )
+    abortControls.clear()
     const result = await applyGenerationResultTransforms(mwCtx, rawResult)
     const duration = Date.now() - startTime
 
@@ -355,10 +389,19 @@ async function runGenerateImage<
 
     return result
   } catch (error) {
-    await runGenerationError(middleware, mwCtx, {
-      error,
-      duration: Date.now() - startTime,
-    })
+    abortControls.clear()
+    const duration = Date.now() - startTime
+    if (isActivityAbortError(error, abortControls.signal)) {
+      await runGenerationAbort(middleware, mwCtx, {
+        reason: abortReasonMessage(error, abortControls.signal),
+        duration,
+      })
+    } else {
+      await runGenerationError(middleware, mwCtx, {
+        error,
+        duration,
+      })
+    }
     logger.errors('generateImage activity failed', {
       error,
       source: 'generateImage',