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
     }
   }
@@ -761,10 +770,12 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
         ...(lastUsage && { usage: lastUsage }),
       }
 
-      chatOptions.logger.errors(`${this.name}.structuredOutputStream fatal`, {
-        error: errorPayload,
-        source: `${this.name}.structuredOutputStream`,
-      })
+      if (!isAbort) {
+        chatOptions.logger.errors(`${this.name}.structuredOutputStream fatal`, {
+          error: errorPayload,
+          source: `${this.name}.structuredOutputStream`,
+        })
+      }
     }
   }
 
@@ -776,6 +787,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 }
@@ -802,8 +816,16 @@ export abstract class OpenAIBaseChatCompletionsTextAdapter<
    */
   protected isAbortError(error: unknown): boolean {
     if (!error || typeof error !== 'object') return false
-    const e = error as { name?: unknown; code?: unknown }
+    const e = error as {
+      name?: unknown
+      code?: unknown
+      constructor?: { name?: unknown }
+    }
     return (
+      // The openai client's APIUserAbortError leaves `name` as 'Error'.
+      error instanceof APIUserAbortError ||
+      // An injected client from another copy of `openai` has its own class.
+      e.constructor?.name === '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'
@@ -599,6 +600,15 @@ export abstract class OpenAIBaseResponsesTextAdapter<
       const rawEvent = toRunErrorRawEvent(error)
       const retryAfterMs = toRetryAfterMs(error)
 
+      // Log before yielding: a consumer that stops at RUN_ERROR closes this
+      // generator at the yield.
+      if (!this.isAbortError(error)) {
+        options.logger.errors(`${this.name}.chatStream fatal`, {
+          error: errorPayload,
+          source: `${this.name}.chatStream`,
+        })
+      }
+
       // Emit RUN_STARTED if not yet emitted
       if (!aguiState.hasEmittedRunStarted) {
         aguiState.hasEmittedRunStarted = true
@@ -637,11 +647,6 @@ export abstract class OpenAIBaseResponsesTextAdapter<
           code: errorPayload.code,
         },
       }
-
-      options.logger.errors(`${this.name}.chatStream fatal`, {
-        error: errorPayload,
-        source: `${this.name}.chatStream`,
-      })
     }
   }
 
@@ -754,10 +759,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
     }
   }
@@ -1195,10 +1205,12 @@ export abstract class OpenAIBaseResponsesTextAdapter<
         },
       }
 
-      chatOptions.logger.errors(`${this.name}.structuredOutputStream fatal`, {
-        error: errorPayload,
-        source: `${this.name}.structuredOutputStream`,
-      })
+      if (!isAbort) {
+        chatOptions.logger.errors(`${this.name}.structuredOutputStream fatal`, {
+          error: errorPayload,
+          source: `${this.name}.structuredOutputStream`,
+        })
+      }
     }
   }
 
@@ -1247,6 +1259,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 }
@@ -1271,8 +1286,16 @@ export abstract class OpenAIBaseResponsesTextAdapter<
    */
   protected isAbortError(error: unknown): boolean {
     if (!error || typeof error !== 'object') return false
-    const e = error as { name?: unknown; code?: unknown }
+    const e = error as {
+      name?: unknown
+      code?: unknown
+      constructor?: { name?: unknown }
+    }
     return (
+      // The openai client's APIUserAbortError leaves `name` as 'Error'.
+      error instanceof APIUserAbortError ||
+      // An injected client from another copy of `openai` has its own class.
+      e.constructor?.name === 'APIUserAbortError' ||
       e.name === 'APIUserAbortError' ||
       e.name === 'AbortError' ||
       e.code === 'ERR_CANCELED'
@@ -2707,10 +2730,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,177 @@
+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 }),
+  },
+]
+
+describe.each(apis)('$api adapter: isAbortError', ({ adapter }) => {
+  it('recognizes an APIUserAbortError from another copy of openai', () => {
+    class APIUserAbortError extends Error {}
+    const isAbortError = (
+      adapter(createOpenAIClient({ apiKey: 'test-key' })) as unknown as {
+        isAbortError: (error: unknown) => boolean
+      }
+    ).isAbortError(new APIUserAbortError('Request was aborted.'))
+    expect(isAbortError).toBe(true)
+  })
+
+  it('returns false for an object with no prototype', () => {
+    const isAbortError = (
+      adapter(createOpenAIClient({ apiKey: 'test-key' })) as unknown as {
+        isAbortError: (error: unknown) => boolean
+      }
+    ).isAbortError(Object.create(null))
+    expect(isAbortError).toBe(false)
+  })
+})
+
+const failingFetch: Fetch = async () =>
+  new Response(JSON.stringify({ error: { message: 'boom' } }), {
+    status: 500,
+    headers: { 'content-type': 'application/json' },
+  })
+
+describe.each([
+  { api: 'Responses', Adapter: Responses },
+  { api: 'Chat Completions', Adapter: Completions },
+])('$api adapter: a request failure', ({ Adapter }) => {
+  it('logs fatal when the consumer stops at RUN_ERROR', async () => {
+    const { error, logger } = errorsLogger()
+    const client = createOpenAIClient({
+      apiKey: 'test-key',
+      maxRetries: 0,
+      fetch: failingFetch,
+    })
+    for await (const chunk of new Adapter(model, 'openai', client, {
+      fetch: failingFetch,
+    }).chatStream({
+      threadId: 'thread-1',
+      logger,
+      model,
+      messages: [{ role: 'user', content: 'Hi' }],
+    })) {
+      if (chunk.type === 'RUN_ERROR') break
+    }
+    expect(error).toHaveBeenCalledWith(
+      expect.stringContaining('.chatStream fatal'),
+      expect.anything(),
+    )
+  })
+})
+
+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([])
+    })
+  },
+)