From 515f33a11711b5ad367c3458c209d732afecb467 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 22:35:37 +0000 Subject: [PATCH 1/2] Add onError callback to durable agent stream calls The DurableAgent.stream() options expose an onError callback that fires on streaming errors. Wire it up for the invocation and heartbeat handlers to log the error and emit a best-effort activity breadcrumb. The realtime ToolLoopAgent runner does not expose onError, so it is left unchanged. --- .../workflows/session/handlers/handle-heartbeat.ts | 9 +++++++++ .../workflows/session/handlers/handle-invocation.ts | 9 +++++++++ 2 files changed, 18 insertions(+) diff --git a/packages/ai/agent-runtime/workflows/session/handlers/handle-heartbeat.ts b/packages/ai/agent-runtime/workflows/session/handlers/handle-heartbeat.ts index a00f1d56..4a7d7862 100644 --- a/packages/ai/agent-runtime/workflows/session/handlers/handle-heartbeat.ts +++ b/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({ diff --git a/packages/ai/agent-runtime/workflows/session/handlers/handle-invocation.ts b/packages/ai/agent-runtime/workflows/session/handlers/handle-invocation.ts index a4488501..ab98e69a 100644 --- a/packages/ai/agent-runtime/workflows/session/handlers/handle-invocation.ts +++ b/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, From 22e3de6718dd1ed66f38cb2f5d65af2e0cab9084 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 8 Jun 2026 22:53:19 +0000 Subject: [PATCH 2/2] Intercept ToolLoopAgent stream errors and surface human-readable messages ToolLoopAgent.stream() does not expose an onError option, so non-fatal errors arrive as 'error' parts on the fullStream rather than thrown. Intercept them in the tap, log them, and swallow the raw chunk so channels never render provider internals. Add a shared humanReadableAgentError() helper that maps known errors (missing credentials, provider 401/403/429/5xx, aborts) to concise user-facing text and falls back to a generic message instead of leaking raw provider output. Wire it into the web chat UI streams (createUIMessageStream + createAgentUIStream onError, replacing the SDK default 'An error occurred.') and the channel text-only path (posted via postText). --- .../server/human-readable-error.test.ts | 52 +++++++++ .../server/human-readable-error.ts | 46 ++++++++ .../server/realtime-chat-runner.test.ts | 103 +++++++++++++++++- .../server/realtime-chat-runner.ts | 59 +++++++++- 4 files changed, 255 insertions(+), 5 deletions(-) create mode 100644 packages/ai/agent-runtime/server/human-readable-error.test.ts create mode 100644 packages/ai/agent-runtime/server/human-readable-error.ts diff --git a/packages/ai/agent-runtime/server/human-readable-error.test.ts b/packages/ai/agent-runtime/server/human-readable-error.test.ts new file mode 100644 index 00000000..bbe1ac0e --- /dev/null +++ b/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.' + ) + }) +}) diff --git a/packages/ai/agent-runtime/server/human-readable-error.ts b/packages/ai/agent-runtime/server/human-readable-error.ts new file mode 100644 index 00000000..16258392 --- /dev/null +++ b/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' +} diff --git a/packages/ai/agent-runtime/server/realtime-chat-runner.test.ts b/packages/ai/agent-runtime/server/realtime-chat-runner.test.ts index 68fc534c..d3e76a04 100644 --- a/packages/ai/agent-runtime/server/realtime-chat-runner.test.ts +++ b/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>[] + 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>[] = [] + 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>, void, unknown> { for (const chunk of chunks) { await Promise.resolve() diff --git a/packages/ai/agent-runtime/server/realtime-chat-runner.ts b/packages/ai/agent-runtime/server/realtime-chat-runner.ts index 72b6e5bd..2e138471 100644 --- a/packages/ai/agent-runtime/server/realtime-chat-runner.ts +++ b/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>>, - accumulator: { text: string } + accumulator: { text: string }, + onError?: (error: unknown) => void ): AsyncIterable>> { - 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, + 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>>, - accumulator: { text: string } + accumulator: { text: string }, + onError?: (error: unknown) => void ): AsyncGenerator>, 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 }