feat(proxy): enhance upstream endpoint handling and error summarization
- Added new utility functions for building minimal JSON headers and normalizing responses for compatibility. - Implemented detailed error handling for unsupported media types and endpoint downgrade scenarios. - Introduced a new endpoint flow mechanism to manage retries and downgrades between different upstream endpoints. - Enhanced tests for endpoint flow and error summarization to cover various edge cases and scenarios. - Updated CSS styles for accounts table to improve layout and responsiveness. - Refactored ModelTester component to handle incremental text updates more effectively.
This commit is contained in:
cita
@@ -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<VerifyFailureReason> => {
|
||||
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|<script/i.test(text)) {
|
||||
if (
|
||||
!skipRawShieldDetection
|
||||
&& ct.includes('text/html')
|
||||
&& /var\s+arg1\s*=|acw_sc__v2|cdn_sec_tc|<script/i.test(text)
|
||||
) {
|
||||
return 'shield-blocked';
|
||||
}
|
||||
|
||||
@@ -516,7 +525,9 @@ export async function accountsRoutes(app: FastifyInstance) {
|
||||
const body = JSON.parse(text) as any;
|
||||
const message = typeof body?.message === 'string' ? body.message : '';
|
||||
if (/mismatch|new-api-user|user id/i.test(message)) return 'needs-user-id';
|
||||
if (/shield|challenge|captcha|acw_sc__v2|arg1/i.test(message)) return 'shield-blocked';
|
||||
if (!skipRawShieldDetection && /shield|challenge|captcha|acw_sc__v2|arg1/i.test(message)) {
|
||||
return 'shield-blocked';
|
||||
}
|
||||
} catch { }
|
||||
|
||||
return null;
|
||||
|
||||
@@ -6,9 +6,11 @@ import { join } from 'node:path';
|
||||
|
||||
const verifyTokenMock = vi.fn();
|
||||
const undiciFetchMock = vi.fn();
|
||||
let adapterPlatformName = 'new-api';
|
||||
|
||||
vi.mock('../../services/platforms/index.js', () => ({
|
||||
getAdapter: () => ({
|
||||
platformName: adapterPlatformName,
|
||||
verifyToken: (...args: unknown[]) => verifyTokenMock(...args),
|
||||
}),
|
||||
}));
|
||||
@@ -42,6 +44,7 @@ describe('accounts verify-token shield detection', () => {
|
||||
beforeEach(() => {
|
||||
verifyTokenMock.mockReset();
|
||||
undiciFetchMock.mockReset();
|
||||
adapterPlatformName = 'new-api';
|
||||
|
||||
db.delete(schema.proxyLogs).run();
|
||||
db.delete(schema.checkinLogs).run();
|
||||
@@ -84,7 +87,7 @@ describe('accounts verify-token shield detection', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('returns shieldBlocked when upstream /api/user/self responds with challenge html', async () => {
|
||||
it('avoids raw shieldBlocked misclassification for new-api when verifyToken returned tokenType unknown', async () => {
|
||||
verifyTokenMock.mockResolvedValueOnce({ tokenType: 'unknown' });
|
||||
undiciFetchMock.mockResolvedValue({
|
||||
text: async () => '<html><script>var arg1="ABC123";</script></html>',
|
||||
@@ -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 () => '<html><script>var arg1="ABC123";</script></html>',
|
||||
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 () => '<html><script>var arg1="ABC123";</script></html>',
|
||||
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,
|
||||
|
||||
@@ -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<Uint8Array>({
|
||||
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 = '<!DOCTYPE html><html><head><title>qaq.al | 502: Bad gateway</title></head><body>Cloudflare Ray ID: 9d6f7c889ffbc8eb</body></html>';
|
||||
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('<!DOCTYPE html>');
|
||||
});
|
||||
|
||||
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(
|
||||
'<!DOCTYPE html><html><head><title>qaq.al | 502: Bad gateway</title></head><body>Cloudflare</body></html>',
|
||||
{
|
||||
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<Uint8Array>({
|
||||
@@ -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<Uint8Array>({
|
||||
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<Uint8Array>({
|
||||
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: [
|
||||
|
||||
+141
-106
@@ -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<ReturnType<typeof fetch>> | 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<string, unknown>,
|
||||
});
|
||||
|
||||
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<string, unknown>,
|
||||
});
|
||||
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<string, unknown>,
|
||||
};
|
||||
},
|
||||
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<string, unknown>,
|
||||
});
|
||||
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<string, unknown>,
|
||||
};
|
||||
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 {
|
||||
|
||||
@@ -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<ReturnType<typeof fetch>> {
|
||||
return response as unknown as Awaited<ReturnType<typeof fetch>>;
|
||||
}
|
||||
|
||||
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');
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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<string, string>;
|
||||
body: Record<string, unknown>;
|
||||
};
|
||||
|
||||
export type EndpointAttemptContext = {
|
||||
endpointIndex: number;
|
||||
endpointCount: number;
|
||||
request: BuiltEndpointRequest;
|
||||
targetUrl: string;
|
||||
response: Awaited<ReturnType<typeof fetch>>;
|
||||
rawErrText: string;
|
||||
};
|
||||
|
||||
export type EndpointRecoverResult = {
|
||||
upstream: Awaited<ReturnType<typeof fetch>>;
|
||||
upstreamPath: string;
|
||||
} | null;
|
||||
|
||||
export type EndpointFlowResult =
|
||||
| {
|
||||
ok: true;
|
||||
upstream: Awaited<ReturnType<typeof fetch>>;
|
||||
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<EndpointRecoverResult>;
|
||||
shouldDowngrade?: (ctx: EndpointAttemptContext) => boolean;
|
||||
onDowngrade?: (ctx: EndpointAttemptContext & { errText: string }) => void | Promise<void>;
|
||||
};
|
||||
|
||||
export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Promise<EndpointFlowResult> {
|
||||
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',
|
||||
};
|
||||
}
|
||||
@@ -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<string, unknown>;
|
||||
const error = (parsed.error && typeof parsed.error === 'object')
|
||||
? parsed.error as Record<string, unknown>
|
||||
: 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<string, unknown>,
|
||||
): Record<string, unknown> | 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<string, unknown>,
|
||||
): Record<string, unknown> | null {
|
||||
const model = typeof body.model === 'string' ? body.model.trim() : '';
|
||||
if (!model) return null;
|
||||
if (body.input === undefined) return null;
|
||||
|
||||
const core: Record<string, unknown> = {
|
||||
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<string, unknown>,
|
||||
): Record<string, unknown> | 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<string, unknown>,
|
||||
): Record<string, unknown>[] {
|
||||
const candidates: Record<string, unknown>[] = [];
|
||||
const seen = new Set<string>();
|
||||
try {
|
||||
const originalKey = JSON.stringify(body);
|
||||
if (originalKey) seen.add(originalKey);
|
||||
} catch {
|
||||
// ignore non-serializable bodies
|
||||
}
|
||||
const push = (next: Record<string, unknown> | 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<string, string>,
|
||||
stream: boolean,
|
||||
): Record<string, string>[] {
|
||||
const candidates: Record<string, string>[] = [];
|
||||
const seen = new Set<string>();
|
||||
const push = (next: Record<string, string>) => {
|
||||
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<string, string> = {};
|
||||
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<string, unknown> {
|
||||
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<ReturnType<typeof fetch>> | 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<string, unknown>,
|
||||
});
|
||||
return {
|
||||
endpoint,
|
||||
path: endpointRequest.path,
|
||||
headers: endpointRequest.headers,
|
||||
body: endpointRequest.body as Record<string, unknown>,
|
||||
};
|
||||
},
|
||||
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<string, unknown>,
|
||||
});
|
||||
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');
|
||||
|
||||
@@ -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');
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<string, unknown> {
|
||||
return {
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text }],
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeResponsesMessageItem(item: Record<string, unknown>): Record<string, unknown> {
|
||||
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<string, unknown>,
|
||||
): Record<string, unknown> {
|
||||
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<string, unknown>,
|
||||
modelName: string,
|
||||
stream: boolean,
|
||||
): Record<string, unknown> {
|
||||
let normalized = normalizeResponsesBodyForCompatibility({
|
||||
...body,
|
||||
model: modelName,
|
||||
stream,
|
||||
});
|
||||
|
||||
if (normalized.input === undefined) {
|
||||
if (Array.isArray((normalized as Record<string, unknown>).messages)) {
|
||||
const converted = convertOpenAiBodyToResponsesBody(normalized, modelName, stream);
|
||||
normalized = normalizeResponsesBodyForCompatibility(converted);
|
||||
} else {
|
||||
const prompt = asTrimmedString((normalized as Record<string, unknown>).prompt);
|
||||
if (prompt) {
|
||||
normalized = {
|
||||
...normalized,
|
||||
input: [toResponsesInputMessageFromText(prompt)],
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const sanitized: Record<string, unknown> = {};
|
||||
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<string, unknown>,
|
||||
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<string, string>): Record<string, string> {
|
||||
const normalized: Record<string, string> = {};
|
||||
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<string, string>;
|
||||
endpoint: UpstreamEndpoint;
|
||||
stream: boolean;
|
||||
}): Record<string, string> {
|
||||
const source = normalizeHeaderMap(input.headers);
|
||||
const minimal: Record<string, string> = {};
|
||||
|
||||
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'
|
||||
|
||||
@@ -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 = `<!DOCTYPE html><html><head><title>qaq.al | 502: Bad gateway</title></head><body>Cloudflare Ray ID: abc</body></html>`;
|
||||
const message = summarizeUpstreamError(502, html);
|
||||
|
||||
expect(message).toContain('Upstream returned HTTP 502');
|
||||
expect(message).toContain('Cloudflare 502: Bad gateway');
|
||||
expect(message).not.toContain('<!DOCTYPE html>');
|
||||
});
|
||||
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
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(/<title[^>]*>([\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 (!/(<!doctype|<html)/i.test(rawText)) return '';
|
||||
|
||||
const cloudflareSummary = extractCloudflareHtmlSummary(rawText, status);
|
||||
if (cloudflareSummary) return cloudflareSummary;
|
||||
|
||||
const title = extractHtmlTitle(rawText);
|
||||
if (title) return title;
|
||||
|
||||
const heading = rawText.match(/<h1[^>]*>([\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)`;
|
||||
}
|
||||
|
||||
+30
-4
@@ -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),
|
||||
|
||||
@@ -765,7 +765,7 @@ export default function Accounts() {
|
||||
)}
|
||||
|
||||
{/* Accounts Table */}
|
||||
<div className="card" style={{ overflowX: 'auto' }}>
|
||||
<div className="card">
|
||||
{accounts.length > 0 ? (
|
||||
<table className="data-table accounts-table">
|
||||
<thead>
|
||||
|
||||
@@ -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') {
|
||||
|
||||
Reference in New Issue
Block a user