diff --git a/.changeset/activity-abort-timeout.md b/.changeset/activity-abort-timeout.md new file mode 100644 index 000000000..e579c49f3 --- /dev/null +++ b/.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. diff --git a/packages/ai-fal/src/adapters/audio.ts b/packages/ai-fal/src/adapters/audio.ts index e06aff913..26c067e93 100644 --- a/packages/ai-fal/src/adapters/audio.ts +++ b/packages/ai-fal/src/adapters/audio.ts @@ -84,7 +84,11 @@ export class FalAudioAdapter 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', { diff --git a/packages/ai-fal/src/adapters/image.ts b/packages/ai-fal/src/adapters/image.ts index d0cfbd969..69dc7c1c7 100644 --- a/packages/ai-fal/src/adapters/image.ts +++ b/packages/ai-fal/src/adapters/image.ts @@ -87,7 +87,12 @@ export class FalImageAdapter 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', { diff --git a/packages/ai-fal/src/adapters/speech.ts b/packages/ai-fal/src/adapters/speech.ts index cdb868806..44f9a05e4 100644 --- a/packages/ai-fal/src/adapters/speech.ts +++ b/packages/ai-fal/src/adapters/speech.ts @@ -56,7 +56,11 @@ export class FalSpeechAdapter 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', { diff --git a/packages/ai-fal/src/adapters/transcription.ts b/packages/ai-fal/src/adapters/transcription.ts index 8d940b7fa..4a0a24d09 100644 --- a/packages/ai-fal/src/adapters/transcription.ts +++ b/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', { diff --git a/packages/ai-fal/src/adapters/video.ts b/packages/ai-fal/src/adapters/video.ts index e2d19e9b0..ec0f75bf9 100644 --- a/packages/ai-fal/src/adapters/video.ts +++ b/packages/ai-fal/src/adapters/video.ts @@ -179,9 +179,11 @@ export class FalVideoAdapter extends BaseVideoAdapter< ...(duration ? { duration } : {}), } as FalModelInput - // 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 { diff --git a/packages/ai-fal/tests/image-adapter.test.ts b/packages/ai-fal/tests/image-adapter.test.ts index b429f1000..dcbe1030d 100644 --- a/packages/ai-fal/tests/image-adapter.test.ts +++ b/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' }, diff --git a/packages/ai/src/activities/generateAudio/index.ts b/packages/ai/src/activities/generateAudio/index.ts index 97ff11283..e02baf4cc 100644 --- a/packages/ai/src/activities/generateAudio/index.ts +++ b/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 | 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', diff --git a/packages/ai/src/activities/generateImage/index.ts b/packages/ai/src/activities/generateImage/index.ts index 336be0842..713ca2cec 100644 --- a/packages/ai/src/activities/generateImage/index.ts +++ b/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 ? { /** 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', diff --git a/packages/ai/src/activities/generateSpeech/index.ts b/packages/ai/src/activities/generateSpeech/index.ts index ef193e1a1..9a96d8221 100644 --- a/packages/ai/src/activities/generateSpeech/index.ts +++ b/packages/ai/src/activities/generateSpeech/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' @@ -92,6 +99,18 @@ export interface TTSActivityOptions< 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 } // =========================== @@ -175,12 +194,18 @@ async function runGenerateSpeech< middleware, threadId, runId, + timeout, + abortSignal: callerAbortSignal, ...rest } = options const model = adapter.model const requestId = createId('speech') 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 ?? @@ -223,7 +248,16 @@ async function runGenerateSpeech< }) try { - const rawResult = await adapter.generateSpeech({ ...rest, model, logger }) + const rawResult = await raceWithAbort( + adapter.generateSpeech({ + ...rest, + model, + logger, + ...(abortControls.signal ? { abortSignal: abortControls.signal } : {}), + }), + abortControls.signal, + ) + abortControls.clear() const result = await applyGenerationResultTransforms(mwCtx, rawResult) const duration = Date.now() - startTime @@ -263,6 +297,7 @@ async function runGenerateSpeech< return result } catch (error) { + abortControls.clear() const duration = Date.now() - startTime const err = error as Error aiEventClient.emit('speech:request:error', { @@ -274,10 +309,17 @@ async function runGenerateSpeech< modelOptions: rest.modelOptions as Record | undefined, timestamp: Date.now(), }) - await runGenerationError(middleware, mwCtx, { - error, - duration, - }) + if (isActivityAbortError(error, abortControls.signal)) { + await runGenerationAbort(middleware, mwCtx, { + reason: abortReasonMessage(error, abortControls.signal), + duration, + }) + } else { + await runGenerationError(middleware, mwCtx, { + error, + duration, + }) + } logger.errors('generateSpeech activity failed', { error, source: 'generateSpeech', diff --git a/packages/ai/src/activities/generateTranscription/index.ts b/packages/ai/src/activities/generateTranscription/index.ts index 02ec87fba..96d85f2e5 100644 --- a/packages/ai/src/activities/generateTranscription/index.ts +++ b/packages/ai/src/activities/generateTranscription/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' @@ -99,6 +106,18 @@ export interface TranscriptionActivityOptions< 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 } // =========================== @@ -211,12 +230,18 @@ async function runGenerateTranscription< middleware, threadId, runId, + timeout, + abortSignal: callerAbortSignal, ...rest } = options const model = adapter.model const requestId = createId('transcription') 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 ?? @@ -258,7 +283,16 @@ async function runGenerateTranscription< }) try { - const rawResult = await adapter.transcribe({ ...rest, model, logger }) + const rawResult = await raceWithAbort( + adapter.transcribe({ + ...rest, + model, + logger, + ...(abortControls.signal ? { abortSignal: abortControls.signal } : {}), + }), + abortControls.signal, + ) + abortControls.clear() const result = await applyGenerationResultTransforms(mwCtx, rawResult) const duration = Date.now() - startTime @@ -286,6 +320,7 @@ async function runGenerateTranscription< return result } catch (error) { + abortControls.clear() const duration = Date.now() - startTime const err = error as Error aiEventClient.emit('transcription:request:error', { @@ -297,10 +332,17 @@ async function runGenerateTranscription< modelOptions: rest.modelOptions as Record | undefined, timestamp: Date.now(), }) - await runGenerationError(middleware, mwCtx, { - error, - duration, - }) + if (isActivityAbortError(error, abortControls.signal)) { + await runGenerationAbort(middleware, mwCtx, { + reason: abortReasonMessage(error, abortControls.signal), + duration, + }) + } else { + await runGenerationError(middleware, mwCtx, { + error, + duration, + }) + } logger.errors('generateTranscription activity failed', { error, source: 'generateTranscription', diff --git a/packages/ai/src/activities/generateVideo/index.ts b/packages/ai/src/activities/generateVideo/index.ts index 433eb79b0..9fdc97745 100644 --- a/packages/ai/src/activities/generateVideo/index.ts +++ b/packages/ai/src/activities/generateVideo/index.ts @@ -19,6 +19,13 @@ import { runGenerationStart, runGenerationUsage, } from '../middleware/run' +import { + abortReasonMessage, + createActivityAbortControls, + isActivityAbortError, + raceWithAbort, + toAbortError, +} from '../../utilities/activity-abort' import type { InternalLogger } from '../../logger/internal-logger' import type { DebugOption } from '../../logger/types' import type { @@ -235,6 +242,24 @@ export type VideoCreateOptions< * job to resume. */ middleware?: Array + /** + * 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. + * + * In stream mode this bounds the full create→poll→complete lifecycle and + * complements {@link maxDuration} (which defaults to 10 minutes). When both + * are set, the shorter limit wins via signal composition against the + * polling deadline. + */ + timeout?: number + /** + * Caller cancellation signal (request disconnects, job/runtime cancellation). + * Composed with {@link timeout} into an effective signal forwarded to the + * adapter on job submission. Request-specific — not stored on global + * provider client config. + */ + abortSignal?: AbortSignal } & ({} extends VideoProviderOptions ? { /** Provider-specific options for video generation */ modelOptions?: VideoProviderOptions @@ -413,11 +438,24 @@ function videoRunIdForJob(provider: string, jobId: string): string { async function runCreateVideoJob< TAdapter extends VideoAdapter, >(options: VideoCreateOptions): Promise { - const { adapter, prompt, size, duration, modelOptions, middleware } = options + const { + adapter, + prompt, + size, + duration, + modelOptions, + middleware, + timeout, + abortSignal: callerAbortSignal, + } = options const model = adapter.model const requestId = createId('video') 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 ?? @@ -451,24 +489,38 @@ async function runCreateVideoJob< let jobResult: VideoJobResult try { - jobResult = await adapter.createVideoJob({ - model, - prompt, - size, - duration, - modelOptions, - logger, - }) + jobResult = await raceWithAbort( + adapter.createVideoJob({ + model, + prompt, + size, + duration, + modelOptions, + logger, + ...(abortControls.signal ? { abortSignal: abortControls.signal } : {}), + }), + abortControls.signal, + ) + abortControls.clear() } catch (error) { + abortControls.clear() // No jobId exists, so this run can only be keyed on the request. Start it // just to fail it: `generationRuns.update` on an unknown run id is a no-op // by contract, so without the `onStart` the failure would persist nowhere. const failedCtx = contextFor() await runGenerationStart(middleware, failedCtx) - await runGenerationError(middleware, failedCtx, { - error, - duration: Date.now() - startTime, - }) + const elapsed = Date.now() - startTime + if (isActivityAbortError(error, abortControls.signal)) { + await runGenerationAbort(middleware, failedCtx, { + reason: abortReasonMessage(error, abortControls.signal), + duration: elapsed, + }) + } else { + await runGenerationError(middleware, failedCtx, { + error, + duration: elapsed, + }) + } logger.errors('generateVideo activity failed', { error, source: 'generateVideo', @@ -489,8 +541,25 @@ async function runCreateVideoJob< return await applyGenerationResultTransforms(mwCtx, jobResult) } -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)) +function sleep(ms: number, signal?: AbortSignal): Promise { + if (!signal) { + return new Promise((resolve) => setTimeout(resolve, ms)) + } + if (signal.aborted) { + return Promise.reject(toAbortError(signal.reason)) + } + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal.removeEventListener('abort', onAbort) + resolve() + }, ms) + const onAbort = () => { + clearTimeout(timer) + signal.removeEventListener('abort', onAbort) + reject(toAbortError(signal.reason)) + } + signal.addEventListener('abort', onAbort, { once: true }) + }) } /** @@ -500,7 +569,16 @@ function sleep(ms: number): Promise { async function* runStreamingVideoGeneration< TAdapter extends VideoAdapter, >(options: VideoCreateOptions): AsyncIterable { - const { adapter, prompt, size, duration, modelOptions, middleware } = options + const { + adapter, + prompt, + size, + duration, + modelOptions, + middleware, + timeout, + abortSignal: callerAbortSignal, + } = options const model = adapter.model const runId = options.runId ?? createId('run') const requestId = createId('video') @@ -508,6 +586,10 @@ async function* runStreamingVideoGeneration< const pollingInterval = options.pollingInterval ?? 2000 const maxDuration = options.maxDuration ?? 600_000 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 ?? @@ -555,19 +637,24 @@ async function* runStreamingVideoGeneration< }, ) - // Tracks whether a terminal observer event (finish/error) has already fired, - // so the `finally` below can fire one on abandonment without double-firing. + // Tracks whether a terminal observer event (finish/error/abort) has already + // fired, so the `finally` below can fire one on abandonment without + // double-firing. let settled = false try { // Create the video generation job - const jobResult = await adapter.createVideoJob({ - model, - prompt, - size, - duration, - modelOptions, - logger, - }) + const jobResult = await raceWithAbort( + adapter.createVideoJob({ + model, + prompt, + size, + duration, + modelOptions, + logger, + ...(abortControls.signal ? { abortSignal: abortControls.signal } : {}), + }), + abortControls.signal, + ) yield { type: 'CUSTOM', @@ -579,7 +666,7 @@ async function* runStreamingVideoGeneration< // Poll for completion const startTime = Date.now() while (Date.now() - startTime < maxDuration) { - await sleep(pollingInterval) + await sleep(pollingInterval, abortControls.signal) const statusResult = await adapter.getVideoStatus(jobResult.jobId) @@ -632,6 +719,7 @@ async function* runStreamingVideoGeneration< usage: urlResult.usage, }) settled = true + abortControls.clear() yield { type: 'CUSTOM', @@ -657,15 +745,24 @@ async function* runStreamingVideoGeneration< throw new Error('Video generation timed out') } catch (error: unknown) { + abortControls.clear() const payload = toRunErrorPayload(error, 'Video generation failed') - // Mark settled before firing onError: if a user error-hook throws, the - // `finally` below must still not double-fire onAbort over the same op + // Mark settled before firing terminal hooks: if a user error-hook throws, + // the `finally` below must still not double-fire onAbort over the same op // (which would mask the original error and end the span twice). settled = true - await runGenerationError(middleware, mwCtx, { - error, - duration: Date.now() - obsStartTime, - }) + const elapsed = Date.now() - obsStartTime + if (isActivityAbortError(error, abortControls.signal)) { + await runGenerationAbort(middleware, mwCtx, { + reason: abortReasonMessage(error, abortControls.signal), + duration: elapsed, + }) + } else { + await runGenerationError(middleware, mwCtx, { + error, + duration: elapsed, + }) + } logger.errors('generateVideo activity failed', { message: payload.message, code: payload.code, @@ -681,6 +778,7 @@ async function* runStreamingVideoGeneration< timestamp: Date.now(), } as StreamChunk } finally { + abortControls.clear() if (!settled) { // The consumer abandoned the stream (broke the `for await` loop or // disconnected) before completion, so the generator is being unwound at diff --git a/packages/ai/src/activities/summarize/index.ts b/packages/ai/src/activities/summarize/index.ts index fc08afb5a..f890844b0 100644 --- a/packages/ai/src/activities/summarize/index.ts +++ b/packages/ai/src/activities/summarize/index.ts @@ -17,6 +17,12 @@ import { 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' @@ -93,6 +99,18 @@ export interface SummarizeActivityOptions< * mid-summary fires `onAbort`. */ middleware?: Array + /** + * 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 /** * Whether to stream the summarization result. * When true, returns an AsyncIterable for streaming output. @@ -216,13 +234,26 @@ export function summarize< async function runSummarize( options: SummarizeActivityOptions, false>, ): Promise { - const { adapter, text, maxLength, style, focus, modelOptions, middleware } = - options + const { + adapter, + text, + maxLength, + style, + focus, + modelOptions, + middleware, + timeout, + abortSignal: callerAbortSignal, + } = options const model = adapter.model const requestId = createId('summarize') const inputLength = text.length const startTime = Date.now() const logger: InternalLogger = resolveDebugOption(options.debug) + const abortControls = createActivityAbortControls({ + timeout, + abortSignal: callerAbortSignal, + }) const mwCtx = createGenerationContext({ requestId, @@ -259,10 +290,15 @@ async function runSummarize( focus, modelOptions, logger, + ...(abortControls.signal ? { abortSignal: abortControls.signal } : {}), } try { - const rawResult = await adapter.summarize(summarizeOptions) + const rawResult = await raceWithAbort( + adapter.summarize(summarizeOptions), + abortControls.signal, + ) + abortControls.clear() // Transforms run before anything observes the result — the same order every // media activity uses — so the run record and the returned value are the // same object. @@ -294,10 +330,19 @@ async function runSummarize( 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('summarize activity failed', { error, source: 'summarize', diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index af1a20059..71ea2a39f 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -1992,6 +1992,12 @@ export interface SummarizationOptions< * call logger.request() before the SDK call and logger.errors() in catch blocks. */ logger: InternalLogger + /** + * Effective abort signal composed by the activity from caller `abortSignal` + * and/or `timeout`. Adapters should forward this to the provider SDK when + * supported. Request-specific — never store on a global client config. + */ + abortSignal?: AbortSignal } export interface SummarizationResult { @@ -2129,6 +2135,12 @@ export interface ImageGenerationOptions< * call logger.request() before the SDK call and logger.errors() in catch blocks. */ logger: InternalLogger + /** + * Effective abort signal composed by the activity from caller `abortSignal` + * and/or `timeout`. Adapters should forward this to the provider SDK when + * supported. Request-specific — never store on a global client config. + */ + abortSignal?: AbortSignal } /** @@ -2242,6 +2254,12 @@ export interface AudioGenerationOptions< * catch blocks. */ logger: InternalLogger + /** + * Effective abort signal composed by the activity from caller `abortSignal` + * and/or `timeout`. Adapters should forward this to the provider SDK when + * supported. Request-specific — never store on a global client config. + */ + abortSignal?: AbortSignal } /** @@ -2311,6 +2329,12 @@ export interface VideoGenerationOptions< * call logger.request() before the SDK call and logger.errors() in catch blocks. */ logger: InternalLogger + /** + * Effective abort signal composed by the activity from caller `abortSignal` + * and/or `timeout`. Adapters should forward this to the provider SDK when + * supported. Request-specific — never store on a global client config. + */ + abortSignal?: AbortSignal } /** @@ -2396,6 +2420,12 @@ export interface TTSOptions { * catch blocks. */ logger: InternalLogger + /** + * Effective abort signal composed by the activity from caller `abortSignal` + * and/or `timeout`. Adapters should forward this to the provider SDK when + * supported. Request-specific — never store on a global client config. + */ + abortSignal?: AbortSignal } /** @@ -2456,6 +2486,12 @@ export interface TranscriptionOptions< * in catch blocks. */ logger: InternalLogger + /** + * Effective abort signal composed by the activity from caller `abortSignal` + * and/or `timeout`. Adapters should forward this to the provider SDK when + * supported. Request-specific — never store on a global client config. + */ + abortSignal?: AbortSignal } /** diff --git a/packages/ai/src/utilities/activity-abort.ts b/packages/ai/src/utilities/activity-abort.ts new file mode 100644 index 000000000..15e375ba3 --- /dev/null +++ b/packages/ai/src/utilities/activity-abort.ts @@ -0,0 +1,197 @@ +/** + * Shared abort/timeout composition for media (and summarize) activities. + * + * Callers pass optional `timeout` and/or `abortSignal` on activity options. + * Core composes them into one effective signal, races the adapter call so a + * hung provider still rejects, clears timeout resources on settle, and + * classifies aborts so lifecycle middleware gets `onAbort` rather than + * `onError`. + */ + +const ABORT_ERROR_NAMES = new Set([ + 'AbortError', + 'TimeoutError', + 'APIUserAbortError', + 'RequestAbortedError', +]) + +/** + * Combine two optional AbortSignals into one that aborts when either does. + * Returns the other signal directly when one is absent or already aborted. + * First abort wins and preserves its reason. + * + * Manual implementation — `AbortSignal.any` requires Node >= 20.3. + */ +export function combineAbortSignals( + a: AbortSignal | undefined, + b: AbortSignal | undefined, +): AbortSignal | undefined { + if (!a) return b + if (!b) return a + if (a.aborted) return a + if (b.aborted) return b + const controller = new AbortController() + const onAbort = (source: AbortSignal) => () => { + controller.abort(source.reason) + } + a.addEventListener('abort', onAbort(a), { once: true }) + b.addEventListener('abort', onAbort(b), { once: true }) + return controller.signal +} + +function createTimeoutReason(ms: number): Error { + if (typeof DOMException !== 'undefined') { + return new DOMException(`Activity timed out after ${ms}ms`, 'TimeoutError') + } + const err = new Error(`Activity timed out after ${ms}ms`) + err.name = 'TimeoutError' + return err +} + +/** Normalize an abort reason into an Error the activity can reject with. */ +export function toAbortError(reason: unknown): Error { + if (reason instanceof Error) return reason + if (typeof reason === 'string' && reason.length > 0) { + const err = new Error(reason) + err.name = 'AbortError' + return err + } + const err = new Error('The operation was aborted') + err.name = 'AbortError' + return err +} + +export interface ActivityAbortControls { + /** Effective signal, or `undefined` when neither timeout nor caller signal. */ + signal: AbortSignal | undefined + /** Clear the timeout timer if one was set. Idempotent. */ + clear: () => void +} + +/** + * Compose an activity-level timeout with a caller AbortSignal. + * + * - No SDK-wide default timeout; omit both for unlimited wait. + * - First of caller cancellation or timeout wins and keeps its reason. + * - Call `clear()` when the activity settles (success or failure) so timers + * do not leak. + */ +export function createActivityAbortControls(options: { + abortSignal?: AbortSignal + timeout?: number +}): ActivityAbortControls { + let timeoutId: ReturnType | undefined + let timeoutSignal: AbortSignal | undefined + + if (options.timeout !== undefined) { + if (!Number.isFinite(options.timeout) || options.timeout < 0) { + throw new Error( + `Invalid activity timeout: expected a non-negative finite number, got ${String(options.timeout)}`, + ) + } + const controller = new AbortController() + timeoutSignal = controller.signal + const ms = options.timeout + timeoutId = setTimeout(() => { + controller.abort(createTimeoutReason(ms)) + }, ms) + } + + const signal = combineAbortSignals(options.abortSignal, timeoutSignal) + + return { + signal, + clear: () => { + if (timeoutId !== undefined) { + clearTimeout(timeoutId) + timeoutId = undefined + } + }, + } +} + +/** + * Reject when `signal` aborts, even if the underlying promise ignores it. + * Ensures activity-level timeouts work for adapters that do not yet forward + * the signal to the provider SDK. + * + * When the signal wins, the adapter promise is observed with an empty handler + * so a later settle cannot surface as an unhandled rejection. + */ +export function raceWithAbort( + promise: Promise, + signal: AbortSignal | undefined, +): Promise { + if (!signal) return promise + + const swallow = () => { + // Observe the adapter promise without acting on its outcome so a late + // reject after we already aborted cannot become an unhandled rejection. + promise.then( + () => undefined, + () => undefined, + ) + } + + if (signal.aborted) { + swallow() + return Promise.reject(toAbortError(signal.reason)) + } + + return new Promise((resolve, reject) => { + let settled = false + const onAbort = () => { + if (settled) return + settled = true + cleanup() + swallow() + reject(toAbortError(signal.reason)) + } + const cleanup = () => { + signal.removeEventListener('abort', onAbort) + } + signal.addEventListener('abort', onAbort, { once: true }) + promise.then( + (value) => { + if (settled) return + settled = true + cleanup() + resolve(value) + }, + (error: unknown) => { + if (settled) return + settled = true + cleanup() + reject(error) + }, + ) + }) +} + +/** + * Whether a thrown value (and optional effective signal) should route to + * middleware `onAbort` instead of `onError`. + */ +export function isActivityAbortError( + error: unknown, + signal?: AbortSignal, +): boolean { + if (signal?.aborted) return true + if (!error || typeof error !== 'object') return false + const name = (error as { name?: unknown }).name + return typeof name === 'string' && ABORT_ERROR_NAMES.has(name) +} + +/** Best-effort string reason for {@link GenerationAbortInfo}. */ +export function abortReasonMessage( + error: unknown, + signal?: AbortSignal, +): string | undefined { + if (signal?.reason !== undefined) { + if (typeof signal.reason === 'string') return signal.reason + if (signal.reason instanceof Error) return signal.reason.message + } + if (error instanceof Error) return error.message + if (typeof error === 'string') return error + return undefined +} diff --git a/packages/ai/tests/activity-abort.test.ts b/packages/ai/tests/activity-abort.test.ts new file mode 100644 index 000000000..12579e0a1 --- /dev/null +++ b/packages/ai/tests/activity-abort.test.ts @@ -0,0 +1,236 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { generateImage } from '../src/index' +import type { GenerationMiddleware } from '../src/activities/middleware/types' +import type { ImageAdapter } from '../src/activities/generateImage/adapter' + +function createMockImageAdapter( + overrides?: Partial<{ + generateImages: (...args: Array) => Promise + }>, +): ImageAdapter { + return { + kind: 'image' as const, + name: 'test-image', + model: 'test-model', + '~types': {} as any, + generateImages: + overrides?.generateImages ?? + vi.fn(async () => ({ + id: 'img-1', + model: 'test-model', + images: [{ url: 'https://example.com/image.png' }], + })), + } +} + +describe('generateImage abortSignal + timeout', () => { + afterEach(() => { + vi.useRealTimers() + }) + + it('forwards the composed abortSignal to the media adapter', async () => { + const generateImages = vi.fn( + async (opts: { abortSignal?: AbortSignal }) => { + expect(opts.abortSignal).toBeInstanceOf(AbortSignal) + expect(opts.abortSignal?.aborted).toBe(false) + return { + id: 'img-1', + model: 'test-model', + images: [{ url: 'https://example.com/image.png' }], + } + }, + ) + const adapter = createMockImageAdapter({ generateImages }) + const controller = new AbortController() + + await generateImage({ + adapter, + prompt: 'a cat', + abortSignal: controller.signal, + }) + + expect(generateImages).toHaveBeenCalledTimes(1) + const passed = generateImages.mock.calls[0]![0] as { + abortSignal?: AbortSignal + } + expect(passed.abortSignal).toBeDefined() + }) + + it('timeout aborts and is forwarded to the media adapter', async () => { + vi.useFakeTimers() + let seenSignal: AbortSignal | undefined + const generateImages = vi.fn( + (_opts: { abortSignal?: AbortSignal }) => + new Promise((_resolve, _reject) => { + seenSignal = _opts.abortSignal + // Never resolves — activity timeout must win. + }), + ) + const adapter = createMockImageAdapter({ generateImages }) + + const promise = generateImage({ + adapter, + prompt: 'a cat', + timeout: 50, + }) + // Attach rejection handler before timers fire to avoid unhandled rejections. + const assertion = expect(promise).rejects.toMatchObject({ + name: 'TimeoutError', + message: expect.stringContaining('timed out'), + }) + + // Allow the adapter call to start, then advance past the timeout. + await vi.advanceTimersByTimeAsync(0) + expect(seenSignal).toBeInstanceOf(AbortSignal) + expect(seenSignal?.aborted).toBe(false) + + await vi.advanceTimersByTimeAsync(50) + await assertion + expect(seenSignal?.aborted).toBe(true) + }) + + it('caller-provided signal aborts and first abort reason wins', async () => { + const generateImages = vi.fn( + (opts: { abortSignal?: AbortSignal }) => + new Promise((_resolve, reject) => { + opts.abortSignal?.addEventListener( + 'abort', + () => { + reject( + opts.abortSignal?.reason instanceof Error + ? opts.abortSignal.reason + : Object.assign(new Error('aborted'), { name: 'AbortError' }), + ) + }, + { once: true }, + ) + }), + ) + const adapter = createMockImageAdapter({ generateImages }) + const controller = new AbortController() + + const promise = generateImage({ + adapter, + prompt: 'a cat', + abortSignal: controller.signal, + timeout: 60_000, + }) + const assertion = expect(promise).rejects.toMatchObject({ + message: 'caller cancelled', + }) + + // Let the adapter attach its listener. + await Promise.resolve() + controller.abort(new Error('caller cancelled')) + + await assertion + }) + + it('successful completion clears the timer (no late abort)', async () => { + vi.useFakeTimers() + const generateImages = vi.fn(async () => ({ + id: 'img-1', + model: 'test-model', + images: [{ url: 'https://example.com/image.png' }], + })) + const adapter = createMockImageAdapter({ generateImages }) + + const resultPromise = generateImage({ + adapter, + prompt: 'a cat', + timeout: 1_000, + }) + + await expect(resultPromise).resolves.toMatchObject({ + id: 'img-1', + }) + + // Advancing past the original timeout must not throw or leave a hanging + // timer that would abort a subsequent unrelated operation. + await vi.advanceTimersByTimeAsync(5_000) + expect(generateImages).toHaveBeenCalledTimes(1) + }) + + it('timeout triggers onAbort exactly once rather than onError', async () => { + vi.useFakeTimers() + const generateImages = vi.fn( + () => + new Promise(() => { + // hang + }), + ) + const adapter = createMockImageAdapter({ generateImages }) + + const onAbort = vi.fn() + const onError = vi.fn() + const onFinish = vi.fn() + const middleware: Array = [ + { + name: 'test-abort', + onAbort, + onError, + onFinish, + }, + ] + + const promise = generateImage({ + adapter, + prompt: 'a cat', + timeout: 25, + middleware, + }) + const assertion = expect(promise).rejects.toMatchObject({ + name: 'TimeoutError', + }) + + await vi.advanceTimersByTimeAsync(25) + await assertion + + expect(onAbort).toHaveBeenCalledTimes(1) + expect(onError).not.toHaveBeenCalled() + expect(onFinish).not.toHaveBeenCalled() + expect(onAbort.mock.calls[0]![1]).toMatchObject({ + reason: expect.stringContaining('timed out'), + duration: expect.any(Number), + }) + }) + + it('caller abort triggers onAbort exactly once', async () => { + const generateImages = vi.fn( + (opts: { abortSignal?: AbortSignal }) => + new Promise((_resolve, reject) => { + opts.abortSignal?.addEventListener( + 'abort', + () => + reject( + Object.assign(new Error('aborted'), { name: 'AbortError' }), + ), + { once: true }, + ) + }), + ) + const adapter = createMockImageAdapter({ generateImages }) + const controller = new AbortController() + + const onAbort = vi.fn() + const onError = vi.fn() + const middleware: Array = [ + { name: 'test-abort', onAbort, onError }, + ] + + const promise = generateImage({ + adapter, + prompt: 'a cat', + abortSignal: controller.signal, + middleware, + }) + const assertion = expect(promise).rejects.toBeTruthy() + + await Promise.resolve() + controller.abort() + await assertion + + expect(onAbort).toHaveBeenCalledTimes(1) + expect(onError).not.toHaveBeenCalled() + }) +})