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

apps/web/src/lib/ai-gateway/processUsage.shared.ts

@@ -91,8 +91,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 response body read failures gracefully and
+ * always releases the reader lock and ends `streamProcessingSpan`.
  *
  * Returns `true` if the stream was aborted before completion.
  */
@@ -106,16 +106,18 @@ export async function drainSseStream(
   let wasAborted = false;
   try {
     while (true) {
-      const { done, value } = await reader.read();
+      let readResult: ReadableStreamReadResult<Uint8Array>;
+      try {
+        readResult = await reader.read();
+      } catch {
+        wasAborted = true;
+        break;
+      }
+
+      const { done, value } = readResult;
       if (done) break;
       onTextChunk(decoder.decode(value, { stream: true }));
     }
-  } catch (error) {
-    if (isResponseInterruptedError(error)) {
-      wasAborted = true;
-    } else {
-      throw error;
-    }
   } finally {
     reader.releaseLock();
     streamProcessingSpan.end();

apps/web/src/lib/ai-gateway/processUsage.test.ts

@@ -175,15 +175,29 @@ 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',
+        }),
+      }),
+    ],
+  ];
+
+  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 +225,19 @@ describe('parseMicrodollarUsageFromStream approval tests', () => {
     }
   );
 
+  test('does not swallow SSE parser errors', async () => {
+    const stream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('data: not-json\n\n'));
+        controller.close();
+      },
+    });
+
+    await expect(
+      parseMicrodollarUsageFromStream(stream, 'fake-user-id', undefined, 'openrouter', 200)
+    ).rejects.toThrow(SyntaxError);
+  });
+
   test.each([
     ['chat_completions', 'ResponseAborted'],
     ['chat_completions', 'TimeoutError'],