fix(ai-gateway): continue usage processing after stream errors

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();

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<Uint8Array>({
         pull(controller) {
@@ -211,6 +226,25 @@ describe('parseMicrodollarUsageFromStream approval tests', () => {
     }
   );
 
+  test('handles SSE parser errors as aborted streams', async () => {
+    const stream = new ReadableStream<Uint8Array>({
+      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'],