diff --git a/src/server/routes/api/accounts.ts b/src/server/routes/api/accounts.ts index 6905af0..9bc2927 100644 --- a/src/server/routes/api/accounts.ts +++ b/src/server/routes/api/accounts.ts @@ -503,12 +503,21 @@ export async function accountsRoutes(app: FastifyInstance) { } // Try to explain unknown failures: missing user id vs anti-bot challenge page. + const normalizedPlatform = String(adapter.platformName || site.platform || '').trim().toLowerCase(); + // New-API family already runs shield-aware probing inside adapters. + // Raw fallback probe below does not include challenge-solving and can + // misclassify valid Cookie/Session flows as shield-blocked. + const skipRawShieldDetection = normalizedPlatform === 'new-api' || normalizedPlatform === 'anyrouter'; type VerifyFailureReason = 'needs-user-id' | 'shield-blocked' | null; const detectVerifyFailureReason = async (): Promise => { const parseFailureReason = (bodyText: string, contentType: string): VerifyFailureReason => { const text = bodyText || ''; const ct = (contentType || '').toLowerCase(); - if (ct.includes('text/html') && /var\s+arg1\s*=|acw_sc__v2|cdn_sec_tc|', @@ -108,6 +111,75 @@ describe('accounts verify-token shield detection', () => { }, }); + expect(response.statusCode).toBe(200); + expect(response.json()).toMatchObject({ + success: false, + message: 'Token invalid: cannot use it as session cookie or API key', + }); + expect(undiciFetchMock).toHaveBeenCalled(); + }); + + it('uses adapter platformName to skip raw shield detection for newapi alias site platform', async () => { + adapterPlatformName = 'new-api'; + verifyTokenMock.mockResolvedValueOnce({ tokenType: 'unknown' }); + undiciFetchMock.mockResolvedValue({ + text: async () => '', + headers: { + get: (name: string) => (name.toLowerCase() === 'content-type' ? 'text/html; charset=utf-8' : null), + }, + }); + + const site = db.insert(schema.sites).values({ + name: 'AnyRouter Alias', + url: 'https://anyrouter-alias.example.com', + platform: 'newapi', + }).returning().get(); + + const response = await app.inject({ + method: 'POST', + url: '/api/accounts/verify-token', + payload: { + siteId: site.id, + accessToken: 'session-or-cookie-token', + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toMatchObject({ + success: false, + message: 'Token invalid: cannot use it as session cookie or API key', + }); + expect(response.json()).not.toMatchObject({ + shieldBlocked: true, + }); + expect(undiciFetchMock).toHaveBeenCalled(); + }); + + it('still returns shieldBlocked for non-new-api platforms when challenge html is detected', async () => { + adapterPlatformName = 'one-api'; + verifyTokenMock.mockResolvedValueOnce({ tokenType: 'unknown' }); + undiciFetchMock.mockResolvedValue({ + text: async () => '', + headers: { + get: (name: string) => (name.toLowerCase() === 'content-type' ? 'text/html; charset=utf-8' : null), + }, + }); + + const site = db.insert(schema.sites).values({ + name: 'Legacy Shielded', + url: 'https://legacy-shield.example.com', + platform: 'one-api', + }).returning().get(); + + const response = await app.inject({ + method: 'POST', + url: '/api/accounts/verify-token', + payload: { + siteId: site.id, + accessToken: 'session-or-cookie-token', + }, + }); + expect(response.statusCode).toBe(200); expect(response.json()).toMatchObject({ success: false, diff --git a/src/server/routes/proxy/chat.stream.test.ts b/src/server/routes/proxy/chat.stream.test.ts index 7665128..156b280 100644 --- a/src/server/routes/proxy/chat.stream.test.ts +++ b/src/server/routes/proxy/chat.stream.test.ts @@ -448,6 +448,41 @@ describe('chat proxy stream behavior', () => { expect(forwardedBody.thinking).toEqual({ type: 'enabled', budget_tokens: 1024 }); }); + it('stops forwarding extra SSE events after message_stop on /v1/messages', async () => { + const encoder = new TextEncoder(); + const upstreamBody = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('event: message_start\ndata: {"type":"message_start","message":{"id":"msg_stop_early","model":"claude-opus-4-6"}}\n\n')); + controller.enqueue(encoder.encode('event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"hello"}}\n\n')); + controller.enqueue(encoder.encode('event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}\n\n')); + controller.enqueue(encoder.encode('event: message_stop\ndata: {"type":"message_stop"}\n\n')); + controller.enqueue(encoder.encode('event: ping\ndata: {"type":"ping"}\n\n')); + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + controller.close(); + }, + }); + + fetchMock.mockResolvedValue(new Response(upstreamBody, { + status: 200, + headers: { 'content-type': 'text/event-stream; charset=utf-8' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/messages', + payload: { + model: 'claude-opus-4-6', + stream: true, + max_tokens: 256, + messages: [{ role: 'user', content: 'hello' }], + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.body).toContain('event: message_stop'); + expect(response.body).not.toContain('event: ping'); + }); + it('normalizes Claude thinking adaptive type for legacy upstreams on /v1/messages', async () => { fetchMock.mockResolvedValue(new Response(JSON.stringify({ id: 'msg_headers_adaptive', @@ -655,7 +690,13 @@ describe('chat proxy stream behavior', () => { expect(targetUrl).toContain('/v1/responses'); const forwarded = JSON.parse(options.body); expect(forwarded.model).toBe('upstream-gpt'); - expect(forwarded.input).toBe('hello'); + expect(forwarded.input).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello' }], + }, + ]); }); it('continues downgrade to /v1/messages when /v1/chat/completions returns messages is required for /v1/responses', async () => { @@ -921,6 +962,409 @@ describe('chat proxy stream behavior', () => { expect(forwardedBody.reasoning).toEqual({ effort: 'high' }); }); + it('retries /v1/responses without metadata when upstream returns empty upstream_error', async () => { + fetchMock + .mockResolvedValueOnce(new Response(JSON.stringify({ + error: { + message: '', + type: 'upstream_error', + }, + }), { + status: 400, + headers: { 'content-type': 'application/json' }, + })) + .mockResolvedValueOnce(new Response(JSON.stringify({ + id: 'resp_retry_without_metadata', + object: 'response', + status: 'completed', + model: 'upstream-gpt', + output_text: 'ok after stripping metadata', + output: [], + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'gpt-5.2', + input: 'hello', + metadata: { trace_id: 'req-1' }, + }, + }); + + expect(response.statusCode).toBe(200); + const body = response.json(); + expect(body.output_text).toContain('ok after stripping metadata'); + + expect(fetchMock).toHaveBeenCalledTimes(2); + const [firstUrl, firstOptions] = fetchMock.mock.calls[0] as [string, any]; + const [secondUrl, secondOptions] = fetchMock.mock.calls[1] as [string, any]; + expect(firstUrl).toContain('/v1/responses'); + expect(secondUrl).toContain('/v1/responses'); + + const firstBody = JSON.parse(firstOptions.body); + const secondBody = JSON.parse(secondOptions.body); + expect(firstBody.metadata).toEqual({ trace_id: 'req-1' }); + expect(secondBody.metadata).toBeUndefined(); + }); + + it('retries /v1/responses with core body when upstream returns empty upstream_error', async () => { + fetchMock + .mockResolvedValueOnce(new Response(JSON.stringify({ + error: { + message: '', + type: 'upstream_error', + }, + }), { + status: 400, + headers: { 'content-type': 'application/json' }, + })) + .mockResolvedValueOnce(new Response(JSON.stringify({ + id: 'resp_retry_core_body', + object: 'response', + status: 'completed', + model: 'upstream-gpt', + output_text: 'ok after core retry', + output: [], + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'gpt-5.2', + input: 'hello', + store: true, + }, + }); + + expect(response.statusCode).toBe(200); + const body = response.json(); + expect(body.output_text).toContain('ok after core retry'); + + expect(fetchMock).toHaveBeenCalledTimes(2); + const [_firstUrl, firstOptions] = fetchMock.mock.calls[0] as [string, any]; + const [_secondUrl, secondOptions] = fetchMock.mock.calls[1] as [string, any]; + const firstBody = JSON.parse(firstOptions.body); + const secondBody = JSON.parse(secondOptions.body); + expect(firstBody.store).toBe(true); + expect(secondBody.store).toBeUndefined(); + expect(secondBody.input).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello' }], + }, + ]); + }); + + it('retries /v1/responses when upstream_error message is literal upstream_error and falls back to strict body', async () => { + fetchMock + .mockResolvedValueOnce(new Response(JSON.stringify({ + error: { + message: 'upstream_error', + type: 'upstream_error', + }, + }), { + status: 400, + headers: { 'content-type': 'application/json' }, + })) + .mockResolvedValueOnce(new Response(JSON.stringify({ + error: { + message: 'upstream_error', + type: 'upstream_error', + }, + }), { + status: 400, + headers: { 'content-type': 'application/json' }, + })) + .mockResolvedValueOnce(new Response(JSON.stringify({ + id: 'resp_retry_strict_body', + object: 'response', + status: 'completed', + model: 'upstream-gpt', + output_text: 'ok after strict retry', + output: [], + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'gpt-5.2', + input: 'hello', + temperature: 0.7, + }, + }); + + expect(response.statusCode).toBe(200); + const body = response.json(); + expect(body.output_text).toContain('ok after strict retry'); + + expect(fetchMock).toHaveBeenCalledTimes(3); + const [, firstOptions] = fetchMock.mock.calls[0] as [string, any]; + const [, secondOptions] = fetchMock.mock.calls[1] as [string, any]; + const [, thirdOptions] = fetchMock.mock.calls[2] as [string, any]; + const firstBody = JSON.parse(firstOptions.body); + const secondBody = JSON.parse(secondOptions.body); + const thirdBody = JSON.parse(thirdOptions.body); + expect(firstBody.temperature).toBe(0.7); + expect(secondBody.temperature).toBe(0.7); + expect(thirdBody.temperature).toBeUndefined(); + expect(thirdBody).toEqual({ + model: 'upstream-gpt', + input: [ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello' }], + }, + ], + stream: false, + }); + }); + + it('retries generic 400 /v1/responses with minimal headers for strict compatibility fallback', async () => { + fetchMock + .mockResolvedValueOnce(new Response(JSON.stringify({ + error: { + message: 'request validation failed', + type: 'invalid_request_error', + }, + }), { + status: 400, + headers: { 'content-type': 'application/json' }, + })) + .mockResolvedValueOnce(new Response(JSON.stringify({ + error: { + message: 'request validation failed', + type: 'invalid_request_error', + }, + }), { + status: 400, + headers: { 'content-type': 'application/json' }, + })) + .mockResolvedValueOnce(new Response(JSON.stringify({ + id: 'resp_retry_minimal_headers', + object: 'response', + status: 'completed', + model: 'upstream-gpt', + output_text: 'ok after minimal headers retry', + output: [], + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + headers: { + 'openai-beta': 'responses-2025-03-11', + }, + payload: { + model: 'gpt-5.2', + input: 'hello', + user: 'user-123', + }, + }); + + expect(response.statusCode).toBe(200); + const body = response.json(); + expect(body.output_text).toContain('ok after minimal headers retry'); + + expect(fetchMock).toHaveBeenCalledTimes(3); + const [, firstOptions] = fetchMock.mock.calls[0] as [string, any]; + const [, secondOptions] = fetchMock.mock.calls[1] as [string, any]; + const [, thirdOptions] = fetchMock.mock.calls[2] as [string, any]; + + expect(firstOptions.headers['openai-beta']).toBe('responses-2025-03-11'); + expect(secondOptions.headers['openai-beta']).toBe('responses-2025-03-11'); + expect(thirdOptions.headers['openai-beta']).toBeUndefined(); + + const firstBody = JSON.parse(firstOptions.body); + const secondBody = JSON.parse(secondOptions.body); + const thirdBody = JSON.parse(thirdOptions.body); + expect(firstBody.user).toBe('user-123'); + expect(secondBody.user).toBeUndefined(); + expect(thirdBody.user).toBeUndefined(); + }); + + it('returns concise Cloudflare host error on /v1/responses 502 html failures', async () => { + const html = 'qaq.al | 502: Bad gatewayCloudflare Ray ID: 9d6f7c889ffbc8eb'; + fetchMock.mockImplementation(() => Promise.resolve(new Response(html, { + status: 502, + headers: { 'content-type': 'text/html; charset=UTF-8' }, + }))); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'gpt-5.2', + input: 'hello', + }, + }); + + expect(response.statusCode).toBe(502); + const body = response.json(); + expect(body.error?.type).toBe('upstream_error'); + expect(body.error?.message).toContain('[upstream:'); + expect(body.error?.message).toContain('Cloudflare 502: Bad gateway'); + expect(body.error?.message).not.toContain(''); + }); + + it('downgrades /v1/responses to /v1/chat/completions when upstream returns 400 upstream_error', async () => { + fetchMock + .mockResolvedValueOnce(new Response(JSON.stringify({ + error: { + message: '', + type: 'upstream_error', + }, + }), { + status: 400, + headers: { 'content-type': 'application/json' }, + })) + .mockResolvedValueOnce(new Response(JSON.stringify({ + id: 'chatcmpl-fallback-upstream-error', + object: 'chat.completion', + model: 'upstream-gpt', + choices: [{ + index: 0, + message: { role: 'assistant', content: 'ok via chat fallback from upstream_error' }, + finish_reason: 'stop', + }], + usage: { prompt_tokens: 4, completion_tokens: 2, total_tokens: 6 }, + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'gpt-5.2', + input: 'hello', + }, + }); + + expect(response.statusCode).toBe(200); + expect(fetchMock).toHaveBeenCalledTimes(2); + const [firstUrl] = fetchMock.mock.calls[0] as [string, any]; + const [secondUrl] = fetchMock.mock.calls[1] as [string, any]; + expect(firstUrl).toContain('/v1/responses'); + expect(secondUrl).toContain('/v1/chat/completions'); + + const body = response.json(); + expect(body.object).toBe('response'); + expect(body.output_text).toContain('ok via chat fallback from upstream_error'); + }); + + it('downgrades /v1/responses to /v1/chat/completions when upstream responses endpoint returns 502', async () => { + fetchMock + .mockResolvedValueOnce(new Response( + 'qaq.al | 502: Bad gatewayCloudflare', + { + status: 502, + headers: { 'content-type': 'text/html; charset=UTF-8' }, + }, + )) + .mockResolvedValueOnce(new Response(JSON.stringify({ + id: 'chatcmpl-fallback-502', + object: 'chat.completion', + model: 'upstream-gpt', + choices: [{ + index: 0, + message: { role: 'assistant', content: 'ok via chat fallback' }, + finish_reason: 'stop', + }], + usage: { prompt_tokens: 3, completion_tokens: 2, total_tokens: 5 }, + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'gpt-5.2', + input: 'hello', + }, + }); + + expect(response.statusCode).toBe(200); + expect(fetchMock).toHaveBeenCalledTimes(2); + const [firstUrl] = fetchMock.mock.calls[0] as [string, any]; + const [secondUrl] = fetchMock.mock.calls[1] as [string, any]; + expect(firstUrl).toContain('/v1/responses'); + expect(secondUrl).toContain('/v1/chat/completions'); + const body = response.json(); + expect(body.object).toBe('response'); + expect(body.output_text).toContain('ok via chat fallback'); + }); + + it('retries /v1/chat/completions with minimal JSON headers on unsupported media type', async () => { + fetchMock + .mockResolvedValueOnce(new Response(JSON.stringify({ + error: { + message: "Unsupported Media Type: Only 'application/json' is allowed", + type: 'invalid_request_error', + }, + }), { + status: 415, + headers: { 'content-type': 'application/json' }, + })) + .mockResolvedValueOnce(new Response(JSON.stringify({ + id: 'chatcmpl-media-type-retry', + object: 'chat.completion', + created: 1_706_123_456, + model: 'upstream-gpt', + choices: [{ + index: 0, + message: { role: 'assistant', content: 'ok after header retry' }, + finish_reason: 'stop', + }], + usage: { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 }, + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/chat/completions', + headers: { + 'openai-beta': 'responses-2025-03-11', + }, + payload: { + model: 'gpt-5.2', + messages: [{ role: 'user', content: 'hello' }], + }, + }); + + expect(response.statusCode).toBe(200); + expect(fetchMock).toHaveBeenCalledTimes(2); + const [firstUrl, firstOptions] = fetchMock.mock.calls[0] as [string, any]; + const [secondUrl, secondOptions] = fetchMock.mock.calls[1] as [string, any]; + expect(firstUrl).toContain('/v1/chat/completions'); + expect(secondUrl).toContain('/v1/chat/completions'); + expect(firstOptions.headers['openai-beta']).toBe('responses-2025-03-11'); + expect(secondOptions.headers['openai-beta']).toBeUndefined(); + expect(secondOptions.headers['content-type']).toBe('application/json'); + expect(secondOptions.headers.accept).toBe('application/json'); + }); + it('sets stream accept header for /v1/responses when downstream omits accept', async () => { const encoder = new TextEncoder(); const upstreamBody = new ReadableStream({ @@ -952,6 +1396,99 @@ describe('chat proxy stream behavior', () => { expect(options.headers.accept).toBe('text/event-stream'); }); + it('deduplicates cumulative text chunks when /v1/responses is converted from /v1/messages stream', async () => { + fetchModelPricingCatalogMock.mockResolvedValue({ + models: [ + { + modelName: 'upstream-gpt', + supportedEndpointTypes: ['/v1/messages'], + }, + ], + groupRatio: {}, + }); + + const fullText = "I'm Claude, an AI assistant made by Anthropic."; + const encoder = new TextEncoder(); + const upstreamBody = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('event: message_start\ndata: {"type":"message_start","message":{"id":"msg_dup_1","model":"upstream-gpt"}}\n\n')); + controller.enqueue(encoder.encode(`event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":${JSON.stringify(fullText)}}}\n\n`)); + controller.enqueue(encoder.encode(`event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":${JSON.stringify(fullText)}}}\n\n`)); + controller.enqueue(encoder.encode(`event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":${JSON.stringify(fullText)}}}\n\n`)); + controller.enqueue(encoder.encode('event: message_stop\ndata: {"type":"message_stop"}\n\n')); + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + controller.close(); + }, + }); + + fetchMock.mockResolvedValue(new Response(upstreamBody, { + status: 200, + headers: { 'content-type': 'text/event-stream; charset=utf-8' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'claude-sonnet-4-6', + stream: true, + input: 'whoru', + }, + }); + + expect(response.statusCode).toBe(200); + const deltaMatches = response.body.match(/event: response\.output_text\.delta/g) || []; + expect(deltaMatches.length).toBe(1); + const textMatches = response.body.match(/I'm Claude, an AI assistant made by Anthropic\./g) || []; + expect(textMatches.length).toBeGreaterThan(0); + expect(textMatches.length).toBeLessThanOrEqual(4); + }); + + it('deduplicates overlapping text windows when /v1/responses is converted from /v1/messages stream', async () => { + fetchModelPricingCatalogMock.mockResolvedValue({ + models: [ + { + modelName: 'upstream-gpt', + supportedEndpointTypes: ['/v1/messages'], + }, + ], + groupRatio: {}, + }); + + const encoder = new TextEncoder(); + const upstreamBody = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('event: message_start\ndata: {"type":"message_start","message":{"id":"msg_overlap_1","model":"upstream-gpt"}}\n\n')); + controller.enqueue(encoder.encode('event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"The user asked \\"whoru\\" which is a common"}}\n\n')); + controller.enqueue(encoder.encode('event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"which is a common internet slang"}}\n\n')); + controller.enqueue(encoder.encode('event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":" internet slang and shorthand for who are you."}}\n\n')); + controller.enqueue(encoder.encode('event: message_stop\ndata: {"type":"message_stop"}\n\n')); + controller.enqueue(encoder.encode('data: [DONE]\n\n')); + controller.close(); + }, + }); + + fetchMock.mockResolvedValue(new Response(upstreamBody, { + status: 200, + headers: { 'content-type': 'text/event-stream; charset=utf-8' }, + })); + + const response = await app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'claude-sonnet-4-6', + stream: true, + input: 'whoru', + }, + }); + + expect(response.statusCode).toBe(200); + expect(response.body).toContain('"delta":" internet slang"'); + expect(response.body).not.toContain('"delta":"which is a common internet slang"'); + expect(response.body).toContain('"text":"The user asked \\"whoru\\" which is a common internet slang and shorthand for who are you."'); + }); + it('preserves function_call/function_call_output when /v1/responses falls back to /v1/chat/completions', async () => { fetchModelPricingCatalogMock.mockResolvedValue({ models: [ diff --git a/src/server/routes/proxy/chat.ts b/src/server/routes/proxy/chat.ts index fa58838..70a3b06 100644 --- a/src/server/routes/proxy/chat.ts +++ b/src/server/routes/proxy/chat.ts @@ -24,8 +24,10 @@ import { buildSyntheticOpenAiChunks, } from './chatFormats.js'; import { + buildMinimalJsonHeadersForCompatibility, buildUpstreamEndpointRequest, isEndpointDowngradeError, + isUnsupportedMediaTypeError, resolveUpstreamEndpointCandidates, } from './upstreamEndpoint.js'; import { @@ -34,6 +36,7 @@ import { recordDownstreamCostUsage, } from './downstreamPolicy.js'; import { composeProxyLogMessage } from './logPathMeta.js'; +import { executeEndpointFlow, withUpstreamPath } from './endpointFlow.js'; const MAX_RETRIES = 2; const CLAUDE_SSE_EVENT_NAMES = new Set([ @@ -47,10 +50,6 @@ const CLAUDE_SSE_EVENT_NAMES = new Set([ 'error', ]); -function withUpstreamPath(path: string, message: string): string { - return `[upstream:${path}] ${message}`; -} - function shouldRetryClaudeMessagesWithNormalizedBody( downstreamFormat: DownstreamFormat, endpointPath: string, @@ -181,45 +180,13 @@ async function handleChatProxyRequest( let startTime = Date.now(); try { - let upstream: Awaited> | null = null; - let successfulUpstreamPath: string | null = null; - let finalStatus = 0; - let finalErrText = 'unknown error'; - - for (let endpointIndex = 0; endpointIndex < endpointCandidates.length; endpointIndex += 1) { - const endpointRequest = buildUpstreamEndpointRequest({ - endpoint: endpointCandidates[endpointIndex], - modelName, - stream: isStream, - tokenValue: selected.tokenValue, - sitePlatform: selected.site.platform, - siteUrl: selected.site.url, - openaiBody: upstreamBody, - downstreamFormat, - claudeOriginalBody, - downstreamHeaders: request.headers as Record, - }); - - const targetUrl = `${selected.site.url}${endpointRequest.path}`; - startTime = Date.now(); - - const response = await fetch(targetUrl, withExplicitProxyRequestInit(selected.site.proxyUrl, { - method: 'POST', - headers: endpointRequest.headers, - body: JSON.stringify(endpointRequest.body), - })); - - if (response.ok) { - upstream = response; - successfulUpstreamPath = endpointRequest.path; - break; - } - - const rawErrText = await response.text().catch(() => 'unknown error'); - - if (shouldRetryClaudeMessagesWithNormalizedBody(downstreamFormat, endpointRequest.path, response.status, rawErrText)) { - const normalizedClaudeRequest = buildUpstreamEndpointRequest({ - endpoint: endpointCandidates[endpointIndex], + const endpointResult = await executeEndpointFlow({ + siteUrl: selected.site.url, + proxyUrl: selected.site.proxyUrl, + endpointCandidates, + buildRequest: (endpoint) => { + const endpointRequest = buildUpstreamEndpointRequest({ + endpoint, modelName, stream: isStream, tokenValue: selected.tokenValue, @@ -227,76 +194,117 @@ async function handleChatProxyRequest( siteUrl: selected.site.url, openaiBody: upstreamBody, downstreamFormat, - claudeOriginalBody: undefined, + claudeOriginalBody, downstreamHeaders: request.headers as Record, }); - const normalizedTargetUrl = `${selected.site.url}${normalizedClaudeRequest.path}`; - const normalizedResponse = await fetch(normalizedTargetUrl, withExplicitProxyRequestInit(selected.site.proxyUrl, { + return { + endpoint, + path: endpointRequest.path, + headers: endpointRequest.headers, + body: endpointRequest.body as Record, + }; + }, + tryRecover: async (ctx) => { + if (shouldRetryClaudeMessagesWithNormalizedBody( + downstreamFormat, + ctx.request.path, + ctx.response.status, + ctx.rawErrText, + )) { + const normalizedClaudeRequest = buildUpstreamEndpointRequest({ + endpoint: ctx.request.endpoint, + modelName, + stream: isStream, + tokenValue: selected.tokenValue, + sitePlatform: selected.site.platform, + siteUrl: selected.site.url, + openaiBody: upstreamBody, + downstreamFormat, + claudeOriginalBody: undefined, + downstreamHeaders: request.headers as Record, + }); + const normalizedTargetUrl = `${selected.site.url}${normalizedClaudeRequest.path}`; + const normalizedResponse = await fetch(normalizedTargetUrl, withExplicitProxyRequestInit(selected.site.proxyUrl, { + method: 'POST', + headers: normalizedClaudeRequest.headers, + body: JSON.stringify(normalizedClaudeRequest.body), + })); + + if (normalizedResponse.ok) { + return { + upstream: normalizedResponse, + upstreamPath: normalizedClaudeRequest.path, + }; + } + + ctx.request = { + ...ctx.request, + path: normalizedClaudeRequest.path, + headers: normalizedClaudeRequest.headers, + body: normalizedClaudeRequest.body as Record, + }; + ctx.response = normalizedResponse; + ctx.rawErrText = await normalizedResponse.text().catch(() => 'unknown error'); + } + + if (!isUnsupportedMediaTypeError(ctx.response.status, ctx.rawErrText)) { + return null; + } + + const minimalHeaders = buildMinimalJsonHeadersForCompatibility({ + headers: ctx.request.headers, + endpoint: ctx.request.endpoint, + stream: isStream, + }); + const normalizedCurrentHeaders = Object.fromEntries( + Object.entries(ctx.request.headers).map(([key, value]) => [key.toLowerCase(), value]), + ); + if (JSON.stringify(minimalHeaders) === JSON.stringify(normalizedCurrentHeaders)) { + return null; + } + + const minimalResponse = await fetch(ctx.targetUrl, withExplicitProxyRequestInit(selected.site.proxyUrl, { method: 'POST', - headers: normalizedClaudeRequest.headers, - body: JSON.stringify(normalizedClaudeRequest.body), + headers: minimalHeaders, + body: JSON.stringify(ctx.request.body), })); - if (normalizedResponse.ok) { - upstream = normalizedResponse; - successfulUpstreamPath = normalizedClaudeRequest.path; - break; + if (minimalResponse.ok) { + return { + upstream: minimalResponse, + upstreamPath: ctx.request.path, + }; } - const normalizedErrText = await normalizedResponse.text().catch(() => 'unknown error'); - const shouldDowngradeEndpoint = ( - endpointIndex < endpointCandidates.length - 1 - && ( - isEndpointDowngradeError(normalizedResponse.status, normalizedErrText) - || isMessagesRequiredError(normalizedErrText) - ) + ctx.request = { + ...ctx.request, + headers: minimalHeaders, + }; + ctx.response = minimalResponse; + ctx.rawErrText = await minimalResponse.text().catch(() => 'unknown error'); + return null; + }, + shouldDowngrade: (ctx) => ( + isEndpointDowngradeError(ctx.response.status, ctx.rawErrText) + || isMessagesRequiredError(ctx.rawErrText) + ), + onDowngrade: (ctx) => { + logProxy( + selected, + requestedModel, + 'failed', + ctx.response.status, + Date.now() - startTime, + ctx.errText, + retryCount, + downstreamPath, ); + }, + }); - if (shouldDowngradeEndpoint) { - const errText = withUpstreamPath( - normalizedClaudeRequest.path, - `normalized-claude-body fallback failed: ${normalizedErrText}`, - ); - logProxy( - selected, - requestedModel, - 'failed', - normalizedResponse.status, - Date.now() - startTime, - errText, - retryCount, - downstreamPath, - ); - continue; - } - - finalStatus = normalizedResponse.status; - finalErrText = withUpstreamPath( - normalizedClaudeRequest.path, - `normalized-claude-body fallback failed: ${normalizedErrText}`, - ); - break; - } - - const errText = withUpstreamPath(endpointRequest.path, rawErrText); - const shouldDowngradeEndpoint = ( - endpointIndex < endpointCandidates.length - 1 - && isEndpointDowngradeError(response.status, rawErrText) - ); - - if (shouldDowngradeEndpoint) { - logProxy(selected, requestedModel, 'failed', response.status, Date.now() - startTime, errText, retryCount, downstreamPath); - continue; - } - - finalStatus = response.status; - finalErrText = errText; - break; - } - - if (!upstream) { - const status = finalStatus || 502; - const errText = finalErrText || 'unknown error'; + if (!endpointResult.ok) { + const status = endpointResult.status || 502; + const errText = endpointResult.errText || 'unknown error'; tokenRouter.recordFailure(selected.channel.id); logProxy(selected, requestedModel, 'failed', status, Date.now() - startTime, errText, retryCount, downstreamPath); @@ -324,6 +332,9 @@ async function handleChatProxyRequest( }); } + const upstream = endpointResult.upstream; + const successfulUpstreamPath = endpointResult.upstreamPath; + if (isStream) { reply.hijack(); reply.raw.statusCode = 200; @@ -451,12 +462,14 @@ async function handleChatProxyRequest( const decoder = new TextDecoder(); let sseBuffer = ''; + let shouldTerminateEarly = false; const consumeSseBuffer = (incoming: string): string => { const pulled = pullSseEventsWithDone(incoming); for (const eventBlock of pulled.events) { if (eventBlock.data === '[DONE]') { writeDone(); + shouldTerminateEarly = true; continue; } @@ -486,6 +499,10 @@ async function handleChatProxyRequest( claudeContext, ); reply.raw.write(serializeRawSseEvent(claudeEventName, eventBlock.data)); + if (claudeContext.doneSent) { + shouldTerminateEarly = true; + break; + } continue; } } @@ -497,6 +514,14 @@ async function handleChatProxyRequest( streamContext, claudeContext, )); + if (downstreamFormat === 'claude' && claudeContext.doneSent) { + shouldTerminateEarly = true; + break; + } + if (downstreamFormat === 'openai' && streamContext.doneSent) { + shouldTerminateEarly = true; + break; + } continue; } @@ -506,6 +531,10 @@ async function handleChatProxyRequest( writeLines(serializeNormalizedStreamEvent('claude', { contentDelta: eventBlock.data, }, streamContext, claudeContext)); + if (claudeContext.doneSent) { + shouldTerminateEarly = true; + break; + } } } @@ -520,10 +549,16 @@ async function handleChatProxyRequest( sseBuffer += decoder.decode(value, { stream: true }); sseBuffer = consumeSseBuffer(sseBuffer); + if (shouldTerminateEarly) { + await reader.cancel().catch(() => {}); + break; + } } - sseBuffer += decoder.decode(); - if (sseBuffer.trim().length > 0) { + if (!shouldTerminateEarly) { + sseBuffer += decoder.decode(); + } + if (!shouldTerminateEarly && sseBuffer.trim().length > 0) { sseBuffer = consumeSseBuffer(`${sseBuffer}\n\n`); } } finally { diff --git a/src/server/routes/proxy/endpointFlow.test.ts b/src/server/routes/proxy/endpointFlow.test.ts new file mode 100644 index 0000000..b029566 --- /dev/null +++ b/src/server/routes/proxy/endpointFlow.test.ts @@ -0,0 +1,133 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { fetch } from 'undici'; +import { executeEndpointFlow, type BuiltEndpointRequest } from './endpointFlow.js'; + +vi.mock('undici', () => ({ + fetch: vi.fn(), +})); + +const fetchMock = vi.mocked(fetch); + +function requestFor(path: string): BuiltEndpointRequest { + return { + endpoint: 'responses', + path, + headers: { 'content-type': 'application/json' }, + body: { model: 'gpt-5.2', input: 'hello' }, + }; +} + +function toUndiciResponse(response: Response): Awaited> { + return response as unknown as Awaited>; +} + +describe('executeEndpointFlow', () => { + beforeEach(() => { + fetchMock.mockReset(); + }); + + it('returns the first successful upstream response', async () => { + fetchMock.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({ ok: true }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }))); + + const result = await executeEndpointFlow({ + siteUrl: 'https://example.com', + endpointCandidates: ['responses'], + buildRequest: () => requestFor('/v1/responses'), + }); + + expect(result.ok).toBe(true); + if (result.ok) { + expect(result.upstreamPath).toBe('/v1/responses'); + } + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('downgrades to next endpoint when policy allows', async () => { + fetchMock + .mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({ + error: { message: 'unsupported endpoint', type: 'invalid_request_error' }, + }), { + status: 404, + headers: { 'content-type': 'application/json' }, + }))) + .mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({ ok: true }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }))); + + const downgradedPaths: string[] = []; + const result = await executeEndpointFlow({ + siteUrl: 'https://example.com', + endpointCandidates: ['responses', 'chat'], + buildRequest: (endpoint) => endpoint === 'responses' + ? requestFor('/v1/responses') + : { ...requestFor('/v1/chat/completions'), endpoint }, + shouldDowngrade: () => true, + onDowngrade: (ctx) => { + downgradedPaths.push(ctx.request.path); + }, + }); + + expect(result.ok).toBe(true); + if (result.ok) { + expect(result.upstreamPath).toBe('/v1/chat/completions'); + } + expect(downgradedPaths).toEqual(['/v1/responses']); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it('accepts recovered response from tryRecover hook', async () => { + fetchMock.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({ + error: { message: 'upstream_error', type: 'upstream_error' }, + }), { + status: 400, + headers: { 'content-type': 'application/json' }, + }))); + + const recovered = toUndiciResponse(new Response(JSON.stringify({ ok: 'recovered' }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const result = await executeEndpointFlow({ + siteUrl: 'https://example.com', + endpointCandidates: ['responses'], + buildRequest: () => requestFor('/v1/responses'), + tryRecover: async () => ({ + upstream: recovered, + upstreamPath: '/v1/responses', + }), + }); + + expect(result.ok).toBe(true); + if (result.ok) { + expect(result.upstreamPath).toBe('/v1/responses'); + } + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('returns normalized final error when all endpoints fail', async () => { + fetchMock.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({ + error: { message: 'upstream_error', type: 'upstream_error' }, + }), { + status: 400, + headers: { 'content-type': 'application/json' }, + }))); + + const result = await executeEndpointFlow({ + siteUrl: 'https://example.com', + endpointCandidates: ['responses'], + buildRequest: () => requestFor('/v1/responses'), + }); + + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.status).toBe(400); + expect(result.errText).toContain('[upstream:/v1/responses]'); + expect(result.errText).toContain('Upstream returned HTTP 400'); + } + }); +}); diff --git a/src/server/routes/proxy/endpointFlow.ts b/src/server/routes/proxy/endpointFlow.ts new file mode 100644 index 0000000..ed0f250 --- /dev/null +++ b/src/server/routes/proxy/endpointFlow.ts @@ -0,0 +1,134 @@ +import { fetch } from 'undici'; +import { withExplicitProxyRequestInit } from '../../services/siteProxy.js'; +import { summarizeUpstreamError } from './upstreamError.js'; +import type { UpstreamEndpoint } from './upstreamEndpoint.js'; + +export type BuiltEndpointRequest = { + endpoint: UpstreamEndpoint; + path: string; + headers: Record; + body: Record; +}; + +export type EndpointAttemptContext = { + endpointIndex: number; + endpointCount: number; + request: BuiltEndpointRequest; + targetUrl: string; + response: Awaited>; + rawErrText: string; +}; + +export type EndpointRecoverResult = { + upstream: Awaited>; + upstreamPath: string; +} | null; + +export type EndpointFlowResult = + | { + ok: true; + upstream: Awaited>; + upstreamPath: string; + } + | { + ok: false; + status: number; + errText: string; + }; + +export function withUpstreamPath(path: string, message: string): string { + return `[upstream:${path}] ${message}`; +} + +type ExecuteEndpointFlowInput = { + siteUrl: string; + proxyUrl?: string | null; + endpointCandidates: UpstreamEndpoint[]; + buildRequest: (endpoint: UpstreamEndpoint, endpointIndex: number) => BuiltEndpointRequest; + tryRecover?: (ctx: EndpointAttemptContext) => Promise; + shouldDowngrade?: (ctx: EndpointAttemptContext) => boolean; + onDowngrade?: (ctx: EndpointAttemptContext & { errText: string }) => void | Promise; +}; + +export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Promise { + const endpointCount = input.endpointCandidates.length; + if (endpointCount <= 0) { + return { + ok: false, + status: 502, + errText: 'Upstream request failed', + }; + } + + let finalStatus = 0; + let finalErrText = 'unknown error'; + + for (let endpointIndex = 0; endpointIndex < endpointCount; endpointIndex += 1) { + const endpoint = input.endpointCandidates[endpointIndex] as UpstreamEndpoint; + const request = input.buildRequest(endpoint, endpointIndex); + const targetUrl = `${input.siteUrl}${request.path}`; + + let response = await fetch(targetUrl, withExplicitProxyRequestInit(input.proxyUrl, { + method: 'POST', + headers: request.headers, + body: JSON.stringify(request.body), + })); + + if (response.ok) { + return { + ok: true, + upstream: response, + upstreamPath: request.path, + }; + } + + let rawErrText = await response.text().catch(() => 'unknown error'); + const baseContext: EndpointAttemptContext = { + endpointIndex, + endpointCount, + request, + targetUrl, + response, + rawErrText, + }; + + if (input.tryRecover) { + const recovered = await input.tryRecover(baseContext); + if (recovered?.upstream?.ok) { + return { + ok: true, + upstream: recovered.upstream, + upstreamPath: recovered.upstreamPath, + }; + } + } + + // Normalize again in case recoverer performed additional probes and updated the response text. + rawErrText = baseContext.rawErrText; + response = baseContext.response; + const errText = withUpstreamPath( + baseContext.request.path, + summarizeUpstreamError(response.status, rawErrText), + ); + + const isLastEndpoint = endpointIndex >= endpointCount - 1; + const shouldDowngrade = !isLastEndpoint && !!input.shouldDowngrade?.(baseContext); + if (shouldDowngrade) { + await input.onDowngrade?.({ + ...baseContext, + errText, + }); + continue; + } + + finalStatus = response.status; + finalErrText = errText; + break; + } + + return { + ok: false, + status: finalStatus || 502, + errText: finalErrText || 'unknown error', + }; +} diff --git a/src/server/routes/proxy/responses.ts b/src/server/routes/proxy/responses.ts index c109d16..1303689 100644 --- a/src/server/routes/proxy/responses.ts +++ b/src/server/routes/proxy/responses.ts @@ -18,20 +18,19 @@ import { type NormalizedStreamEvent, } from './chatFormats.js'; import { + buildMinimalJsonHeadersForCompatibility, buildUpstreamEndpointRequest, isEndpointDowngradeError, + isUnsupportedMediaTypeError, resolveUpstreamEndpointCandidates, type UpstreamEndpoint, } from './upstreamEndpoint.js'; import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js'; import { composeProxyLogMessage } from './logPathMeta.js'; +import { executeEndpointFlow, withUpstreamPath } from './endpointFlow.js'; const MAX_RETRIES = 2; -function withUpstreamPath(path: string, message: string): string { - return `[upstream:${path}] ${message}`; -} - function shouldDowngradeFromChatToMessagesForResponses( endpointPath: string, status: number, @@ -42,6 +41,186 @@ function shouldDowngradeFromChatToMessagesForResponses( return /messages\s+is\s+required/i.test(upstreamErrorText); } +function parseUpstreamErrorShape(rawText: string): { + type: string; + code: string; + message: string; +} { + try { + const parsed = JSON.parse(rawText) as Record; + const error = (parsed.error && typeof parsed.error === 'object') + ? parsed.error as Record + : parsed; + return { + type: typeof error.type === 'string' ? error.type.trim().toLowerCase() : '', + code: typeof error.code === 'string' ? error.code.trim().toLowerCase() : '', + message: typeof error.message === 'string' ? error.message.trim() : '', + }; + } catch { + return { type: '', code: '', message: '' }; + } +} + +function stripResponsesMetadata( + body: Record, +): Record | null { + if (!Object.prototype.hasOwnProperty.call(body, 'metadata')) return null; + const next = { ...body }; + delete next.metadata; + return next; +} + +function toFiniteNumber(value: unknown): number | null { + return typeof value === 'number' && Number.isFinite(value) ? value : null; +} + +function buildCoreResponsesBody( + body: Record, +): Record | null { + const model = typeof body.model === 'string' ? body.model.trim() : ''; + if (!model) return null; + if (body.input === undefined) return null; + + const core: Record = { + model, + input: body.input, + stream: body.stream === true, + }; + + const maxOutputTokens = toFiniteNumber(body.max_output_tokens); + if (maxOutputTokens !== null && maxOutputTokens > 0) { + core.max_output_tokens = Math.trunc(maxOutputTokens); + } + + const temperature = toFiniteNumber(body.temperature); + if (temperature !== null) core.temperature = temperature; + + const topP = toFiniteNumber(body.top_p); + if (topP !== null) core.top_p = topP; + + const instructions = typeof body.instructions === 'string' ? body.instructions.trim() : ''; + if (instructions) core.instructions = instructions; + + return core; +} + +function buildStrictResponsesBody( + body: Record, +): Record | null { + const model = typeof body.model === 'string' ? body.model.trim() : ''; + if (!model) return null; + if (body.input === undefined) return null; + + return { + model, + input: body.input, + stream: body.stream === true, + }; +} + +function buildResponsesCompatibilityBodies( + body: Record, +): Record[] { + const candidates: Record[] = []; + const seen = new Set(); + try { + const originalKey = JSON.stringify(body); + if (originalKey) seen.add(originalKey); + } catch { + // ignore non-serializable bodies + } + const push = (next: Record | null) => { + if (!next) return; + let key = ''; + try { + key = JSON.stringify(next); + } catch { + return; + } + if (!key || seen.has(key)) return; + seen.add(key); + candidates.push(next); + }; + + push(stripResponsesMetadata(body)); + push(buildCoreResponsesBody(body)); + push(buildStrictResponsesBody(body)); + return candidates; +} + +function buildResponsesCompatibilityHeaderCandidates( + headers: Record, + stream: boolean, +): Record[] { + const candidates: Record[] = []; + const seen = new Set(); + const push = (next: Record) => { + const normalizedEntries = Object.entries(next) + .filter(([, value]) => typeof value === 'string' && value.trim().length > 0) + .map(([key, value]) => [key.toLowerCase(), value] as const) + .sort(([a], [b]) => a.localeCompare(b)); + const key = JSON.stringify(normalizedEntries); + if (!key || seen.has(key)) return; + seen.add(key); + candidates.push(Object.fromEntries(normalizedEntries)); + }; + + push(headers); + + const minimal: Record = {}; + for (const [rawKey, rawValue] of Object.entries(headers)) { + const key = rawKey.toLowerCase(); + if ( + key === 'authorization' + || key === 'x-api-key' + || key === 'content-type' + || key === 'accept' + ) { + minimal[key] = rawValue; + } + } + if (!minimal['content-type']) minimal['content-type'] = 'application/json'; + if (stream && !minimal.accept) minimal.accept = 'text/event-stream'; + push(minimal); + + return candidates; +} + +function shouldRetryResponsesCompatibility(input: { + endpoint: UpstreamEndpoint; + status: number; + rawErrText: string; +}): boolean { + if (input.endpoint !== 'responses') return false; + if (input.status !== 400) return false; + const parsedError = parseUpstreamErrorShape(input.rawErrText); + const type = parsedError.type.trim().toLowerCase(); + const code = parsedError.code.trim().toLowerCase(); + const message = parsedError.message.trim().toLowerCase(); + const compact = `${type} ${code} ${message}`.trim(); + const rawCompact = (input.rawErrText || '').toLowerCase(); + + // Authentication/authorization failures should not enter compatibility retries. + if ( + compact.includes('invalid_api_key') + || compact.includes('authentication') + || compact.includes('unauthorized') + || compact.includes('forbidden') + || compact.includes('insufficient_quota') + || compact.includes('rate_limit') + ) { + return false; + } + + if (type === 'upstream_error' || code === 'upstream_error') return true; + if (message === 'upstream_error' || message === 'upstream request failed') return true; + if (rawCompact.includes('upstream_error')) return true; + + // Many sub2api-compatible gateways return generic 400 for field incompatibilities. + // Retry with progressively stricter payload/header candidates to maximize compatibility. + return true; +} + function isRecord(value: unknown): value is Record { return !!value && typeof value === 'object'; } @@ -677,6 +856,29 @@ function ensureMessageItem(state: ResponsesStreamState): string[] { ]; } +function toIncrementalText(existingText: string, incomingText: string): string { + if (!incomingText) return ''; + if (!existingText) return incomingText; + + if (incomingText === existingText) return ''; + if (incomingText.startsWith(existingText)) { + return incomingText.slice(existingText.length); + } + if (existingText.endsWith(incomingText)) return ''; + + // Some upstreams emit overlapping windows instead of strict deltas. + // Keep only the non-overlapping suffix to avoid duplicated text. + const maxOverlap = Math.min(existingText.length, incomingText.length); + const MIN_OVERLAP = 8; + for (let overlap = maxOverlap; overlap >= MIN_OVERLAP; overlap -= 1) { + if (existingText.slice(-overlap) === incomingText.slice(0, overlap)) { + return incomingText.slice(overlap); + } + } + + return incomingText; +} + function appendMessageDelta(state: ResponsesStreamState, delta: string): string[] { if (!delta) return []; @@ -684,12 +886,15 @@ function appendMessageDelta(state: ResponsesStreamState, delta: string): string[ events.push(...ensureMessageItem(state)); if (!state.messageItem) return events; - state.messageItem.text += delta; + const normalizedDelta = toIncrementalText(state.messageItem.text, delta); + if (!normalizedDelta) return events; + + state.messageItem.text += normalizedDelta; events.push(emitResponsesEvent(state, 'response.output_text.delta', { item_id: state.messageItem.itemId, output_index: state.messageItem.outputIndex, content_index: state.messageItem.contentIndex, - delta, + delta: normalizedDelta, })); return events; } @@ -989,66 +1194,128 @@ export async function responsesProxyRoute(app: FastifyInstance) { const startTime = Date.now(); try { - let upstream: Awaited> | null = null; - let successfulUpstreamPath: string | null = null; - let finalStatus = 0; - let finalErrText = 'unknown error'; + const endpointResult = await executeEndpointFlow({ + siteUrl: selected.site.url, + proxyUrl: selected.site.proxyUrl, + endpointCandidates, + buildRequest: (endpoint) => { + const endpointRequest = buildUpstreamEndpointRequest({ + endpoint, + modelName, + stream: isStream, + tokenValue: selected.tokenValue, + sitePlatform: selected.site.platform, + siteUrl: selected.site.url, + openaiBody: openAiBody, + downstreamFormat: 'responses', + responsesOriginalBody: body, + downstreamHeaders: request.headers as Record, + }); + return { + endpoint, + path: endpointRequest.path, + headers: endpointRequest.headers, + body: endpointRequest.body as Record, + }; + }, + tryRecover: async (ctx) => { + if (shouldRetryResponsesCompatibility({ + endpoint: ctx.request.endpoint, + status: ctx.response.status, + rawErrText: ctx.rawErrText, + })) { + const compatibilityBodies = buildResponsesCompatibilityBodies(ctx.request.body); + const compatibilityHeaders = buildResponsesCompatibilityHeaderCandidates( + ctx.request.headers, + isStream, + ); - for (let endpointIndex = 0; endpointIndex < endpointCandidates.length; endpointIndex += 1) { - const endpoint = endpointCandidates[endpointIndex] as UpstreamEndpoint; - const endpointRequest = buildUpstreamEndpointRequest({ - endpoint, - modelName, - stream: isStream, - tokenValue: selected.tokenValue, - sitePlatform: selected.site.platform, - siteUrl: selected.site.url, - openaiBody: openAiBody, - downstreamFormat: 'responses', - responsesOriginalBody: body, - downstreamHeaders: request.headers as Record, - }); - const targetUrl = `${selected.site.url}${endpointRequest.path}`; + for (const compatibilityHeadersCandidate of compatibilityHeaders) { + for (const compatibilityBody of compatibilityBodies) { + const compatibilityResponse = await fetch( + ctx.targetUrl, + withExplicitProxyRequestInit(selected.site.proxyUrl, { + method: 'POST', + headers: compatibilityHeadersCandidate, + body: JSON.stringify(compatibilityBody), + }), + ); + if (compatibilityResponse.ok) { + return { + upstream: compatibilityResponse, + upstreamPath: ctx.request.path, + }; + } - const response = await fetch(targetUrl, withExplicitProxyRequestInit(selected.site.proxyUrl, { - method: 'POST', - headers: endpointRequest.headers, - body: JSON.stringify(endpointRequest.body), - })); + ctx.request = { + ...ctx.request, + headers: compatibilityHeadersCandidate, + body: compatibilityBody, + }; + ctx.response = compatibilityResponse; + ctx.rawErrText = await compatibilityResponse.text().catch(() => 'unknown error'); + } + } + } - if (response.ok) { - upstream = response; - successfulUpstreamPath = endpointRequest.path; - break; - } + if (!isUnsupportedMediaTypeError(ctx.response.status, ctx.rawErrText)) { + return null; + } - const rawErrText = await response.text().catch(() => 'unknown error'); - const errText = withUpstreamPath(endpointRequest.path, rawErrText); - const shouldDowngradeEndpoint = ( - endpointIndex < endpointCandidates.length - 1 - && ( - isEndpointDowngradeError(response.status, rawErrText) - || shouldDowngradeFromChatToMessagesForResponses( - endpointRequest.path, - response.status, - rawErrText, - ) + const minimalHeaders = buildMinimalJsonHeadersForCompatibility({ + headers: ctx.request.headers, + endpoint: ctx.request.endpoint, + stream: isStream, + }); + const minimalResponse = await fetch( + ctx.targetUrl, + withExplicitProxyRequestInit(selected.site.proxyUrl, { + method: 'POST', + headers: minimalHeaders, + body: JSON.stringify(ctx.request.body), + }), + ); + if (minimalResponse.ok) { + return { + upstream: minimalResponse, + upstreamPath: ctx.request.path, + }; + } + + ctx.request = { + ...ctx.request, + headers: minimalHeaders, + }; + ctx.response = minimalResponse; + ctx.rawErrText = await minimalResponse.text().catch(() => 'unknown error'); + return null; + }, + shouldDowngrade: (ctx) => ( + ctx.response.status >= 500 + || isEndpointDowngradeError(ctx.response.status, ctx.rawErrText) + || shouldDowngradeFromChatToMessagesForResponses( + ctx.request.path, + ctx.response.status, + ctx.rawErrText, ) - ); + ), + onDowngrade: (ctx) => { + logProxy( + selected, + requestedModel, + 'failed', + ctx.response.status, + Date.now() - startTime, + ctx.errText, + retryCount, + downstreamPath, + ); + }, + }); - if (shouldDowngradeEndpoint) { - logProxy(selected, requestedModel, 'failed', response.status, Date.now() - startTime, errText, retryCount, downstreamPath); - continue; - } - - finalStatus = response.status; - finalErrText = errText; - break; - } - - if (!upstream) { - const status = finalStatus || 502; - const errText = finalErrText || 'unknown error'; + if (!endpointResult.ok) { + const status = endpointResult.status || 502; + const errText = endpointResult.errText || 'unknown error'; tokenRouter.recordFailure(selected.channel.id); logProxy(selected, requestedModel, 'failed', status, Date.now() - startTime, errText, retryCount, downstreamPath); @@ -1073,6 +1340,9 @@ export async function responsesProxyRoute(app: FastifyInstance) { return reply.code(status).send({ error: { message: errText, type: 'upstream_error' } }); } + const upstream = endpointResult.upstream; + const successfulUpstreamPath = endpointResult.upstreamPath; + if (isStream) { reply.raw.statusCode = 200; reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8'); diff --git a/src/server/routes/proxy/upstreamEndpoint.test.ts b/src/server/routes/proxy/upstreamEndpoint.test.ts index f94d20a..4cfe7f4 100644 --- a/src/server/routes/proxy/upstreamEndpoint.test.ts +++ b/src/server/routes/proxy/upstreamEndpoint.test.ts @@ -6,7 +6,13 @@ vi.mock('../../services/modelPricingService.js', () => ({ fetchModelPricingCatalog: (arg: unknown) => fetchModelPricingCatalogMock(arg), })); -import { isEndpointDowngradeError, resolveUpstreamEndpointCandidates } from './upstreamEndpoint.js'; +import { + buildMinimalJsonHeadersForCompatibility, + buildUpstreamEndpointRequest, + isUnsupportedMediaTypeError, + isEndpointDowngradeError, + resolveUpstreamEndpointCandidates, +} from './upstreamEndpoint.js'; const baseContext = { site: { @@ -58,6 +64,16 @@ describe('resolveUpstreamEndpointCandidates', () => { 'responses', ); expect(responsesOrder).toEqual(['responses', 'chat', 'messages']); + + const claudeResponsesOrder = await resolveUpstreamEndpointCandidates( + { + ...baseContext, + site: { ...baseContext.site, platform: 'new-api' }, + }, + 'claude-haiku-4-5-20251001', + 'responses', + ); + expect(claudeResponsesOrder).toEqual(['messages', 'chat', 'responses']); }); it('prioritizes messages-first for claude-family models on openai downstream', async () => { @@ -186,6 +202,9 @@ describe('resolveUpstreamEndpointCandidates', () => { expect(isEndpointDowngradeError(404, '{"error":{"message":"Not Found","type":"not_found_error"}}')).toBe(true); expect(isEndpointDowngradeError(405, '{"error":{"message":"Method Not Allowed"}}')).toBe(true); expect(isEndpointDowngradeError(400, '{"error":{"message":"unsupported endpoint","type":"invalid_request_error"}}')).toBe(true); + expect(isEndpointDowngradeError(400, '{"error":{"message":"","type":"upstream_error"}}')).toBe(true); + expect(isEndpointDowngradeError(400, '{"error":{"message":"openai_error","type":"bad_response_status_code"}}')).toBe(true); + expect(isEndpointDowngradeError(415, '{"error":{"message":"Unsupported Media Type: Only \\"application/json\\" is allowed"}}')).toBe(true); }); it('treats Claude Code CLI-only restriction on responses as downgrade candidate', () => { @@ -199,4 +218,195 @@ describe('resolveUpstreamEndpointCandidates', () => { expect(isEndpointDowngradeError(400, upstreamError)).toBe(true); }); + + it('detects unsupported media type errors as compatibility retry candidates', () => { + expect(isUnsupportedMediaTypeError(415, '{"error":{"message":"Unsupported Media Type"}}')).toBe(true); + expect(isUnsupportedMediaTypeError(400, '{"error":{"message":"Only \\"application/json\\" is allowed"}}')).toBe(true); + expect(isUnsupportedMediaTypeError(400, '{"error":{"message":"messages is required"}}')).toBe(false); + }); +}); + +describe('buildUpstreamEndpointRequest', () => { + it('builds minimal JSON compatibility headers for messages endpoint', () => { + const headers = buildMinimalJsonHeadersForCompatibility({ + endpoint: 'messages', + stream: false, + headers: { + Authorization: 'Bearer sk-demo', + 'X-Api-Key': 'sk-claude', + 'anthropic-version': '2023-06-01', + 'anthropic-beta': 'prompt-caching-2024-07-31', + 'OpenAI-Beta': 'responses-2025-03-11', + }, + }); + + expect(headers).toEqual({ + authorization: 'Bearer sk-demo', + 'x-api-key': 'sk-claude', + 'anthropic-version': '2023-06-01', + 'anthropic-beta': 'prompt-caching-2024-07-31', + 'content-type': 'application/json', + accept: 'application/json', + }); + }); + + it('normalizes single-message OpenAI requests to structured responses input', () => { + const request = buildUpstreamEndpointRequest({ + endpoint: 'responses', + modelName: 'upstream-gpt', + stream: false, + tokenValue: 'sk-test', + sitePlatform: 'sub2api', + siteUrl: 'https://example.com', + openaiBody: { + model: 'gpt-5.2', + messages: [{ role: 'user', content: 'hello' }], + }, + downstreamFormat: 'openai', + }); + + expect(request.path).toBe('/v1/responses'); + expect(request.body.input).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello' }], + }, + ]); + }); + + it('normalizes downstream responses input string before forwarding upstream', () => { + const request = buildUpstreamEndpointRequest({ + endpoint: 'responses', + modelName: 'upstream-gpt', + stream: false, + tokenValue: 'sk-test', + sitePlatform: 'sub2api', + siteUrl: 'https://example.com', + openaiBody: {}, + downstreamFormat: 'responses', + responsesOriginalBody: { + model: 'gpt-5.2', + input: 'hello', + metadata: { trace: 'abc123' }, + }, + }); + + expect(request.path).toBe('/v1/responses'); + expect(request.body.metadata).toEqual({ trace: 'abc123' }); + expect(request.body.input).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello' }], + }, + ]); + }); + + it('applies global responses standardization and drops non-standard fields', () => { + const request = buildUpstreamEndpointRequest({ + endpoint: 'responses', + modelName: 'upstream-gpt', + stream: false, + tokenValue: 'sk-test', + sitePlatform: 'openai', + siteUrl: 'https://example.com', + openaiBody: {}, + downstreamFormat: 'responses', + responsesOriginalBody: { + model: 'gpt-5.2', + input: 'hello', + metadata: { trace: 'abc123' }, + max_completion_tokens: 512, + custom_vendor_flag: 'drop-me', + }, + }); + + expect(request.path).toBe('/v1/responses'); + expect(request.body.metadata).toEqual({ trace: 'abc123' }); + expect(request.body.custom_vendor_flag).toBeUndefined(); + expect(request.body.max_completion_tokens).toBeUndefined(); + expect(request.body.max_output_tokens).toBe(512); + expect(request.body.input).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello' }], + }, + ]); + }); + + it('backfills responses input from legacy messages payload', () => { + const request = buildUpstreamEndpointRequest({ + endpoint: 'responses', + modelName: 'upstream-gpt', + stream: false, + tokenValue: 'sk-test', + sitePlatform: 'openai', + siteUrl: 'https://example.com', + openaiBody: {}, + downstreamFormat: 'responses', + responsesOriginalBody: { + model: 'gpt-5.2', + messages: [{ role: 'user', content: 'hello from messages' }], + }, + }); + + expect(request.path).toBe('/v1/responses'); + expect(request.body.input).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello from messages' }], + }, + ]); + }); + + it('backfills responses input from prompt field when input/messages are missing', () => { + const request = buildUpstreamEndpointRequest({ + endpoint: 'responses', + modelName: 'upstream-gpt', + stream: false, + tokenValue: 'sk-test', + sitePlatform: 'openai', + siteUrl: 'https://example.com', + openaiBody: {}, + downstreamFormat: 'responses', + responsesOriginalBody: { + model: 'gpt-5.2', + prompt: 'hello from prompt', + }, + }); + + expect(request.path).toBe('/v1/responses'); + expect(request.body.input).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello from prompt' }], + }, + ]); + }); + + it('blocks downstream content-type passthrough and forces json content-type upstream', () => { + const request = buildUpstreamEndpointRequest({ + endpoint: 'responses', + modelName: 'upstream-gpt', + stream: false, + tokenValue: 'sk-test', + sitePlatform: 'new-api', + siteUrl: 'https://example.com', + openaiBody: { + model: 'gpt-5.2', + messages: [{ role: 'user', content: 'hello' }], + }, + downstreamFormat: 'openai', + downstreamHeaders: { + 'content-type': 'text/plain', + }, + }); + + expect(request.headers['content-type']).toBeUndefined(); + expect(request.headers['Content-Type']).toBe('application/json'); + }); }); diff --git a/src/server/routes/proxy/upstreamEndpoint.ts b/src/server/routes/proxy/upstreamEndpoint.ts index af022b6..b6a8328 100644 --- a/src/server/routes/proxy/upstreamEndpoint.ts +++ b/src/server/routes/proxy/upstreamEndpoint.ts @@ -66,6 +66,7 @@ const HOP_BY_HOP_HEADERS = new Set([ const BLOCKED_PASSTHROUGH_HEADERS = new Set([ 'host', + 'content-type', 'content-length', 'accept-encoding', 'cookie', @@ -382,6 +383,162 @@ function convertOpenAiToolChoiceToResponses(rawToolChoice: unknown): unknown { return rawToolChoice; } +function toResponsesInputMessageFromText(text: string): Record { + return { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text }], + }; +} + +function normalizeResponsesMessageItem(item: Record): Record { + const type = asTrimmedString(item.type).toLowerCase(); + if (type === 'function_call' || type === 'function_call_output') { + return item; + } + + const role = asTrimmedString(item.role).toLowerCase(); + const textContent = normalizeContentText(item.content ?? item.text).trim(); + + if (type === 'message') { + if (!textContent) return item; + const normalizedRole = role || 'user'; + const textType = normalizedRole === 'assistant' ? 'output_text' : 'input_text'; + return { + ...item, + role: normalizedRole, + content: [{ type: textType, text: textContent }], + }; + } + + if (role) { + if (!textContent) return item; + const textType = role === 'assistant' ? 'output_text' : 'input_text'; + return { + type: 'message', + role, + content: [{ type: textType, text: textContent }], + }; + } + + if (textContent) { + return toResponsesInputMessageFromText(textContent); + } + + return item; +} + +function normalizeResponsesInputForCompatibility(input: unknown): unknown { + if (typeof input === 'string') { + const normalized = input.trim(); + if (!normalized) return input; + return [toResponsesInputMessageFromText(normalized)]; + } + + if (Array.isArray(input)) { + return input.map((item) => { + if (typeof item === 'string') { + const normalized = item.trim(); + return normalized ? toResponsesInputMessageFromText(normalized) : item; + } + if (!isRecord(item)) return item; + return normalizeResponsesMessageItem(item); + }); + } + + if (isRecord(input)) { + return [normalizeResponsesMessageItem(input)]; + } + + return input; +} + +function normalizeResponsesBodyForCompatibility( + body: Record, +): Record { + const nextInput = normalizeResponsesInputForCompatibility(body.input); + if (nextInput === body.input) return body; + return { + ...body, + input: nextInput, + }; +} + +const ALLOWED_RESPONSES_FIELDS = new Set([ + 'model', + 'input', + 'instructions', + 'max_output_tokens', + 'max_completion_tokens', + 'temperature', + 'top_p', + 'truncation', + 'tools', + 'tool_choice', + 'parallel_tool_calls', + 'metadata', + 'reasoning', + 'store', + 'stream', + 'user', + 'previous_response_id', + 'text', + 'audio', + 'include', + 'response_format', + 'service_tier', + 'stop', + 'n', +]); + +function sanitizeResponsesBodyForProxy( + body: Record, + modelName: string, + stream: boolean, +): Record { + let normalized = normalizeResponsesBodyForCompatibility({ + ...body, + model: modelName, + stream, + }); + + if (normalized.input === undefined) { + if (Array.isArray((normalized as Record).messages)) { + const converted = convertOpenAiBodyToResponsesBody(normalized, modelName, stream); + normalized = normalizeResponsesBodyForCompatibility(converted); + } else { + const prompt = asTrimmedString((normalized as Record).prompt); + if (prompt) { + normalized = { + ...normalized, + input: [toResponsesInputMessageFromText(prompt)], + }; + } + } + } + + const sanitized: Record = {}; + for (const [key, value] of Object.entries(normalized)) { + if (!ALLOWED_RESPONSES_FIELDS.has(key)) continue; + if (key === 'max_completion_tokens') continue; + sanitized[key] = value; + } + + const maxOutputTokens = toFiniteNumber(normalized.max_output_tokens); + if (maxOutputTokens !== null && maxOutputTokens > 0) { + sanitized.max_output_tokens = Math.trunc(maxOutputTokens); + } else { + const maxCompletionTokens = toFiniteNumber(normalized.max_completion_tokens); + if (maxCompletionTokens !== null && maxCompletionTokens > 0) { + sanitized.max_output_tokens = Math.trunc(maxCompletionTokens); + } + } + + sanitized.model = modelName; + sanitized.stream = stream; + return sanitized; +} + function convertOpenAiBodyToMessagesBody( openaiBody: Record, modelName: string, @@ -632,22 +789,6 @@ function convertOpenAiBodyToResponsesBody( }); } - let input: unknown = inputItems; - if (systemContents.length === 0 && inputItems.length === 1) { - const first = inputItems[0]; - const firstRole = asTrimmedString(first.role).toLowerCase(); - const firstContent = Array.isArray(first.content) ? first.content : []; - if ( - firstRole === 'user' - && firstContent.length === 1 - && isRecord(firstContent[0]) - && asTrimmedString(firstContent[0].type).toLowerCase() === 'input_text' - && asTrimmedString(firstContent[0].text) - ) { - input = asTrimmedString(firstContent[0].text); - } - } - const maxOutputTokens = ( toFiniteNumber(openaiBody.max_output_tokens) ?? toFiniteNumber(openaiBody.max_completion_tokens) @@ -659,7 +800,7 @@ function convertOpenAiBodyToResponsesBody( model: modelName, stream, max_output_tokens: maxOutputTokens, - input, + input: inputItems, }; if (systemContents.length > 0) { @@ -681,7 +822,7 @@ function convertOpenAiBodyToResponsesBody( const responsesToolChoice = convertOpenAiToolChoiceToResponses(openaiBody.tool_choice); if (responsesToolChoice !== undefined) body.tool_choice = responsesToolChoice; - return body; + return normalizeResponsesBodyForCompatibility(body); } function normalizeEndpointTypes(value: unknown): UpstreamEndpoint[] { @@ -757,6 +898,11 @@ function preferredEndpointOrder( // Unknown/generic upstreams: prefer endpoint family that matches the // downstream API surface, then degrade progressively. if (downstreamFormat === 'responses') { + if (preferMessagesForClaudeModel) { + // Claude-family models on generic/new-api upstreams are commonly + // messages-first even when downstream API is /v1/responses. + return ['messages', 'chat', 'responses']; + } return ['responses', 'chat', 'messages']; } @@ -830,7 +976,7 @@ export async function resolveUpstreamEndpointCandidates( const shouldIgnoreCatalogOrderingForClaudeMessages = ( preferMessagesForClaudeModel - && downstreamFormat !== 'responses' + && (downstreamFormat !== 'responses' || sitePlatform !== 'openai') ); if (shouldIgnoreCatalogOrderingForClaudeMessages) { return preferred; @@ -980,7 +1126,7 @@ export function buildUpstreamEndpointRequest(input: { const responsesHeaders = input.downstreamFormat === 'responses' ? extractResponsesPassthroughHeaders(input.downstreamHeaders) : {}; - const body = ( + const rawBody = ( input.downstreamFormat === 'responses' && input.responsesOriginalBody ? { ...input.responsesOriginalBody, @@ -989,6 +1135,7 @@ export function buildUpstreamEndpointRequest(input: { } : convertOpenAiBodyToResponsesBody(input.openaiBody, input.modelName, input.stream) ); + const body = sanitizeResponsesBodyForProxy(rawBody, input.modelName, input.stream); const headers = ensureStreamAcceptHeader({ ...commonHeaders, @@ -1014,10 +1161,63 @@ export function buildUpstreamEndpointRequest(input: { }; } +function normalizeHeaderMap(headers: Record): Record { + const normalized: Record = {}; + for (const [rawKey, rawValue] of Object.entries(headers)) { + const key = rawKey.trim().toLowerCase(); + if (!key) continue; + const value = headerValueToString(rawValue); + if (!value) continue; + normalized[key] = value; + } + return normalized; +} + +export function buildMinimalJsonHeadersForCompatibility(input: { + headers: Record; + endpoint: UpstreamEndpoint; + stream: boolean; +}): Record { + const source = normalizeHeaderMap(input.headers); + const minimal: Record = {}; + + if (source.authorization) minimal.authorization = source.authorization; + if (source['x-api-key']) minimal['x-api-key'] = source['x-api-key']; + + if (input.endpoint === 'messages') { + for (const [key, value] of Object.entries(source)) { + if (!key.startsWith('anthropic-')) continue; + minimal[key] = value; + } + if (!minimal['anthropic-version']) { + minimal['anthropic-version'] = '2023-06-01'; + } + } + + minimal['content-type'] = 'application/json'; + minimal.accept = input.stream ? 'text/event-stream' : 'application/json'; + return minimal; +} + +export function isUnsupportedMediaTypeError(status: number, upstreamErrorText?: string | null): boolean { + if (status < 400) return false; + if (status !== 400 && status !== 415) return false; + const text = (upstreamErrorText || '').toLowerCase(); + if (!text) return status === 415; + + return ( + text.includes('unsupported media type') + || text.includes("only 'application/json' is allowed") + || text.includes('only "application/json" is allowed') + || text.includes('application/json') + || text.includes('content-type') + ); +} + export function isEndpointDowngradeError(status: number, upstreamErrorText?: string | null): boolean { if (status < 400) return false; const text = (upstreamErrorText || '').toLowerCase(); - if (status === 404 || status === 405 || status === 501) return true; + if (status === 404 || status === 405 || status === 415 || status === 501) return true; if (!text) return false; let parsedCode = ''; @@ -1046,6 +1246,12 @@ export function isEndpointDowngradeError(status: number, upstreamErrorText?: str || text.includes('unrecognized request url') || text.includes('no route matched') || text.includes('does not exist') + || text.includes('openai_error') + || text.includes('upstream_error') + || text.includes('bad_response_status_code') + || text.includes('unsupported media type') + || text.includes("only 'application/json' is allowed") + || text.includes('only "application/json" is allowed') || (status === 400 && text.includes('unsupported')) || text.includes('not implemented') || text.includes('api not implemented') @@ -1056,11 +1262,15 @@ export function isEndpointDowngradeError(status: number, upstreamErrorText?: str || parsedCode === 'unknown_endpoint' || parsedCode === 'unsupported_endpoint' || parsedCode === 'bad_response_status_code' + || parsedCode === 'openai_error' + || parsedCode === 'upstream_error' || parsedType === 'not_found_error' || parsedType === 'invalid_request_error' || parsedType === 'unsupported_endpoint' || parsedType === 'unsupported_path' || parsedType === 'bad_response_status_code' + || parsedType === 'openai_error' + || parsedType === 'upstream_error' || parsedMessage.includes('unknown endpoint') || parsedMessage.includes('unsupported endpoint') || parsedMessage.includes('unsupported path') @@ -1068,6 +1278,11 @@ export function isEndpointDowngradeError(status: number, upstreamErrorText?: str || parsedMessage.includes('no route matched') || parsedMessage.includes('does not exist') || parsedMessage.includes('bad_response_status_code') + || parsedMessage === 'openai_error' + || parsedMessage === 'upstream_error' + || parsedMessage.includes('unsupported media type') + || parsedMessage.includes("only 'application/json' is allowed") + || parsedMessage.includes('only "application/json" is allowed') || ( status === 400 && parsedCode === 'invalid_request' diff --git a/src/server/routes/proxy/upstreamError.test.ts b/src/server/routes/proxy/upstreamError.test.ts new file mode 100644 index 0000000..9df83c7 --- /dev/null +++ b/src/server/routes/proxy/upstreamError.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from 'vitest'; +import { summarizeUpstreamError } from './upstreamError.js'; + +describe('summarizeUpstreamError', () => { + it('extracts concise message from JSON error payload', () => { + const message = summarizeUpstreamError(400, JSON.stringify({ + error: { + message: 'messages is required', + type: 'bad_request', + }, + })); + + expect(message).toBe('Upstream returned HTTP 400: messages is required'); + }); + + it('summarizes Cloudflare 5xx HTML page without dumping full body', () => { + const html = `qaq.al | 502: Bad gatewayCloudflare Ray ID: abc`; + const message = summarizeUpstreamError(502, html); + + expect(message).toContain('Upstream returned HTTP 502'); + expect(message).toContain('Cloudflare 502: Bad gateway'); + expect(message).not.toContain(''); + }); + + it('truncates oversized plain text payloads', () => { + const longText = 'x'.repeat(800); + const message = summarizeUpstreamError(500, longText); + + expect(message).toContain('Upstream returned HTTP 500:'); + expect(message).toContain('...(truncated)'); + expect(message.length).toBeLessThan(500); + }); +}); + diff --git a/src/server/routes/proxy/upstreamError.ts b/src/server/routes/proxy/upstreamError.ts new file mode 100644 index 0000000..6799035 --- /dev/null +++ b/src/server/routes/proxy/upstreamError.ts @@ -0,0 +1,81 @@ +function isRecord(value: unknown): value is Record { + return !!value && typeof value === 'object'; +} + +function collapseWhitespace(text: string): string { + return text.replace(/\s+/g, ' ').trim(); +} + +function extractJsonErrorMessage(rawText: string): string { + try { + const parsed = JSON.parse(rawText) as unknown; + const root = isRecord(parsed) ? parsed : null; + const error = (root && isRecord(root.error)) ? root.error : root; + if (!error) return ''; + + const message = typeof error.message === 'string' ? collapseWhitespace(error.message) : ''; + if (message) return message; + + const code = typeof error.code === 'string' ? collapseWhitespace(error.code) : ''; + const type = typeof error.type === 'string' ? collapseWhitespace(error.type) : ''; + return [type, code].filter((part) => part.length > 0).join('/'); + } catch { + return ''; + } +} + +function extractHtmlTitle(rawText: string): string { + const match = rawText.match(/]*>([\s\S]*?)<\/title>/i); + if (!match?.[1]) return ''; + return collapseWhitespace(match[1].replace(/<[^>]+>/g, '')); +} + +function extractCloudflareHtmlSummary(rawText: string, status: number): string { + if (!/cloudflare/i.test(rawText)) return ''; + const title = extractHtmlTitle(rawText); + const codeMatch = ( + title.match(/\b(\d{3,4})\s*:\s*([^\|<]+)/i) + || rawText.match(/Error code\s*(\d{3,4})/i) + ); + const code = codeMatch?.[1] || (status > 0 ? String(status) : ''); + const reason = collapseWhitespace( + (typeof codeMatch?.[2] === 'string' ? codeMatch[2] : '') + || (status >= 500 ? 'origin host error' : 'request blocked') + ); + if (code) return `Cloudflare ${code}: ${reason}`; + return `Cloudflare: ${reason}`; +} + +function extractHtmlSummary(rawText: string, status: number): string { + if (!/(]*>([\s\S]*?)<\/h1>/i)?.[1] || ''; + return collapseWhitespace(heading.replace(/<[^>]+>/g, '')); +} + +export function summarizeUpstreamError(status: number, rawErrorText: string): string { + const statusPrefix = status > 0 + ? `Upstream returned HTTP ${status}` + : 'Upstream request failed'; + + const raw = typeof rawErrorText === 'string' ? rawErrorText.trim() : ''; + if (!raw) return statusPrefix; + + const jsonMessage = extractJsonErrorMessage(raw); + if (jsonMessage) return `${statusPrefix}: ${jsonMessage}`; + + const htmlMessage = extractHtmlSummary(raw, status); + if (htmlMessage) return `${statusPrefix}: ${htmlMessage}`; + + const compact = collapseWhitespace(raw); + if (!compact) return statusPrefix; + if (compact.length <= 400) return `${statusPrefix}: ${compact}`; + return `${statusPrefix}: ${compact.slice(0, 400)}...(truncated)`; +} + diff --git a/src/web/index.css b/src/web/index.css index 2dfdf64..3634db7 100644 --- a/src/web/index.css +++ b/src/web/index.css @@ -1308,11 +1308,24 @@ body { font-size: 12px; } +.accounts-table { + width: 100%; + table-layout: fixed; +} + +.accounts-table th, +.accounts-table td { + white-space: normal; + word-break: break-word; +} + .accounts-actions-col { - min-width: 220px; + width: 24%; + min-width: 0; } .accounts-actions-cell { + min-width: 0; white-space: normal !important; } @@ -1320,7 +1333,12 @@ body { display: flex; flex-wrap: wrap; justify-content: flex-end; - gap: 2px; + gap: 4px 6px; + width: 100%; +} + +.accounts-row-actions .btn-link { + white-space: nowrap; } .sites-actions-col { @@ -3125,7 +3143,11 @@ textarea { } .accounts-actions-col { - min-width: 180px; + width: 28%; + } + + .accounts-actions-cell { + min-width: 0; } .accounts-table th, @@ -3190,7 +3212,11 @@ textarea { } .accounts-actions-col { - min-width: 150px; + width: 30%; + } + + .accounts-actions-cell { + min-width: 0; } .accounts-table th:nth-child(5), diff --git a/src/web/pages/Accounts.tsx b/src/web/pages/Accounts.tsx index f1fd750..06de665 100644 --- a/src/web/pages/Accounts.tsx +++ b/src/web/pages/Accounts.tsx @@ -765,7 +765,7 @@ export default function Accounts() { )} {/* Accounts Table */} -
+
{accounts.length > 0 ? ( diff --git a/src/web/pages/ModelTester.tsx b/src/web/pages/ModelTester.tsx index 8176bb7..ba44f87 100644 --- a/src/web/pages/ModelTester.tsx +++ b/src/web/pages/ModelTester.tsx @@ -296,20 +296,51 @@ const applyAssistantDelta = ( status: MESSAGE_STATUS.INCOMPLETE, }; + const toIncrementalText = (existingText: string, incomingText?: string): string => { + if (!incomingText) return ''; + if (!existingText) return incomingText; + + if (incomingText === existingText) return ''; + if (incomingText.startsWith(existingText)) { + return incomingText.slice(existingText.length); + } + if (existingText.endsWith(incomingText)) return ''; + + const maxOverlap = Math.min(existingText.length, incomingText.length); + const MIN_OVERLAP = 8; + for (let overlap = maxOverlap; overlap >= MIN_OVERLAP; overlap -= 1) { + if (existingText.slice(-overlap) === incomingText.slice(0, overlap)) { + return incomingText.slice(overlap); + } + } + + return incomingText; + }; + if (delta.reasoningDelta) { - next = { - ...next, - reasoningContent: (next.reasoningContent || '') + delta.reasoningDelta, - isThinkingComplete: false, - }; + const existingReasoning = next.reasoningContent || ''; + const reasoningAppend = toIncrementalText(existingReasoning, delta.reasoningDelta); + if (reasoningAppend) { + next = { + ...next, + reasoningContent: existingReasoning + reasoningAppend, + isThinkingComplete: false, + }; + } } if (delta.contentDelta) { + const existingContent = next.content || ''; + const contentAppend = toIncrementalText(existingContent, delta.contentDelta); + if (!contentAppend) { + return replaceMessageAt(messages, targetIndex, next); + } + const hasReasoning = Boolean(next.reasoningContent); const shouldAutoCollapse = hasReasoning && !next.hasAutoCollapsed; next = { ...next, - content: (next.content || '') + delta.contentDelta, + content: existingContent + contentAppend, isReasoningExpanded: shouldAutoCollapse ? false : next.isReasoningExpanded, hasAutoCollapsed: shouldAutoCollapse || next.hasAutoCollapsed, }; @@ -383,21 +414,15 @@ const parseAnyStreamDelta = (eventPayload: any): { } if (typeof eventPayload.type === 'string') { + // Responses stream emits a full-text summary again in several "done" events + // (output_text.done/content_part.done/output_item.done/response.completed). + // Treat those as structural events only; otherwise UI appends duplicate text. if (eventPayload.type === 'response.output_item.added' || eventPayload.type === 'response.output_item.done') { - const parsed = extractResponsesContent(eventPayload.item || eventPayload.output_item || eventPayload.response || eventPayload); - return { - contentDelta: parsed.content || undefined, - reasoningDelta: parsed.reasoningContent || undefined, - }; + return {}; } if (eventPayload.type === 'response.content_part.added' || eventPayload.type === 'response.content_part.done') { - const part = eventPayload.part || eventPayload.content_part || eventPayload; - const parsed = extractResponsesContent(part); - return { - contentDelta: parsed.content || undefined, - reasoningDelta: parsed.reasoningContent || undefined, - }; + return {}; } if (eventPayload.type === 'response.content_part.delta') { @@ -434,18 +459,10 @@ const parseAnyStreamDelta = (eventPayload: any): { return { reasoningDelta: text || undefined }; } - if (eventPayload.type === 'response.output_text.done') { - const text = typeof eventPayload.text === 'string' ? eventPayload.text : ''; - return { contentDelta: text || undefined }; - } + if (eventPayload.type === 'response.output_text.done') return {}; if (eventPayload.type === 'response.completed' || eventPayload.type === 'response.failed') { - const parsed = extractResponsesContent(eventPayload.response || eventPayload); - return { - contentDelta: parsed.content || undefined, - reasoningDelta: parsed.reasoningContent || undefined, - done: true, - }; + return { done: true }; } if (eventPayload.type === 'content_block_delta') {