fix(ai-router): handle stream error failover

Author: moonrailgun

src/server/model/aiRouter.spec.ts

@@ -645,6 +645,43 @@ describe('AI Router buffered attempt mapping', () => {
     });
   });
 
+  test('converts buffered stream error payload into an uncommitted retryable failure', () => {
+    const result = buildBufferedAIGatewayAttemptResult({
+      protocol: AI_ROUTER_PROTOCOLS.OPENAI_CHAT,
+      failOnEmptyContent: true,
+      gatewayId: 'gw-stream-error',
+      logId: 'log-stream-error',
+      response: {
+        statusCode: 200,
+        headers: {
+          'content-type': 'text/event-stream',
+        },
+        chunks: [
+          Buffer.from(AI_GATEWAY_STREAM_PING_COMMENT),
+          Buffer.from(
+            'data: {"error":{"message":"LLM returned empty response from stream","type":"server_error"}}\n\n'
+          ),
+          Buffer.from('data: [DONE]\n\n'),
+        ],
+        wroteBody: true,
+        bodyStartedBeforeFailure: false,
+        ended: true,
+      },
+    });
+
+    expect(result).toMatchObject({
+      ok: false,
+      committed: false,
+      gatewayId: 'gw-stream-error',
+      logId: 'log-stream-error',
+      statusCode: 502,
+      failure: {
+        message: 'LLM returned empty response from stream',
+        errorType: 'upstream',
+      },
+    });
+  });
+
   test('detects empty OpenAI chat content while allowing tool calls', () => {
     expect(
       inspectAIRouterBufferedResponseContent(

src/server/model/aiRouter.ts

@@ -1219,7 +1219,8 @@ function buildAIRouterRuntimeHandler(args: {
           writeAIRouterStreamError(
             res,
             args.protocol,
-            getBufferedFailureMessage(result.response),
+            result.failure?.message ??
+              getBufferedFailureMessage(result.response),
             'router_failed'
           );
           return;
@@ -1592,13 +1593,22 @@ export function buildBufferedAIGatewayAttemptResult(args: {
     };
   }
 
+  const streamFailureMessage = readSSEErrorMessage(args.response.chunks);
   const statusCode =
     args.response.statusCode >= 200 &&
     args.response.statusCode < 300 &&
     partialFailure
       ? 500
-      : args.response.statusCode;
-  const ok = statusCode >= 200 && statusCode < 300 && !partialFailure;
+      : args.response.statusCode >= 200 &&
+          args.response.statusCode < 300 &&
+          streamFailureMessage
+        ? 502
+        : args.response.statusCode;
+  const ok =
+    statusCode >= 200 &&
+    statusCode < 300 &&
+    !partialFailure &&
+    !streamFailureMessage;
   const emptyContentFailure =
     ok &&
     args.failOnEmptyContent === true &&
@@ -1631,7 +1641,8 @@ export function buildBufferedAIGatewayAttemptResult(args: {
       : {
           message: partialFailure
             ? 'AI Gateway attempt ended after partial output'
-            : getBufferedFailureMessage(args.response),
+            : streamFailureMessage ??
+              getBufferedFailureMessage(args.response),
           errorType: 'upstream',
         },
     response: args.response,
@@ -1863,6 +1874,12 @@ function getBufferedFailureMessage(snapshot: BufferedResponseSnapshot): string {
     return jsonMessage;
   }
 
+  const sseMessage = readSSEErrorMessage(snapshot.chunks);
+
+  if (sseMessage) {
+    return sseMessage;
+  }
+
   const text = Buffer.concat(snapshot.chunks).toString('utf8').trim();
 
   return text || `AI Gateway attempt failed with status ${snapshot.statusCode}`;
@@ -1883,6 +1900,47 @@ function readJsonErrorMessage(value: unknown): string | null {
   return typeof message === 'string' && message ? message : null;
 }
 
+function readSSEErrorMessage(chunks: Buffer[]): string | null {
+  for (const event of parseAIRouterSSEEvents(chunks)) {
+    const message = readSSEEventErrorMessage(event);
+
+    if (message) {
+      return message;
+    }
+  }
+
+  return null;
+}
+
+function readSSEEventErrorMessage(value: unknown): string | null {
+  if (!isAIRouterRecord(value)) {
+    return null;
+  }
+
+  const error = value.error;
+
+  if (typeof error === 'string' && error) {
+    return error;
+  }
+
+  if (isAIRouterRecord(error)) {
+    const nestedError = error.error;
+    const message =
+      error.message ??
+      (isAIRouterRecord(nestedError) ? nestedError.message : undefined);
+
+    if (typeof message === 'string' && message) {
+      return message;
+    }
+  }
+
+  if (value.type === 'error' && typeof value.message === 'string') {
+    return value.message;
+  }
+
+  return null;
+}
+
 export interface AIRouterBufferedContentInspection {
   parsed: boolean;
   empty: boolean;