diff --git a/open-sse/config/errorConfig.js b/open-sse/config/errorConfig.js index 71491a4d2c..2889aa6a28 100644 --- a/open-sse/config/errorConfig.js +++ b/open-sse/config/errorConfig.js @@ -50,17 +50,30 @@ const COOLDOWN = { /** * Unified error classification rules. * Checked top-to-bottom: text rules first (by order), then status rules. - * Each rule: { text?, status?, cooldownMs?, backoff? } + * Each rule: { text?, status?, cooldownMs?, backoff?, pass? } * - text: substring match (case-insensitive) on error message * - status: HTTP status code match * - cooldownMs: fixed cooldown duration * - backoff: true = use exponential backoff (rate limit) + * - pass: true = client-side error, do NOT lock the account (shouldFallback: false) */ export const ERROR_RULES = [ // --- Text-based rules (checked first, order = priority) --- { text: "no credentials", cooldownMs: COOLDOWN.long }, { text: "request not allowed", cooldownMs: COOLDOWN.short }, - { text: "improperly formed request", cooldownMs: COOLDOWN.long }, + // Client-side request errors: request is invalid / malformed — no account can fix this, don't lock + { text: "invalid_request_error", pass: true }, + { text: "invalid_request", pass: true }, + { text: "improperly formed request", pass: true }, + { text: "bad request", pass: true }, + { text: "unsupported parameter", pass: true }, + { text: "unsupported_parameter", pass: true }, + { text: "invalid parameter", pass: true }, + { text: "invalid_parameter", pass: true }, + { text: "maximum context length", pass: true }, + { text: "context_length_exceeded", pass: true }, + { text: "prompt is too long", pass: true }, + { text: "exceeds the limit", pass: true }, { text: "rate limit", backoff: true }, { text: "too many requests", backoff: true }, { text: "quota exceeded", backoff: true }, @@ -68,6 +81,7 @@ export const ERROR_RULES = [ { text: "overloaded", backoff: true }, // --- Status-based rules (fallback when text doesn't match) --- + { status: 400, pass: true }, { status: 401, cooldownMs: COOLDOWN.long }, { status: 402, cooldownMs: COOLDOWN.long }, { status: 403, cooldownMs: COOLDOWN.long }, diff --git a/open-sse/handlers/chatCore.js b/open-sse/handlers/chatCore.js index cb6ce96146..351665ce6c 100644 --- a/open-sse/handlers/chatCore.js +++ b/open-sse/handlers/chatCore.js @@ -310,6 +310,7 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred connectionProxyUrl: credentials?.providerSpecificData?.connectionProxyUrl || "", connectionNoProxy: credentials?.providerSpecificData?.connectionNoProxy || "", vercelRelayUrl: credentials?.providerSpecificData?.vercelRelayUrl || "", + strictProxy: credentials?.providerSpecificData?.strictProxy === true, }; if (proxyOptions.vercelRelayUrl) { diff --git a/open-sse/handlers/chatCore/streamingHandler.js b/open-sse/handlers/chatCore/streamingHandler.js index be008e7dae..af53ca26b6 100644 --- a/open-sse/handlers/chatCore/streamingHandler.js +++ b/open-sse/handlers/chatCore/streamingHandler.js @@ -44,14 +44,6 @@ function buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, * Handle streaming response — pipe provider SSE through transform stream to client. */ export async function handleStreamingResponse({ providerResponse, provider, model, sourceFormat, targetFormat, userAgent, body, stream, translatedBody, finalBody, requestStartTime, connectionId, apiKey, clientRawRequest, onRequestSuccess, reqLogger, toolNameMap, customToolNames, streamController, onStreamComplete, streamDetailId, pxpipe, reqTag, log }) { - if (onRequestSuccess) { - Promise.resolve() - .then(onRequestSuccess) - .catch(err => { - console.error("[ChatCore] onRequestSuccess failed:", err?.message || err); - }); - } - // When upstream returns HTML/text instead of SSE (e.g. Cloudflare 5xx error // page), piping it through the SSE transform stream causes Next.js // "failed to pipe response" and crashes the chat router. Read the body, @@ -59,8 +51,14 @@ export async function handleStreamingResponse({ providerResponse, provider, mode // return a clean JSON error instead. The message is stripped of HTML tags // and clamped so untrusted upstream text never reaches the client verbatim // (the UI may render error.message as HTML). - const upstreamContentType = (providerResponse.headers.get('content-type') || '').toLowerCase(); - if (upstreamContentType && !upstreamContentType.includes('text/event-stream') && !upstreamContentType.includes('application/json')) { + const upstreamContentType = (providerResponse.headers?.get?.('content-type') || '').toLowerCase(); + if ( + upstreamContentType && + !upstreamContentType.includes('text/event-stream') && + !upstreamContentType.includes('application/json') && + !upstreamContentType.includes('application/x-ndjson') && + !upstreamContentType.includes('application/stream+json') + ) { const bodyText = await providerResponse.text().catch(() => ''); const titleMatch = bodyText.match(/([^<]+)<\/title>/i); const sanitizedTitle = (titleMatch?.[1] || '').replace(/<[^>]*>/g, '').replace(/[\r\n]+/g, ' ').trim().slice(0, 160); @@ -72,6 +70,8 @@ export async function handleStreamingResponse({ providerResponse, provider, mode streamController?.handleError?.(new Error(`upstream non-SSE: ${status}`)); return { success: false, + status, + error: shortMsg, response: new Response(JSON.stringify({ error: { message: `[${status}]: ${shortMsg}` } }), { status, headers: { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*' }, @@ -79,13 +79,143 @@ export async function handleStreamingResponse({ providerResponse, provider, mode }; } + // First-valid-event gate: buffer the first chunk from upstream before confirming success. + // This prevents empty streams (0 bytes) or immediate error objects disguised as 200 OK + // from falsely clearing account errors or committing an unusable stream to the client. + if (!providerResponse.body) { + const status = 502; + const shortMsg = "Upstream returned no response body"; + if (log?.errorLine) log.errorLine(reqTag, "✗", `BLOCKED ${status} · ${provider}/${model} · ${shortMsg}`); + streamController?.handleError?.(new Error(shortMsg)); + return { + success: false, + status, + error: shortMsg, + response: new Response(JSON.stringify({ error: { message: `[${status}]: ${shortMsg}` } }), { + status, + headers: { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*' }, + }), + }; + } + + let reader = null; + let firstChunk = null; + try { + reader = providerResponse.body.getReader(); + const { done, value } = await reader.read(); + if (done || !value || value.length === 0) { + try { reader.releaseLock?.(); } catch {} + const status = 502; + const shortMsg = "Upstream stream ended before a valid event (empty stream)"; + if (log?.errorLine) log.errorLine(reqTag, "✗", `BLOCKED ${status} · ${provider}/${model} · ${shortMsg}`); + streamController?.handleError?.(new Error(shortMsg)); + return { + success: false, + status, + error: shortMsg, + response: new Response(JSON.stringify({ error: { message: `[${status}]: ${shortMsg}` } }), { + status, + headers: { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*' }, + }), + }; + } + firstChunk = value; + } catch (readErr) { + const status = 502; + const shortMsg = `Upstream stream read error: ${readErr?.message || readErr}`; + if (log?.errorLine) log.errorLine(reqTag, "✗", `BLOCKED ${status} · ${provider}/${model} · ${shortMsg}`); + streamController?.handleError?.(readErr); + return { + success: false, + status, + error: shortMsg, + response: new Response(JSON.stringify({ error: { message: `[${status}]: ${shortMsg}` } }), { + status, + headers: { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*' }, + }), + }; + } + + // Check if first chunk contains a structured JSON error object returned as 200 OK + if (firstChunk) { + const chunkStr = new TextDecoder().decode(firstChunk); + const trimmed = chunkStr.trim(); + if (trimmed.startsWith("{") && (trimmed.includes('"error"') || trimmed.includes('"error_code"') || trimmed.includes('"detail"'))) { + try { + const parsed = JSON.parse(trimmed); + if (parsed.error || parsed.error_code || (parsed.detail && !parsed.choices && !parsed.delta)) { + const errMsg = typeof parsed.error === "string" + ? parsed.error + : parsed.error?.message || parsed.error_msg || parsed.detail || JSON.stringify(parsed); + const rawStatus = parsed.error?.status || parsed.status || 502; + const status = typeof rawStatus === "number" && rawStatus >= 400 && rawStatus < 600 ? rawStatus : 502; + if (log?.errorLine) log.errorLine(reqTag, "✗", `ERROR ${status} · ${provider}/${model} · ${errMsg}`); + streamController?.handleError?.(new Error(errMsg)); + return { + success: false, + status, + error: errMsg, + response: new Response(JSON.stringify({ error: { message: `[${status}]: ${errMsg}` } }), { + status, + headers: { 'Content-Type': 'application/json', 'Access-Control-Allow-Origin': '*' }, + }), + }; + } + } catch { + // Not a pure JSON error object, treat as valid streaming content + } + } + } + + // First valid event/chunk confirmed — notify request success callback + if (onRequestSuccess) { + Promise.resolve() + .then(onRequestSuccess) + .catch(err => { + console.error("[ChatCore] onRequestSuccess failed:", err?.message || err); + }); + } + + // Reconstruct ReadableStream with the buffered firstChunk prepended + let responseBodyStream = providerResponse.body; + if (reader && firstChunk) { + let yieldedFirst = false; + responseBodyStream = new ReadableStream({ + async pull(controller) { + if (!yieldedFirst) { + yieldedFirst = true; + controller.enqueue(firstChunk); + return; + } + try { + const { done, value } = await reader.read(); + if (done) { + controller.close(); + return; + } + controller.enqueue(value); + } catch (err) { + controller.error(err); + } + }, + cancel(reason) { + return reader.cancel(reason); + } + }); + } + const transformStream = buildTransformStream({ provider, sourceFormat, targetFormat, userAgent, reqLogger, toolNameMap, customToolNames, model, connectionId, body, onStreamComplete, apiKey }); // Responses passthrough: synthesize response.failed + [DONE] if the stream aborts/stalls before a terminal event const isResponsesPassthrough = sourceFormat === FORMATS.OPENAI_RESPONSES && targetFormat === FORMATS.OPENAI_RESPONSES; const onAbortTerminal = isResponsesPassthrough ? buildAbortedResponsesTerminalBytes : null; const stallTimeoutMs = PROVIDERS[provider]?.stallTimeoutMs || STREAM_STALL_TIMEOUT_MS; - const transformedBody = pipeWithDisconnect(providerResponse, transformStream, streamController, onAbortTerminal, stallTimeoutMs); + const wrappedResponse = { + ...providerResponse, + body: responseBodyStream, + headers: providerResponse.headers, + }; + const transformedBody = pipeWithDisconnect(wrappedResponse, transformStream, streamController, onAbortTerminal, stallTimeoutMs); saveRequestDetail(buildRequestDetail({ provider, model, connectionId, diff --git a/open-sse/providers/registry/nvidia.js b/open-sse/providers/registry/nvidia.js index 4d375a1744..afbddea86a 100644 --- a/open-sse/providers/registry/nvidia.js +++ b/open-sse/providers/registry/nvidia.js @@ -20,6 +20,7 @@ export default { transport: { baseUrl: "https://integrate.api.nvidia.com/v1/chat/completions", validateUrl: "https://integrate.api.nvidia.com/v1/models", + stallTimeoutMs: 600000, // 10 min — reasoning models (step-3.7-flash, nemotron-ultra) can have long silent thinking phases; observed >5 min on step-3.7-flash }, models: [ { id: "minimaxai/minimax-m2.7", name: "MiniMax M2.7" }, @@ -28,6 +29,7 @@ export default { { id: "deepseek-ai/deepseek-v4-pro", name: "DeepSeek V4 Pro" }, { id: "deepseek-ai/deepseek-v4-flash", name: "DeepSeek V4 Flash" }, { id: "moonshotai/kimi-k2.6", name: "Kimi K2.6" }, + { id: "stepfun-ai/step-3.7-flash", name: "Step 3.7 Flash" }, { id: "nvidia/nemotron-3-ultra-550b-a55b", name: "Nemotron 3 Ultra" }, { id: "nvidia/nv-embedqa-e5-v5", name: "NV EmbedQA E5 v5", kind: "embedding" }, { id: "nvidia/parakeet-ctc-1.1b-asr", name: "Parakeet CTC 1.1B", params: ["language"], kind: "stt" }, diff --git a/open-sse/services/accountFallback.js b/open-sse/services/accountFallback.js index 8d280da412..9e0e74f4a3 100644 --- a/open-sse/services/accountFallback.js +++ b/open-sse/services/accountFallback.js @@ -28,6 +28,8 @@ export function checkFallbackError(status, errorText, backoffLevel = 0) { for (const rule of ERROR_RULES) { // Text-based rule: match substring in error message if (rule.text && lowerError && lowerError.includes(rule.text)) { + // pass: true = client-side error, do not lock the account + if (rule.pass) return { shouldFallback: false, cooldownMs: 0 }; if (rule.backoff) { const newLevel = Math.min(backoffLevel + 1, BACKOFF_CONFIG.maxLevel); return { shouldFallback: true, cooldownMs: getQuotaCooldown(newLevel), newBackoffLevel: newLevel }; @@ -37,6 +39,8 @@ export function checkFallbackError(status, errorText, backoffLevel = 0) { // Status-based rule: match HTTP status code if (rule.status && rule.status === status) { + // pass: true = client-side error, do not lock the account + if (rule.pass) return { shouldFallback: false, cooldownMs: 0 }; if (rule.backoff) { const newLevel = Math.min(backoffLevel + 1, BACKOFF_CONFIG.maxLevel); return { shouldFallback: true, cooldownMs: getQuotaCooldown(newLevel), newBackoffLevel: newLevel }; diff --git a/package.json b/package.json index f1c2c97f57..7616d8f1ea 100644 --- a/package.json +++ b/package.json @@ -4,13 +4,13 @@ "description": "9Router web dashboard", "private": true, "scripts": { - "dev": "next dev --port 20127", - "dev:webpack": "next dev --webpack --port 20127", + "dev": "next dev --port 20128", + "dev:webpack": "next dev --webpack --port 20128", "build": "next build --webpack", "postbuild": "node scripts/copy-standalone-assets.mjs", "postbuild:bun": "node scripts/copy-standalone-assets.mjs", - "start": "node custom-server.js --port 20127", - "dev:bun": "bun --bun next dev --webpack --port 20127", + "start": "node custom-server.js --port 20128", + "dev:bun": "bun --bun next dev --webpack --port 20128", "build:bun": "bun --bun next build --webpack", "start:bun": "bun ./.next/standalone/custom-server.js", "cli:pack": "npm --prefix cli run pack:cli", diff --git a/src/sse/services/auth.js b/src/sse/services/auth.js index feaaa2ab23..f9aaf539b5 100644 --- a/src/sse/services/auth.js +++ b/src/sse/services/auth.js @@ -64,6 +64,7 @@ export async function getProviderCredentials(provider, excludeConnectionIds = nu connectionNoProxy: resolvedProxy.connectionNoProxy, connectionProxyPoolId: resolvedProxy.proxyPoolId || null, vercelRelayUrl: resolvedProxy.vercelRelayUrl || "", + strictProxy: resolvedProxy.strictProxy === true, }, }; } @@ -193,6 +194,7 @@ export async function getProviderCredentials(provider, excludeConnectionIds = nu connectionNoProxy: resolvedProxy.connectionNoProxy, connectionProxyPoolId: resolvedProxy.proxyPoolId || null, vercelRelayUrl: resolvedProxy.vercelRelayUrl || "", + strictProxy: resolvedProxy.strictProxy === true, }, connectionId: connection.id, // Include current status for optimization check diff --git a/tests/unit/account-fallback-rules.test.js b/tests/unit/account-fallback-rules.test.js new file mode 100644 index 0000000000..6f080a364e --- /dev/null +++ b/tests/unit/account-fallback-rules.test.js @@ -0,0 +1,67 @@ +import { describe, it, expect } from "vitest"; +import { checkFallbackError, getQuotaCooldown } from "open-sse/services/accountFallback.js"; +import { TRANSIENT_COOLDOWN_MS } from "open-sse/config/errorConfig.js"; + +describe("accountFallback: checkFallbackError rules", () => { + describe("pass: true client-side errors (context length exceeded)", () => { + it("does not fallback or lock account when message contains 'maximum context length'", () => { + const res = checkFallbackError(400, "This model's maximum context length is 128000 tokens. However, your messages resulted in 130000 tokens."); + expect(res).toEqual({ shouldFallback: false, cooldownMs: 0 }); + }); + + it("does not fallback or lock account when message contains 'context_length_exceeded'", () => { + const res = checkFallbackError(400, JSON.stringify({ error: { code: "context_length_exceeded", message: "Too many tokens" } })); + expect(res).toEqual({ shouldFallback: false, cooldownMs: 0 }); + }); + + it("does not fallback or lock account when message contains 'prompt is too long'", () => { + const res = checkFallbackError(400, "The prompt is too long for the requested model."); + expect(res).toEqual({ shouldFallback: false, cooldownMs: 0 }); + }); + + it("does not fallback or lock account on status 400 (Bad Request)", () => { + const res = checkFallbackError(400, "Bad Request"); + expect(res).toEqual({ shouldFallback: false, cooldownMs: 0 }); + }); + + it("does not fallback or lock account when message contains 'invalid_request_error'", () => { + const res = checkFallbackError(400, JSON.stringify({ error: { type: "invalid_request_error", message: "tools[0].function.parameters is invalid" } })); + expect(res).toEqual({ shouldFallback: false, cooldownMs: 0 }); + }); + + it("does not fallback or lock account when message contains 'improperly formed request'", () => { + const res = checkFallbackError(400, "Improperly formed request"); + expect(res).toEqual({ shouldFallback: false, cooldownMs: 0 }); + }); + + it("does not fallback or lock account when message contains 'unsupported parameter'", () => { + const res = checkFallbackError(400, "Unsupported parameter: reasoning_effort"); + expect(res).toEqual({ shouldFallback: false, cooldownMs: 0 }); + }); + }); + + describe("standard rate-limit and auth error rules", () => { + it("handles 401 unauthorized with fixed long cooldown", () => { + const res = checkFallbackError(401, "Unauthorized"); + expect(res.shouldFallback).toBe(true); + expect(res.cooldownMs).toBeGreaterThan(0); + }); + + it("handles rate limit (429) with exponential backoff", () => { + const res1 = checkFallbackError(429, "Too many requests", 0); + expect(res1.shouldFallback).toBe(true); + expect(res1.newBackoffLevel).toBe(1); + expect(res1.cooldownMs).toBe(getQuotaCooldown(1)); + + const res2 = checkFallbackError(429, "rate limit", 1); + expect(res2.shouldFallback).toBe(true); + expect(res2.newBackoffLevel).toBe(2); + expect(res2.cooldownMs).toBe(getQuotaCooldown(2)); + }); + + it("returns transient cooldown for unknown errors", () => { + const res = checkFallbackError(500, "Internal server glitch"); + expect(res).toEqual({ shouldFallback: true, cooldownMs: TRANSIENT_COOLDOWN_MS }); + }); + }); +}); diff --git a/tests/unit/ollama-ndjson-stream.test.js b/tests/unit/ollama-ndjson-stream.test.js new file mode 100644 index 0000000000..244472f14c --- /dev/null +++ b/tests/unit/ollama-ndjson-stream.test.js @@ -0,0 +1,51 @@ +import { describe, it, expect } from "vitest"; +import { handleStreamingResponse } from "../../open-sse/handlers/chatCore/streamingHandler.js"; + +describe("Ollama stream content type support", () => { + it("allows application/x-ndjson and application/stream+json without blocking as non-SSE error", async () => { + const mockProviderResponse = { + status: 200, + headers: new Map([["content-type", "application/x-ndjson"]]), + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('{"message":{"role":"assistant","content":"hello"}}\n')); + controller.close(); + }, + }), + }; + mockProviderResponse.headers.get = (name) => (name.toLowerCase() === "content-type" ? "application/x-ndjson" : null); + + const mockStreamController = { + signal: new AbortController().signal, + handleError: () => {}, + }; + + const res = await handleStreamingResponse({ + providerResponse: mockProviderResponse, + provider: "ollama", + model: "llama3", + sourceFormat: "openai", + targetFormat: "openai", + userAgent: "", + body: { stream: true }, + translatedBody: {}, + finalBody: {}, + requestStartTime: Date.now(), + connectionId: "conn_1", + apiKey: null, + clientRawRequest: null, + onRequestSuccess: null, + reqLogger: { logTargetRequest: () => {} }, + toolNameMap: null, + customToolNames: null, + streamController: mockStreamController, + onStreamComplete: null, + streamDetailId: "detail_1", + pxpipe: null, + reqTag: "tag_1", + log: { debug: () => {} }, + }); + + expect(res?.success).not.toBe(false); + }); +}); diff --git a/tests/unit/stream-first-valid-event-gate.test.js b/tests/unit/stream-first-valid-event-gate.test.js new file mode 100644 index 0000000000..e1fe4abf57 --- /dev/null +++ b/tests/unit/stream-first-valid-event-gate.test.js @@ -0,0 +1,253 @@ +import { describe, it, expect, vi } from "vitest"; +import { handleStreamingResponse } from "open-sse/handlers/chatCore/streamingHandler.js"; + +describe("Streaming first-valid-event gate (Issue 2951 Finding 3)", () => { + const baseParams = { + provider: "nvidia", + model: "meta/llama-3.1-70b-instruct", + sourceFormat: "openai", + targetFormat: "openai", + userAgent: "test-agent", + body: { stream: true }, + translatedBody: {}, + finalBody: {}, + requestStartTime: Date.now(), + connectionId: "conn-nv-1", + apiKey: "nv-key", + clientRawRequest: null, + reqLogger: { logTargetRequest: () => {}, logError: () => {} }, + toolNameMap: null, + customToolNames: null, + streamController: { + signal: new AbortController().signal, + isConnected: () => true, + handleComplete: vi.fn(), + handleError: vi.fn(), + handleDisconnect: vi.fn(), + abort: vi.fn(), + }, + onStreamComplete: vi.fn(), + streamDetailId: "detail-test", + pxpipe: null, + reqTag: "REQ_TEST", + log: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), errorLine: vi.fn(), line: vi.fn() }, + }; + + it("Case 1: Empty stream (0 bytes) returns success=false and does NOT call onRequestSuccess", async () => { + const onRequestSuccess = vi.fn(); + const mockProviderResponse = { + status: 200, + headers: new Map([["content-type", "text/event-stream"]]), + body: new ReadableStream({ + start(controller) { + controller.close(); + }, + }), + }; + mockProviderResponse.headers.get = (k) => (k.toLowerCase() === "content-type" ? "text/event-stream" : null); + + const res = await handleStreamingResponse({ + ...baseParams, + providerResponse: mockProviderResponse, + onRequestSuccess, + }); + + expect(res.success).toBe(false); + expect(res.status).toBe(502); + expect(res.error).toMatch(/empty stream/i); + expect(onRequestSuccess).not.toHaveBeenCalled(); + }); + + it("Case 2: JSON error disguised in 200 stream returns success=false and does NOT call onRequestSuccess", async () => { + const onRequestSuccess = vi.fn(); + const errorJson = JSON.stringify({ error: { message: "Model overloaded", status: 503 } }); + const mockProviderResponse = { + status: 200, + headers: new Map([["content-type", "text/event-stream"]]), + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(errorJson)); + controller.close(); + }, + }), + }; + mockProviderResponse.headers.get = (k) => (k.toLowerCase() === "content-type" ? "text/event-stream" : null); + + const res = await handleStreamingResponse({ + ...baseParams, + providerResponse: mockProviderResponse, + onRequestSuccess, + }); + + expect(res.success).toBe(false); + expect(res.status).toBe(503); + expect(res.error).toMatch(/Model overloaded/i); + expect(onRequestSuccess).not.toHaveBeenCalled(); + }); + + it("Case 3: Non-SSE HTML response returns success=false and does NOT call onRequestSuccess", async () => { + const onRequestSuccess = vi.fn(); + const mockProviderResponse = { + status: 500, + headers: new Map([["content-type", "text/html"]]), + text: async () => "<html><head><title>Internal Cloudflare Error500", + }; + mockProviderResponse.headers.get = (k) => (k.toLowerCase() === "content-type" ? "text/html" : null); + + const res = await handleStreamingResponse({ + ...baseParams, + providerResponse: mockProviderResponse, + onRequestSuccess, + }); + + expect(res.success).toBe(false); + expect(res.status).toBe(500); + expect(res.error).toMatch(/Internal Cloudflare Error/i); + expect(onRequestSuccess).not.toHaveBeenCalled(); + }); + + it("Case 4: Valid stream with data returns success=true and calls onRequestSuccess", async () => { + const onRequestSuccess = vi.fn(); + const sseChunk = 'data: {"id":"1","choices":[{"delta":{"content":"Hi"}}]}\n\n'; + const sseDone = "data: [DONE]\n\n"; + + const mockProviderResponse = { + status: 200, + headers: new Map([["content-type", "text/event-stream"]]), + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(sseChunk)); + controller.enqueue(new TextEncoder().encode(sseDone)); + controller.close(); + }, + }), + }; + mockProviderResponse.headers.get = (k) => (k.toLowerCase() === "content-type" ? "text/event-stream" : null); + + const res = await handleStreamingResponse({ + ...baseParams, + providerResponse: mockProviderResponse, + onRequestSuccess, + }); + + expect(res.success).toBe(true); + expect(res.response).toBeDefined(); + + // Verify onRequestSuccess was triggered + await new Promise((r) => setTimeout(r, 10)); + expect(onRequestSuccess).toHaveBeenCalledTimes(1); + + // Verify response body can be read and contains the original stream data + const reader = res.response.body.getReader(); + const { value } = await reader.read(); + const text = new TextDecoder().decode(value); + expect(text).toContain("data:"); + }); + + it("Case 5: Null body returns success=false and 502", async () => { + const onRequestSuccess = vi.fn(); + const mockProviderResponse = { + status: 200, + headers: new Map([["content-type", "text/event-stream"]]), + body: null, + }; + mockProviderResponse.headers.get = (k) => (k.toLowerCase() === "content-type" ? "text/event-stream" : null); + + const res = await handleStreamingResponse({ + ...baseParams, + providerResponse: mockProviderResponse, + onRequestSuccess, + }); + + expect(res.success).toBe(false); + expect(res.status).toBe(502); + expect(res.error).toMatch(/no response body/i); + expect(onRequestSuccess).not.toHaveBeenCalled(); + }); + + it("Case 6: Raw string error in JSON returns success=false and 502", async () => { + const onRequestSuccess = vi.fn(); + const rawErrorJson = JSON.stringify({ error: "Invalid API key format" }); + const mockProviderResponse = { + status: 200, + headers: new Map([["content-type", "text/event-stream"]]), + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(rawErrorJson)); + controller.close(); + }, + }), + }; + mockProviderResponse.headers.get = (k) => (k.toLowerCase() === "content-type" ? "text/event-stream" : null); + + const res = await handleStreamingResponse({ + ...baseParams, + providerResponse: mockProviderResponse, + onRequestSuccess, + }); + + expect(res.success).toBe(false); + expect(res.status).toBe(502); + expect(res.error).toBe("Invalid API key format"); + expect(onRequestSuccess).not.toHaveBeenCalled(); + }); + + it("Case 7: FastAPI detail error payload returns success=false and 502", async () => { + const onRequestSuccess = vi.fn(); + const detailJson = JSON.stringify({ detail: "Gateway timeout upstream" }); + const mockProviderResponse = { + status: 200, + headers: new Map([["content-type", "text/event-stream"]]), + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(detailJson)); + controller.close(); + }, + }), + }; + mockProviderResponse.headers.get = (k) => (k.toLowerCase() === "content-type" ? "text/event-stream" : null); + + const res = await handleStreamingResponse({ + ...baseParams, + providerResponse: mockProviderResponse, + onRequestSuccess, + }); + + expect(res.success).toBe(false); + expect(res.status).toBe(502); + expect(res.error).toBe("Gateway timeout upstream"); + expect(onRequestSuccess).not.toHaveBeenCalled(); + }); + + it("Case 8: Assistant output containing the word 'error' in normal payload is NOT treated as an error", async () => { + const onRequestSuccess = vi.fn(); + const normalPayload = JSON.stringify({ + id: "chatcmpl-1", + choices: [{ delta: { content: "Here is how to fix the error in your code" } }] + }); + const sseChunk = `data: ${normalPayload}\n\n`; + + const mockProviderResponse = { + status: 200, + headers: new Map([["content-type", "text/event-stream"]]), + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(sseChunk)); + controller.enqueue(new TextEncoder().encode("data: [DONE]\n\n")); + controller.close(); + }, + }), + }; + mockProviderResponse.headers.get = (k) => (k.toLowerCase() === "content-type" ? "text/event-stream" : null); + + const res = await handleStreamingResponse({ + ...baseParams, + providerResponse: mockProviderResponse, + onRequestSuccess, + }); + + expect(res.success).toBe(true); + await new Promise((r) => setTimeout(r, 10)); + expect(onRequestSuccess).toHaveBeenCalledTimes(1); + }); +}); diff --git a/tests/unit/strict-proxy-propagation.test.js b/tests/unit/strict-proxy-propagation.test.js new file mode 100644 index 0000000000..298cd326bb --- /dev/null +++ b/tests/unit/strict-proxy-propagation.test.js @@ -0,0 +1,178 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; + +// Mocks for database & proxy config resolution +const dbMocks = vi.hoisted(() => ({ + getProxyPools: vi.fn(), + getProxyPoolById: vi.fn(), + getProviderConnections: vi.fn(), + getSettings: vi.fn(() => ({})), +})); + +vi.mock('@/lib/localDb', () => ({ + getProxyPools: dbMocks.getProxyPools, + getProxyPoolById: dbMocks.getProxyPoolById, + getProviderConnections: dbMocks.getProviderConnections, + getSettings: dbMocks.getSettings, +})); + +import { getProviderCredentials } from '@/sse/services/auth.js'; + +describe('PR A: strictProxy Propagation', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + describe('Credential Propagation (auth.js)', () => { + it('propagates strictProxy=true for normal provider connection path', async () => { + const mockPool = { + id: 'pool-strict', + name: 'Strict Proxy Pool', + isActive: true, + proxyUrl: 'http://127.0.0.1:9999', + strictProxy: true, + }; + dbMocks.getProviderConnections.mockResolvedValue([ + { + id: 'conn-1', + provider: 'openai', + name: 'OpenAI Conn', + testStatus: 'active', + apiKey: 'sk-test-key', + providerSpecificData: { proxyPoolId: 'pool-strict' }, + }, + ]); + dbMocks.getProxyPools.mockResolvedValue([mockPool]); + dbMocks.getProxyPoolById.mockResolvedValue(mockPool); + + const creds = await getProviderCredentials('openai', {}, null); + expect(creds).toBeDefined(); + expect(creds.providerSpecificData.connectionProxyEnabled).toBe(true); + expect(creds.providerSpecificData.connectionProxyUrl).toBe('http://127.0.0.1:9999'); + expect(creds.providerSpecificData.strictProxy).toBe(true); + }); + + it('propagates strictProxy=false for normal provider connection path', async () => { + const mockPool = { + id: 'pool-lenient', + name: 'Lenient Proxy Pool', + isActive: true, + proxyUrl: 'http://127.0.0.1:9999', + strictProxy: false, + }; + dbMocks.getProviderConnections.mockResolvedValue([ + { + id: 'conn-2', + provider: 'openai', + name: 'OpenAI Conn 2', + testStatus: 'active', + apiKey: 'sk-test-key-2', + providerSpecificData: { proxyPoolId: 'pool-lenient' }, + }, + ]); + dbMocks.getProxyPools.mockResolvedValue([mockPool]); + dbMocks.getProxyPoolById.mockResolvedValue(mockPool); + + const creds = await getProviderCredentials('openai', {}, null); + expect(creds).toBeDefined(); + expect(creds.providerSpecificData.strictProxy).toBe(false); + }); + + it('propagates strictProxy=true for no-auth virtual connection path', async () => { + const mockPool = { + id: 'pool-noauth-strict', + name: 'NoAuth Strict Proxy Pool', + isActive: true, + proxyUrl: 'http://127.0.0.1:9999', + strictProxy: true, + }; + dbMocks.getProviderConnections.mockResolvedValue([]); + dbMocks.getProxyPools.mockResolvedValue([mockPool]); + dbMocks.getProxyPoolById.mockResolvedValue(mockPool); + dbMocks.getSettings.mockResolvedValue({ + providerStrategies: { + 'mimo-free': { rotateStrategy: 'round-robin', proxyPoolId: 'pool-noauth-strict' }, + }, + }); + + const creds = await getProviderCredentials('mimo-free', {}, null); + expect(creds).toBeDefined(); + expect(creds.id).toBe('noauth'); + expect(creds.providerSpecificData.connectionProxyEnabled).toBe(true); + expect(creds.providerSpecificData.strictProxy).toBe(true); + }); + }); + + describe('Full Propagation & Fetch Behavior (chatCore -> proxyAwareFetch)', () => { + it('Case A: strictProxy=true + proxy failure -> direct fetch invocation count = 0', async () => { + let callCount = 0; + // Proxy fetch fails with connection error + const mockProxyFetch = vi.fn(async () => { + callCount++; + throw new Error('connect ECONNREFUSED 127.0.0.1:19999'); + }); + + const origFetch = globalThis.fetch; + globalThis.fetch = mockProxyFetch; + + vi.resetModules(); + const { proxyAwareFetch } = await import('open-sse/utils/proxyFetch.js'); + + const proxyOptions = { + connectionProxyEnabled: true, + connectionProxyUrl: 'http://127.0.0.1:19999', + connectionNoProxy: '', + vercelRelayUrl: '', + strictProxy: true, + }; + + try { + await expect( + proxyAwareFetch('https://example.com/v1/chat/completions', { method: 'POST' }, proxyOptions) + ).rejects.toThrow(/strictProxy=true/); + + // When strictProxy=true and proxy fails, it MUST NOT fall back to direct fetch. + // The single call that failed was the proxy fetch attempt. + expect(callCount).toBe(1); + } finally { + globalThis.fetch = origFetch; + } + }); + + it('Case B: strictProxy=false + proxy failure -> falls back to direct fetch (count = 1)', async () => { + let proxyAttempts = 0; + let directFallbackAttempts = 0; + + const fakeFetch = vi.fn(async (url, init) => { + if (init?.dispatcher) { + proxyAttempts++; + throw new Error('connect ECONNREFUSED 127.0.0.1:19999'); + } + directFallbackAttempts++; + return new Response('{"ok":true}', { status: 200 }); + }); + + const origFetch = globalThis.fetch; + globalThis.fetch = fakeFetch; + + vi.resetModules(); + const { proxyAwareFetch } = await import('open-sse/utils/proxyFetch.js'); + + const proxyOptions = { + connectionProxyEnabled: true, + connectionProxyUrl: 'http://127.0.0.1:19999', + connectionNoProxy: '', + vercelRelayUrl: '', + strictProxy: false, + }; + + try { + const res = await proxyAwareFetch('https://example.com/v1/chat/completions', { method: 'POST' }, proxyOptions); + expect(res.status).toBe(200); + // Direct fetch fallback MUST be invoked exactly once when strictProxy=false + expect(directFallbackAttempts).toBe(1); + } finally { + globalThis.fetch = origFetch; + } + }); + }); +});