fix: decode zstd responses while restoring claude encoding
This commit is contained in:
@@ -0,0 +1,56 @@
|
||||
import { gzipSync, zstdCompressSync } from 'node:zlib';
|
||||
import { Response } from 'undici';
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { materializeErrorResponse, readRuntimeResponseText } from './types.js';
|
||||
|
||||
describe('readRuntimeResponseText', () => {
|
||||
it('decompresses zstd responses before reading the body text', async () => {
|
||||
const payload = JSON.stringify({ ok: true, text: 'hello zstd' });
|
||||
const response = new Response(zstdCompressSync(Buffer.from(payload)), {
|
||||
status: 200,
|
||||
headers: {
|
||||
'content-encoding': 'zstd',
|
||||
'content-type': 'application/json; charset=utf-8',
|
||||
},
|
||||
});
|
||||
|
||||
await expect(readRuntimeResponseText(response)).resolves.toBe(payload);
|
||||
});
|
||||
|
||||
it('decompresses stacked content-encodings in reverse order', async () => {
|
||||
const payload = JSON.stringify({ ok: true, text: 'stacked' });
|
||||
const response = new Response(
|
||||
zstdCompressSync(gzipSync(Buffer.from(payload))),
|
||||
{
|
||||
status: 200,
|
||||
headers: {
|
||||
'content-encoding': 'gzip, zstd',
|
||||
'content-type': 'application/json; charset=utf-8',
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
await expect(readRuntimeResponseText(response)).resolves.toBe(payload);
|
||||
});
|
||||
});
|
||||
|
||||
describe('materializeErrorResponse', () => {
|
||||
it('decodes compressed error bodies and strips compression headers', async () => {
|
||||
const payload = JSON.stringify({ error: { message: 'upstream failed' } });
|
||||
const response = new Response(zstdCompressSync(Buffer.from(payload)), {
|
||||
status: 503,
|
||||
headers: {
|
||||
'content-encoding': 'zstd',
|
||||
'content-length': '999',
|
||||
'content-type': 'application/json; charset=utf-8',
|
||||
},
|
||||
});
|
||||
|
||||
const materialized = await materializeErrorResponse(response);
|
||||
|
||||
await expect(materialized.text()).resolves.toBe(payload);
|
||||
expect(materialized.headers.get('content-encoding')).toBeNull();
|
||||
expect(materialized.headers.get('content-length')).toBeNull();
|
||||
expect(materialized.headers.get('content-type')).toBe('application/json; charset=utf-8');
|
||||
});
|
||||
});
|
||||
@@ -1,3 +1,9 @@
|
||||
import {
|
||||
brotliDecompressSync,
|
||||
gunzipSync,
|
||||
inflateSync,
|
||||
zstdDecompressSync,
|
||||
} from 'node:zlib';
|
||||
import {
|
||||
Response,
|
||||
fetch,
|
||||
@@ -63,13 +69,79 @@ export async function performFetch(
|
||||
return fetch(requestUrl, init);
|
||||
}
|
||||
|
||||
function hasZstdContentEncoding(contentEncoding: string | null): boolean {
|
||||
if (!contentEncoding) return false;
|
||||
return contentEncoding
|
||||
.split(',')
|
||||
.some((encoding) => encoding.trim().toLowerCase() === 'zstd');
|
||||
}
|
||||
|
||||
function looksLikeZstdFrame(buffer: Buffer): boolean {
|
||||
return buffer.length >= 4
|
||||
&& buffer[0] === 0x28
|
||||
&& buffer[1] === 0xb5
|
||||
&& buffer[2] === 0x2f
|
||||
&& buffer[3] === 0xfd;
|
||||
}
|
||||
|
||||
function decodeRuntimeResponseBuffer(buffer: Buffer, contentEncoding: string | null): Buffer {
|
||||
if (!contentEncoding) return buffer;
|
||||
|
||||
let decoded = buffer;
|
||||
const encodings = contentEncoding
|
||||
.split(',')
|
||||
.map((encoding) => encoding.trim().toLowerCase())
|
||||
.filter(Boolean)
|
||||
.reverse();
|
||||
|
||||
for (const encoding of encodings) {
|
||||
if (encoding === 'zstd') {
|
||||
decoded = zstdDecompressSync(decoded);
|
||||
continue;
|
||||
}
|
||||
if (encoding === 'br') {
|
||||
decoded = brotliDecompressSync(decoded);
|
||||
continue;
|
||||
}
|
||||
if (encoding === 'gzip' || encoding === 'x-gzip') {
|
||||
decoded = gunzipSync(decoded);
|
||||
continue;
|
||||
}
|
||||
if (encoding === 'deflate') {
|
||||
decoded = inflateSync(decoded);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
return decoded;
|
||||
}
|
||||
|
||||
export async function readRuntimeResponseText(
|
||||
response: RuntimeResponse,
|
||||
): Promise<string> {
|
||||
const contentEncoding = response.headers.get('content-encoding');
|
||||
if (!hasZstdContentEncoding(contentEncoding)) {
|
||||
return response.text().catch(() => '');
|
||||
}
|
||||
|
||||
const rawBuffer = Buffer.from(await response.arrayBuffer());
|
||||
try {
|
||||
return decodeRuntimeResponseBuffer(rawBuffer, contentEncoding).toString('utf8');
|
||||
} catch {
|
||||
return looksLikeZstdFrame(rawBuffer) ? '' : rawBuffer.toString('utf8');
|
||||
}
|
||||
}
|
||||
|
||||
export async function materializeErrorResponse(
|
||||
response: RuntimeResponse,
|
||||
): Promise<RuntimeResponse> {
|
||||
if (response.ok) return response;
|
||||
const text = await response.text().catch(() => '');
|
||||
const text = await readRuntimeResponseText(response);
|
||||
const headers = new Headers(response.headers);
|
||||
headers.delete('content-encoding');
|
||||
headers.delete('content-length');
|
||||
return new Response(text, {
|
||||
status: response.status,
|
||||
headers: response.headers,
|
||||
headers,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -83,7 +83,7 @@ function buildClaudeRuntimeHeaders(input: {
|
||||
'User-Agent': getInputHeader(input.claudeHeaders, 'user-agent') || CLAUDE_DEFAULT_USER_AGENT,
|
||||
Connection: 'keep-alive',
|
||||
Accept: input.stream ? 'text/event-stream' : 'application/json',
|
||||
'Accept-Encoding': 'identity',
|
||||
'Accept-Encoding': 'gzip, deflate, br, zstd',
|
||||
};
|
||||
if (input.isClaudeOauthUpstream) {
|
||||
headers.Authorization = `Bearer ${input.tokenValue}`;
|
||||
|
||||
@@ -70,7 +70,7 @@ describe('resolveProviderProfile', () => {
|
||||
expect(result.headers.Authorization).toBe('Bearer oauth-access-token');
|
||||
expect(result.headers['x-api-key']).toBeUndefined();
|
||||
expect(result.headers['anthropic-version']).toBe('2023-06-01');
|
||||
expect(result.headers['Accept-Encoding']).toBe('identity');
|
||||
expect(result.headers['Accept-Encoding']).toBe('gzip, deflate, br, zstd');
|
||||
expect(result.runtime).toMatchObject({
|
||||
executor: 'claude',
|
||||
modelName: 'claude-opus-4-6',
|
||||
|
||||
@@ -34,6 +34,7 @@ import { dispatchRuntimeRequest } from '../../routes/proxy/runtimeExecutor.js';
|
||||
import { detectDownstreamClientContext, type DownstreamClientContext } from '../../routes/proxy/downstreamClientContext.js';
|
||||
import { insertProxyLog } from '../../services/proxyLogStore.js';
|
||||
import { summarizeConversationFileInputsInOpenAiBody } from '../capabilities/conversationFileCapabilities.js';
|
||||
import { readRuntimeResponseText } from '../executors/types.js';
|
||||
|
||||
const MAX_RETRIES = 2;
|
||||
const GEMINI_MODEL_PROBES = [
|
||||
@@ -294,7 +295,7 @@ export async function geminiProxyRoute(app: FastifyInstance) {
|
||||
geminiGenerateContentTransformer.resolveModelsUrl(selected.site.url, apiVersion, selected.tokenValue),
|
||||
{ method: 'GET' },
|
||||
);
|
||||
const text = await upstream.text();
|
||||
const text = await readRuntimeResponseText(upstream);
|
||||
if (!upstream.ok) {
|
||||
lastStatus = upstream.status;
|
||||
lastText = text;
|
||||
@@ -504,7 +505,7 @@ export async function geminiProxyRoute(app: FastifyInstance) {
|
||||
if (!upstream.ok) {
|
||||
lastStatus = upstream.status;
|
||||
lastContentType = contentType;
|
||||
lastText = await upstream.text();
|
||||
lastText = await readRuntimeResponseText(upstream);
|
||||
await tokenRouter.recordFailure?.(selected.channel.id, {
|
||||
status: upstream.status,
|
||||
errorText: lastText,
|
||||
@@ -612,7 +613,7 @@ export async function geminiProxyRoute(app: FastifyInstance) {
|
||||
return;
|
||||
}
|
||||
|
||||
const text = await upstream.text();
|
||||
const text = await readRuntimeResponseText(upstream);
|
||||
const aggregateState = geminiGenerateContentTransformer.stream.createAggregateState();
|
||||
let parsedUsage = EMPTY_PROXY_USAGE;
|
||||
try {
|
||||
@@ -822,7 +823,7 @@ export async function geminiProxyRoute(app: FastifyInstance) {
|
||||
|
||||
upstreamPath = endpointResult.upstreamPath;
|
||||
const upstream = endpointResult.upstream;
|
||||
const rawText = await upstream.text();
|
||||
const rawText = await readRuntimeResponseText(upstream);
|
||||
let upstreamData: unknown = rawText;
|
||||
try {
|
||||
upstreamData = JSON.parse(rawText);
|
||||
|
||||
@@ -145,7 +145,7 @@ describe('claude count_tokens proxy route', () => {
|
||||
expect(targetUrl).toBe('https://api.anthropic.com/v1/messages/count_tokens?beta=true');
|
||||
expect(options.headers['anthropic-version']).toBe('2023-06-01');
|
||||
expect(options.headers['anthropic-beta']).toContain('claude-code-20250219');
|
||||
expect(options.headers['Accept-Encoding']).toBe('identity');
|
||||
expect(options.headers['Accept-Encoding']).toBe('gzip, deflate, br, zstd');
|
||||
|
||||
const forwardedBody = JSON.parse(String(options.body));
|
||||
expect(forwardedBody.model).toBe('claude-opus-4-6');
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { fetch } from 'undici';
|
||||
import { readRuntimeResponseText } from '../../proxy-core/executors/types.js';
|
||||
import { withSiteProxyRequestInit } from '../../services/siteProxy.js';
|
||||
import { summarizeUpstreamError } from './upstreamError.js';
|
||||
import type { UpstreamEndpoint } from './upstreamEndpoint.js';
|
||||
@@ -114,7 +115,7 @@ export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Prom
|
||||
};
|
||||
}
|
||||
|
||||
let rawErrText = await response.text().catch(() => 'unknown error');
|
||||
let rawErrText = await readRuntimeResponseText(response).catch(() => 'unknown error');
|
||||
const baseContext: EndpointAttemptContext = {
|
||||
endpointIndex,
|
||||
endpointCount,
|
||||
|
||||
@@ -864,7 +864,7 @@ describe('buildUpstreamEndpointRequest', () => {
|
||||
expect(request.headers['User-Agent']).toBe('claude-cli/2.1.63 (external, cli)');
|
||||
expect(request.headers.Connection).toBe('keep-alive');
|
||||
expect(request.headers.Accept).toBe('text/event-stream');
|
||||
expect(request.headers['Accept-Encoding']).toBe('identity');
|
||||
expect(request.headers['Accept-Encoding']).toBe('gzip, deflate, br, zstd');
|
||||
});
|
||||
|
||||
it('uses claude-code beta headers and uncompressed non-stream responses for claude upstream requests', () => {
|
||||
@@ -887,7 +887,7 @@ describe('buildUpstreamEndpointRequest', () => {
|
||||
expect(request.headers['anthropic-beta']).toContain('oauth-2025-04-20');
|
||||
expect(request.headers['anthropic-beta']).toContain('context-management-2025-06-27');
|
||||
expect(request.headers.Accept).toBe('application/json');
|
||||
expect(request.headers['Accept-Encoding']).toBe('identity');
|
||||
expect(request.headers['Accept-Encoding']).toBe('gzip, deflate, br, zstd');
|
||||
});
|
||||
|
||||
it('converts system roles to developer in native codex responses bodies', () => {
|
||||
|
||||
@@ -443,7 +443,7 @@ function buildClaudeRuntimeHeaders(input: {
|
||||
'User-Agent': getInputHeader(input.claudeHeaders, 'user-agent') || CLAUDE_DEFAULT_USER_AGENT,
|
||||
Connection: 'keep-alive',
|
||||
Accept: input.stream ? 'text/event-stream' : 'application/json',
|
||||
'Accept-Encoding': 'identity',
|
||||
'Accept-Encoding': 'gzip, deflate, br, zstd',
|
||||
};
|
||||
if (input.isClaudeOauthUpstream) {
|
||||
headers.Authorization = `Bearer ${input.tokenValue}`;
|
||||
|
||||
Reference in New Issue
Block a user