diff --git a/apps/web/src/lib/ai-gateway/processUsage.shared.ts b/apps/web/src/lib/ai-gateway/processUsage.shared.ts index e33c5180e8..f21734b5e4 100644 --- a/apps/web/src/lib/ai-gateway/processUsage.shared.ts +++ b/apps/web/src/lib/ai-gateway/processUsage.shared.ts @@ -1,6 +1,7 @@ import { captureMessage } from '@sentry/nextjs'; import type { Span } from '@sentry/nextjs'; import { toMicrodollars } from '../utils'; +import { errorExceptInTest } from '@/lib/utils.server'; import { OPENROUTER_BYOK_COST_MULTIPLIER } from '@/lib/ai-gateway/processUsage.constants'; import type { NotYetCostedUsageStats, @@ -91,8 +92,8 @@ export function isResponseInterruptedError(error: unknown): boolean { /** * Drains a ReadableStream of binary chunks, calling `onTextChunk` for each - * decoded piece of text. Handles client aborts and upstream timeouts gracefully - * and always releases the reader lock and ends `streamProcessingSpan`. + * decoded piece of text. Handles stream processing failures gracefully and + * always releases the reader lock and ends `streamProcessingSpan`. * * Returns `true` if the stream was aborted before completion. */ @@ -111,11 +112,8 @@ export async function drainSseStream( onTextChunk(decoder.decode(value, { stream: true })); } } catch (error) { - if (isResponseInterruptedError(error)) { - wasAborted = true; - } else { - throw error; - } + errorExceptInTest('[processUsage] treating stream processing error as aborted', error); + wasAborted = true; } finally { reader.releaseLock(); streamProcessingSpan.end(); diff --git a/apps/web/src/lib/ai-gateway/processUsage.test.ts b/apps/web/src/lib/ai-gateway/processUsage.test.ts index 3710633f89..6e3a203dd2 100644 --- a/apps/web/src/lib/ai-gateway/processUsage.test.ts +++ b/apps/web/src/lib/ai-gateway/processUsage.test.ts @@ -175,15 +175,30 @@ describe('parseMicrodollarUsageFromStream approval tests', () => { await verifyApproval(resultString, approvalFilePath); }); - test.each(['ResponseAborted', 'TimeoutError'])( + const interruptedStreamErrors: [name: string, error: Error][] = [ + [ + 'ResponseAborted', + Object.assign(new Error('Response interrupted'), { name: 'ResponseAborted' }), + ], + ['TimeoutError', Object.assign(new Error('Response interrupted'), { name: 'TimeoutError' })], + [ + 'Undici body timeout', + new TypeError('terminated', { + cause: Object.assign(new Error('Body Timeout Error'), { + name: 'BodyTimeoutError', + code: 'UND_ERR_BODY_TIMEOUT', + }), + }), + ], + ['unexpected read failure', new Error('Unexpected stream failure')], + ]; + + test.each(interruptedStreamErrors)( 'handles %s gracefully and returns partial data', - async errorName => { + async (_name, streamError) => { // Create a stream that emits some SSE data then fails. const partialSSEData = `data: {"id":"gen-123","model":"anthropic/claude-3-5-sonnet","choices":[{"delta":{"content":"Hello"}}]}\n\ndata: {"id":"gen-123","model":"anthropic/claude-3-5-sonnet","choices":[{"delta":{"content":" world"}}]}\n\n`; - const streamError = new Error('Response interrupted'); - streamError.name = errorName; - let pullCount = 0; const stream = new ReadableStream({ pull(controller) { @@ -211,6 +226,25 @@ describe('parseMicrodollarUsageFromStream approval tests', () => { } ); + test('handles SSE parser errors as aborted streams', async () => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: not-json\n\n')); + controller.close(); + }, + }); + + const result = await parseMicrodollarUsageFromStream( + stream, + 'fake-user-id', + undefined, + 'openrouter', + 200 + ); + + expect(result.hasError).toBe(true); + }); + test.each([ ['chat_completions', 'ResponseAborted'], ['chat_completions', 'TimeoutError'],