fix(openai-base): treat a caller abort as an abort, not a fatal error

.changeset/openai-base-abort-log.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/openai-base': patch
+---
+
+A caller abort no longer logs a `fatal` entry on the `errors` debug channel. A direct `chatStream()` call that is aborted before the response arrives now yields `RUN_ERROR` with `code: 'aborted'`.

packages/openai-base/src/adapters/chat-completions-text.ts

@@ -17,6 +17,7 @@ import {
   transformMessagesForReplay,
 } from '@tanstack/ai/adapter-internals'
 import { generateId } from '@tanstack/ai-utils'
+import { APIUserAbortError } from 'openai/core/error'
 import { clientFor, extractRequestOptions } from '../utils/request-options'
 import type { Fetch } from '../utils/request-options'
 import {
@@ -271,10 +272,13 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
       return
     }
 
-    options.logger.errors(`${this.name}.${source} fatal`, {
-      error: errorPayload,
-      source: `${this.name}.${source}`,
-    })
+    // A caller abort is not a failure, so it stays off the errors channel.
+    if (!this.isAbortError(error)) {
+      options.logger.errors(`${this.name}.${source} fatal`, {
+        error: errorPayload,
+        source: `${this.name}.${source}`,
+      })
+    }
 
     yield {
       type: EventType.RUN_ERROR,
@@ -427,10 +431,15 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
     } catch (error: unknown) {
       // Narrow before logging: raw SDK errors can carry request metadata
       // (including auth headers) which we must never surface to user loggers.
-      chatOptions.logger.errors(`${this.name}.structuredOutput fatal`, {
-        error: toRunErrorPayload(error, `${this.name}.structuredOutput failed`),
-        source: `${this.name}.structuredOutput`,
-      })
+      if (!this.isAbortError(error)) {
+        chatOptions.logger.errors(`${this.name}.structuredOutput fatal`, {
+          error: toRunErrorPayload(
+            error,
+            `${this.name}.structuredOutput failed`,
+          ),
+          source: `${this.name}.structuredOutput`,
+        })
+      }
       throw error
     }
   }
@@ -776,6 +785,9 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
     error: unknown,
     fallbackMessage: string,
   ): { message: string; code: string | undefined } {
+    if (this.isAbortError(error)) {
+      return { message: 'Request aborted', code: 'aborted' }
+    }
     const payload = toRunErrorPayload(error, fallbackMessage)
     return this.isContextOverflow(payload)
       ? { ...payload, code: CONTEXT_OVERFLOW_CODE }
@@ -804,6 +816,8 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
     if (!error || typeof error !== 'object') return false
     const e = error as { name?: unknown; code?: unknown }
     return (
+      // The openai client's APIUserAbortError leaves `name` as 'Error'.
+      error instanceof APIUserAbortError ||
       e.name === 'APIUserAbortError' ||
       e.name === 'AbortError' ||
       e.code === 'ERR_CANCELED'

packages/openai-base/src/adapters/responses-text.ts

@@ -19,6 +19,7 @@ import {
   transformMessagesForReplay,
 } from '@tanstack/ai/adapter-internals'
 import { drainAfterTerminal, generateId } from '@tanstack/ai-utils'
+import { APIUserAbortError } from 'openai/core/error'
 import { Stream } from 'openai/streaming'
 import { clientFor, extractRequestOptions } from '../utils/request-options'
 import type { Fetch } from '../utils/request-options'
@@ -638,10 +639,12 @@ export abstract class OpenAIBaseResponsesTextAdapter<
         },
       }
 
-      options.logger.errors(`${this.name}.chatStream fatal`, {
-        error: errorPayload,
-        source: `${this.name}.chatStream`,
-      })
+      if (!this.isAbortError(error)) {
+        options.logger.errors(`${this.name}.chatStream fatal`, {
+          error: errorPayload,
+          source: `${this.name}.chatStream`,
+        })
+      }
     }
   }
 
@@ -754,10 +757,15 @@ export abstract class OpenAIBaseResponsesTextAdapter<
     } catch (error: unknown) {
       // Narrow before logging: raw SDK errors can carry request metadata
       // (including auth headers) which we must never surface to user loggers.
-      chatOptions.logger.errors(`${this.name}.structuredOutput fatal`, {
-        error: toRunErrorPayload(error, `${this.name}.structuredOutput failed`),
-        source: `${this.name}.structuredOutput`,
-      })
+      if (!this.isAbortError(error)) {
+        chatOptions.logger.errors(`${this.name}.structuredOutput fatal`, {
+          error: toRunErrorPayload(
+            error,
+            `${this.name}.structuredOutput failed`,
+          ),
+          source: `${this.name}.structuredOutput`,
+        })
+      }
       throw error
     }
   }
@@ -1247,6 +1255,9 @@ export abstract class OpenAIBaseResponsesTextAdapter<
     error: unknown,
     fallbackMessage: string,
   ): { message: string; code: string | undefined } {
+    if (this.isAbortError(error)) {
+      return { message: 'Request aborted', code: 'aborted' }
+    }
     const payload = toRunErrorPayload(error, fallbackMessage)
     return this.isContextOverflow(payload)
       ? { ...payload, code: CONTEXT_OVERFLOW_CODE }
@@ -1273,6 +1284,8 @@ export abstract class OpenAIBaseResponsesTextAdapter<
     if (!error || typeof error !== 'object') return false
     const e = error as { name?: unknown; code?: unknown }
     return (
+      // The openai client's APIUserAbortError leaves `name` as 'Error'.
+      error instanceof APIUserAbortError ||
       e.name === 'APIUserAbortError' ||
       e.name === 'AbortError' ||
       e.code === 'ERR_CANCELED'
@@ -2707,10 +2720,13 @@ export abstract class OpenAIBaseResponsesTextAdapter<
         error,
         `${this.name}.processStreamChunks failed`,
       )
-      options.logger.errors(`${this.name}.processStreamChunks fatal`, {
-        error: errorPayload,
-        source: `${this.name}.processStreamChunks`,
-      })
+      // A caller abort is not a failure, so it stays off the errors channel.
+      if (!this.isAbortError(error)) {
+        options.logger.errors(`${this.name}.processStreamChunks fatal`, {
+          error: errorPayload,
+          source: `${this.name}.processStreamChunks`,
+        })
+      }
       // A held refusal wins over a later stream error. A caller abort stays
       // an abort.
       if (pendingRunError && !this.isAbortError(error)) {

packages/openai-base/tests/abort-before-response.test.ts

@@ -0,0 +1,122 @@
+import { describe, expect, it, vi } from 'vitest'
+import OpenAI from 'openai'
+import { resolveDebugOption } from '@tanstack/ai/adapter-internals'
+import { createOpenAIClient } from '../src/client'
+import { OpenAIBaseChatCompletionsTextAdapter } from '../src/adapters/chat-completions-text'
+import { OpenAIBaseResponsesTextAdapter } from '../src/adapters/responses-text'
+import type { ClientOptions } from 'openai'
+import type { OpenAIClientLike } from '../src/client'
+
+type Fetch = NonNullable<ClientOptions['fetch']>
+
+const model = 'test-model'
+
+/** A fetch that never answers and rejects when the request aborts. */
+const hangingFetch: Fetch = (_url, init) =>
+  new Promise((_resolve, reject) => {
+    init?.signal?.addEventListener('abort', () =>
+      reject(init.signal?.reason ?? new Error('aborted')),
+    )
+  })
+
+class Responses extends OpenAIBaseResponsesTextAdapter<string> {}
+class Completions extends OpenAIBaseChatCompletionsTextAdapter<string> {}
+
+const clients = [
+  {
+    client: 'createOpenAIClient()',
+    make: () =>
+      createOpenAIClient({
+        apiKey: 'test-key',
+        maxRetries: 0,
+        fetch: hangingFetch,
+      }),
+  },
+  {
+    client: 'new OpenAI()',
+    make: () =>
+      new OpenAI({ apiKey: 'test-key', maxRetries: 0, fetch: hangingFetch }),
+  },
+]
+
+const apis = [
+  {
+    api: 'Responses',
+    adapter: (client: OpenAIClientLike) =>
+      new Responses(model, 'openai', client, { fetch: hangingFetch }),
+  },
+  {
+    api: 'Chat Completions',
+    adapter: (client: OpenAIClientLike) =>
+      new Completions(model, 'openai', client, { fetch: hangingFetch }),
+  },
+]
+
+const cases = clients.flatMap((c) => apis.map((a) => ({ ...c, ...a })))
+
+function errorsLogger() {
+  const error = vi.fn()
+  const logger = resolveDebugOption({
+    logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error },
+  })
+  return { error, logger }
+}
+
+describe.each(cases)(
+  '$api adapter on $client: a caller abort before the response',
+  ({ make, adapter }) => {
+    async function runAborted() {
+      const { error, logger } = errorsLogger()
+      const controller = new AbortController()
+      setTimeout(() => controller.abort(), 10)
+      const chunks = []
+      for await (const chunk of adapter(make()).chatStream({
+        threadId: 'thread-1',
+        logger,
+        model,
+        messages: [{ role: 'user', content: 'Hi' }],
+        request: { signal: controller.signal },
+      })) {
+        chunks.push(chunk)
+      }
+      return { chunks, error }
+    }
+
+    it('yields a RUN_ERROR classified as aborted', async () => {
+      const { chunks } = await runAborted()
+      expect(chunks.at(-1)).toMatchObject({
+        type: 'RUN_ERROR',
+        code: 'aborted',
+        metadata: { tanstack: { stopReason: 'aborted' } },
+      })
+    })
+
+    it('logs nothing on the errors channel', async () => {
+      const { error } = await runAborted()
+      expect(error.mock.calls).toEqual([])
+    })
+
+    it('structuredOutput logs nothing on the errors channel', async () => {
+      const { error, logger } = errorsLogger()
+      const controller = new AbortController()
+      setTimeout(() => controller.abort(), 10)
+      await expect(
+        adapter(make()).structuredOutput({
+          chatOptions: {
+            threadId: 'thread-1',
+            logger,
+            model,
+            messages: [{ role: 'user', content: 'Hi' }],
+            request: { signal: controller.signal },
+          },
+          outputSchema: {
+            type: 'object',
+            properties: { a: { type: 'string' } },
+            required: ['a'],
+          },
+        }),
+      ).rejects.toThrow('Request was aborted.')
+      expect(error.mock.calls).toEqual([])
+    })
+  },
+)

packages/openai-base/tests/caller-abort-logging.test.ts

@@ -0,0 +1,122 @@
+import { describe, expect, it, vi } from 'vitest'
+import { chat } from '@tanstack/ai'
+import { createOpenAIClient } from '../src/client'
+import { OpenAIBaseChatCompletionsTextAdapter } from '../src/adapters/chat-completions-text'
+import { OpenAIBaseResponsesTextAdapter } from '../src/adapters/responses-text'
+import type { ClientOptions } from 'openai'
+import type { Logger } from '@tanstack/ai'
+
+type Fetch = NonNullable<ClientOptions['fetch']>
+
+const model = 'test-model'
+
+/** An SSE body that sends `events`, then stays open until the request aborts. */
+function openStream(events: Array<Record<string, unknown>>): Fetch {
+  return async (_url, init) => {
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        for (const event of events) {
+          controller.enqueue(
+            new TextEncoder().encode(`data: ${JSON.stringify(event)}\n\n`),
+          )
+        }
+        init?.signal?.addEventListener('abort', () => {
+          controller.error(init.signal?.reason ?? new Error('aborted'))
+        })
+      },
+    })
+    return new Response(body, {
+      headers: { 'content-type': 'text/event-stream' },
+    })
+  }
+}
+
+class Responses extends OpenAIBaseResponsesTextAdapter<string> {}
+class Completions extends OpenAIBaseChatCompletionsTextAdapter<string> {}
+
+const adapters = [
+  {
+    name: 'Responses',
+    make: () => {
+      const fetch = openStream([
+        {
+          type: 'response.created',
+          response: { id: 'resp-1', model, status: 'in_progress' },
+        },
+        { type: 'response.output_text.delta', delta: 'The answer is ' },
+      ])
+      return new Responses(
+        model,
+        'openai',
+        createOpenAIClient({ apiKey: 'test-key', maxRetries: 0, fetch }),
+        { fetch },
+      )
+    },
+  },
+  {
+    name: 'Chat Completions',
+    make: () => {
+      const fetch = openStream([
+        {
+          id: 'chatcmpl-1',
+          object: 'chat.completion.chunk',
+          created: 1,
+          model,
+          choices: [
+            { index: 0, delta: { content: 'partial' }, finish_reason: null },
+          ],
+        },
+      ])
+      return new Completions(
+        model,
+        'openai',
+        createOpenAIClient({ apiKey: 'test-key', maxRetries: 0, fetch }),
+        { fetch },
+      )
+    },
+  },
+]
+
+describe.each(adapters)('$name: a caller abort mid-stream', ({ make }) => {
+  it('reaches onAbort and logs nothing on the errors channel', async () => {
+    const logger: Logger = {
+      debug: vi.fn(),
+      info: vi.fn(),
+      warn: vi.fn(),
+      error: vi.fn(),
+    }
+    const onAbort = vi.fn()
+    const onError = vi.fn()
+    const controller = new AbortController()
+
+    for await (const event of chat({
+      threadId: 'thread-1',
+      adapter: make(),
+      messages: [{ role: 'user', content: 'Hello' }],
+      abortController: controller,
+      // Default categories: only `errors` is on, as when `debug` is left out.
+      debug: {
+        logger,
+        provider: false,
+        output: false,
+        middleware: false,
+        tools: false,
+        agentLoop: false,
+        config: false,
+        request: false,
+        sandbox: false,
+      },
+      middleware: [{ name: 'observe', onAbort, onError }],
+    })) {
+      // Abort while the adapter waits for the next SSE event, as a user's
+      // Stop does between two tokens.
+      if (event.type === 'TEXT_MESSAGE_CONTENT') {
+        setTimeout(() => controller.abort(), 10)
+      }
+    }
+
+    expect(onAbort).toHaveBeenCalledTimes(1)
+    expect(onError).not.toHaveBeenCalled()
+    expect(vi.mocked(logger.error).mock.calls).toEqual([])
+  })
+})