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 } 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,