fix: handle streamed provider errors

apps/gateway/src/api.spec.ts

@@ -985,6 +985,62 @@ describe("api", () => {
 			expect(logs[0].errorDetails?.statusText).toBe("TimeoutError");
 		}, 15000);
 
+		test("streaming request surfaces truncated upstream streams", async () => {
+			await db.insert(tables.apiKey).values({
+				id: "token-id",
+				token: "real-token",
+				projectId: "project-id",
+				description: "Test API Key",
+				createdBy: "user-id",
+			});
+
+			await db.insert(tables.providerKey).values({
+				id: "provider-key-id",
+				token: "sk-test-key",
+				provider: "llmgateway",
+				organizationId: "org-id",
+				baseUrl: mockServerUrl,
+			});
+
+			const res = await app.request("/v1/chat/completions", {
+				method: "POST",
+				headers: {
+					"Content-Type": "application/json",
+					Authorization: `Bearer real-token`,
+				},
+				body: JSON.stringify({
+					model: "llmgateway/custom",
+					messages: [
+						{
+							role: "user",
+							content: "TRIGGER_TRUNCATED_STREAM",
+						},
+					],
+					stream: true,
+				}),
+			});
+
+			expect(res.status).toBe(200);
+
+			const streamResult = await readAll(res.body);
+
+			expect(streamResult.hasContent).toBe(true);
+			expect(streamResult.hasError).toBe(true);
+			expect(streamResult.errorEvents.length).toBeGreaterThan(0);
+			expect(streamResult.errorEvents[0].error.type).toBe("upstream_error");
+			expect(streamResult.errorEvents[0].error.code).toBe("stream_truncated");
+
+			const logs = await waitForLogs(1);
+			expect(logs.length).toBe(1);
+			expect(logs[0].finishReason).toBe("upstream_error");
+			expect(logs[0].unifiedFinishReason).toBe("upstream_error");
+			expect(logs[0].hasError).toBe(true);
+			expect(logs[0].errorDetails?.statusCode).toBe(502);
+			expect(logs[0].errorDetails?.statusText).toBe(
+				"Upstream Stream Terminated",
+			);
+		});
+
 		test("request with short delay under timeout succeeds", async () => {
 			await db.insert(tables.apiKey).values({
 				id: "token-id",

apps/gateway/src/chat/chat.ts

@@ -5150,6 +5150,76 @@ chat.openapi(completions, async (c) => {
 								estimateTokensFromContent(fullReasoningContent);
 						}
 					}
+
+					const streamEndedWithoutTerminalEvent =
+						!streamingError && !canceled && finishReason === null;
+					if (streamEndedWithoutTerminalEvent) {
+						const responseText =
+							buffer.trim().length > 0
+								? buffer.substring(0, 5000)
+								: "Stream ended before a terminal finish reason or [DONE] event";
+						const errorMessage =
+							"Upstream stream terminated unexpectedly before completion";
+
+						logger.warn("[streaming] Stream ended without terminal event", {
+							provider: usedProvider,
+							model: usedModel,
+							bufferLength: buffer.length,
+							fullContentLength: fullContent.length,
+							hasToolCalls:
+								!!streamingToolCalls && streamingToolCalls.length > 0,
+							unifiedFinishReason: getUnifiedFinishReason(
+								"upstream_error",
+								usedProvider,
+							),
+						});
+
+						streamingError = {
+							message: errorMessage,
+							type: "upstream_error",
+							code: "stream_truncated",
+							details: {
+								statusCode: 502,
+								statusText: "Upstream Stream Terminated",
+								responseText,
+								timestamp: new Date().toISOString(),
+								provider: usedProvider,
+								model: usedModel,
+								bufferLength: buffer.length,
+							},
+						};
+						finishReason = "upstream_error";
+
+						try {
+							await writeSSEAndCache({
+								event: "error",
+								data: JSON.stringify({
+									error: {
+										message: errorMessage,
+										type: "upstream_error",
+										code: "stream_truncated",
+										param: null,
+										responseText,
+									},
+								}),
+								id: String(eventId++),
+							});
+							await writeSSEAndCache({
+								event: "done",
+								data: "[DONE]",
+								id: String(eventId++),
+							});
+							doneSent = true;
+						} catch (sseError) {
+							logger.error(
+								"Failed to send truncated stream error SSE",
+								sseError instanceof Error
+									? sseError
+									: new Error(String(sseError)),
+							);
+						}
+					}
+
 					// Check if the response finished successfully but has no content, tokens, or tool calls
 					// This indicates an empty response which should be marked as an error
 					// Do this check BEFORE sending usage chunks to ensure proper event ordering
@@ -5232,7 +5302,7 @@ chat.openapi(completions, async (c) => {
 									: new Error(String(sseError)),
 							);
 						}
-					} else {
+					} else if (!streamingError) {
 						// Calculate costs before sending usage chunk so we can include cost data
 						const billCancelledRequestsEarly = shouldBillCancelledRequests();
 						streamingCostsEarly =

apps/gateway/src/test-utils/mock-openai-server.ts

@@ -1,5 +1,6 @@
 import { serve } from "@hono/node-server";
 import { Hono } from "hono";
+import { streamSSE } from "hono/streaming";
 
 // Create a mock OpenAI API server
 export const mockOpenAIServer = new Hono();
@@ -537,6 +538,91 @@ mockOpenAIServer.post("/v1/chat/completions", async (c) => {
 		(msg: any) => msg.role === "user" && msg.content.includes("ZERO_TOKENS"),
 	);
 
+	const shouldTruncateStream = body.messages.some(
+		(msg: any) =>
+			msg.role === "user" && msg.content.includes("TRIGGER_TRUNCATED_STREAM"),
+	);
+
+	const assistantContent = `Hello! I received your message: "${userMessage}". This is a mock response from the test server.`;
+
+	if (body.stream === true) {
+		return streamSSE(c, async (stream) => {
+			let eventId = 0;
+
+			await stream.writeSSE({
+				data: JSON.stringify({
+					id: "chatcmpl-123",
+					object: "chat.completion.chunk",
+					created: Math.floor(Date.now() / 1000),
+					model: body.model ?? "gpt-4o-mini",
+					choices: [
+						{
+							index: 0,
+							delta: {
+								role: "assistant",
+							},
+							finish_reason: null,
+						},
+					],
+				}),
+				id: String(eventId++),
+			});
+
+			await stream.writeSSE({
+				data: JSON.stringify({
+					id: "chatcmpl-123",
+					object: "chat.completion.chunk",
+					created: Math.floor(Date.now() / 1000),
+					model: body.model ?? "gpt-4o-mini",
+					choices: [
+						{
+							index: 0,
+							delta: {
+								content: assistantContent,
+							},
+							finish_reason: null,
+						},
+					],
+				}),
+				id: String(eventId++),
+			});
+
+			if (shouldTruncateStream) {
+				return;
+			}
+
+			await stream.writeSSE({
+				data: JSON.stringify({
+					id: "chatcmpl-123",
+					object: "chat.completion.chunk",
+					created: Math.floor(Date.now() / 1000),
+					model: body.model ?? "gpt-4o-mini",
+					choices: [
+						{
+							index: 0,
+							delta: {},
+							finish_reason: "stop",
+						},
+					],
+					usage: shouldReturnZeroTokens
+						? {
+								prompt_tokens: 0,
+								completion_tokens: 20,
+								total_tokens: 20,
+							}
+						: sampleChatCompletionResponse.usage,
+				}),
+				id: String(eventId++),
+			});
+
+			await stream.writeSSE({
+				event: "done",
+				data: "[DONE]",
+				id: String(eventId++),
+			});
+		});
+	}
+
 	// Create a custom response that includes the user's message
 	const response = {
 		...sampleChatCompletionResponse,
@@ -545,7 +631,7 @@ mockOpenAIServer.post("/v1/chat/completions", async (c) => {
 				...sampleChatCompletionResponse.choices[0],
 				message: {
 					role: "assistant",
-					content: `Hello! I received your message: "${userMessage}". This is a mock response from the test server.`,
+					content: assistantContent,
 				},
 			},
 		],