fix(ai, ai-client): make a cancel stop the provider request, not only the caller

.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()` or a token refresh no longer leaves a connection or a token refresh timer behind.

.changeset/client-cancel-like-stop.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-client': patch
+---
+
+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.

.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 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.

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<TanStackMessageMetadata>` | 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<ModelMessage>` | 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). |

packages/ai-client/src/chat-client.ts

@@ -110,6 +110,22 @@ type SchedulerWithYield = {
   yield?: () => Promise<void>
 }
 
+/**
+ * Wait for `promise`, but reject with an AbortError once `signal` aborts.
+ * A client tool that never settles must not block stop().
+ */
+function untilAborted<T>(promise: Promise<T>, signal: AbortSignal): Promise<T> {
+  return new Promise<T>((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<void> {
   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).
@@ -2899,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) {
@@ -2913,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) {

packages/ai-client/src/connection-adapters.ts

@@ -328,6 +328,38 @@ 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.
+ */
+function eitherSignal(
+  a: AbortSignal | undefined,
+  b: AbortSignal | undefined,
+): AbortSignal | undefined {
+  if (!a || !b) return a ?? 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<AbortSignal, AbortSignal>()
+
+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])
+  // 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 })
+  return controller.signal
+}
+
 /** Resolve after `ms`, or immediately once `signal` aborts. Never rejects. */
 function abortableDelay(ms: number, signal?: AbortSignal): Promise<void> {
   if (ms <= 0 || signal?.aborted) return Promise.resolve()
@@ -1278,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,
@@ -1303,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(
               {
@@ -1318,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 = {
@@ -1500,7 +1541,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 +1584,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 +1701,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 +1741,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 +1948,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 +2298,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 +2320,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 +2382,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 +2404,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 +2937,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 +3029,11 @@ async function* abortableIterable<T>(
       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 +3080,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