diff --git a/apps/gateway/src/api.spec.ts b/apps/gateway/src/api.spec.ts index 69c72579a6..4e71752867 100644 --- a/apps/gateway/src/api.spec.ts +++ b/apps/gateway/src/api.spec.ts @@ -897,6 +897,7 @@ describe("api", () => { headers: { "Content-Type": "application/json", Authorization: `Bearer real-token`, + "x-debug": "true", }, body: JSON.stringify({ model: "llmgateway/custom", @@ -985,6 +986,190 @@ 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("streaming request surfaces missing upstream done sentinel", 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_FINISH_WITHOUT_DONE", + }, + ], + 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("streaming request surfaces inline provider SSE errors", 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_STREAM_PROVIDER_ERROR", + }, + ], + stream: true, + }), + }); + + expect(res.status).toBe(200); + + const streamResult = await readAll(res.body); + + expect(streamResult.hasError).toBe(false); + expect(streamResult.errorEvents).toHaveLength(0); + expect(streamResult.chunks.some((chunk) => chunk.error)).toBe(false); + expect( + streamResult.chunks.some( + (chunk) => chunk.choices?.[0]?.finish_reason === "content_filter", + ), + ).toBe(true); + expect( + streamResult.chunks.some( + (chunk) => (chunk.usage?.prompt_tokens ?? 0) > 0, + ), + ).toBe(true); + expect(streamResult.hasUsage).toBe(true); + expect( + (streamResult.fullContent?.match(/data: \[DONE\]/g) ?? []).length, + ).toBe(1); + + const logs = await waitForLogs(1); + expect(logs.length).toBe(1); + expect(logs[0].finishReason).toBe("content_filter"); + expect(logs[0].unifiedFinishReason).toBe("content_filter"); + expect(logs[0].hasError).toBe(false); + expect(logs[0].errorDetails).toBeNull(); + expect(logs[0].promptTokens).not.toBeNull(); + expect(logs[0].completionTokens).toBeNull(); + expect(typeof logs[0].rawResponse).toBe("string"); + expect(logs[0].rawResponse).toContain("data_inspection_failed"); + expect(logs[0].rawResponse).toContain('"finish_reason":"content_filter"'); + expect(typeof logs[0].upstreamResponse).toBe("string"); + expect(logs[0].upstreamResponse).toContain("data_inspection_failed"); + }); + test("request with short delay under timeout succeeds", async () => { await db.insert(tables.apiKey).values({ id: "token-id", diff --git a/apps/gateway/src/chat/chat.ts b/apps/gateway/src/chat/chat.ts index bec216219d..d881e71e44 100644 --- a/apps/gateway/src/chat/chat.ts +++ b/apps/gateway/src/chat/chat.ts @@ -2935,6 +2935,7 @@ chat.openapi(completions, async (c) => { let eventId = 0; let canceled = false; let streamingError: unknown = null; + let doneSent = false; // Track if [DONE] has been sent downstream // Raw logging variables let streamingRawResponseData = ""; // Raw SSE data sent back to the client @@ -2992,6 +2993,103 @@ chat.openapi(completions, async (c) => { } }; + const writeStreamingContentFilterResponse = async ({ + billingModel, + billingProvider, + responseModel, + metadata, + }: { + billingModel: string; + billingProvider: Provider; + responseModel: string; + metadata?: Record; + }) => { + const { calculatedPromptTokens } = estimateTokens( + billingProvider, + messages, + null, + null, + 0, + ); + const promptTokenCount = Math.max( + 1, + Math.round(calculatedPromptTokens ?? 1), + ); + const streamingCosts = await calculateCosts( + billingModel, + billingProvider, + promptTokenCount, + 0, + null, + { + prompt: messages + .map((m) => messageContentToString(m.content)) + .join("\n"), + completion: "", + }, + null, + 0, + image_config?.image_size, + inputImageCount, + 0, + project.organizationId, + ); + + await writeSSEAndCache({ + data: JSON.stringify({ + id: `chatcmpl-${Date.now()}`, + object: "chat.completion.chunk", + created: Math.floor(Date.now() / 1000), + model: responseModel, + choices: [ + { + index: 0, + delta: {}, + finish_reason: "content_filter", + }, + ], + ...(metadata && { metadata }), + }), + id: String(eventId++), + }); + + await writeSSEAndCache({ + data: JSON.stringify({ + id: `chatcmpl-${Date.now()}`, + object: "chat.completion.chunk", + created: Math.floor(Date.now() / 1000), + model: responseModel, + choices: [ + { + index: 0, + delta: {}, + finish_reason: null, + }, + ], + usage: { + prompt_tokens: promptTokenCount, + completion_tokens: 0, + total_tokens: promptTokenCount, + cost_usd_total: streamingCosts.totalCost, + cost_usd_input: streamingCosts.inputCost, + cost_usd_output: streamingCosts.outputCost, + cost_usd_cached_input: streamingCosts.cachedInputCost, + cost_usd_request: streamingCosts.requestCost, + cost_usd_image_input: streamingCosts.imageInputCost, + cost_usd_image_output: streamingCosts.imageOutputCost, + }, + }), + id: String(eventId++), + }); + + await writeSSEAndCache({ + event: "done", + data: "[DONE]", + id: String(eventId++), + }); + doneSent = true; + }; + // Set up cancellation handling const controller = new AbortController(); // Set up a listener for the request being aborted @@ -3685,9 +3783,21 @@ chat.openapi(completions, async (c) => { content: null, reasoningContent: null, finishReason, - promptTokens: null, + promptTokens: + finishReason === "content_filter" + ? ( + estimateTokens(usedProvider, messages, null, null, 0) + .calculatedPromptTokens ?? null + )?.toString() + : null, completionTokens: null, - totalTokens: null, + totalTokens: + finishReason === "content_filter" + ? ( + estimateTokens(usedProvider, messages, null, null, 0) + .calculatedPromptTokens ?? null + )?.toString() + : null, reasoningTokens: null, cachedTokens: null, hasError: finishReason !== "content_filter", // content_filter is not an error @@ -3743,18 +3853,10 @@ chat.openapi(completions, async (c) => { // For content_filter, return a proper completion chunk (not an error) // This handles Azure ResponsibleAIPolicyViolation and similar content filtering errors if (finishReason === "content_filter") { - const contentFilterChunk = { - id: `chatcmpl-${Date.now()}`, - object: "chat.completion.chunk", - created: Math.floor(Date.now() / 1000), - model: `${usedProvider}/${baseModelName}`, - choices: [ - { - index: 0, - delta: {}, - finish_reason: "content_filter", - }, - ], + await writeStreamingContentFilterResponse({ + billingModel: usedModel, + billingProvider: usedProvider, + responseModel: `${usedProvider}/${baseModelName}`, metadata: { requested_model: initialRequestedModel, requested_provider: requestedProvider, @@ -3763,42 +3865,6 @@ chat.openapi(completions, async (c) => { ...(usedRegion && { used_region: usedRegion }), underlying_used_model: usedModel, }, - }; - - await writeSSEAndCache({ - data: JSON.stringify(contentFilterChunk), - id: String(eventId++), - }); - - // Send a usage chunk for SDK compatibility (stream_options: { include_usage: true }) - const contentFilterUsageChunk = { - id: `chatcmpl-${Date.now()}`, - object: "chat.completion.chunk", - created: Math.floor(Date.now() / 1000), - model: `${usedProvider}/${baseModelName}`, - choices: [ - { - index: 0, - delta: {}, - finish_reason: null, - }, - ], - usage: { - prompt_tokens: 0, - completion_tokens: 0, - total_tokens: 0, - }, - }; - - await writeSSEAndCache({ - data: JSON.stringify(contentFilterUsageChunk), - id: String(eventId++), - }); - - await writeSSEAndCache({ - event: "done", - data: "[DONE]", - id: String(eventId++), }); } else { // For client errors, return the original provider error response @@ -3956,7 +4022,9 @@ chat.openapi(completions, async (c) => { let outputImageCount = 0; // Track number of output images for cost calculation let webSearchCount = 0; // Track web search calls for cost calculation const serverToolUseIndices = new Set(); // Track Anthropic server_tool_use block indices - let doneSent = false; // Track if [DONE] has been sent + let sawUpstreamDoneSentinel = false; + let sawProviderTerminalEvent = false; + let handledTerminalProviderEvent = false; let buffer = ""; // Buffer for accumulating partial data across chunks (string for SSE) let binaryBuffer = new Uint8Array(0); // Buffer for binary event streams (AWS Bedrock) let rawUpstreamData = ""; // Raw data received from upstream provider @@ -4286,6 +4354,7 @@ chat.openapi(completions, async (c) => { } if (eventData === "[DONE]") { + sawUpstreamDoneSentinel = true; // Set default finish_reason if not provided by the stream // Some providers (like Novita) don't send finish_reason in streaming chunks if (finishReason === null) { @@ -4466,6 +4535,117 @@ chat.openapi(completions, async (c) => { usedProvider === "aws-bedrock" ? extractAwsBedrockStreamError(data) : null; + const openAiCompatibleStreamError = + !awsBedrockStreamError && + data && + typeof data === "object" && + "error" in data && + data.error && + typeof data.error === "object" + ? (data.error as Record) + : null; + if (openAiCompatibleStreamError) { + const errorResponseText = JSON.stringify(data); + if ( + debugMode && + streamingRawResponseData.length < MAX_RAW_DATA_SIZE + ) { + const rawProviderSseEvent = `data: ${errorResponseText}\n\n`; + streamingRawResponseData += rawProviderSseEvent.substring( + 0, + Math.max( + 0, + MAX_RAW_DATA_SIZE - streamingRawResponseData.length, + ), + ); + } + const inferredStatusCode = + typeof openAiCompatibleStreamError.status_code === "number" + ? openAiCompatibleStreamError.status_code + : typeof openAiCompatibleStreamError.status === "number" + ? openAiCompatibleStreamError.status + : 400; + const errorType = getFinishReasonFromError( + inferredStatusCode, + errorResponseText, + ); + const errorMessage = + typeof openAiCompatibleStreamError.message === "string" + ? openAiCompatibleStreamError.message + : "Upstream provider returned a streaming error"; + const errorCode = + typeof openAiCompatibleStreamError.code === "string" + ? openAiCompatibleStreamError.code + : typeof openAiCompatibleStreamError.type === "string" + ? openAiCompatibleStreamError.type + : errorType; + + logger.info("[streaming] Provider SSE error received", { + requestId, + provider: usedProvider, + model: usedModel, + errorType, + errorCode, + inferredStatusCode, + errorMessage, + errorPayload: errorResponseText.substring(0, 5000), + }); + + finishReason = errorType; + + if (errorType === "content_filter") { + await writeStreamingContentFilterResponse({ + billingModel: usedModel, + billingProvider: usedProvider, + responseModel: data.model ?? usedModel, + }); + handledTerminalProviderEvent = true; + } else { + streamingError = { + message: errorMessage, + type: errorType, + code: errorCode, + details: { + statusCode: inferredStatusCode, + statusText: + typeof openAiCompatibleStreamError.type === "string" + ? openAiCompatibleStreamError.type + : "stream_error", + responseText: errorResponseText, + }, + }; + + await writeSSEAndCache({ + event: "error", + data: JSON.stringify({ + error: { + message: errorMessage, + type: errorType, + code: errorCode, + param: + "param" in openAiCompatibleStreamError + ? (openAiCompatibleStreamError.param ?? null) + : null, + responseText: errorResponseText, + }, + }), + id: String(eventId++), + }); + } + + if (!doneSent) { + await writeSSEAndCache({ + event: "done", + data: "[DONE]", + id: String(eventId++), + }); + doneSent = true; + } + shouldTerminateStream = true; + processedLength = eventEnd; + searchStart = eventEnd; + break; + } if (awsBedrockStreamError) { const errorType = getFinishReasonFromError( awsBedrockStreamError.statusCode, @@ -4859,8 +5039,10 @@ chat.openapi(completions, async (c) => { // Preserve original Google finish reason for logging if (data.promptFeedback?.blockReason) { finishReason = data.promptFeedback.blockReason; + sawProviderTerminalEvent = true; } else if (data.candidates?.[0]?.finishReason) { finishReason = data.candidates[0].finishReason; + sawProviderTerminalEvent = true; } break; case "anthropic": @@ -4869,13 +5051,16 @@ chat.openapi(completions, async (c) => { data.delta?.stop_reason ) { finishReason = data.delta.stop_reason; + sawProviderTerminalEvent = true; } else if ( data.type === "message_stop" || data.stop_reason ) { finishReason = data.stop_reason ?? "end_turn"; + sawProviderTerminalEvent = true; } else if (data.delta?.stop_reason) { finishReason = data.delta.stop_reason; + sawProviderTerminalEvent = true; } break; default: // OpenAI format @@ -5150,6 +5335,82 @@ chat.openapi(completions, async (c) => { estimateTokensFromContent(fullReasoningContent); } } + + const streamHasVerifiedTerminalEvent = + sawUpstreamDoneSentinel || + sawProviderTerminalEvent || + handledTerminalProviderEvent; + const streamEndedWithoutTerminalEvent = + !streamingError && + !canceled && + (!streamHasVerifiedTerminalEvent || finishReason === null); + if (streamEndedWithoutTerminalEvent) { + const hasBufferedNonWhitespace = /\S/u.test(buffer); + const responseText = hasBufferedNonWhitespace + ? buffer.slice(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 +5493,7 @@ chat.openapi(completions, async (c) => { : new Error(String(sseError)), ); } - } else { + } else if (!streamingError && !doneSent) { // Calculate costs before sending usage chunk so we can include cost data const billCancelledRequestsEarly = shouldBillCancelledRequests(); streamingCostsEarly = diff --git a/apps/gateway/src/chat/tools/get-finish-reason-from-error.spec.ts b/apps/gateway/src/chat/tools/get-finish-reason-from-error.spec.ts index 3c6f3ccd21..708d0810e9 100644 --- a/apps/gateway/src/chat/tools/get-finish-reason-from-error.spec.ts +++ b/apps/gateway/src/chat/tools/get-finish-reason-from-error.spec.ts @@ -61,6 +61,19 @@ describe("getFinishReasonFromError", () => { ); }); + it("returns content_filter for Alibaba data inspection errors", () => { + const alibabaError = JSON.stringify({ + error: { + code: "data_inspection_failed", + message: + "Input data may contain inappropriate content. Please ensure that your input complies with the usage policy of DashScope LLM.", + param: null, + type: "data_inspection_failed", + }, + }); + expect(getFinishReasonFromError(400, alibabaError)).toBe("content_filter"); + }); + it("returns client_error for zai content filter", () => { expect( getFinishReasonFromError( diff --git a/apps/gateway/src/chat/tools/get-finish-reason-from-error.ts b/apps/gateway/src/chat/tools/get-finish-reason-from-error.ts index eb0d44ef9a..8386b14611 100644 --- a/apps/gateway/src/chat/tools/get-finish-reason-from-error.ts +++ b/apps/gateway/src/chat/tools/get-finish-reason-from-error.ts @@ -39,6 +39,14 @@ export function getFinishReasonFromError( return "content_filter"; } + // Alibaba / DashScope moderation block + if ( + errorText?.includes("data_inspection_failed") || + errorText?.includes("Input data may contain inappropriate content") + ) { + return "content_filter"; + } + // xAI (Grok) content safety violations (e.g. SAFETY_CHECK_TYPE_CSAM, usage guidelines) if ( statusCode === 403 && diff --git a/apps/gateway/src/test-utils/mock-openai-server.ts b/apps/gateway/src/test-utils/mock-openai-server.ts index 5183fc134b..cc0aabfb1c 100644 --- a/apps/gateway/src/test-utils/mock-openai-server.ts +++ b/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(); @@ -192,6 +193,35 @@ async function extractMockVideoFileImage(file: File): Promise<{ }; } +interface MockChatMessage { + role: string; + content: string; +} + +function isMockChatMessage(value: unknown): value is MockChatMessage { + if (!value || typeof value !== "object") { + return false; + } + + const message = value as Record; + return ( + typeof message.role === "string" && typeof message.content === "string" + ); +} + +function getMockChatMessages(messages: unknown): MockChatMessage[] { + return Array.isArray(messages) ? messages.filter(isMockChatMessage) : []; +} + +function hasUserMessageTrigger( + messages: MockChatMessage[], + trigger: string, +): boolean { + return messages.some( + (message) => message.role === "user" && message.content.includes(trigger), + ); +} + // Counter for TRIGGER_FAIL_ONCE - tracks how many times a request with this // trigger has been received. First request fails with 500, subsequent succeed. // NOTE: This is module-level mutable state shared across all tests using this server. @@ -486,11 +516,10 @@ mockOpenAIServer.post("/v1/responses", async (c) => { // Handle chat completions endpoint mockOpenAIServer.post("/v1/chat/completions", async (c) => { const body = await c.req.json(); + const chatMessages = getMockChatMessages(body.messages); // Check if this request should trigger an error response - const shouldError = body.messages.some( - (msg: any) => msg.role === "user" && msg.content.includes("TRIGGER_ERROR"), - ); + const shouldError = hasUserMessageTrigger(chatMessages, "TRIGGER_ERROR"); if (shouldError) { c.status(500); @@ -499,7 +528,7 @@ mockOpenAIServer.post("/v1/chat/completions", async (c) => { // Get the user's message to include in the response const userMessage = - body.messages.find((msg: any) => msg.role === "user")?.content ?? ""; + chatMessages.find((message) => message.role === "user")?.content ?? ""; // Check if this request should trigger a specific HTTP status code error const statusTrigger = extractStatusCodeTrigger(userMessage); @@ -533,9 +562,123 @@ mockOpenAIServer.post("/v1/chat/completions", async (c) => { } // Check if this request should trigger zero tokens response - const shouldReturnZeroTokens = body.messages.some( - (msg: any) => msg.role === "user" && msg.content.includes("ZERO_TOKENS"), + const shouldReturnZeroTokens = hasUserMessageTrigger( + chatMessages, + "ZERO_TOKENS", ); + const shouldTruncateStream = hasUserMessageTrigger( + chatMessages, + "TRIGGER_TRUNCATED_STREAM", + ); + const shouldFinishWithoutDone = hasUserMessageTrigger( + chatMessages, + "TRIGGER_FINISH_WITHOUT_DONE", + ); + const shouldReturnStreamedProviderError = hasUserMessageTrigger( + chatMessages, + "TRIGGER_STREAM_PROVIDER_ERROR", + ); + + 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++), + }); + + if (shouldReturnStreamedProviderError) { + await stream.writeSSE({ + data: JSON.stringify({ + id: "chatcmpl-err-123", + error: { + code: "data_inspection_failed", + message: + "Input data may contain inappropriate content. Please ensure that your input complies with the usage policy of DashScope LLM.", + param: null, + type: "data_inspection_failed", + }, + }), + id: String(eventId++), + }); + 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: { + 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++), + }); + + if (shouldFinishWithoutDone) { + return; + } + + await stream.writeSSE({ + event: "done", + data: "[DONE]", + id: String(eventId++), + }); + }); + } // Create a custom response that includes the user's message const response = { @@ -545,7 +688,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, }, }, ],