Add onError callback to durable agent stream calls

packages/ai/agent-runtime/server/human-readable-error.test.ts

@@ -0,0 +1,52 @@
+import { MissingInferenceCredentialError } from '@outname/shared/server/inference-provider-errors'
+import { APICallError } from 'ai'
+import { describe, expect, it, vi } from 'vitest'
+import { humanReadableAgentError } from './human-readable-error'
+
+vi.mock('server-only', () => ({}))
+
+function apiError(statusCode?: number): APICallError {
+  return new APICallError({
+    isRetryable: false,
+    message: 'raw provider message',
+    requestBodyValues: {},
+    statusCode,
+    url: 'https://example.test/v1/messages',
+  })
+}
+
+describe('humanReadableAgentError', () => {
+  it('passes through the already user-facing missing-credential message', () => {
+    const error = new MissingInferenceCredentialError('vercel-ai-gateway')
+    expect(humanReadableAgentError(error)).toBe(error.message)
+  })
+
+  it('maps auth status codes to a credentials hint', () => {
+    expect(humanReadableAgentError(apiError(401))).toContain('API key')
+    expect(humanReadableAgentError(apiError(403))).toContain('API key')
+  })
+
+  it('maps 429 to a rate-limit message', () => {
+    expect(humanReadableAgentError(apiError(429))).toContain('rate limiting')
+  })
+
+  it('maps 5xx to a temporary-error message', () => {
+    expect(humanReadableAgentError(apiError(503))).toContain('temporary error')
+  })
+
+  it('never leaks the raw provider message for unknown errors', () => {
+    const message = humanReadableAgentError(new Error('stack trace: secret'))
+    expect(message).not.toContain('secret')
+    expect(message).toBe(
+      'Something went wrong while the agent was responding. Please try again in a moment.'
+    )
+  })
+
+  it('reports aborted responses distinctly', () => {
+    const aborted = new Error('aborted')
+    aborted.name = 'AbortError'
+    expect(humanReadableAgentError(aborted)).toBe(
+      'The response was stopped before it finished.'
+    )
+  })
+})

packages/ai/agent-runtime/server/human-readable-error.ts

@@ -0,0 +1,46 @@
+import 'server-only'
+import { MissingInferenceCredentialError } from '@outname/shared/server/inference-provider-errors'
+import { APICallError } from 'ai'
+
+const GENERIC_MESSAGE =
+  'Something went wrong while the agent was responding. Please try again in a moment.'
+
+/**
+ * Maps an arbitrary error — thrown or streamed during an agent run — to a
+ * concise, user-facing message.
+ *
+ * Raw provider/internal errors are intentionally never surfaced verbatim: they
+ * tend to leak request bodies, stack traces, or provider jargon. Callers are
+ * expected to log the original error separately for debugging.
+ */
+export function humanReadableAgentError(error: unknown): string {
+  if (error instanceof MissingInferenceCredentialError) {
+    // Already written for the end user ("Add your key in Settings…").
+    return error.message
+  }
+  if (APICallError.isInstance(error)) {
+    return messageForApiCallError(error)
+  }
+  if (isAbortError(error)) {
+    return 'The response was stopped before it finished.'
+  }
+  return GENERIC_MESSAGE
+}
+
+function messageForApiCallError(error: APICallError): string {
+  const status = error.statusCode
+  if (status === 401 || status === 403) {
+    return 'Your inference provider rejected the request. Check that your API key is still valid in Settings.'
+  }
+  if (status === 429) {
+    return 'The inference provider is rate limiting requests right now. Please wait a moment and try again.'
+  }
+  if (status !== undefined && status >= 500) {
+    return 'The inference provider had a temporary error. Please try again in a moment.'
+  }
+  return 'The inference provider could not complete the request. Please try again.'
+}
+
+function isAbortError(error: unknown): boolean {
+  return error instanceof Error && error.name === 'AbortError'
+}

packages/ai/agent-runtime/server/realtime-chat-runner.test.ts

@@ -90,6 +90,38 @@ describe('tapFullStream', () => {
     expect(accumulator.text).toBe('hello')
   })
 
+  it('intercepts error chunks via onError and swallows them from the output', async () => {
+    const failure = new Error('provider exploded')
+    const chunks = [
+      { type: 'text-delta', text: 'partial' },
+      { type: 'error', error: failure },
+      { type: 'text-delta', text: ' more' },
+    ] as TextStreamPart<Record<string, Tool>>[]
+    const source = (async function* () {
+      for (const chunk of chunks) {
+        await Promise.resolve()
+        yield chunk
+      }
+    })()
+    const accumulator = { text: '' }
+    const seen: unknown[] = []
+
+    const { tapFullStream } = await import('./realtime-chat-runner')
+    const output: TextStreamPart<Record<string, Tool>>[] = []
+    for await (const chunk of tapFullStream(source, accumulator, (error) => {
+      seen.push(error)
+    })) {
+      output.push(chunk)
+    }
+
+    expect(seen).toEqual([failure])
+    expect(output.map((chunk) => chunk.type)).toEqual([
+      'text-delta',
+      'text-delta',
+    ])
+    expect(accumulator.text).toBe('partial more')
+  })
+
   it('propagates upstream stream errors after preserving accumulated text', async () => {
     const accumulator = { text: '' }
     const source = (async function* () {
@@ -389,6 +421,73 @@ describe('realtime chat runner persistence policy', () => {
     })
   })
 
+  it('posts a human-readable notice and hides raw error chunks on the channel', async () => {
+    mocks.preflightBudget.mockResolvedValue(null)
+    mocks.buildAgentRuntimeSpec.mockResolvedValue(runtimeSpec())
+    mocks.buildRealtimeAgentRuntime.mockImplementation(
+      async (
+        _spec: AgentRuntimeSpec,
+        options: { onFinish?: (event: unknown) => void }
+      ) => {
+        await Promise.resolve()
+        options.onFinish?.({ generations: [testGeneration()], steps: [] })
+        return {
+          agent: {
+            stream: async () => ({
+              fullStream: streamFromChunks([
+                { text: 'partial', type: 'text-delta' },
+                { type: 'error', error: new Error('provider exploded') },
+              ]),
+            }),
+          },
+          meta: {
+            model: 'openai/gpt-5.1',
+            name: 'Agent',
+            stepLimitCustom: null,
+            stepLimitMode: 'medium',
+            userId: 'user_123',
+          },
+          tools: {},
+        }
+      }
+    )
+    const forwarded: Array<{ type: string }> = []
+    const delivery = buildDelivery({
+      postAgentStream: async (stream) => {
+        for await (const chunk of stream) {
+          forwarded.push(chunk as { type: string })
+        }
+      },
+    })
+    const errorSpy = vi
+      .spyOn(console, 'error')
+      .mockImplementation(() => undefined)
+
+    const { runRealtimeChatTurn } = await import('./realtime-chat-runner')
+    await runRealtimeChatTurn({
+      abortSignal: new AbortController().signal,
+      agentId: 'agent_123',
+      assistantMessageId: 'msg_assistant',
+      conversationId: 'conv_123',
+      delivery,
+      messages: [] satisfies ModelMessage[],
+      persistMode: 'text-only',
+      runId: 'rt_123',
+      source: 'slack',
+      titleMessages: [userMessage('msg_user', 'hello')],
+      userId: 'user_123',
+    })
+
+    // Raw provider error never reaches the channel stream.
+    expect(forwarded.map((chunk) => chunk.type)).toEqual(['text-delta'])
+    // A readable notice is posted instead.
+    expect(delivery.postText).toHaveBeenCalledWith(
+      'Something went wrong while the agent was responding. Please try again in a moment.'
+    )
+    expect(errorSpy).toHaveBeenCalled()
+    errorSpy.mockRestore()
+  })
+
   it('contains usage recording failures inside the scheduled task', async () => {
     mocks.preflightBudget.mockResolvedValue(null)
     mocks.buildAgentRuntimeSpec.mockResolvedValue(runtimeSpec())
@@ -584,7 +683,9 @@ function testGeneration() {
 }
 
 async function* streamFromChunks(
-  chunks: Array<{ text: string; type: 'text-delta' }>
+  chunks: Array<
+    { text: string; type: 'text-delta' } | { type: 'error'; error: unknown }
+  >
 ): AsyncGenerator<TextStreamPart<Record<string, Tool>>, void, unknown> {
   for (const chunk of chunks) {
     await Promise.resolve()

packages/ai/agent-runtime/server/realtime-chat-runner.ts

@@ -45,6 +45,7 @@ import {
   preflightBudget,
   recordTokenUsageStep,
 } from '../workflows/session/steps/budget'
+import { humanReadableAgentError } from './human-readable-error'
 import { buildRealtimeAgentRuntime } from './realtime-agent-runtime'
 
 export type RealtimePersistMode = 'ui-message-full' | 'text-only'
@@ -120,15 +121,32 @@ function createRealtimeUiMessageResponse(
         }
       })
     },
+    // Without this, the AI SDK surfaces a generic "An error occurred." to the UI.
+    onError: (error) =>
+      reportRealtimeStreamError(
+        { agentId: input.agentId, conversationId: input.conversationId },
+        error
+      ),
   })
   return createUIMessageStreamResponse({ stream })
 }
 
 export function tapFullStream(
   stream: AsyncIterable<TextStreamPart<Record<string, Tool>>>,
-  accumulator: { text: string }
+  accumulator: { text: string },
+  onError?: (error: unknown) => void
 ): AsyncIterable<TextStreamPart<Record<string, Tool>>> {
-  return tapFullStreamGenerator(stream, accumulator)
+  return tapFullStreamGenerator(stream, accumulator, onError)
+}
+
+// Logs a stream error and returns a user-facing message. Shared by the web UI
+// stream error handlers and the channel (text-only) path.
+function reportRealtimeStreamError(
+  context: Record<string, unknown>,
+  error: unknown
+): string {
+  console.error('[realtime-chat] agent stream error', { ...context, error })
+  return humanReadableAgentError(error)
 }
 
 async function runRealtimeChatTurnInsideContext(
@@ -246,6 +264,12 @@ async function streamUiMessageTurn(input: {
     // sub-agent outputs to the model. We only persist the new responseMessage.
     originalMessages: turn.messages as never,
     generateMessageId: () => `msg_${nanoid(12)}`,
+    // Replace the SDK default ("An error occurred.") with a readable message.
+    onError: (error) =>
+      reportRealtimeStreamError(
+        { agentId: turn.agentId, conversationId: turn.conversationId },
+        error
+      ),
     onFinish: async ({ responseMessage, isAborted, finishReason }) => {
       await handleUiMessageFinish({
         agentId: turn.agentId,
@@ -304,10 +328,22 @@ async function runTextOnlyTurn(input: {
     abortSignal: turn.abortSignal,
   })
   const accumulator = { text: '' }
+  const streamError: { value: unknown } = { value: null }
 
   try {
     await turn.delivery.postAgentStream(
-      tapFullStream(result.fullStream, accumulator)
+      tapFullStream(result.fullStream, accumulator, (error) => {
+        streamError.value = error
+        console.error('[realtime-chat] agent stream error', {
+          agentId: turn.agentId,
+          conversationId: turn.conversationId,
+          error,
+          externalScopeId: turn.externalScopeId,
+          externalThreadId: turn.externalThreadId,
+          runId: turn.runId,
+          source: turn.source,
+        })
+      })
     )
   } catch (err) {
     console.error('[realtime-chat] channel stream failed', {
@@ -322,6 +358,13 @@ async function runTextOnlyTurn(input: {
     throw err
   }
 
+  // A non-fatal stream error was intercepted (and swallowed) by the tap above.
+  // Surface a human-readable notice to the channel; any text streamed before
+  // the error has already been delivered and is still persisted below.
+  if (streamError.value !== null) {
+    await turn.delivery.postText?.(humanReadableAgentError(streamError.value))
+  }
+
   let assistantText = accumulator.text.trim()
   const finishEvent = finishState.event
   if (
@@ -565,11 +608,19 @@ const missingRealtimeSubAgentTool: BuildAgentTool = () => {
 
 async function* tapFullStreamGenerator(
   stream: AsyncIterable<TextStreamPart<Record<string, Tool>>>,
-  accumulator: { text: string }
+  accumulator: { text: string },
+  onError?: (error: unknown) => void
 ): AsyncGenerator<TextStreamPart<Record<string, Tool>>, void, unknown> {
   for await (const chunk of stream) {
     if (chunk.type === 'text-delta') {
       accumulator.text += chunk.text
+    } else if (chunk.type === 'error') {
+      // ToolLoopAgent.stream() has no onError option: non-fatal errors surface
+      // as `error` parts on the fullStream rather than thrown. Intercept them
+      // here and swallow the raw chunk so the channel never renders provider
+      // internals — the caller posts a human-readable notice instead.
+      onError?.(chunk.error)
+      continue
     }
     yield chunk
   }

packages/ai/agent-runtime/workflows/session/handlers/handle-heartbeat.ts

@@ -114,6 +114,15 @@ export async function handleHeartbeat(input: {
       messages: [{ role: 'user', content: kickoff }],
       writable,
       stopWhen: resolveStepLimit(stepLimitInput),
+      onError: async ({ error }) => {
+        const message = error instanceof Error ? error.message : String(error)
+        console.error('handleHeartbeat: stream error', error)
+        await emitActivity(runId, activityMessage(mode, 'Stream error'), {
+          message,
+        }).catch(() => {
+          // Best-effort breadcrumb; the surrounding catch finalizes the run.
+        })
+      },
     })
     if (budgetCheck.userId) {
       await recordTokenUsageStep({

packages/ai/agent-runtime/workflows/session/handlers/handle-invocation.ts

@@ -140,6 +140,15 @@ export async function handleInvocation(input: {
       collectUIMessages: true,
       preventClose: true,
       sendFinish: false,
+      onError: async ({ error }) => {
+        const message = error instanceof Error ? error.message : String(error)
+        console.error('handleInvocation: stream error', error)
+        await emitActivity(runId, 'Sub-agent: Stream error', { message }).catch(
+          () => {
+            // Best-effort breadcrumb; the surrounding catch finalizes the run.
+          }
+        )
+      },
     })
     await recordTokenUsageStep({
       userId: built.meta.userId,