From e33f99b5af6eb02b0f0e85b18263e5eb989e6987 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 2 Nov 2026 15:16:57 +0100 Subject: [PATCH 01/12] fix(ai): make cancels reach the provider request A cancel rejected the caller, but the provider request kept running. - summarize(): pass the abort signal to the wrapped chatStream request. - summarize({ stream: true }): compose abortSignal and timeout like the non-stream path and pass the signal to summarizeStream. - chat(): middleware ctx.abort() now aborts the adapter request (stream and structured output) and ctx.signal. - Subagent children follow the tool context signal. - A server tool that ignores its signal no longer blocks an aborted run. - generateVideo(): race getVideoStatus() and getVideo() with the signal. - toWebSocket(): stop pumping an aborted turn. Adds unit repros, an E2E spec for streaming summarize timeout, and a changeset. --- .changeset/core-abort-reaches-provider.md | 5 + docs/advanced/middleware-advanced.md | 2 +- .../ai/src/activities/chat/agents/spawn.ts | 4 +- packages/ai/src/activities/chat/index.ts | 9 +- .../src/activities/chat/middleware/types.ts | 2 +- .../src/activities/chat/tools/tool-calls.ts | 8 +- .../ai/src/activities/generateVideo/index.ts | 10 +- .../summarize/chat-stream-summarize.ts | 5 + packages/ai/src/activities/summarize/index.ts | 24 +- packages/ai/src/stream-to-websocket.ts | 2 + packages/ai/tests/abort-signal-reach.test.ts | 380 ++++++++++++++++++ .../e2e/fixtures/summarize-abort/basic.json | 14 + testing/e2e/src/routes/api.summarize.ts | 4 + testing/e2e/tests/summarize-abort.spec.ts | 26 ++ 14 files changed, 480 insertions(+), 15 deletions(-) create mode 100644 .changeset/core-abort-reaches-provider.md create mode 100644 packages/ai/tests/abort-signal-reach.test.ts create mode 100644 testing/e2e/fixtures/summarize-abort/basic.json create mode 100644 testing/e2e/tests/summarize-abort.spec.ts diff --git a/.changeset/core-abort-reaches-provider.md b/.changeset/core-abort-reaches-provider.md new file mode 100644 index 0000000000..5f1d8cb523 --- /dev/null +++ b/.changeset/core-abort-reaches-provider.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai': patch +--- + +A cancel now stops the provider request, not only the caller. `summarize()` (also with `stream: true`, which now honors `abortSignal` and `timeout`), middleware `ctx.abort()` (adapter request, structured output, and `ctx.signal`), subagent children, server tools that ignore their signal, `generateVideo` status polls, and aborted `toWebSocket` turns now stop when the run is cancelled. diff --git a/docs/advanced/middleware-advanced.md b/docs/advanced/middleware-advanced.md index c69c4038fe..42ab4db7e4 100644 --- a/docs/advanced/middleware-advanced.md +++ b/docs/advanced/middleware-advanced.md @@ -838,7 +838,7 @@ Each hook gets a `ChatMiddlewareContext` as its first argument. The fields you u | `chunkIndex` | `number` | How many chunks the run sent so far | | `callMetadata` | `Readonly` | The `metadata.tanstack` of the current model call (`source`, `stopReason`, `responseId`, and more). It changes as chunks arrive. Each assistant message of the call gets a copy. | | `messages` | `ReadonlyArray` | The current messages (read-only) | -| `signal` | `AbortSignal \| undefined` | The external abort signal | +| `signal` | `AbortSignal \| undefined` | Aborts when the caller aborts the run or a middleware calls `ctx.abort()` | | `abort(reason?)` | `function` | Abort the run from middleware | | `retry()` | `function` | Call the model again. Works only in [`onModelError`](#onmodelerror). | | `emitCustomEvent(name, value, options?)` | `function` | Push a `CUSTOM` chunk onto the stream now. See [Live custom events](#live-custom-events). | diff --git a/packages/ai/src/activities/chat/agents/spawn.ts b/packages/ai/src/activities/chat/agents/spawn.ts index 6ed92991d2..a103544ff9 100644 --- a/packages/ai/src/activities/chat/agents/spawn.ts +++ b/packages/ai/src/activities/chat/agents/spawn.ts @@ -1355,6 +1355,7 @@ export function createSyntheticSubagentTools( const toolContext = context as | { toolCallId?: string + abortSignal?: AbortSignal [EMIT_STREAM_CHUNK]?: (chunk: StreamChunk) => void } | undefined @@ -1419,7 +1420,8 @@ export function createSyntheticSubagentTools( } active += 1 const sink = createSubagentSink() - const link = linkAbort(parent.abortSignal) + // The tool context signal also fires on a middleware ctx.abort(). + const link = linkAbort(toolContext?.abortSignal ?? parent.abortSignal) let timedOut = false const timer = timeout === undefined diff --git a/packages/ai/src/activities/chat/index.ts b/packages/ai/src/activities/chat/index.ts index c41beeab47..ec71466a92 100644 --- a/packages/ai/src/activities/chat/index.ts +++ b/packages/ai/src/activities/chat/index.ts @@ -1275,9 +1275,6 @@ class TextEngine< >(this.tools) this.requestId = this.createId('chat') this.streamId = this.createId('stream') - this.effectiveRequest = config.params.abortController - ? { signal: config.params.abortController.signal } - : undefined this.effectiveSignal = config.params.abortController?.signal this.threadId = config.params.threadId this.runIdOverride = config.params.runId @@ -1296,6 +1293,10 @@ class TextEngine< this.effectiveSignal, this.middlewareAbortController.signal, ) + // The provider request follows ctx.abort() too, not only the caller. + this.effectiveRequest = this.toolAbortSignal + ? { signal: this.toolAbortSignal } + : undefined this.middlewareCtx = { requestId: this.requestId, streamId: this.streamId, @@ -1312,7 +1313,7 @@ class TextEngine< iteration: 0, chunkIndex: 0, callMetadata: this.currentCallMetadata, - signal: this.effectiveSignal, + signal: this.toolAbortSignal, abort: (reason?: string) => { this.abortReason = reason this.middlewareAbortController?.abort(reason) diff --git a/packages/ai/src/activities/chat/middleware/types.ts b/packages/ai/src/activities/chat/middleware/types.ts index 76e78c24ac..f31a3dc91e 100644 --- a/packages/ai/src/activities/chat/middleware/types.ts +++ b/packages/ai/src/activities/chat/middleware/types.ts @@ -234,7 +234,7 @@ export interface ChatMiddlewareContext { * message of the call gets a copy of it. */ callMetadata: Readonly - /** Abort signal from the chat request */ + /** Aborts when the caller aborts the chat request or a middleware calls `abort()`. */ signal?: AbortSignal /** Abort the chat run with a reason */ abort: (reason?: string) => void diff --git a/packages/ai/src/activities/chat/tools/tool-calls.ts b/packages/ai/src/activities/chat/tools/tool-calls.ts index fa984b6d52..e8b7ab800f 100644 --- a/packages/ai/src/activities/chat/tools/tool-calls.ts +++ b/packages/ai/src/activities/chat/tools/tool-calls.ts @@ -3,6 +3,7 @@ import { normalizeToolResult } from '../../../utilities/tool-result' import { tanstackMetadata } from '../../../utilities/merge-metadata' import { isProviderExecutedToolCall } from '../../../utilities/provider-executed' import { mergeStreams } from '../../../utilities/merge-streams' +import { raceWithAbort } from '../../../utilities/activity-abort' import { validateToolInput } from './input-validation' import type { AdapterYieldChunk } from '../../../utilities/adapter-yield-chunk' import { @@ -796,7 +797,12 @@ export async function* executeServerTool( [EMIT_STREAM_CHUNK]: (chunk: StreamChunk) => pendingEvents.push(chunk), }) } - const executionPromise = Promise.resolve(tool.execute(input, context)) + const execution = Promise.resolve(tool.execute(input, context)) + // Do not wait for a tool that ignores its signal. An abort that came + // during execute() itself keeps the result the tool already returned. + const executionPromise = context.abortSignal?.aborted + ? execution + : raceWithAbort(execution, context.abortSignal) let result = yield* executeWithEventPolling(executionPromise, pendingEvents) const duration = Date.now() - startTime diff --git a/packages/ai/src/activities/generateVideo/index.ts b/packages/ai/src/activities/generateVideo/index.ts index 44ec2c93b1..dc36e03a86 100644 --- a/packages/ai/src/activities/generateVideo/index.ts +++ b/packages/ai/src/activities/generateVideo/index.ts @@ -683,7 +683,10 @@ async function* runStreamingVideoGeneration< while (Date.now() - startTime < maxDuration) { await sleep(pollingInterval, abortControls.signal) - const statusResult = await adapter.getVideoStatus(jobResult.jobId) + const statusResult = await raceWithAbort( + adapter.getVideoStatus(jobResult.jobId), + abortControls.signal, + ) yield { type: 'CUSTOM', @@ -698,7 +701,10 @@ async function* runStreamingVideoGeneration< } if (statusResult.status === 'completed') { - const video = await adapter.getVideo(jobResult.jobId) + const video = await raceWithAbort( + adapter.getVideo(jobResult.jobId), + abortControls.signal, + ) logger.output( `activity=generateVideo jobId=${jobResult.jobId} status=completed`, diff --git a/packages/ai/src/activities/summarize/chat-stream-summarize.ts b/packages/ai/src/activities/summarize/chat-stream-summarize.ts index 3e7f2fa967..e958f2469c 100644 --- a/packages/ai/src/activities/summarize/chat-stream-summarize.ts +++ b/packages/ai/src/activities/summarize/chat-stream-summarize.ts @@ -407,6 +407,11 @@ export class ChatStreamSummarizeAdapter< // `runId` absent when unset, under `exactOptionalPropertyTypes`. ...(options.runId !== undefined ? { runId: options.runId } : {}), threadId: options.threadId, + // Without this, a cancel or timeout rejects summarize() but the wrapped + // provider request keeps running. + ...(options.abortSignal + ? { request: { signal: options.abortSignal } } + : {}), } } diff --git a/packages/ai/src/activities/summarize/index.ts b/packages/ai/src/activities/summarize/index.ts index fa83c57f0d..8119e10721 100644 --- a/packages/ai/src/activities/summarize/index.ts +++ b/packages/ai/src/activities/summarize/index.ts @@ -404,11 +404,25 @@ async function* runStreamingSummarize( // Use real streaming if the adapter supports it if (adapter.summarizeStream) { - yield* runNativeSummarizeStream( - options, - adapter.summarizeStream(summarizeOptions), - logger, - ) + // Same abort/timeout composition as the non-streaming path. + const abortControls = createActivityAbortControls({ + timeout: options.timeout, + abortSignal: options.abortSignal, + }) + try { + yield* runNativeSummarizeStream( + options, + adapter.summarizeStream({ + ...summarizeOptions, + ...(abortControls.signal + ? { abortSignal: abortControls.signal } + : {}), + }), + logger, + ) + } finally { + abortControls.clear() + } return } diff --git a/packages/ai/src/stream-to-websocket.ts b/packages/ai/src/stream-to-websocket.ts index 290bac4afa..399fa44156 100644 --- a/packages/ai/src/stream-to-websocket.ts +++ b/packages/ai/src/stream-to-websocket.ts @@ -272,6 +272,8 @@ export function toWebSocketStream( } } else { for await (const chunk of init.onRun(ctx)) { + // Ends the onRun stream (iterator.return()) even if it ignores ctx.signal. + if (turnAbort.signal.aborted) break socket.send(encodeWsFrame(chunk, undefined)) } } diff --git a/packages/ai/tests/abort-signal-reach.test.ts b/packages/ai/tests/abort-signal-reach.test.ts new file mode 100644 index 0000000000..74e18f8d1a --- /dev/null +++ b/packages/ai/tests/abort-signal-reach.test.ts @@ -0,0 +1,380 @@ +/** + * A cancel (caller signal, timeout, middleware `ctx.abort()`) must reach the + * provider request, not only the caller. Each fake below rejects at once when + * it gets no signal, and otherwise hangs until its signal aborts — like a real + * fetch that keeps running (and billing) until something aborts it. + */ +import { describe, expect, it, vi } from 'vitest' +import { z } from 'zod' +import { chat } from '../src/activities/chat/index' +import { defineAgent } from '../src/activities/chat/agents/define-agent' +import { summarize } from '../src/activities/summarize/index' +import { ChatStreamSummarizeAdapter } from '../src/activities/summarize/chat-stream-summarize' +import { generateVideo } from '../src/activities/generateVideo/index' +import { toWebSocketStream } from '../src/stream-to-websocket' +import { collectChunks, createMockAdapter, ev } from './test-utils' +import type { ChatStreamCapable } from '../src/activities/summarize/chat-stream-summarize' +import type { ChatMiddleware } from '../src/activities/chat/middleware/types' +import type { WebSocketLike } from '../src/stream-to-websocket' +import type { StreamChunk } from '../src/types' + +/** A provider request: no signal fails at once, else hangs until abort. */ +function hangUntilAborted(signal: AbortSignal | undefined): Promise { + if (!signal) { + return Promise.reject(new Error('request sent without an abort signal')) + } + return new Promise((_, reject) => { + const fail = () => + reject(Object.assign(new Error('aborted'), { name: 'AbortError' })) + if (signal.aborted) fail() + else signal.addEventListener('abort', fail, { once: true }) + }) +} + +/** 'done' when `work` settles within `ms`, else 'hung'. */ +function settlesWithin(work: Promise, ms = 500) { + return Promise.race([ + work.then( + () => 'done', + () => 'done', + ), + new Promise((resolve) => setTimeout(() => resolve('hung'), ms)), + ]) +} + +function summarizeAdapter(seen: { signal?: AbortSignal }) { + const text: ChatStreamCapable = { + chatStream: (options) => + (async function* () { + seen.signal = options.request?.signal ?? undefined + yield* [] + await hangUntilAborted(seen.signal) + })(), + } + return new ChatStreamSummarizeAdapter( + text, + 'test-model', + 'openai', + ) +} + +describe('summarize() abort reaches the wrapped chatStream', () => { + it('non-stream: caller abort aborts the request (C1)', async () => { + const seen: { signal?: AbortSignal } = {} + const controller = new AbortController() + const run = summarize({ + adapter: summarizeAdapter(seen), + threadId: 'thread-1', + text: 'long text', + abortSignal: controller.signal, + }) + setTimeout(() => controller.abort(), 10) + await expect(run).rejects.toThrow() + expect(seen.signal?.aborted).toBe(true) + }) + + it('stream: caller abort aborts the request (C2)', async () => { + const seen: { signal?: AbortSignal } = {} + const controller = new AbortController() + const stream = summarize({ + adapter: summarizeAdapter(seen), + threadId: 'thread-1', + text: 'long text', + stream: true, + abortSignal: controller.signal, + }) + setTimeout(() => controller.abort(), 10) + await expect(collectChunks(stream)).rejects.toThrow() + expect(seen.signal?.aborted).toBe(true) + }) + + it('stream: timeout aborts the request (C2)', async () => { + const seen: { signal?: AbortSignal } = {} + const stream = summarize({ + adapter: summarizeAdapter(seen), + threadId: 'thread-1', + text: 'long text', + stream: true, + timeout: 10, + }) + await expect(collectChunks(stream)).rejects.toThrow() + expect(seen.signal?.aborted).toBe(true) + }) +}) + +describe('middleware ctx.abort() reaches the provider', () => { + it('aborts the chatStream request (C3)', async () => { + let seen: AbortSignal | undefined + const { adapter } = createMockAdapter({ + chatStreamFn: (options) => + (async function* () { + seen = options.request?.signal + yield ev.runStarted() + // Like a real adapter: a failed request becomes a RUN_ERROR chunk. + try { + await hangUntilAborted(seen) + } catch (error) { + yield ev.runError((error as Error).message) + } + })(), + }) + const onAbort = vi.fn() + const onError = vi.fn() + const middleware: ChatMiddleware = { + name: 'aborter', + onStart: (ctx) => { + setTimeout(() => ctx.abort('stop'), 10) + }, + onAbort, + onError, + } + const run = collectChunks( + chat({ + threadId: 'thread-1', + adapter, + messages: [{ role: 'user', content: 'Hi' }], + middleware: [middleware], + }) as AsyncIterable, + ) + expect(await settlesWithin(run)).toBe('done') + expect(seen?.aborted).toBe(true) + expect(onAbort).toHaveBeenCalledOnce() + expect(onError).not.toHaveBeenCalled() + }) + + it('aborts the structured output request (C3)', async () => { + let seen: AbortSignal | undefined + let abortRun: ((reason?: string) => void) | undefined + const { adapter } = createMockAdapter({ + iterations: [ + [ + ev.runStarted(), + ev.textStart(), + ev.textContent('{}'), + ev.textEnd(), + ev.runFinished('stop'), + ], + ], + structuredOutput: (options) => { + seen = options.chatOptions.request?.signal + abortRun?.('stop') + return hangUntilAborted(seen) + }, + }) + const run = chat({ + threadId: 'thread-1', + adapter, + messages: [{ role: 'user', content: 'Hi' }], + outputSchema: z.object({ a: z.string() }), + middleware: [ + { + name: 'aborter', + onStart: (ctx) => { + abortRun = ctx.abort + }, + }, + ], + }) + expect(await settlesWithin(run)).toBe('done') + expect(seen?.aborted).toBe(true) + }) + + it('aborts ctx.signal (C4)', async () => { + let signal: AbortSignal | undefined + const { adapter } = createMockAdapter({ + iterations: [[ev.runStarted(), ev.runFinished('stop')]], + }) + await collectChunks( + chat({ + threadId: 'thread-1', + adapter, + messages: [{ role: 'user', content: 'Hi' }], + middleware: [ + { + name: 'aborter', + onStart: (ctx) => { + signal = ctx.signal + ctx.abort('stop') + }, + }, + ], + }) as AsyncIterable, + ) + expect(signal?.aborted).toBe(true) + }) + + it('stops a subagent child (C5)', async () => { + let outcome = 'running' + const worker = defineAgent({ + name: 'worker', + description: 'Waits for a stop', + run: (ctx) => + new Promise((resolve) => { + ctx.abortSignal?.addEventListener('abort', () => { + outcome = 'stopped' + resolve('stopped') + }) + setTimeout(() => { + if (outcome === 'running') outcome = 'never stopped' + resolve('done') + }, 300) + }), + }) + const { adapter } = createMockAdapter({ + iterations: [ + [ + ev.runStarted(), + ev.toolStart('c1', 'worker'), + ev.toolArgs('c1', '{}'), + ev.runFinished('tool_calls'), + ], + [ev.runStarted(), ev.runFinished('stop')], + ], + }) + await collectChunks( + chat({ + threadId: 'thread-1', + adapter, + messages: [{ role: 'user', content: 'Go' }], + subagents: { agents: [worker] }, + middleware: [ + { + name: 'aborter', + onBeforeToolCall: (ctx) => { + setTimeout(() => ctx.abort('stop'), 10) + return undefined + }, + }, + ], + }) as AsyncIterable, + ) + expect(outcome).toBe('stopped') + }) + + it('does not wait for a server tool that ignores the signal (C6)', async () => { + const { adapter } = createMockAdapter({ + iterations: [ + [ + ev.runStarted(), + ev.toolStart('c1', 'stuck'), + ev.toolArgs('c1', '{}'), + ev.toolEnd('c1', 'stuck', { input: {} }), + ev.runFinished('tool_calls'), + ], + ], + }) + const run = collectChunks( + chat({ + threadId: 'thread-1', + adapter, + messages: [{ role: 'user', content: 'Hi' }], + tools: [ + { + name: 'stuck', + description: 'Ignores its abort signal', + execute: () => new Promise(() => {}), + }, + ], + middleware: [ + { + name: 'aborter', + onBeforeToolCall: (ctx) => { + setTimeout(() => ctx.abort('stop'), 10) + return undefined + }, + }, + ], + }) as AsyncIterable, + ) + expect(await settlesWithin(run)).toBe('done') + }) +}) + +describe('generateVideo() polling', () => { + // `stuck` is the adapter call that hangs while the caller aborts. + for (const stuck of ['getVideoStatus', 'getVideo'] as const) { + it(`stops on abort while ${stuck} hangs (C8)`, async () => { + const controller = new AbortController() + const hang = () => { + setTimeout(() => controller.abort(), 10) + return new Promise(() => {}) + } + const adapter = { + kind: 'video' as const, + name: 'mock', + model: 'mock-video-model', + createVideoJob: async () => ({ + jobId: 'job-1', + model: 'mock-video-model', + }), + getVideoStatus: () => + stuck === 'getVideoStatus' + ? hang() + : Promise.resolve({ status: 'completed' as const }), + getVideo: () => + stuck === 'getVideo' + ? hang() + : Promise.resolve({ url: 'https://example.com/v.mp4' }), + } + const stream = generateVideo({ + threadId: 'thread-1', + adapter: adapter as any, + prompt: 'a cat', + stream: true, + pollingInterval: 1, + abortSignal: controller.signal, + }) + expect(await settlesWithin(collectChunks(stream))).toBe('done') + }) + } +}) + +describe('toWebSocketStream() (non-durable)', () => { + it('stops pumping a turn after an abort frame (C11)', async () => { + const sent: Array = [] + const handlers: Record void>> = {} + const socket: WebSocketLike = { + send: (data) => sent.push(data), + close: () => {}, + addEventListener: (type, handler) => { + ;(handlers[type] ??= []).push(handler) + }, + } + const emit = (data: string) => { + for (const h of handlers.message ?? []) h({ data }) + } + let closed = false + toWebSocketStream(socket, new Request('https://x/api/chat'), { + // Ignores ctx.signal, like a hand-rolled stream. + onRun: ({ runId, threadId }): AsyncIterable => + (async function* () { + try { + yield ev.runStarted(runId, threadId) + await new Promise((resolve) => setTimeout(resolve, 20)) + yield ev.textContent('late') + await new Promise((resolve) => setTimeout(resolve, 20)) + yield ev.textContent('later') + } finally { + closed = true + } + })(), + }) + emit( + JSON.stringify({ + threadId: 't', + runId: 'run-1', + messages: [], + tools: [], + context: [], + forwardedProps: {}, + state: {}, + }), + ) + await new Promise((resolve) => setTimeout(resolve, 5)) + emit(JSON.stringify({ type: 'abort', runId: 'run-1' })) + await new Promise((resolve) => setTimeout(resolve, 80)) + const deltas = sent.map((s) => JSON.parse(s).delta).filter(Boolean) + expect(deltas).toEqual([]) + expect(closed).toBe(true) + }) +}) diff --git a/testing/e2e/fixtures/summarize-abort/basic.json b/testing/e2e/fixtures/summarize-abort/basic.json new file mode 100644 index 0000000000..9b9d801724 --- /dev/null +++ b/testing/e2e/fixtures/summarize-abort/basic.json @@ -0,0 +1,14 @@ +{ + "fixtures": [ + { + "match": { + "userMessage": "[summarize-abort] A slow summary that the timeout must stop" + }, + "response": { + "content": "This summary streams very slowly so that a short timeout ends the run before the model finishes and the request to the provider is aborted." + }, + "chunkSize": 2, + "streamingProfile": { "tps": 5 } + } + ] +} diff --git a/testing/e2e/src/routes/api.summarize.ts b/testing/e2e/src/routes/api.summarize.ts index 7e8040d7bf..b138fafd97 100644 --- a/testing/e2e/src/routes/api.summarize.ts +++ b/testing/e2e/src/routes/api.summarize.ts @@ -186,6 +186,7 @@ export const Route = createFileRoute('/api/summarize')({ testId, aimockPort, maxLength, + timeout, } = data as { text: string provider: Provider @@ -193,6 +194,7 @@ export const Route = createFileRoute('/api/summarize')({ testId?: string aimockPort?: number maxLength?: number + timeout?: number } try { @@ -216,6 +218,7 @@ export const Route = createFileRoute('/api/summarize')({ adapter, text, maxLength, + timeout, stream: false, }) return new Response(JSON.stringify({ summary }), { @@ -227,6 +230,7 @@ export const Route = createFileRoute('/api/summarize')({ adapter, text, maxLength, + timeout, stream: true, }) return toServerSentEventsResponse(stream) diff --git a/testing/e2e/tests/summarize-abort.spec.ts b/testing/e2e/tests/summarize-abort.spec.ts new file mode 100644 index 0000000000..b072c2dcc5 --- /dev/null +++ b/testing/e2e/tests/summarize-abort.spec.ts @@ -0,0 +1,26 @@ +import { test, expect } from './fixtures' + +// fixtures/summarize-abort/basic.json streams at 5 tokens per second, so the +// full summary takes many seconds. A 300 ms timeout must end the run first. +test('streaming summarize stops at its timeout', async ({ + request, + aimockPort, + testId, +}) => { + const started = Date.now() + const response = await request.post('/api/summarize', { + data: { + provider: 'openai', + text: '[summarize-abort] A slow summary that the timeout must stop', + stream: true, + timeout: 300, + aimockPort, + testId, + }, + }) + const body = await response.text() + + expect(body).toContain('RUN_ERROR') + expect(body).not.toContain('RUN_FINISHED') + expect(Date.now() - started).toBeLessThan(5000) +}) From 40c83b336a5005e44ff5b76e5c089ed252c4b728 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 2 Nov 2026 15:16:58 +0100 Subject: [PATCH 02/12] fix(ai-client): make client cancels reach the request - ChatClient.stop(): do not wait for a client tool that never settles. - Fetch and XHR connection adapters: abort on the client's signal and on the user's own options.signal. - stream() and rpcStream(): end on abort even if the source ignores it. - RealtimeClient: disconnect() during connect() no longer keeps a late connection or a token refresh timer. Adds unit repros and a changeset. --- .changeset/client-abort-reaches-request.md | 5 + packages/ai-client/src/chat-client.ts | 21 +- packages/ai-client/src/connection-adapters.ts | 48 +++- packages/ai-client/src/realtime-client.ts | 16 +- .../tests/abort-signal-reach.test.ts | 244 ++++++++++++++++++ .../tests/connection-adapters-abort.test.ts | 14 +- 6 files changed, 328 insertions(+), 20 deletions(-) create mode 100644 .changeset/client-abort-reaches-request.md create mode 100644 packages/ai-client/tests/abort-signal-reach.test.ts diff --git a/.changeset/client-abort-reaches-request.md b/.changeset/client-abort-reaches-request.md new file mode 100644 index 0000000000..a210b934e0 --- /dev/null +++ b/.changeset/client-abort-reaches-request.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-client': patch +--- + +`stop()` no longer waits for a client tool that never settles. Fetch and XHR connection adapters now abort on the client's signal and on your own `options.signal`. `stream()` and `rpcStream()` end on abort even if the source ignores the signal. `RealtimeClient.disconnect()` during `connect()` no longer leaves a connection or a token refresh behind. diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index 368dd7373f..59737c7e3d 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -110,6 +110,22 @@ type SchedulerWithYield = { yield?: () => Promise } +/** + * Wait for `promise`, but reject with an AbortError once `signal` aborts. + * A client tool that never settles must not block stop(). + */ +function untilAborted(promise: Promise, signal: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const onAbort = () => + reject(Object.assign(new Error('Aborted'), { name: 'AbortError' })) + if (signal.aborted) return onAbort() + signal.addEventListener('abort', onAbort, { once: true }) + promise.then(resolve, reject).finally(() => { + signal.removeEventListener('abort', onAbort) + }) + }) +} + function yieldToHost(): Promise { const { scheduler } = globalThis as typeof globalThis & { scheduler?: SchedulerWithYield @@ -2884,7 +2900,10 @@ export class ChatClient< // Wait for pending client tool executions if (this.pendingToolExecutions.size > 0) { - await Promise.all(this.pendingToolExecutions.values()) + await untilAborted( + Promise.all(this.pendingToolExecutions.values()), + signal, + ) } // Finalize (idempotent — may already be done by RUN_FINISHED handler). diff --git a/packages/ai-client/src/connection-adapters.ts b/packages/ai-client/src/connection-adapters.ts index 00d7b8551d..5eb58ae5ea 100644 --- a/packages/ai-client/src/connection-adapters.ts +++ b/packages/ai-client/src/connection-adapters.ts @@ -328,6 +328,24 @@ export function createReconnectTracker( } } +/** + * A signal that aborts when either one does. The client's signal (stop()) and + * the user's own `options.signal` must both stop the request. + */ +// ponytail: listeners stay on a long-lived signal until it aborts; AbortSignal.any when the browser floor allows. +function eitherSignal( + a: AbortSignal | undefined, + b: AbortSignal | undefined, +): AbortSignal | undefined { + if (!a || !b) return a ?? b + if (a.aborted) return a + if (b.aborted) return b + const controller = new AbortController() + a.addEventListener('abort', () => controller.abort(a.reason), { once: true }) + b.addEventListener('abort', () => controller.abort(b.reason), { once: true }) + return controller.signal +} + /** Resolve after `ms`, or immediately once `signal` aborts. Never rejects. */ function abortableDelay(ms: number, signal?: AbortSignal): Promise { if (ms <= 0 || signal?.aborted) return Promise.resolve() @@ -1500,7 +1518,7 @@ export function fetchServerSentEvents( // `RequestInit.signal` is typed `AbortSignal | null` (no `undefined` // under `exactOptionalPropertyTypes`), so spread it conditionally // rather than passing `undefined` explicitly. - const signal = abortSignal || resolvedOptions.signal + const signal = eitherSignal(abortSignal, resolvedOptions.signal) // POST URL is byte-identical to a plain request; the run id (when set) // rides in the X-Run-Id header so durability can key the log by it // without changing the request URL existing clients rely on. @@ -1543,7 +1561,7 @@ export function fetchServerSentEvents( ...mergeHeaders(resolvedOptions.headers), } const fetchClient = resolvedOptions.fetchClient ?? fetch - const signal = abortSignal || resolvedOptions.signal + const signal = eitherSignal(abortSignal, resolvedOptions.signal) yield* resumableStream( fetchEventSource( @@ -1660,7 +1678,7 @@ export function fetchHttpStream( // `RequestInit.signal` is typed `AbortSignal | null` (no `undefined` // under `exactOptionalPropertyTypes`), so spread it conditionally // rather than passing `undefined` explicitly. - const signal = abortSignal || resolvedOptions.signal + const signal = eitherSignal(abortSignal, resolvedOptions.signal) // POST URL is byte-identical to a plain request; the run id (when set) // rides in the X-Run-Id header so durability can key the log by it // without changing the request URL existing clients rely on. @@ -1700,7 +1718,7 @@ export function fetchHttpStream( ...mergeHeaders(resolvedOptions.headers), } const fetchClient = resolvedOptions.fetchClient ?? fetch - const signal = abortSignal || resolvedOptions.signal + const signal = eitherSignal(abortSignal, resolvedOptions.signal) yield* resumableStream( fetchEventSource( @@ -1907,7 +1925,7 @@ export function fetchJson( runId: string, abortSignal: AbortSignal | undefined, ) { - const signal = abortSignal || resolvedOptions.signal + const signal = eitherSignal(abortSignal, resolvedOptions.signal) return { fetchClient: resolvedOptions.fetchClient ?? fetch, url: resolvedUrl, @@ -2257,7 +2275,7 @@ export function xhrServerSentEvents( async *connect(messages, data, abortSignal, runContext) { const resolvedUrl = typeof url === 'function' ? url() : url const resolvedOptions = await resolveXhrConnectionOptions(options) - const signal = abortSignal || resolvedOptions.signal + const signal = eitherSignal(abortSignal, resolvedOptions.signal) // POST URL is byte-identical to a plain request; the run id (when set) // rides in the X-Run-Id header so durability can key the log by it // without changing the request URL existing clients rely on. @@ -2279,7 +2297,7 @@ export function xhrServerSentEvents( async *joinRun(runId, abortSignal) { const resolvedUrl = typeof url === 'function' ? url() : url const resolvedOptions = await resolveXhrConnectionOptions(options) - const signal = abortSignal || resolvedOptions.signal + const signal = eitherSignal(abortSignal, resolvedOptions.signal) const joinUrl = withSearchParams(resolvedUrl, { offset: '-1', runId }) yield* resumableStream( xhrEventSource( @@ -2341,7 +2359,7 @@ export function xhrHttpStream( async *connect(messages, data, abortSignal, runContext) { const resolvedUrl = typeof url === 'function' ? url() : url const resolvedOptions = await resolveXhrConnectionOptions(options) - const signal = abortSignal || resolvedOptions.signal + const signal = eitherSignal(abortSignal, resolvedOptions.signal) // POST URL is byte-identical to a plain request; the run id (when set) // rides in the X-Run-Id header so durability can key the log by it // without changing the request URL existing clients rely on. @@ -2363,7 +2381,7 @@ export function xhrHttpStream( async *joinRun(runId, abortSignal) { const resolvedUrl = typeof url === 'function' ? url() : url const resolvedOptions = await resolveXhrConnectionOptions(options) - const signal = abortSignal || resolvedOptions.signal + const signal = eitherSignal(abortSignal, resolvedOptions.signal) const joinUrl = withSearchParams(resolvedUrl, { offset: '-1', runId }) yield* resumableStream( xhrEventSource( @@ -2896,12 +2914,13 @@ export function stream( async *connect(messages, data, abortSignal, runContext) { // Pass messages as-is (UIMessages with parts preserved) // Server-side chat() handles conversion to ModelMessages - yield* streamFactory( + const source = streamFactory( messages, data, abortSignal, buildRunAgentInputBody(messages, data, runContext, {}), ) + yield* abortSignal ? abortableIterable(source, abortSignal) : source }, ...(handlers?.hydrate ? { hydrate: handlers.hydrate } : {}), ...(handlers?.hydrateGeneration @@ -2987,7 +3006,11 @@ async function* abortableIterable( yield result.value } } finally { - await iterator.return?.() + const closing = iterator.return?.() + // A source stuck in an await runs return() only after that await ends, + // so do not wait for it once the signal has aborted. + if (signal.aborted) void Promise.resolve(closing).catch(() => {}) + else await closing } } @@ -3034,12 +3057,13 @@ export function rpcStream( async *connect(messages, data, abortSignal, runContext) { // Pass messages as-is (UIMessages with parts preserved) // Server-side chat() handles conversion to ModelMessages - yield* rpcCall( + const source = rpcCall( messages, data, abortSignal, buildRunAgentInputBody(messages, data, runContext, {}), ) + yield* abortSignal ? abortableIterable(source, abortSignal) : source }, ...(handlers?.hydrate ? { hydrate: handlers.hydrate } : {}), ...(handlers?.hydrateGeneration diff --git a/packages/ai-client/src/realtime-client.ts b/packages/ai-client/src/realtime-client.ts index 0bac78f317..9787397ec8 100644 --- a/packages/ai-client/src/realtime-client.ts +++ b/packages/ai-client/src/realtime-client.ts @@ -51,6 +51,7 @@ export class RealtimeClient { private connection: RealtimeConnection | null = null private token: RealtimeToken | null = null private tokenRefreshTimeout: ReturnType | null = null + private connectGeneration = 0 private readonly clientTools: Map private readonly stateChangeCallbacks: Set = new Set() @@ -104,10 +105,14 @@ export class RealtimeClient { status: 'connecting', error: null, }) + // disconnect() bumps this. A connect() it overtook must stop. + const generation = ++this.connectGeneration try { // Fetch token from server - this.token = await this.options.getToken() + const token = await this.options.getToken() + if (generation !== this.connectGeneration) return + this.token = token // Schedule token refresh this.scheduleTokenRefresh() @@ -118,10 +123,15 @@ export class RealtimeClient { ? Array.from(this.clientTools.values()) : undefined - this.connection = await this.options.adapter.connect( + const connection = await this.options.adapter.connect( this.token, toolsList, ) + if (generation !== this.connectGeneration) { + await connection.disconnect() + return + } + this.connection = connection // A new session starts here. Clear the last transcript before events // flow, so a greeting that arrives while audio capture starts is kept. @@ -141,6 +151,7 @@ export class RealtimeClient { // Start audio capture if configured if (this.options.autoCapture) { await this.connection.startAudioCapture() + if (generation !== this.connectGeneration) return } this.updateState({ @@ -160,6 +171,7 @@ export class RealtimeClient { * Disconnect from the realtime session. */ async disconnect(): Promise { + this.connectGeneration++ if (this.tokenRefreshTimeout) { clearTimeout(this.tokenRefreshTimeout) this.tokenRefreshTimeout = null diff --git a/packages/ai-client/tests/abort-signal-reach.test.ts b/packages/ai-client/tests/abort-signal-reach.test.ts new file mode 100644 index 0000000000..4a5eb4eb05 --- /dev/null +++ b/packages/ai-client/tests/abort-signal-reach.test.ts @@ -0,0 +1,244 @@ +/** + * A client cancel (`stop()`, the user's own `options.signal`, `disconnect()`) + * must reach the work it should stop. The fake requests below hang until + * their signal aborts, like a real fetch that keeps the run going. + */ +import { describe, expect, it, vi } from 'vitest' +import { ChatClient } from '../src/chat-client' +import { RealtimeClient } from '../src/realtime-client' +import { + fetchHttpStream, + fetchJson, + fetchServerSentEvents, + rpcStream, + stream, + xhrHttpStream, + xhrServerSentEvents, +} from '../src/connection-adapters' +import { createToolCallChunks } from './test-utils' +import type { ConnectConnectionAdapter } from '../src/connection-adapters' +import type { StreamChunk } from '@tanstack/ai/client' + +const RUN_CONTEXT = { threadId: 'thread-1', runId: 'run-1' } + +/** 'done' when `work` settles within `ms`, else 'hung'. */ +function settlesWithin(work: Promise, ms = 500) { + return Promise.race([ + work.then( + () => 'done', + () => 'done', + ), + new Promise((resolve) => setTimeout(() => resolve('hung'), ms)), + ]) +} + +async function drain(iterable: AsyncIterable) { + for await (const _ of iterable) { + // drain + } +} + +/** A fetch that hangs until its signal aborts (rejects at once without one). */ +function hangingFetch() { + const seen: { signal?: AbortSignal } = {} + const fetchClient = ((_url: RequestInfo | URL, init?: RequestInit) => { + const signal = init?.signal ?? undefined + seen.signal = signal + if (!signal) { + return Promise.reject(new Error('request sent without an abort signal')) + } + return new Promise((_, reject) => { + const fail = () => + reject(Object.assign(new Error('aborted'), { name: 'AbortError' })) + if (signal.aborted) fail() + else signal.addEventListener('abort', fail, { once: true }) + }) + }) as typeof fetch + return { fetchClient, aborted: () => seen.signal?.aborted === true } +} + +/** An XHR whose response never comes; `abort()` ends it. */ +function hangingXhr() { + const xhr = { + status: 0, + statusText: '', + responseText: '', + withCredentials: false, + onprogress: null as null | ((ev: unknown) => void), + onload: null as null | ((ev: unknown) => void), + onerror: null as null | ((ev: unknown) => void), + onabort: null as null | ((ev: unknown) => void), + onloadend: null as null | ((ev: unknown) => void), + open: () => {}, + send: () => {}, + setRequestHeader: () => {}, + getResponseHeader: () => null, + abort: vi.fn(() => { + xhr.onabort?.({}) + xhr.onloadend?.({}) + }), + } + return { + xhrFactory: () => xhr as unknown as XMLHttpRequest, + aborted: () => xhr.abort.mock.calls.length > 0, + } +} + +describe("connection adapters follow the user's options.signal (L2)", () => { + const cases = [ + ['fetchServerSentEvents', fetchServerSentEvents], + ['fetchHttpStream', fetchHttpStream], + ['fetchJson', fetchJson], + ] as const + for (const [name, factory] of cases) { + for (const method of ['connect', 'joinRun'] as const) { + it(`${name}.${method}`, async () => { + const user = new AbortController() + const client = new AbortController() + const fake = hangingFetch() + const adapter = factory('/api/chat', { + fetchClient: fake.fetchClient, + signal: user.signal, + }) + const run = drain( + method === 'connect' + ? adapter.connect([], undefined, client.signal, RUN_CONTEXT) + : adapter.joinRun!('run-1', client.signal), + ) + setTimeout(() => user.abort(), 10) + expect(await settlesWithin(run)).toBe('done') + expect(fake.aborted()).toBe(true) + }) + } + } + + const xhrCases = [ + ['xhrServerSentEvents', xhrServerSentEvents], + ['xhrHttpStream', xhrHttpStream], + ] as const + for (const [name, factory] of xhrCases) { + for (const method of ['connect', 'joinRun'] as const) { + it(`${name}.${method}`, async () => { + const user = new AbortController() + const client = new AbortController() + const fake = hangingXhr() + const adapter = factory('/api/chat', { + xhrFactory: fake.xhrFactory, + signal: user.signal, + }) + const run = drain( + method === 'connect' + ? adapter.connect([], undefined, client.signal, RUN_CONTEXT) + : adapter.joinRun!('run-1', client.signal), + ) + setTimeout(() => user.abort(), 10) + expect(await settlesWithin(run)).toBe('done') + expect(fake.aborted()).toBe(true) + }) + } + } +}) + +describe('stream() and rpcStream() stop on abort (L3)', () => { + // A source that ignores its signal and never ends. + const ignoresSignal = () => + (async function* (): AsyncGenerator { + await new Promise(() => {}) + })() + + for (const [name, adapter] of [ + ['stream', stream(ignoresSignal)], + ['rpcStream', rpcStream(ignoresSignal)], + ] as const) { + it(name, async () => { + const client = new AbortController() + const run = drain( + adapter.connect([], undefined, client.signal, RUN_CONTEXT), + ) + setTimeout(() => client.abort(), 10) + expect(await settlesWithin(run)).toBe('done') + }) + } +}) + +describe('ChatClient.stop() (L1)', () => { + it('does not wait for a client tool that never ends', async () => { + let toolStarted = false + const chunks = createToolCallChunks([ + { id: 'tc-1', name: 'stuck', arguments: '{}' }, + ]) + const connection: ConnectConnectionAdapter = { + async *connect(_messages, _data, abortSignal) { + for (const chunk of chunks) { + if (abortSignal?.aborted) return + yield chunk + } + }, + } + const client = new ChatClient({ + threadId: 'thread-1', + connection, + tools: [ + { + __toolSide: 'client' as const, + name: 'stuck', + description: 'Never finishes', + execute: () => { + toolStarted = true + return new Promise(() => {}) + }, + }, + ], + }) + const send = client.sendMessage('Go') + await vi.waitFor(() => expect(toolStarted).toBe(true)) + client.stop() + expect(await settlesWithin(send)).toBe('done') + }) +}) + +describe('RealtimeClient.disconnect() during connect() (L4)', () => { + // `gate` is the connect() step that waits while disconnect() runs. + for (const gate of ['getToken', 'adapter.connect', 'startAudioCapture']) { + it(`stays disconnected when disconnect() runs during ${gate}`, async () => { + let release!: () => void + const held = new Promise((resolve) => { + release = resolve + }) + const wait = (step: string) => (step === gate ? held : undefined) + const connection = { + on: () => () => {}, + updateSession: () => {}, + disconnect: vi.fn(async () => {}), + startAudioCapture: async () => { + await wait('startAudioCapture') + }, + } + const adapterConnect = vi.fn(async () => { + await wait('adapter.connect') + return connection + }) + const client = new RealtimeClient({ + getToken: async () => { + await wait('getToken') + return { token: 't', expiresAt: Date.now() + 60_000 } as any + }, + adapter: { connect: adapterConnect } as any, + }) + + const connecting = client.connect() + await vi.waitFor(() => expect(client.status).toBe('connecting')) + await new Promise((resolve) => setTimeout(resolve, 0)) + await client.disconnect() + release() + await connecting.catch(() => {}) + + const opened = adapterConnect.mock.calls.length > 0 + // Either it never connects, or it closes what it opened. + expect(!opened || connection.disconnect.mock.calls.length > 0).toBe(true) + expect(client.status).toBe('idle') + // No token refresh is armed for a session that does not exist. + expect(Reflect.get(client, 'tokenRefreshTimeout')).toBeNull() + }) + } +}) diff --git a/packages/ai-client/tests/connection-adapters-abort.test.ts b/packages/ai-client/tests/connection-adapters-abort.test.ts index 766a6a38f9..85cb621165 100644 --- a/packages/ai-client/tests/connection-adapters-abort.test.ts +++ b/packages/ai-client/tests/connection-adapters-abort.test.ts @@ -58,9 +58,11 @@ describe('Connection Adapters - Abort Signal Handling', () => { expect(fetchCall?.[1]?.signal).toBe(abortSignal) }) - it('should use provided abortSignal over options.signal', async () => { - const providedSignal = new AbortController().signal - const optionsSignal = new AbortController().signal + it('should combine provided abortSignal with options.signal', async () => { + const provided = new AbortController() + const providedSignal = provided.signal + const optionsController = new AbortController() + const optionsSignal = optionsController.signal const mockResponse = { ok: true, @@ -88,8 +90,10 @@ describe('Connection Adapters - Abort Signal Handling', () => { // Consume all chunks } - const fetchCall = fetchMock.mock.calls[0] - expect(fetchCall?.[1]?.signal).toBe(providedSignal) + const fetchSignal: AbortSignal = fetchMock.mock.calls[0]?.[1]?.signal + expect(fetchSignal.aborted).toBe(false) + optionsController.abort() + expect(fetchSignal.aborted).toBe(true) }) it('should stop reading stream when aborted', async () => { From b995e2fd0b00d63ce851dae525a45ddb33354943 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 2 Nov 2026 16:11:41 +0100 Subject: [PATCH 03/12] fix(ai): report a streamed summarize cancel to onAbort An aborted or timed-out summarize({ stream: true }) called middleware onError. It now calls onAbort, the same as the non-stream path. --- .changeset/core-abort-reaches-provider.md | 2 +- packages/ai/src/activities/summarize/index.ts | 15 +++++++++++---- packages/ai/tests/abort-signal-reach.test.ts | 18 ++++++++++++++++++ 3 files changed, 30 insertions(+), 5 deletions(-) diff --git a/.changeset/core-abort-reaches-provider.md b/.changeset/core-abort-reaches-provider.md index 5f1d8cb523..e31a6cb2be 100644 --- a/.changeset/core-abort-reaches-provider.md +++ b/.changeset/core-abort-reaches-provider.md @@ -2,4 +2,4 @@ '@tanstack/ai': patch --- -A cancel now stops the provider request, not only the caller. `summarize()` (also with `stream: true`, which now honors `abortSignal` and `timeout`), middleware `ctx.abort()` (adapter request, structured output, and `ctx.signal`), subagent children, server tools that ignore their signal, `generateVideo` status polls, and aborted `toWebSocket` turns now stop when the run is cancelled. +A cancel now stops the provider request, not only the caller. `summarize()` (also with `stream: true`, which now honors `abortSignal` and `timeout` and reports a cancel to middleware `onAbort`, not `onError`), middleware `ctx.abort()` (adapter request, structured output, and `ctx.signal`), subagent children, server tools that ignore their signal, `generateVideo` status polls, and aborted `toWebSocket` turns now stop when the run is cancelled. diff --git a/packages/ai/src/activities/summarize/index.ts b/packages/ai/src/activities/summarize/index.ts index 8119e10721..414bfc9429 100644 --- a/packages/ai/src/activities/summarize/index.ts +++ b/packages/ai/src/activities/summarize/index.ts @@ -419,6 +419,7 @@ async function* runStreamingSummarize( : {}), }), logger, + abortControls.signal, ) } finally { abortControls.clear() @@ -461,6 +462,7 @@ async function* runNativeSummarizeStream( options: SummarizeActivityOptions, true>, stream: AsyncIterable, logger: InternalLogger, + signal: AbortSignal | undefined, ): AsyncIterable { const { adapter, middleware, modelOptions } = options const mwCtx = createGenerationContext({ @@ -509,10 +511,15 @@ async function* runNativeSummarizeStream( } } catch (error) { settled = true - await runGenerationError(middleware, mwCtx, { - error, - duration: Date.now() - startTime, - }) + const duration = Date.now() - startTime + if (isActivityAbortError(error, signal)) { + await runGenerationAbort(middleware, mwCtx, { + reason: abortReasonMessage(error, signal), + duration, + }) + } else { + await runGenerationError(middleware, mwCtx, { error, duration }) + } logger.errors('summarize activity failed', { error, source: 'summarize', diff --git a/packages/ai/tests/abort-signal-reach.test.ts b/packages/ai/tests/abort-signal-reach.test.ts index 74e18f8d1a..5a4b3c5dbe 100644 --- a/packages/ai/tests/abort-signal-reach.test.ts +++ b/packages/ai/tests/abort-signal-reach.test.ts @@ -100,6 +100,24 @@ describe('summarize() abort reaches the wrapped chatStream', () => { await expect(collectChunks(stream)).rejects.toThrow() expect(seen.signal?.aborted).toBe(true) }) + + it('stream: caller abort reports onAbort, not onError', async () => { + const onAbort = vi.fn() + const onError = vi.fn() + const controller = new AbortController() + const stream = summarize({ + adapter: summarizeAdapter({}), + threadId: 'thread-1', + text: 'long text', + stream: true, + abortSignal: controller.signal, + middleware: [{ name: 'observe', onAbort, onError }], + }) + setTimeout(() => controller.abort(), 10) + await expect(collectChunks(stream)).rejects.toThrow() + expect(onError).not.toHaveBeenCalled() + expect(onAbort).toHaveBeenCalledTimes(1) + }) }) describe('middleware ctx.abort() reaches the provider', () => { From 39a62b80f73da2b68c08e257513aeb6d3cd1bbd2 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 2 Nov 2026 16:11:42 +0100 Subject: [PATCH 04/12] fix(ai-client): stop waits and token refresh follow the cancel - ChatClient.stop(): do not wait for a client tool that finalizeStream() started (a run that ends with finishReason tool_calls). - RealtimeClient: disconnect() during a token refresh no longer re-arms the refresh timer. --- .changeset/client-abort-reaches-request.md | 2 +- packages/ai-client/src/chat-client.ts | 5 +- packages/ai-client/src/realtime-client.ts | 6 +- .../tests/abort-signal-reach.test.ts | 109 +++++++++++++----- 4 files changed, 87 insertions(+), 35 deletions(-) diff --git a/.changeset/client-abort-reaches-request.md b/.changeset/client-abort-reaches-request.md index a210b934e0..43e466f904 100644 --- a/.changeset/client-abort-reaches-request.md +++ b/.changeset/client-abort-reaches-request.md @@ -2,4 +2,4 @@ '@tanstack/ai-client': patch --- -`stop()` no longer waits for a client tool that never settles. Fetch and XHR connection adapters now abort on the client's signal and on your own `options.signal`. `stream()` and `rpcStream()` end on abort even if the source ignores the signal. `RealtimeClient.disconnect()` during `connect()` no longer leaves a connection or a token refresh behind. +`stop()` no longer waits for a client tool that never settles. Fetch and XHR connection adapters now abort on the client's signal and on your own `options.signal`. `stream()` and `rpcStream()` end on abort even if the source ignores the signal. `RealtimeClient.disconnect()` during `connect()` or a token refresh no longer leaves a connection or a token refresh timer behind. diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index 59737c7e3d..c49909da67 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -2918,7 +2918,10 @@ export class ChatClient< this.raiseLocalApprovals() // Wait for the tools that finalizeStream() started, as above. if (this.pendingToolExecutions.size > 0) { - await Promise.all(this.pendingToolExecutions.values()) + await untilAborted( + Promise.all(this.pendingToolExecutions.values()), + signal, + ) } streamCompletedSuccessfully = true } catch (err: unknown) { diff --git a/packages/ai-client/src/realtime-client.ts b/packages/ai-client/src/realtime-client.ts index 9787397ec8..b408946485 100644 --- a/packages/ai-client/src/realtime-client.ts +++ b/packages/ai-client/src/realtime-client.ts @@ -457,8 +457,12 @@ export class RealtimeClient { } private async refreshToken(): Promise { + const generation = this.connectGeneration try { - this.token = await this.options.getToken() + const token = await this.options.getToken() + // disconnect() ran during the refresh: do not re-arm the timer. + if (generation !== this.connectGeneration) return + this.token = token this.scheduleTokenRefresh() this.connection?.updateToken?.(this.token) // Note: Some providers may require reconnection with new token diff --git a/packages/ai-client/tests/abort-signal-reach.test.ts b/packages/ai-client/tests/abort-signal-reach.test.ts index 4a5eb4eb05..5a19949399 100644 --- a/packages/ai-client/tests/abort-signal-reach.test.ts +++ b/packages/ai-client/tests/abort-signal-reach.test.ts @@ -162,39 +162,48 @@ describe('stream() and rpcStream() stop on abort (L3)', () => { }) describe('ChatClient.stop() (L1)', () => { - it('does not wait for a client tool that never ends', async () => { - let toolStarted = false - const chunks = createToolCallChunks([ - { id: 'tc-1', name: 'stuck', arguments: '{}' }, - ]) - const connection: ConnectConnectionAdapter = { - async *connect(_messages, _data, abortSignal) { - for (const chunk of chunks) { - if (abortSignal?.aborted) return - yield chunk - } - }, - } - const client = new ChatClient({ - threadId: 'thread-1', - connection, - tools: [ - { - __toolSide: 'client' as const, - name: 'stuck', - description: 'Never finishes', - execute: () => { - toolStarted = true - return new Promise(() => {}) - }, + // 'stop' starts the tool while chunks stream. 'tool_calls' starts it in + // finalizeStream(), after the stream ends. + it.each(['stop', 'tool_calls'] as const)( + 'does not wait for a client tool that never ends (finishReason %s)', + async (finishReason) => { + let toolStarted = false + const chunks = createToolCallChunks([ + { id: 'tc-1', name: 'stuck', arguments: '{}' }, + ]) + chunks[chunks.length - 1] = { + ...chunks[chunks.length - 1]!, + finishReason, + } as StreamChunk + const connection: ConnectConnectionAdapter = { + async *connect(_messages, _data, abortSignal) { + for (const chunk of chunks) { + if (abortSignal?.aborted) return + yield chunk + } }, - ], - }) - const send = client.sendMessage('Go') - await vi.waitFor(() => expect(toolStarted).toBe(true)) - client.stop() - expect(await settlesWithin(send)).toBe('done') - }) + } + const client = new ChatClient({ + threadId: 'thread-1', + connection, + tools: [ + { + __toolSide: 'client' as const, + name: 'stuck', + description: 'Never finishes', + execute: () => { + toolStarted = true + return new Promise(() => {}) + }, + }, + ], + }) + const send = client.sendMessage('Go') + await vi.waitFor(() => expect(toolStarted).toBe(true)) + client.stop() + expect(await settlesWithin(send)).toBe('done') + }, + ) }) describe('RealtimeClient.disconnect() during connect() (L4)', () => { @@ -242,3 +251,39 @@ describe('RealtimeClient.disconnect() during connect() (L4)', () => { }) } }) + +describe('RealtimeClient.disconnect() during a token refresh', () => { + it('does not re-arm the refresh timer', async () => { + let release!: () => void + const held = new Promise((resolve) => { + release = resolve + }) + // First token expires now, so the refresh starts at once and then waits. + const getToken = vi + .fn() + .mockResolvedValueOnce({ token: 't1', expiresAt: Date.now() }) + .mockImplementationOnce(async () => { + await held + return { token: 't2', expiresAt: Date.now() } + }) + const client = new RealtimeClient({ + getToken, + adapter: { + connect: async () => ({ + on: () => () => {}, + updateSession: () => {}, + disconnect: async () => {}, + startAudioCapture: async () => {}, + }), + } as any, + }) + + await client.connect() + await vi.waitFor(() => expect(getToken).toHaveBeenCalledTimes(2)) + await client.disconnect() + release() + await new Promise((resolve) => setTimeout(resolve, 0)) + + expect(Reflect.get(client, 'tokenRefreshTimeout')).toBeNull() + }) +}) From f2523ab1833cfedebbad31dbfe16b1c964020b98 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 2 Nov 2026 16:43:19 +0100 Subject: [PATCH 05/12] test(ai): prove a streamed summarize cancel needs the signal check A provider can throw a plain Error after the abort. Only the aborted signal shows it was a cancel, so onAbort needs the signal, not only the error name. --- packages/ai/tests/abort-signal-reach.test.ts | 33 ++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/packages/ai/tests/abort-signal-reach.test.ts b/packages/ai/tests/abort-signal-reach.test.ts index 5a4b3c5dbe..e5c94ad9fc 100644 --- a/packages/ai/tests/abort-signal-reach.test.ts +++ b/packages/ai/tests/abort-signal-reach.test.ts @@ -118,6 +118,39 @@ describe('summarize() abort reaches the wrapped chatStream', () => { expect(onError).not.toHaveBeenCalled() expect(onAbort).toHaveBeenCalledTimes(1) }) + + it('stream: a plain Error after the abort still reports onAbort', async () => { + // Many SDKs throw a plain Error (no AbortError name) once the request is + // cancelled, so only the aborted signal shows it was a cancel. + const text: ChatStreamCapable = { + chatStream: (options) => + (async function* () { + yield* [] + await hangUntilAborted(options.request?.signal).catch(() => { + throw new Error('socket closed') + }) + })(), + } + const onAbort = vi.fn() + const onError = vi.fn() + const controller = new AbortController() + const stream = summarize({ + adapter: new ChatStreamSummarizeAdapter( + text, + 'test-model', + 'openai', + ), + threadId: 'thread-1', + text: 'long text', + stream: true, + abortSignal: controller.signal, + middleware: [{ name: 'observe', onAbort, onError }], + }) + setTimeout(() => controller.abort(), 10) + await expect(collectChunks(stream)).rejects.toThrow() + expect(onError).not.toHaveBeenCalled() + expect(onAbort).toHaveBeenCalledTimes(1) + }) }) describe('middleware ctx.abort() reaches the provider', () => { From b29e42a6990428612d00b575d46ca3b38aaf780d Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 2 Nov 2026 16:56:04 +0100 Subject: [PATCH 06/12] test(ai): type the plain-Error cancel fake like the other fakes --- packages/ai/tests/abort-signal-reach.test.ts | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/packages/ai/tests/abort-signal-reach.test.ts b/packages/ai/tests/abort-signal-reach.test.ts index e5c94ad9fc..4be4f88d92 100644 --- a/packages/ai/tests/abort-signal-reach.test.ts +++ b/packages/ai/tests/abort-signal-reach.test.ts @@ -126,9 +126,11 @@ describe('summarize() abort reaches the wrapped chatStream', () => { chatStream: (options) => (async function* () { yield* [] - await hangUntilAborted(options.request?.signal).catch(() => { - throw new Error('socket closed') - }) + await hangUntilAborted(options.request?.signal ?? undefined).catch( + () => { + throw new Error('socket closed') + }, + ) })(), } const onAbort = vi.fn() From 7ec7e22b9182a092f14c54cd22a1af51c256e992 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 2 Nov 2026 17:40:05 +0100 Subject: [PATCH 07/12] fix(ai): keep the provider signal off when the caller gives none; drop the tool-wait race CI found two regressions in this PR: - The provider request always got a signal, even with no caller signal. Some SDKs drop their own timeout when a request has a signal (the OpenRouter SDK ignores timeoutMs then). The linked signal (caller signal plus ctx.abort()) is now used only when the caller gives a signal. - Racing a server tool against the abort let the turn end before its tool. The harness durable-step fence relies on that wait: a host that lost its log then ran a side effect a second time (ownership.test.ts). The race is dropped. --- .changeset/core-abort-reaches-provider.md | 2 +- .../ai-gemini/tests/gemini-adapter.test.ts | 7 +- packages/ai/src/activities/chat/index.ts | 13 ++-- .../src/activities/chat/tools/tool-calls.ts | 8 +-- packages/ai/tests/abort-signal-reach.test.ts | 72 +++++++++---------- 5 files changed, 51 insertions(+), 51 deletions(-) diff --git a/.changeset/core-abort-reaches-provider.md b/.changeset/core-abort-reaches-provider.md index e31a6cb2be..e51795f712 100644 --- a/.changeset/core-abort-reaches-provider.md +++ b/.changeset/core-abort-reaches-provider.md @@ -2,4 +2,4 @@ '@tanstack/ai': patch --- -A cancel now stops the provider request, not only the caller. `summarize()` (also with `stream: true`, which now honors `abortSignal` and `timeout` and reports a cancel to middleware `onAbort`, not `onError`), middleware `ctx.abort()` (adapter request, structured output, and `ctx.signal`), subagent children, server tools that ignore their signal, `generateVideo` status polls, and aborted `toWebSocket` turns now stop when the run is cancelled. +A cancel now stops the provider request, not only the caller. `summarize()` (also with `stream: true`, which now honors `abortSignal` and `timeout` and reports a cancel to middleware `onAbort`, not `onError`), middleware `ctx.abort()` when the caller passed an `abortController` (adapter request, structured output, and `ctx.signal`), subagent children, `generateVideo` status polls, and aborted `toWebSocket` turns now stop when the run is cancelled. diff --git a/packages/ai-gemini/tests/gemini-adapter.test.ts b/packages/ai-gemini/tests/gemini-adapter.test.ts index d857fb4621..7126e9f987 100644 --- a/packages/ai-gemini/tests/gemini-adapter.test.ts +++ b/packages/ai-gemini/tests/gemini-adapter.test.ts @@ -241,7 +241,12 @@ describe('GeminiAdapter through AI', () => { expect(mocks.generateContentStreamSpy).toHaveBeenCalledTimes(1) const [payload] = mocks.generateContentStreamSpy.mock.calls[0]! - expect(payload.config.abortSignal).toBe(abortController.signal) + // chat() links the caller signal with middleware ctx.abort(), so the SDK + // gets a linked signal that aborts with the caller. + const sdkSignal: AbortSignal = payload.config.abortSignal + expect(sdkSignal.aborted).toBe(false) + abortController.abort() + expect(sdkSignal.aborted).toBe(true) }) it('omits config.abortSignal when no abort controller is supplied', async () => { diff --git a/packages/ai/src/activities/chat/index.ts b/packages/ai/src/activities/chat/index.ts index ec71466a92..379235ed4f 100644 --- a/packages/ai/src/activities/chat/index.ts +++ b/packages/ai/src/activities/chat/index.ts @@ -1061,6 +1061,8 @@ class TextEngine< private readonly streamId: string private readonly effectiveRequest?: Request | RequestInit private readonly effectiveSignal?: AbortSignal + /** The caller signal plus ctx.abort(), only when there is a caller signal. */ + private readonly linkedSignal?: AbortSignal private messages: Array private activities: Array = [] @@ -1293,9 +1295,12 @@ class TextEngine< this.effectiveSignal, this.middlewareAbortController.signal, ) - // The provider request follows ctx.abort() too, not only the caller. - this.effectiveRequest = this.toolAbortSignal - ? { signal: this.toolAbortSignal } + // With a caller signal, the provider request and ctx.signal follow + // ctx.abort() too. Without one they get no signal, as before: some SDKs + // drop their own timeout when a request has a signal. + this.linkedSignal = this.effectiveSignal ? this.toolAbortSignal : undefined + this.effectiveRequest = this.linkedSignal + ? { signal: this.linkedSignal } : undefined this.middlewareCtx = { requestId: this.requestId, @@ -1313,7 +1318,7 @@ class TextEngine< iteration: 0, chunkIndex: 0, callMetadata: this.currentCallMetadata, - signal: this.toolAbortSignal, + signal: this.linkedSignal, abort: (reason?: string) => { this.abortReason = reason this.middlewareAbortController?.abort(reason) diff --git a/packages/ai/src/activities/chat/tools/tool-calls.ts b/packages/ai/src/activities/chat/tools/tool-calls.ts index e8b7ab800f..fa984b6d52 100644 --- a/packages/ai/src/activities/chat/tools/tool-calls.ts +++ b/packages/ai/src/activities/chat/tools/tool-calls.ts @@ -3,7 +3,6 @@ import { normalizeToolResult } from '../../../utilities/tool-result' import { tanstackMetadata } from '../../../utilities/merge-metadata' import { isProviderExecutedToolCall } from '../../../utilities/provider-executed' import { mergeStreams } from '../../../utilities/merge-streams' -import { raceWithAbort } from '../../../utilities/activity-abort' import { validateToolInput } from './input-validation' import type { AdapterYieldChunk } from '../../../utilities/adapter-yield-chunk' import { @@ -797,12 +796,7 @@ export async function* executeServerTool( [EMIT_STREAM_CHUNK]: (chunk: StreamChunk) => pendingEvents.push(chunk), }) } - const execution = Promise.resolve(tool.execute(input, context)) - // Do not wait for a tool that ignores its signal. An abort that came - // during execute() itself keeps the result the tool already returned. - const executionPromise = context.abortSignal?.aborted - ? execution - : raceWithAbort(execution, context.abortSignal) + const executionPromise = Promise.resolve(tool.execute(input, context)) let result = yield* executeWithEventPolling(executionPromise, pendingEvents) const duration = Date.now() - startTime diff --git a/packages/ai/tests/abort-signal-reach.test.ts b/packages/ai/tests/abort-signal-reach.test.ts index 4be4f88d92..f70d254746 100644 --- a/packages/ai/tests/abort-signal-reach.test.ts +++ b/packages/ai/tests/abort-signal-reach.test.ts @@ -187,6 +187,7 @@ describe('middleware ctx.abort() reaches the provider', () => { adapter, messages: [{ role: 'user', content: 'Hi' }], middleware: [middleware], + abortController: new AbortController(), }) as AsyncIterable, ) expect(await settlesWithin(run)).toBe('done') @@ -227,6 +228,7 @@ describe('middleware ctx.abort() reaches the provider', () => { }, }, ], + abortController: new AbortController(), }) expect(await settlesWithin(run)).toBe('done') expect(seen?.aborted).toBe(true) @@ -251,11 +253,43 @@ describe('middleware ctx.abort() reaches the provider', () => { }, }, ], + abortController: new AbortController(), }) as AsyncIterable, ) expect(signal?.aborted).toBe(true) }) + it('gives the provider no signal when the caller gives none', async () => { + // Some SDKs drop their own timeout when a request has a signal. + let request: unknown = 'not called' + let ctxSignal: AbortSignal | undefined + const { adapter } = createMockAdapter({ + chatStreamFn: (options) => + (async function* () { + request = options.request + yield ev.runStarted() + yield ev.runFinished('stop') + })(), + }) + await collectChunks( + chat({ + threadId: 'thread-1', + adapter, + messages: [{ role: 'user', content: 'Hi' }], + middleware: [ + { + name: 'observe', + onStart: (ctx) => { + ctxSignal = ctx.signal + }, + }, + ], + }) as AsyncIterable, + ) + expect(request).toBeUndefined() + expect(ctxSignal).toBeUndefined() + }) + it('stops a subagent child (C5)', async () => { let outcome = 'running' const worker = defineAgent({ @@ -303,44 +337,6 @@ describe('middleware ctx.abort() reaches the provider', () => { ) expect(outcome).toBe('stopped') }) - - it('does not wait for a server tool that ignores the signal (C6)', async () => { - const { adapter } = createMockAdapter({ - iterations: [ - [ - ev.runStarted(), - ev.toolStart('c1', 'stuck'), - ev.toolArgs('c1', '{}'), - ev.toolEnd('c1', 'stuck', { input: {} }), - ev.runFinished('tool_calls'), - ], - ], - }) - const run = collectChunks( - chat({ - threadId: 'thread-1', - adapter, - messages: [{ role: 'user', content: 'Hi' }], - tools: [ - { - name: 'stuck', - description: 'Ignores its abort signal', - execute: () => new Promise(() => {}), - }, - ], - middleware: [ - { - name: 'aborter', - onBeforeToolCall: (ctx) => { - setTimeout(() => ctx.abort('stop'), 10) - return undefined - }, - }, - ], - }) as AsyncIterable, - ) - expect(await settlesWithin(run)).toBe('done') - }) }) describe('generateVideo() polling', () => { From 5702a2a1ab22df811ebdf62a6541485b090e7375 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 2 Nov 2026 17:47:51 +0100 Subject: [PATCH 08/12] fix(ai, ai-client): close abort gaps found in review - connection adapters: AbortSignal.any joins the client and user signals when the runtime has it, so a reused options.signal keeps no listener from a finished request. - RealtimeClient: a connect() that a later connect() replaced during audio startup closes its own connection. - generateVideo: no download starts when the caller aborts on the completed status. - streamed summarize: an abort that the adapter ends with a RUN_ERROR chunk reports onAbort, not onFinish. - e2e: the summarize timeout spec fails on an error that comes at once. --- packages/ai-client/src/connection-adapters.ts | 6 +- packages/ai-client/src/realtime-client.ts | 11 ++- .../tests/abort-signal-reach.test.ts | 54 ++++++++++++++ .../ai/src/activities/generateVideo/index.ts | 2 + packages/ai/src/activities/summarize/index.ts | 15 +++- packages/ai/tests/abort-signal-reach.test.ts | 73 +++++++++++++++++++ testing/e2e/tests/summarize-abort.spec.ts | 6 +- 7 files changed, 160 insertions(+), 7 deletions(-) diff --git a/packages/ai-client/src/connection-adapters.ts b/packages/ai-client/src/connection-adapters.ts index 5eb58ae5ea..249785f070 100644 --- a/packages/ai-client/src/connection-adapters.ts +++ b/packages/ai-client/src/connection-adapters.ts @@ -332,7 +332,6 @@ export function createReconnectTracker( * A signal that aborts when either one does. The client's signal (stop()) and * the user's own `options.signal` must both stop the request. */ -// ponytail: listeners stay on a long-lived signal until it aborts; AbortSignal.any when the browser floor allows. function eitherSignal( a: AbortSignal | undefined, b: AbortSignal | undefined, @@ -340,6 +339,11 @@ function eitherSignal( if (!a || !b) return a ?? b if (a.aborted) return a if (b.aborted) return b + // `AbortSignal.any` links weakly, so a long-lived `options.signal` keeps + // nothing from a finished request. + if (typeof AbortSignal.any === 'function') return AbortSignal.any([a, b]) + // ponytail: older runtimes keep these listeners on a long-lived signal + // until it aborts. const controller = new AbortController() a.addEventListener('abort', () => controller.abort(a.reason), { once: true }) b.addEventListener('abort', () => controller.abort(b.reason), { once: true }) diff --git a/packages/ai-client/src/realtime-client.ts b/packages/ai-client/src/realtime-client.ts index b408946485..8633884d70 100644 --- a/packages/ai-client/src/realtime-client.ts +++ b/packages/ai-client/src/realtime-client.ts @@ -150,8 +150,15 @@ export class RealtimeClient { // Start audio capture if configured if (this.options.autoCapture) { - await this.connection.startAudioCapture() - if (generation !== this.connectGeneration) return + await connection.startAudioCapture() + if (generation !== this.connectGeneration) { + // A later connect() replaced this connection: close it. After + // disconnect(), `this.connection` is null and it is closed already. + if (this.connection && this.connection !== connection) { + await connection.disconnect() + } + return + } } this.updateState({ diff --git a/packages/ai-client/tests/abort-signal-reach.test.ts b/packages/ai-client/tests/abort-signal-reach.test.ts index 5a19949399..07655f5382 100644 --- a/packages/ai-client/tests/abort-signal-reach.test.ts +++ b/packages/ai-client/tests/abort-signal-reach.test.ts @@ -3,6 +3,7 @@ * must reach the work it should stop. The fake requests below hang until * their signal aborts, like a real fetch that keeps the run going. */ +import { getEventListeners } from 'node:events' import { describe, expect, it, vi } from 'vitest' import { ChatClient } from '../src/chat-client' import { RealtimeClient } from '../src/realtime-client' @@ -137,6 +138,23 @@ describe("connection adapters follow the user's options.signal (L2)", () => { }) } } + + it('leaves no listener on a reused options.signal after each request', async () => { + const user = new AbortController() + const fetchClient = (() => + Promise.resolve( + new Response('', { headers: { 'content-type': 'text/event-stream' } }), + )) as typeof fetch + const adapter = fetchServerSentEvents('/api/chat', { + fetchClient, + signal: user.signal, + }) + for (let run = 0; run < 3; run++) { + const client = new AbortController() + await drain(adapter.connect([], undefined, client.signal, RUN_CONTEXT)) + } + expect(getEventListeners(user.signal, 'abort')).toHaveLength(0) + }) }) describe('stream() and rpcStream() stop on abort (L3)', () => { @@ -250,6 +268,42 @@ describe('RealtimeClient.disconnect() during connect() (L4)', () => { expect(Reflect.get(client, 'tokenRefreshTimeout')).toBeNull() }) } + + it('closes a connection that a second connect() replaced during audio startup', async () => { + let release!: () => void + const held = new Promise((resolve) => { + release = resolve + }) + const makeConnection = (startAudioCapture: () => Promise) => ({ + on: () => () => {}, + updateSession: () => {}, + disconnect: vi.fn(async () => {}), + startAudioCapture, + }) + const first = makeConnection(() => held) + const second = makeConnection(async () => {}) + const adapterConnect = vi + .fn() + .mockResolvedValueOnce(first) + .mockResolvedValueOnce(second) + const client = new RealtimeClient({ + getToken: async () => + ({ token: 't', expiresAt: Date.now() + 60_000 }) as any, + adapter: { connect: adapterConnect } as any, + autoCapture: true, + }) + + const firstConnect = client.connect() + await vi.waitFor(() => expect(adapterConnect).toHaveBeenCalledTimes(1)) + await new Promise((resolve) => setTimeout(resolve, 0)) + await client.connect() + release() + await firstConnect + + expect(first.disconnect).toHaveBeenCalled() + expect(second.disconnect).not.toHaveBeenCalled() + await client.disconnect() + }) }) describe('RealtimeClient.disconnect() during a token refresh', () => { diff --git a/packages/ai/src/activities/generateVideo/index.ts b/packages/ai/src/activities/generateVideo/index.ts index dc36e03a86..943a4cdf46 100644 --- a/packages/ai/src/activities/generateVideo/index.ts +++ b/packages/ai/src/activities/generateVideo/index.ts @@ -701,6 +701,8 @@ async function* runStreamingVideoGeneration< } if (statusResult.status === 'completed') { + // The caller can abort when it reads the status above. + abortControls.signal?.throwIfAborted() const video = await raceWithAbort( adapter.getVideo(jobResult.jobId), abortControls.signal, diff --git a/packages/ai/src/activities/summarize/index.ts b/packages/ai/src/activities/summarize/index.ts index 414bfc9429..09c28ab810 100644 --- a/packages/ai/src/activities/summarize/index.ts +++ b/packages/ai/src/activities/summarize/index.ts @@ -504,10 +504,19 @@ async function* runNativeSummarizeStream( yield chunk } if (!settled) { - await runGenerationFinish(middleware, mwCtx, { - duration: Date.now() - startTime, - }) settled = true + // An adapter can end an aborted request with a RUN_ERROR chunk instead + // of a throw. The consumer got that chunk; middleware gets onAbort. + if (signal?.aborted) { + await runGenerationAbort(middleware, mwCtx, { + reason: abortReasonMessage(undefined, signal), + duration: Date.now() - startTime, + }) + } else { + await runGenerationFinish(middleware, mwCtx, { + duration: Date.now() - startTime, + }) + } } } catch (error) { settled = true diff --git a/packages/ai/tests/abort-signal-reach.test.ts b/packages/ai/tests/abort-signal-reach.test.ts index f70d254746..13ad480ea1 100644 --- a/packages/ai/tests/abort-signal-reach.test.ts +++ b/packages/ai/tests/abort-signal-reach.test.ts @@ -153,6 +153,42 @@ describe('summarize() abort reaches the wrapped chatStream', () => { expect(onError).not.toHaveBeenCalled() expect(onAbort).toHaveBeenCalledTimes(1) }) + + it('stream: an abort that comes as a RUN_ERROR chunk reports onAbort', async () => { + // Like openai-base: a failed request becomes a RUN_ERROR chunk. + const text: ChatStreamCapable = { + chatStream: (options) => + (async function* () { + yield ev.runStarted() + try { + await hangUntilAborted(options.request?.signal ?? undefined) + } catch { + yield ev.runError('Request was aborted.') + } + })(), + } + const onAbort = vi.fn() + const onFinish = vi.fn() + const controller = new AbortController() + const stream = summarize({ + adapter: new ChatStreamSummarizeAdapter( + text, + 'test-model', + 'openai', + ), + threadId: 'thread-1', + text: 'long text', + stream: true, + abortSignal: controller.signal, + middleware: [{ name: 'observe', onAbort, onFinish }], + }) + setTimeout(() => controller.abort(), 10) + const chunks = await collectChunks(stream).catch(() => []) + expect(onFinish).not.toHaveBeenCalled() + expect(onAbort).toHaveBeenCalledTimes(1) + // The error chunk still reaches the consumer. + expect(chunks.some((chunk) => chunk.type === 'RUN_ERROR')).toBe(true) + }) }) describe('middleware ctx.abort() reaches the provider', () => { @@ -376,6 +412,43 @@ describe('generateVideo() polling', () => { expect(await settlesWithin(collectChunks(stream))).toBe('done') }) } + + it('starts no download when the caller aborts on the completed status', async () => { + const controller = new AbortController() + const getVideo = vi.fn(async () => ({ url: 'https://example.com/v.mp4' })) + const adapter = { + kind: 'video' as const, + name: 'mock', + model: 'mock-video-model', + createVideoJob: async () => ({ + jobId: 'job-1', + model: 'mock-video-model', + }), + getVideoStatus: async () => ({ status: 'completed' as const }), + getVideo, + } + const stream = generateVideo({ + threadId: 'thread-1', + adapter: adapter as any, + prompt: 'a cat', + stream: true, + pollingInterval: 1, + abortSignal: controller.signal, + }) + // The stream ends the run itself; it may not throw. + await (async () => { + for await (const chunk of stream as AsyncIterable) { + if ( + chunk.type === 'CUSTOM' && + chunk.name === 'video:status' && + (chunk.value as { status?: string }).status === 'completed' + ) { + controller.abort() + } + } + })().catch(() => undefined) + expect(getVideo).not.toHaveBeenCalled() + }) }) describe('toWebSocketStream() (non-durable)', () => { diff --git a/testing/e2e/tests/summarize-abort.spec.ts b/testing/e2e/tests/summarize-abort.spec.ts index b072c2dcc5..16999f0e56 100644 --- a/testing/e2e/tests/summarize-abort.spec.ts +++ b/testing/e2e/tests/summarize-abort.spec.ts @@ -20,7 +20,11 @@ test('streaming summarize stops at its timeout', async ({ }) const body = await response.text() + const elapsed = Date.now() - started expect(body).toContain('RUN_ERROR') expect(body).not.toContain('RUN_FINISHED') - expect(Date.now() - started).toBeLessThan(5000) + // An error that is not the timeout (a failed connection, a bad route) + // comes at once. The timeout ends the run only after 300 ms. + expect(elapsed).toBeGreaterThanOrEqual(250) + expect(elapsed).toBeLessThan(5000) }) From 7c712c9f00b4862e1087e7a4e60827153b44c09e Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Mon, 2 Nov 2026 17:50:32 +0100 Subject: [PATCH 09/12] fix(ai): cancel a video stream that comes after the abort getVideo can return a VideoStreamResult. When the abort wins the race, nobody reads its body, so cancel it to close the download. --- .../ai/src/activities/generateVideo/index.ts | 16 ++++++-- packages/ai/tests/abort-signal-reach.test.ts | 38 +++++++++++++++++++ 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/packages/ai/src/activities/generateVideo/index.ts b/packages/ai/src/activities/generateVideo/index.ts index 943a4cdf46..3cfd2c3612 100644 --- a/packages/ai/src/activities/generateVideo/index.ts +++ b/packages/ai/src/activities/generateVideo/index.ts @@ -703,9 +703,19 @@ async function* runStreamingVideoGeneration< if (statusResult.status === 'completed') { // The caller can abort when it reads the status above. abortControls.signal?.throwIfAborted() - const video = await raceWithAbort( - adapter.getVideo(jobResult.jobId), - abortControls.signal, + const download = adapter.getVideo(jobResult.jobId) + const video = await raceWithAbort(download, abortControls.signal).catch( + (error: unknown) => { + // Nobody reads a video stream that comes after the abort, so + // cancel its body to close the download. + download.then( + (late) => { + if ('body' in late) late.body?.cancel().catch(() => undefined) + }, + () => undefined, + ) + throw error + }, ) logger.output( diff --git a/packages/ai/tests/abort-signal-reach.test.ts b/packages/ai/tests/abort-signal-reach.test.ts index 13ad480ea1..134ca6aa84 100644 --- a/packages/ai/tests/abort-signal-reach.test.ts +++ b/packages/ai/tests/abort-signal-reach.test.ts @@ -449,6 +449,44 @@ describe('generateVideo() polling', () => { })().catch(() => undefined) expect(getVideo).not.toHaveBeenCalled() }) + + it('cancels the body of a video stream that comes after the abort', async () => { + const controller = new AbortController() + const cancel = vi.fn() + let deliver = () => {} + const adapter = { + kind: 'video' as const, + name: 'mock', + model: 'mock-video-model', + createVideoJob: async () => ({ + jobId: 'job-1', + model: 'mock-video-model', + }), + getVideoStatus: async () => ({ status: 'completed' as const }), + // The download answers only after the caller aborted. + getVideo: () => + new Promise((resolve) => { + deliver = () => + resolve({ + jobId: 'job-1', + contentType: 'video/mp4', + body: new ReadableStream({ cancel }), + }) + setTimeout(() => controller.abort(), 10) + }), + } + const stream = generateVideo({ + threadId: 'thread-1', + adapter: adapter as any, + prompt: 'a cat', + stream: true, + pollingInterval: 1, + abortSignal: controller.signal, + }) + await collectChunks(stream).catch(() => undefined) + deliver() + await vi.waitFor(() => expect(cancel).toHaveBeenCalled()) + }) }) describe('toWebSocketStream() (non-durable)', () => { From dbbd59e300ba3a4f8e6d9816793a48dc0970af61 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Tue, 3 Nov 2026 12:19:09 +0100 Subject: [PATCH 10/12] fix(ai-client): end a user signal abort like stop(); ignore a replaced realtime connection - normalizeConnectionAdapter: send() reads the request signal of the built-in adapters. An abort of options.signal no longer pushes a synthetic RUN_FINISHED or RUN_ERROR. ChatClient ends it like stop(). - RealtimeClient: events from a connection that a later connect() replaced have no effect, and a failed close of that connection does not set the error state. --- .changeset/client-cancel-like-stop.md | 5 + packages/ai-client/src/chat-client.ts | 5 + packages/ai-client/src/connection-adapters.ts | 27 +++- packages/ai-client/src/realtime-client.ts | 31 ++-- .../tests/abort-signal-reach.test.ts | 136 ++++++++++++++++++ 5 files changed, 189 insertions(+), 15 deletions(-) create mode 100644 .changeset/client-cancel-like-stop.md diff --git a/.changeset/client-cancel-like-stop.md b/.changeset/client-cancel-like-stop.md new file mode 100644 index 0000000000..54b2c43904 --- /dev/null +++ b/.changeset/client-cancel-like-stop.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-client': patch +--- + +A cancel now stops the request. `stop()`, an abort of the connection `options.signal`, and `disconnect()` reach the fetch, XHR, `stream()`, and `rpcStream()` requests and the realtime connect. An abort of `options.signal` now ends the chat like `stop()`, not as a finished run. A realtime connection that a later `connect()` replaced no longer changes the status of the live session. diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index c49909da67..381b9bdfe0 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -2935,6 +2935,11 @@ export class ChatClient< ) runTerminalEventEmitted = true } + // The abort came from the user's own `options.signal`, not from + // stop(). End the same way stop() does. + if (!signal.aborted && generation === this.streamGeneration) { + this.stop() + } return false } if (error instanceof ByokMissingError) { diff --git a/packages/ai-client/src/connection-adapters.ts b/packages/ai-client/src/connection-adapters.ts index 249785f070..3dcc9bb0ea 100644 --- a/packages/ai-client/src/connection-adapters.ts +++ b/packages/ai-client/src/connection-adapters.ts @@ -337,8 +337,18 @@ function eitherSignal( b: AbortSignal | undefined, ): AbortSignal | undefined { if (!a || !b) return a ?? b - if (a.aborted) return a - if (b.aborted) return b + const signal = a.aborted ? a : b.aborted ? b : anySignal(a, b) + requestSignals.set(a, signal) + return signal +} + +/** + * The client's signal -> the signal its request follows. `send()` reads it, + * so an abort of `options.signal` is a stop, not an end or an error. + */ +const requestSignals = new WeakMap() + +function anySignal(a: AbortSignal, b: AbortSignal): AbortSignal { // `AbortSignal.any` links weakly, so a long-lived `options.signal` keeps // nothing from a finished request. if (typeof AbortSignal.any === 'function') return AbortSignal.any([a, b]) @@ -1300,6 +1310,10 @@ export function normalizeConnectionAdapter( let hasTerminalEvent = false let upstreamThreadId: string | undefined let upstreamRunId: string | undefined + const aborted = () => + abortSignal?.aborted === true || + (abortSignal !== undefined && + requestSignals.get(abortSignal)?.aborted === true) try { const stream = connection.connect( messages, @@ -1325,7 +1339,12 @@ export function normalizeConnectionAdapter( // The event payload may carry an upstream/provider runId when one was // observed, but stamp the caller's request runId so getChunkRunId() // correlates to activeRunIds / currentRunId (same as real stream chunks). - if (!abortSignal?.aborted && !hasTerminalEvent) { + // Only the user's `options.signal` aborted: end like the client's own + // stop(), which rejects with an AbortError. + if (aborted() && !abortSignal?.aborted) { + throw Object.assign(new Error('Aborted'), { name: 'AbortError' }) + } + if (!aborted() && !hasTerminalEvent) { push( withTanstackMetadata( { @@ -1340,7 +1359,7 @@ export function normalizeConnectionAdapter( ) } } catch (err) { - if (!abortSignal?.aborted && !hasTerminalEvent) { + if (!aborted() && !hasTerminalEvent) { const message = err instanceof Error ? err.message : 'Unknown error in connect()' const synthetic: RunErrorEvent = { diff --git a/packages/ai-client/src/realtime-client.ts b/packages/ai-client/src/realtime-client.ts index 8633884d70..8355f268b5 100644 --- a/packages/ai-client/src/realtime-client.ts +++ b/packages/ai-client/src/realtime-client.ts @@ -155,7 +155,9 @@ export class RealtimeClient { // A later connect() replaced this connection: close it. After // disconnect(), `this.connection` is null and it is closed already. if (this.connection && this.connection !== connection) { - await connection.disconnect() + // A failed close of the replaced connection is not an + // error of the live session. + await connection.disconnect().catch(() => {}) } return } @@ -482,18 +484,25 @@ export class RealtimeClient { } private subscribeToConnectionEvents(): void { - if (!this.connection) return + const connection = this.connection + if (!connection) return + // A later connect() can replace this connection. Its late events (such as + // `idle` from its disconnect) must not change the live session. + const on: RealtimeConnection['on'] = (event, handler) => + connection.on(event, (payload) => { + if (this.connection === connection) handler(payload) + }) // Status changes this.unsubscribers.push( - this.connection.on('status_change', ({ status }) => { + on('status_change', ({ status }) => { this.updateState({ status }) }), ) // Mode changes this.unsubscribers.push( - this.connection.on('mode_change', ({ mode }) => { + on('mode_change', ({ mode }) => { this.updateState({ mode }) }), ) @@ -502,7 +511,7 @@ export class RealtimeClient { // User transcripts are added as messages when final (no separate message_complete for user input) // Assistant transcripts are streamed, final message comes via message_complete this.unsubscribers.push( - this.connection.on('transcript', ({ role, transcript, isFinal }) => { + on('transcript', ({ role, transcript, isFinal }) => { if (role === 'user') { this.updateState({ pendingUserTranscript: isFinal ? null : transcript, @@ -527,7 +536,7 @@ export class RealtimeClient { // Tool calls this.unsubscribers.push( - this.connection.on('tool_call', ({ toolCallId, toolName, input }) => { + on('tool_call', ({ toolCallId, toolName, input }) => { if (!toolCallId) { console.error( '[RealtimeClient] tool_call missing toolCallId', @@ -567,7 +576,7 @@ export class RealtimeClient { // Message complete this.unsubscribers.push( - this.connection.on('message_complete', ({ message }) => { + on('message_complete', ({ message }) => { // Replace pending message with final version if needed const existingIndex = this.state.messages.findIndex( (m) => m.id === message.id, @@ -584,7 +593,7 @@ export class RealtimeClient { // Interruption this.unsubscribers.push( - this.connection.on('interrupted', ({ messageId }) => { + on('interrupted', ({ messageId }) => { if (messageId) { const newMessages = this.state.messages.map((m) => m.id === messageId ? { ...m, interrupted: true } : m, @@ -601,20 +610,20 @@ export class RealtimeClient { // Errors this.unsubscribers.push( - this.connection.on('error', ({ error }) => { + on('error', ({ error }) => { this.updateState({ error }) this.options.onError?.(error) }), ) this.unsubscribers.push( - this.connection.on('go_away', ({ timeLeft }) => { + on('go_away', ({ timeLeft }) => { this.options.onGoAway?.(timeLeft) }), ) this.unsubscribers.push( - this.connection.on('usage', (usage) => { + on('usage', (usage) => { this.options.onUsage?.(usage) }), ) diff --git a/packages/ai-client/tests/abort-signal-reach.test.ts b/packages/ai-client/tests/abort-signal-reach.test.ts index 07655f5382..054d92d38f 100644 --- a/packages/ai-client/tests/abort-signal-reach.test.ts +++ b/packages/ai-client/tests/abort-signal-reach.test.ts @@ -157,6 +157,78 @@ describe("connection adapters follow the user's options.signal (L2)", () => { }) }) +describe('ChatClient ends a user abort of options.signal like stop()', () => { + /** An SSE reply that sends some text, then waits until its signal aborts. */ + const partialFetch = (() => (_url: RequestInfo | URL, init?: RequestInit) => { + const encoder = new TextEncoder() + const body = new ReadableStream({ + start(controller) { + for (const event of [ + { type: 'RUN_STARTED', ...RUN_CONTEXT, timestamp: 1 }, + { + type: 'TEXT_MESSAGE_START', + messageId: 'm1', + role: 'assistant', + timestamp: 1, + }, + { + type: 'TEXT_MESSAGE_CONTENT', + messageId: 'm1', + delta: 'Hel', + timestamp: 1, + }, + ]) { + controller.enqueue( + encoder.encode(`data: ${JSON.stringify(event)} + +`), + ) + } + init?.signal?.addEventListener('abort', () => + controller.error( + Object.assign(new Error('aborted'), { name: 'AbortError' }), + ), + ) + }, + }) + return Promise.resolve( + new Response(body, { + headers: { 'content-type': 'text/event-stream' }, + }), + ) + }) as unknown as () => typeof fetch + + it.each(['options.signal', 'stop()'] as const)( + 'cancel with %s', + async (cancel) => { + const user = new AbortController() + const onError = vi.fn() + const onFinish = vi.fn() + const client = new ChatClient({ + threadId: 'thread-1', + connection: fetchServerSentEvents('/api/chat', { + fetchClient: partialFetch(), + signal: user.signal, + }), + onError, + onFinish, + }) + const send = client.sendMessage('Go') + await vi.waitFor(() => + expect(JSON.stringify(client.getMessages())).toContain('Hel'), + ) + if (cancel === 'stop()') client.stop() + else user.abort() + + expect(await settlesWithin(send)).toBe('done') + expect(client.getStatus()).toBe('ready') + expect(client.getIsLoading()).toBe(false) + expect(onError).not.toHaveBeenCalled() + expect(onFinish).not.toHaveBeenCalled() + }, + ) +}) + describe('stream() and rpcStream() stop on abort (L3)', () => { // A source that ignores its signal and never ends. const ignoresSignal = () => @@ -306,6 +378,70 @@ describe('RealtimeClient.disconnect() during connect() (L4)', () => { }) }) +describe('RealtimeClient keeps the live session when a second connect() wins', () => { + // The first connection emits `idle` when it closes (like the OpenAI + // adapter), and its close can fail. + it.each(['emits idle', 'throws'] as const)( + 'the replaced connection %s on close', + async (onClose) => { + let release!: () => void + const held = new Promise((resolve) => { + release = resolve + }) + const makeConnection = (startAudioCapture: () => Promise) => { + const handlers = new Map void>>() + const connection = { + on: (event: string, handler: (payload: any) => void) => { + const set = handlers.get(event) ?? new Set() + set.add(handler) + handlers.set(event, set) + return () => set.delete(handler) + }, + emit: (event: string, payload: unknown) => + handlers.get(event)?.forEach((handler) => handler(payload)), + updateSession: () => {}, + disconnect: vi.fn(async () => {}), + startAudioCapture, + } + return connection + } + const first = makeConnection(() => held) + first.disconnect.mockImplementation(async () => { + if (onClose === 'throws') throw new Error('close failed') + first.emit('status_change', { status: 'idle' }) + }) + const second = makeConnection(async () => {}) + const onError = vi.fn() + const client = new RealtimeClient({ + getToken: async () => + ({ token: 't', expiresAt: Date.now() + 60_000 }) as any, + adapter: { + connect: vi + .fn() + .mockResolvedValueOnce(first) + .mockResolvedValueOnce(second), + } as any, + autoCapture: true, + onError, + }) + + const firstConnect = client.connect() + await vi.waitFor(() => + expect(Reflect.get(client, 'connection')).toBe(first), + ) + await client.connect() + release() + await firstConnect + + expect(first.disconnect).toHaveBeenCalled() + expect(client.status).toBe('connected') + expect(client.error).toBeNull() + expect(onError).not.toHaveBeenCalled() + await client.disconnect() + }, + ) +}) + describe('RealtimeClient.disconnect() during a token refresh', () => { it('does not re-arm the refresh timer', async () => { let release!: () => void From e26b4647f97b5b86c2b7f1e41b1ba56a22458964 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Tue, 3 Nov 2026 12:21:01 +0100 Subject: [PATCH 11/12] fix(ai-client): a failed close of a late connection does not set error When disconnect() runs while adapter.connect() is pending, connect() closes the late connection. If that close fails, the client now stays idle and does not call onError. --- packages/ai-client/src/realtime-client.ts | 4 +- .../tests/abort-signal-reach.test.ts | 39 +++++++++++++++++++ 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/packages/ai-client/src/realtime-client.ts b/packages/ai-client/src/realtime-client.ts index 8355f268b5..b77bd66ab9 100644 --- a/packages/ai-client/src/realtime-client.ts +++ b/packages/ai-client/src/realtime-client.ts @@ -128,7 +128,9 @@ export class RealtimeClient { toolsList, ) if (generation !== this.connectGeneration) { - await connection.disconnect() + // A failed close of a connection that was never used is not an + // error of the client. + await connection.disconnect().catch(() => {}) return } this.connection = connection diff --git a/packages/ai-client/tests/abort-signal-reach.test.ts b/packages/ai-client/tests/abort-signal-reach.test.ts index 054d92d38f..c797178ce5 100644 --- a/packages/ai-client/tests/abort-signal-reach.test.ts +++ b/packages/ai-client/tests/abort-signal-reach.test.ts @@ -341,6 +341,45 @@ describe('RealtimeClient.disconnect() during connect() (L4)', () => { }) } + it('stays idle when the late connection fails to close', async () => { + let release!: () => void + const held = new Promise((resolve) => { + release = resolve + }) + const connection = { + on: () => () => {}, + updateSession: () => {}, + disconnect: vi.fn(async () => { + throw new Error('close failed') + }), + startAudioCapture: async () => {}, + } + const onError = vi.fn() + const client = new RealtimeClient({ + getToken: async () => + ({ token: 't', expiresAt: Date.now() + 60_000 }) as any, + adapter: { + connect: async () => { + await held + return connection + }, + } as any, + onError, + }) + + const connecting = client.connect() + await vi.waitFor(() => expect(client.status).toBe('connecting')) + await new Promise((resolve) => setTimeout(resolve, 0)) + await client.disconnect() + release() + await connecting + + expect(connection.disconnect).toHaveBeenCalled() + expect(client.status).toBe('idle') + expect(client.error).toBeNull() + expect(onError).not.toHaveBeenCalled() + }) + it('closes a connection that a second connect() replaced during audio startup', async () => { let release!: () => void const held = new Promise((resolve) => { From a6aca12e186b2335330405ba4ac95d12107375c7 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Tue, 3 Nov 2026 12:36:13 +0100 Subject: [PATCH 12/12] fix(ai-client): clear the refresh timer of a replaced realtime connect A connect() that a later connect() replaced left its token refresh timer armed, so it fetched tokens after disconnect(). Also tighten two abort tests and correct the changeset scope for realtime and summarize. --- .changeset/client-cancel-like-stop.md | 2 +- .changeset/core-abort-reaches-provider.md | 2 +- packages/ai-client/src/realtime-client.ts | 2 + .../tests/abort-signal-reach.test.ts | 42 +++++++++++++++++++ packages/ai/tests/abort-signal-reach.test.ts | 11 ++++- 5 files changed, 56 insertions(+), 3 deletions(-) diff --git a/.changeset/client-cancel-like-stop.md b/.changeset/client-cancel-like-stop.md index 54b2c43904..a02c24fa38 100644 --- a/.changeset/client-cancel-like-stop.md +++ b/.changeset/client-cancel-like-stop.md @@ -2,4 +2,4 @@ '@tanstack/ai-client': patch --- -A cancel now stops the request. `stop()`, an abort of the connection `options.signal`, and `disconnect()` reach the fetch, XHR, `stream()`, and `rpcStream()` requests and the realtime connect. An abort of `options.signal` now ends the chat like `stop()`, not as a finished run. A realtime connection that a later `connect()` replaced no longer changes the status of the live session. +A cancel now stops the request. `stop()` and an abort of the connection `options.signal` reach the fetch, XHR, `stream()`, and `rpcStream()` requests. A realtime `disconnect()` during `connect()` does not stop a pending `getToken()` or `adapter.connect()` call: it drops the late result and closes the connection that comes back. An abort of `options.signal` now ends the chat like `stop()`, not as a finished run. A realtime connection that a later `connect()` replaced no longer changes the status of the live session. diff --git a/.changeset/core-abort-reaches-provider.md b/.changeset/core-abort-reaches-provider.md index e51795f712..a0b066ba8e 100644 --- a/.changeset/core-abort-reaches-provider.md +++ b/.changeset/core-abort-reaches-provider.md @@ -2,4 +2,4 @@ '@tanstack/ai': patch --- -A cancel now stops the provider request, not only the caller. `summarize()` (also with `stream: true`, which now honors `abortSignal` and `timeout` and reports a cancel to middleware `onAbort`, not `onError`), middleware `ctx.abort()` when the caller passed an `abortController` (adapter request, structured output, and `ctx.signal`), subagent children, `generateVideo` status polls, and aborted `toWebSocket` turns now stop when the run is cancelled. +A cancel now stops the provider request, not only the caller. `summarize()` (also with `stream: true`, which now passes `abortSignal` and `timeout` to the adapter and reports a cancel to middleware `onAbort`, not `onError`), middleware `ctx.abort()` when the caller passed an `abortController` (adapter request, structured output, and `ctx.signal`), subagent children, `generateVideo` status polls, and aborted `toWebSocket` turns now stop when the run is cancelled. diff --git a/packages/ai-client/src/realtime-client.ts b/packages/ai-client/src/realtime-client.ts index b77bd66ab9..9c2d0f7614 100644 --- a/packages/ai-client/src/realtime-client.ts +++ b/packages/ai-client/src/realtime-client.ts @@ -458,6 +458,8 @@ export class RealtimeClient { private scheduleTokenRefresh(): void { if (!this.token) return + // A connect() that replaced an earlier one must not leave its timer. + if (this.tokenRefreshTimeout) clearTimeout(this.tokenRefreshTimeout) const timeUntilExpiry = this.token.expiresAt - Date.now() const refreshIn = Math.max(0, timeUntilExpiry - TOKEN_REFRESH_BUFFER_MS) diff --git a/packages/ai-client/tests/abort-signal-reach.test.ts b/packages/ai-client/tests/abort-signal-reach.test.ts index c797178ce5..6700f012e2 100644 --- a/packages/ai-client/tests/abort-signal-reach.test.ts +++ b/packages/ai-client/tests/abort-signal-reach.test.ts @@ -515,4 +515,46 @@ describe('RealtimeClient.disconnect() during a token refresh', () => { expect(Reflect.get(client, 'tokenRefreshTimeout')).toBeNull() }) + + it('leaves no refresh timer of a connect() that a later connect() replaced', async () => { + vi.useFakeTimers() + try { + let release!: () => void + const held = new Promise((resolve) => { + release = resolve + }) + const connection = { + on: () => () => {}, + updateSession: () => {}, + disconnect: async () => {}, + startAudioCapture: async () => {}, + } + const getToken = vi.fn( + async () => ({ token: 't', expiresAt: Date.now() + 120_000 }) as any, + ) + const client = new RealtimeClient({ + getToken, + adapter: { + connect: vi + .fn() + .mockImplementationOnce(async () => { + await held + return connection + }) + .mockResolvedValueOnce(connection), + } as any, + }) + + const first = client.connect() + await vi.waitFor(() => expect(getToken).toHaveBeenCalledTimes(1)) + await client.connect() + release() + await first + await client.disconnect() + + expect(vi.getTimerCount()).toBe(0) + } finally { + vi.useRealTimers() + } + }) }) diff --git a/packages/ai/tests/abort-signal-reach.test.ts b/packages/ai/tests/abort-signal-reach.test.ts index 134ca6aa84..558cfc4b17 100644 --- a/packages/ai/tests/abort-signal-reach.test.ts +++ b/packages/ai/tests/abort-signal-reach.test.ts @@ -187,7 +187,13 @@ describe('summarize() abort reaches the wrapped chatStream', () => { expect(onFinish).not.toHaveBeenCalled() expect(onAbort).toHaveBeenCalledTimes(1) // The error chunk still reaches the consumer. - expect(chunks.some((chunk) => chunk.type === 'RUN_ERROR')).toBe(true) + expect( + chunks.some( + (chunk) => + chunk.type === 'RUN_ERROR' && + chunk.message === 'Request was aborted.', + ), + ).toBe(true) }) }) @@ -435,6 +441,7 @@ describe('generateVideo() polling', () => { pollingInterval: 1, abortSignal: controller.signal, }) + let sawCompleted = false // The stream ends the run itself; it may not throw. await (async () => { for await (const chunk of stream as AsyncIterable) { @@ -443,10 +450,12 @@ describe('generateVideo() polling', () => { chunk.name === 'video:status' && (chunk.value as { status?: string }).status === 'completed' ) { + sawCompleted = true controller.abort() } } })().catch(() => undefined) + expect(sawCompleted).toBe(true) expect(getVideo).not.toHaveBeenCalled() })