mirror of
https://github.com/tiennm99/ccs.git
synced 2026-10-13 06:15:03 +00:00
refactor(proxy): internalize sse translation and cleanup handlers
This commit is contained in:
1 parent
17187c4abd
commit
a3407093d7
7 files changed
+476
-278
No files matched your search
@@ -1,249 +1,4 @@
|
|||||||
import { DeltaAccumulator } from '../glmt/delta-accumulator';
|
export {
|
||||||
import { GlmtTransformer } from '../glmt/glmt-transformer';
|
createAnthropicErrorResponse,
|
||||||
import { SSEParser } from '../glmt/sse-parser';
|
createAnthropicProxyResponse,
|
||||||
import type { OpenAIResponse, SSEEvent } from '../glmt/pipeline';
|
} from '../proxy/transformers/sse-stream-transformer';
|
||||||
|
|
||||||
const JSON_TRANSLATION_ERROR_MESSAGE = 'Failed to translate Cursor JSON response';
|
|
||||||
const STREAM_TRANSLATION_ERROR_MESSAGE = 'Failed to translate Cursor SSE response';
|
|
||||||
type ResponseHeaders = Headers | Record<string, string> | Array<[string, string]>;
|
|
||||||
|
|
||||||
interface AnthropicErrorPayload {
|
|
||||||
type: 'error';
|
|
||||||
error: {
|
|
||||||
type: string;
|
|
||||||
message: string;
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
function createAnthropicErrorPayload(type: string, message: string): AnthropicErrorPayload {
|
|
||||||
return {
|
|
||||||
type: 'error',
|
|
||||||
error: {
|
|
||||||
type,
|
|
||||||
message,
|
|
||||||
},
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
function formatErrorForLog(error: unknown): string {
|
|
||||||
if (error instanceof Error) {
|
|
||||||
return error.message;
|
|
||||||
}
|
|
||||||
|
|
||||||
try {
|
|
||||||
return JSON.stringify(error);
|
|
||||||
} catch {
|
|
||||||
return String(error);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
function logTranslationError(context: string, error: unknown): void {
|
|
||||||
console.error(`[cursor-anthropic-response] ${context}: ${formatErrorForLog(error)}`);
|
|
||||||
}
|
|
||||||
|
|
||||||
export function createAnthropicErrorResponse(
|
|
||||||
status: number,
|
|
||||||
type: string,
|
|
||||||
message: string,
|
|
||||||
headers?: ResponseHeaders
|
|
||||||
): Response {
|
|
||||||
const responseHeaders = new Headers(headers);
|
|
||||||
responseHeaders.set('Content-Type', 'application/json');
|
|
||||||
responseHeaders.delete('Content-Length');
|
|
||||||
|
|
||||||
return new Response(JSON.stringify(createAnthropicErrorPayload(type, message)), {
|
|
||||||
status,
|
|
||||||
headers: responseHeaders,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
function formatSseEvent(event: string, data: unknown): string {
|
|
||||||
return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
|
|
||||||
}
|
|
||||||
|
|
||||||
function hasTranslatableChoices(value: unknown): value is OpenAIResponse {
|
|
||||||
if (typeof value !== 'object' || value === null) {
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
const { choices } = value as OpenAIResponse;
|
|
||||||
if (!Array.isArray(choices) || choices.length === 0) {
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
const firstChoice = choices[0];
|
|
||||||
if (typeof firstChoice !== 'object' || firstChoice === null) {
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
const message = (firstChoice as { message?: unknown }).message;
|
|
||||||
return typeof message === 'object' && message !== null;
|
|
||||||
}
|
|
||||||
|
|
||||||
function isSyntheticTransformationFallback(value: unknown): boolean {
|
|
||||||
return (
|
|
||||||
typeof value === 'object' &&
|
|
||||||
value !== null &&
|
|
||||||
typeof (value as { id?: unknown }).id === 'string' &&
|
|
||||||
(value as { id: string }).id.startsWith('msg_error_')
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
async function createAnthropicErrorProxyResponse(response: Response): Promise<Response> {
|
|
||||||
const headers = new Headers(response.headers);
|
|
||||||
headers.delete('Content-Type');
|
|
||||||
headers.delete('Content-Length');
|
|
||||||
|
|
||||||
let type =
|
|
||||||
response.status === 401
|
|
||||||
? 'authentication_error'
|
|
||||||
: response.status === 429
|
|
||||||
? 'rate_limit_error'
|
|
||||||
: response.status >= 400 && response.status < 500
|
|
||||||
? 'invalid_request_error'
|
|
||||||
: 'api_error';
|
|
||||||
let message = `Cursor request failed with status ${response.status}`;
|
|
||||||
|
|
||||||
try {
|
|
||||||
const contentType = (response.headers.get('content-type') || '').toLowerCase();
|
|
||||||
if (contentType.includes('application/json')) {
|
|
||||||
const payload = (await response.json()) as {
|
|
||||||
error?: { type?: string; message?: string };
|
|
||||||
message?: string;
|
|
||||||
};
|
|
||||||
|
|
||||||
if (typeof payload?.error?.type === 'string' && payload.error.type.trim().length > 0) {
|
|
||||||
type = payload.error.type;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (typeof payload?.error?.message === 'string' && payload.error.message.trim().length > 0) {
|
|
||||||
message = payload.error.message;
|
|
||||||
} else if (typeof payload?.message === 'string' && payload.message.trim().length > 0) {
|
|
||||||
message = payload.message;
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
const text = (await response.text()).trim();
|
|
||||||
if (text.length > 0) {
|
|
||||||
message = text;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} catch (error) {
|
|
||||||
logTranslationError('Failed to parse Cursor error response', error);
|
|
||||||
}
|
|
||||||
|
|
||||||
return createAnthropicErrorResponse(response.status, type, message, headers);
|
|
||||||
}
|
|
||||||
|
|
||||||
async function createAnthropicJsonResponse(response: Response): Promise<Response> {
|
|
||||||
try {
|
|
||||||
const openAiResponse = await response.json();
|
|
||||||
if (!hasTranslatableChoices(openAiResponse)) {
|
|
||||||
return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE);
|
|
||||||
}
|
|
||||||
|
|
||||||
const anthropicResponse = new GlmtTransformer().transformResponse(openAiResponse);
|
|
||||||
if (isSyntheticTransformationFallback(anthropicResponse)) {
|
|
||||||
logTranslationError(
|
|
||||||
'Cursor JSON translation produced synthetic fallback response',
|
|
||||||
anthropicResponse
|
|
||||||
);
|
|
||||||
return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE);
|
|
||||||
}
|
|
||||||
|
|
||||||
return new Response(JSON.stringify(anthropicResponse), {
|
|
||||||
status: response.status,
|
|
||||||
headers: { 'Content-Type': 'application/json' },
|
|
||||||
});
|
|
||||||
} catch (error) {
|
|
||||||
logTranslationError('Cursor JSON translation failed', error);
|
|
||||||
return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
function createAnthropicStreamingResponse(response: Response): Response {
|
|
||||||
const body = response.body;
|
|
||||||
if (!body) {
|
|
||||||
return createAnthropicErrorResponse(
|
|
||||||
502,
|
|
||||||
'api_error',
|
|
||||||
'Cursor stream ended before a response body was available'
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
const parser = new SSEParser({ throwOnMalformedJson: true });
|
|
||||||
const transformer = new GlmtTransformer();
|
|
||||||
const accumulator = new DeltaAccumulator({});
|
|
||||||
const encoder = new TextEncoder();
|
|
||||||
|
|
||||||
const readable = new ReadableStream<Uint8Array>({
|
|
||||||
async start(controller) {
|
|
||||||
const reader = body.getReader();
|
|
||||||
|
|
||||||
try {
|
|
||||||
while (true) {
|
|
||||||
const { done, value } = await reader.read();
|
|
||||||
if (done) {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
if (!value) {
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
const events = parser.parse(Buffer.from(value));
|
|
||||||
events.forEach((event) => {
|
|
||||||
const anthropicEvents = transformer.transformDelta(event as SSEEvent, accumulator);
|
|
||||||
anthropicEvents.forEach((anthropicEvent) => {
|
|
||||||
controller.enqueue(
|
|
||||||
encoder.encode(formatSseEvent(anthropicEvent.event, anthropicEvent.data))
|
|
||||||
);
|
|
||||||
});
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!accumulator.isFinalized() && accumulator.isMessageStarted()) {
|
|
||||||
transformer.finalizeDelta(accumulator).forEach((anthropicEvent) => {
|
|
||||||
controller.enqueue(
|
|
||||||
encoder.encode(formatSseEvent(anthropicEvent.event, anthropicEvent.data))
|
|
||||||
);
|
|
||||||
});
|
|
||||||
}
|
|
||||||
} catch (error) {
|
|
||||||
logTranslationError('Cursor SSE translation failed', error);
|
|
||||||
controller.enqueue(
|
|
||||||
encoder.encode(
|
|
||||||
formatSseEvent(
|
|
||||||
'error',
|
|
||||||
createAnthropicErrorPayload('api_error', STREAM_TRANSLATION_ERROR_MESSAGE)
|
|
||||||
)
|
|
||||||
)
|
|
||||||
);
|
|
||||||
} finally {
|
|
||||||
reader.releaseLock();
|
|
||||||
controller.close();
|
|
||||||
}
|
|
||||||
},
|
|
||||||
});
|
|
||||||
|
|
||||||
return new Response(readable, {
|
|
||||||
status: response.status,
|
|
||||||
headers: {
|
|
||||||
'Content-Type': 'text/event-stream',
|
|
||||||
'Cache-Control': 'no-cache',
|
|
||||||
Connection: 'keep-alive',
|
|
||||||
},
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
export async function createAnthropicProxyResponse(response: Response): Promise<Response> {
|
|
||||||
if (!response.ok) {
|
|
||||||
return createAnthropicErrorProxyResponse(response);
|
|
||||||
}
|
|
||||||
|
|
||||||
const contentType = (response.headers.get('content-type') || '').toLowerCase();
|
|
||||||
const isEventStream =
|
|
||||||
contentType === 'text/event-stream' || contentType.startsWith('text/event-stream;');
|
|
||||||
|
|
||||||
return isEventStream
|
|
||||||
? createAnthropicStreamingResponse(response)
|
|
||||||
: createAnthropicJsonResponse(response);
|
|
||||||
}
|
|
||||||
@@ -49,7 +49,7 @@ export class SSEParser {
|
|||||||
* @returns Array of parsed events
|
* @returns Array of parsed events
|
||||||
*/
|
*/
|
||||||
parse(chunk: Buffer | string): SSEEvent[] {
|
parse(chunk: Buffer | string): SSEEvent[] {
|
||||||
this.buffer += chunk.toString().replace(/\r\n/g, '\n');
|
this.buffer += chunk.toString().replace(/\r\n?/g, '\n');
|
||||||
|
|
||||||
// C-01 Fix: Prevent unbounded buffer growth (DoS protection)
|
// C-01 Fix: Prevent unbounded buffer growth (DoS protection)
|
||||||
if (this.buffer.length > this.maxBufferSize) {
|
if (this.buffer.length > this.maxBufferSize) {
|
||||||
|
|||||||
@@ -109,6 +109,61 @@ function formatTimeoutDuration(timeoutMs: number): string {
|
|||||||
return timeoutMs % 1000 === 0 ? `${timeoutMs / 1000} seconds` : `${timeoutMs}ms`;
|
return timeoutMs % 1000 === 0 ? `${timeoutMs / 1000} seconds` : `${timeoutMs}ms`;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function registerOnceListener(
|
||||||
|
emitter: NodeJS.EventEmitter | null | undefined,
|
||||||
|
event: string,
|
||||||
|
handler: () => void
|
||||||
|
): () => void {
|
||||||
|
if (!emitter) {
|
||||||
|
return () => {};
|
||||||
|
}
|
||||||
|
|
||||||
|
emitter.once(event, handler);
|
||||||
|
return () => {
|
||||||
|
emitter.removeListener(event, handler);
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
export function attachDisconnectAbortHandlers(
|
||||||
|
req: http.IncomingMessage,
|
||||||
|
res: http.ServerResponse,
|
||||||
|
controller: AbortController,
|
||||||
|
onDisconnect: (source: string) => void
|
||||||
|
): () => void {
|
||||||
|
const abortOnDisconnect = (source: string) => {
|
||||||
|
if (!controller.signal.aborted && !res.writableEnded) {
|
||||||
|
onDisconnect(source);
|
||||||
|
controller.abort();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
const cleanupFns = [
|
||||||
|
registerOnceListener(req, 'aborted', () => abortOnDisconnect('req.aborted')),
|
||||||
|
registerOnceListener(req, 'close', () => abortOnDisconnect('req.close')),
|
||||||
|
registerOnceListener(req.socket, 'close', () => abortOnDisconnect('req.socket.close')),
|
||||||
|
registerOnceListener(res, 'close', () => abortOnDisconnect('res.close')),
|
||||||
|
registerOnceListener(res.socket, 'close', () => abortOnDisconnect('res.socket.close')),
|
||||||
|
];
|
||||||
|
|
||||||
|
const disconnectPoll = setInterval(() => {
|
||||||
|
if (
|
||||||
|
req.destroyed ||
|
||||||
|
res.destroyed ||
|
||||||
|
req.socket?.destroyed === true ||
|
||||||
|
res.socket?.destroyed === true
|
||||||
|
) {
|
||||||
|
abortOnDisconnect('poll.destroyed');
|
||||||
|
}
|
||||||
|
}, 50);
|
||||||
|
|
||||||
|
return () => {
|
||||||
|
clearInterval(disconnectPoll);
|
||||||
|
for (const cleanup of cleanupFns) {
|
||||||
|
cleanup();
|
||||||
|
}
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
export async function handleProxyMessagesRequest(
|
export async function handleProxyMessagesRequest(
|
||||||
req: http.IncomingMessage,
|
req: http.IncomingMessage,
|
||||||
res: http.ServerResponse,
|
res: http.ServerResponse,
|
||||||
@@ -145,8 +200,11 @@ export async function handleProxyMessagesRequest(
|
|||||||
const controller = new AbortController();
|
const controller = new AbortController();
|
||||||
timeoutMs = getRequestTimeoutMs();
|
timeoutMs = getRequestTimeoutMs();
|
||||||
const timeout = setTimeout(() => controller.abort(), timeoutMs);
|
const timeout = setTimeout(() => controller.abort(), timeoutMs);
|
||||||
const abortOnDisconnect = (source: string) => {
|
const cleanupDisconnectHandlers = attachDisconnectAbortHandlers(
|
||||||
if (!controller.signal.aborted && !res.writableEnded) {
|
req,
|
||||||
|
res,
|
||||||
|
controller,
|
||||||
|
(source) => {
|
||||||
logger.info(
|
logger.info(
|
||||||
'request.disconnect',
|
'request.disconnect',
|
||||||
'Aborting upstream request after local client disconnect',
|
'Aborting upstream request after local client disconnect',
|
||||||
@@ -155,25 +213,8 @@ export async function handleProxyMessagesRequest(
|
|||||||
source,
|
source,
|
||||||
}
|
}
|
||||||
);
|
);
|
||||||
controller.abort();
|
|
||||||
}
|
}
|
||||||
};
|
);
|
||||||
|
|
||||||
req.on('aborted', () => abortOnDisconnect('req.aborted'));
|
|
||||||
req.on('close', () => abortOnDisconnect('req.close'));
|
|
||||||
req.socket?.on('close', () => abortOnDisconnect('req.socket.close'));
|
|
||||||
res.on('close', () => abortOnDisconnect('res.close'));
|
|
||||||
res.socket?.on('close', () => abortOnDisconnect('res.socket.close'));
|
|
||||||
const disconnectPoll = setInterval(() => {
|
|
||||||
if (
|
|
||||||
req.destroyed ||
|
|
||||||
res.destroyed ||
|
|
||||||
req.socket?.destroyed === true ||
|
|
||||||
res.socket?.destroyed === true
|
|
||||||
) {
|
|
||||||
abortOnDisconnect('poll.destroyed');
|
|
||||||
}
|
|
||||||
}, 50);
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const upstreamResponse = await fetch(
|
const upstreamResponse = await fetch(
|
||||||
@@ -181,7 +222,7 @@ export async function handleProxyMessagesRequest(
|
|||||||
buildFetchInit(upstream.route.profile, upstream.body, controller.signal, insecureDispatcher)
|
buildFetchInit(upstream.route.profile, upstream.body, controller.signal, insecureDispatcher)
|
||||||
);
|
);
|
||||||
clearTimeout(timeout);
|
clearTimeout(timeout);
|
||||||
clearInterval(disconnectPoll);
|
cleanupDisconnectHandlers();
|
||||||
logger.info('response.received', 'Received upstream response', {
|
logger.info('response.received', 'Received upstream response', {
|
||||||
profileName: profile.profileName,
|
profileName: profile.profileName,
|
||||||
routedProfileName: upstream.route.profile.profileName,
|
routedProfileName: upstream.route.profile.profileName,
|
||||||
@@ -191,7 +232,7 @@ export async function handleProxyMessagesRequest(
|
|||||||
await pipeWebResponseToNode(response, res);
|
await pipeWebResponseToNode(response, res);
|
||||||
} finally {
|
} finally {
|
||||||
clearTimeout(timeout);
|
clearTimeout(timeout);
|
||||||
clearInterval(disconnectPoll);
|
cleanupDisconnectHandlers();
|
||||||
}
|
}
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
const message = error instanceof Error ? error.message : 'Unknown proxy error';
|
const message = error instanceof Error ? error.message : 'Unknown proxy error';
|
||||||
|
|||||||
@@ -1,7 +1,253 @@
|
|||||||
import {
|
import { DeltaAccumulator } from '../../glmt/delta-accumulator';
|
||||||
createAnthropicErrorResponse,
|
import { GlmtTransformer } from '../../glmt/glmt-transformer';
|
||||||
createAnthropicProxyResponse,
|
import { SSEParser } from '../../glmt/sse-parser';
|
||||||
} from '../../cursor/cursor-anthropic-response';
|
import type { OpenAIResponse, SSEEvent } from '../../glmt/pipeline';
|
||||||
|
|
||||||
|
const JSON_TRANSLATION_ERROR_MESSAGE = 'Failed to translate Cursor JSON response';
|
||||||
|
const STREAM_TRANSLATION_ERROR_MESSAGE = 'Failed to translate Cursor SSE response';
|
||||||
|
|
||||||
|
type ResponseHeaders = Headers | Record<string, string> | Array<[string, string]>;
|
||||||
|
|
||||||
|
interface AnthropicErrorPayload {
|
||||||
|
type: 'error';
|
||||||
|
error: {
|
||||||
|
type: string;
|
||||||
|
message: string;
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
function createAnthropicErrorPayload(type: string, message: string): AnthropicErrorPayload {
|
||||||
|
return {
|
||||||
|
type: 'error',
|
||||||
|
error: {
|
||||||
|
type,
|
||||||
|
message,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
function formatErrorForLog(error: unknown): string {
|
||||||
|
if (error instanceof Error) {
|
||||||
|
return error.message;
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
return JSON.stringify(error);
|
||||||
|
} catch {
|
||||||
|
return String(error);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function logTranslationError(context: string, error: unknown): void {
|
||||||
|
console.error(`[proxy-sse-transformer] ${context}: ${formatErrorForLog(error)}`);
|
||||||
|
}
|
||||||
|
|
||||||
|
export function createAnthropicErrorResponse(
|
||||||
|
status: number,
|
||||||
|
type: string,
|
||||||
|
message: string,
|
||||||
|
headers?: ResponseHeaders
|
||||||
|
): Response {
|
||||||
|
const responseHeaders = new Headers(headers);
|
||||||
|
responseHeaders.set('Content-Type', 'application/json');
|
||||||
|
responseHeaders.delete('Content-Length');
|
||||||
|
|
||||||
|
return new Response(JSON.stringify(createAnthropicErrorPayload(type, message)), {
|
||||||
|
status,
|
||||||
|
headers: responseHeaders,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
function formatSseEvent(event: string, data: unknown): string {
|
||||||
|
return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
|
||||||
|
}
|
||||||
|
|
||||||
|
function hasTranslatableChoices(value: unknown): value is OpenAIResponse {
|
||||||
|
if (typeof value !== 'object' || value === null) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
const { choices } = value as OpenAIResponse;
|
||||||
|
if (!Array.isArray(choices) || choices.length === 0) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
const firstChoice = choices[0];
|
||||||
|
if (typeof firstChoice !== 'object' || firstChoice === null) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
const message = (firstChoice as { message?: unknown }).message;
|
||||||
|
return typeof message === 'object' && message !== null;
|
||||||
|
}
|
||||||
|
|
||||||
|
function isSyntheticTransformationFallback(value: unknown): boolean {
|
||||||
|
return (
|
||||||
|
typeof value === 'object' &&
|
||||||
|
value !== null &&
|
||||||
|
typeof (value as { id?: unknown }).id === 'string' &&
|
||||||
|
(value as { id: string }).id.startsWith('msg_error_')
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
async function createAnthropicErrorProxyResponse(response: Response): Promise<Response> {
|
||||||
|
const headers = new Headers(response.headers);
|
||||||
|
headers.delete('Content-Type');
|
||||||
|
headers.delete('Content-Length');
|
||||||
|
|
||||||
|
let type =
|
||||||
|
response.status === 401
|
||||||
|
? 'authentication_error'
|
||||||
|
: response.status === 429
|
||||||
|
? 'rate_limit_error'
|
||||||
|
: response.status >= 400 && response.status < 500
|
||||||
|
? 'invalid_request_error'
|
||||||
|
: 'api_error';
|
||||||
|
let message = `Cursor request failed with status ${response.status}`;
|
||||||
|
|
||||||
|
try {
|
||||||
|
const contentType = (response.headers.get('content-type') || '').toLowerCase();
|
||||||
|
if (contentType.includes('application/json')) {
|
||||||
|
const payload = (await response.json()) as {
|
||||||
|
error?: { type?: string; message?: string };
|
||||||
|
message?: string;
|
||||||
|
};
|
||||||
|
|
||||||
|
if (typeof payload?.error?.type === 'string' && payload.error.type.trim().length > 0) {
|
||||||
|
type = payload.error.type;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (typeof payload?.error?.message === 'string' && payload.error.message.trim().length > 0) {
|
||||||
|
message = payload.error.message;
|
||||||
|
} else if (typeof payload?.message === 'string' && payload.message.trim().length > 0) {
|
||||||
|
message = payload.message;
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
const text = (await response.text()).trim();
|
||||||
|
if (text.length > 0) {
|
||||||
|
message = text;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
logTranslationError('Failed to parse upstream error response', error);
|
||||||
|
}
|
||||||
|
|
||||||
|
return createAnthropicErrorResponse(response.status, type, message, headers);
|
||||||
|
}
|
||||||
|
|
||||||
|
async function createAnthropicJsonResponse(response: Response): Promise<Response> {
|
||||||
|
try {
|
||||||
|
const openAIResponse = await response.json();
|
||||||
|
if (!hasTranslatableChoices(openAIResponse)) {
|
||||||
|
return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE);
|
||||||
|
}
|
||||||
|
|
||||||
|
const anthropicResponse = new GlmtTransformer().transformResponse(openAIResponse);
|
||||||
|
if (isSyntheticTransformationFallback(anthropicResponse)) {
|
||||||
|
logTranslationError(
|
||||||
|
'Cursor JSON translation produced synthetic fallback response',
|
||||||
|
anthropicResponse
|
||||||
|
);
|
||||||
|
return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE);
|
||||||
|
}
|
||||||
|
|
||||||
|
return new Response(JSON.stringify(anthropicResponse), {
|
||||||
|
status: response.status,
|
||||||
|
headers: { 'Content-Type': 'application/json' },
|
||||||
|
});
|
||||||
|
} catch (error) {
|
||||||
|
logTranslationError('Cursor JSON translation failed', error);
|
||||||
|
return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function createAnthropicStreamingResponse(response: Response): Response {
|
||||||
|
const body = response.body;
|
||||||
|
if (!body) {
|
||||||
|
return createAnthropicErrorResponse(
|
||||||
|
502,
|
||||||
|
'api_error',
|
||||||
|
'Cursor stream ended before a response body was available'
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
const parser = new SSEParser({ throwOnMalformedJson: true });
|
||||||
|
const transformer = new GlmtTransformer();
|
||||||
|
const accumulator = new DeltaAccumulator({});
|
||||||
|
const encoder = new TextEncoder();
|
||||||
|
|
||||||
|
const readable = new ReadableStream<Uint8Array>({
|
||||||
|
async start(controller) {
|
||||||
|
const reader = body.getReader();
|
||||||
|
|
||||||
|
try {
|
||||||
|
while (true) {
|
||||||
|
const { done, value } = await reader.read();
|
||||||
|
if (done) {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
if (!value) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
const events = parser.parse(Buffer.from(value));
|
||||||
|
for (const event of events) {
|
||||||
|
const anthropicEvents = transformer.transformDelta(event as SSEEvent, accumulator);
|
||||||
|
for (const anthropicEvent of anthropicEvents) {
|
||||||
|
controller.enqueue(
|
||||||
|
encoder.encode(formatSseEvent(anthropicEvent.event, anthropicEvent.data))
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (!accumulator.isFinalized() && accumulator.isMessageStarted()) {
|
||||||
|
for (const anthropicEvent of transformer.finalizeDelta(accumulator)) {
|
||||||
|
controller.enqueue(
|
||||||
|
encoder.encode(formatSseEvent(anthropicEvent.event, anthropicEvent.data))
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
logTranslationError('Cursor SSE translation failed', error);
|
||||||
|
controller.enqueue(
|
||||||
|
encoder.encode(
|
||||||
|
formatSseEvent(
|
||||||
|
'error',
|
||||||
|
createAnthropicErrorPayload('api_error', STREAM_TRANSLATION_ERROR_MESSAGE)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
);
|
||||||
|
} finally {
|
||||||
|
reader.releaseLock();
|
||||||
|
controller.close();
|
||||||
|
}
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
return new Response(readable, {
|
||||||
|
status: response.status,
|
||||||
|
headers: {
|
||||||
|
'Content-Type': 'text/event-stream',
|
||||||
|
'Cache-Control': 'no-cache',
|
||||||
|
Connection: 'keep-alive',
|
||||||
|
},
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function createAnthropicProxyResponse(response: Response): Promise<Response> {
|
||||||
|
if (!response.ok) {
|
||||||
|
return createAnthropicErrorProxyResponse(response);
|
||||||
|
}
|
||||||
|
|
||||||
|
const contentType = (response.headers.get('content-type') || '').toLowerCase();
|
||||||
|
const isEventStream =
|
||||||
|
contentType === 'text/event-stream' || contentType.startsWith('text/event-stream;');
|
||||||
|
|
||||||
|
return isEventStream
|
||||||
|
? createAnthropicStreamingResponse(response)
|
||||||
|
: createAnthropicJsonResponse(response);
|
||||||
|
}
|
||||||
|
|
||||||
export class ProxySseStreamTransformer {
|
export class ProxySseStreamTransformer {
|
||||||
async transform(response: Response): Promise<Response> {
|
async transform(response: Response): Promise<Response> {
|
||||||
|
|||||||
@@ -28,8 +28,22 @@ describe('SSEParser', () => {
|
|||||||
const events = parser.parse('\n\n');
|
const events = parser.parse('\n\n');
|
||||||
|
|
||||||
expect(events).toHaveLength(1);
|
expect(events).toHaveLength(1);
|
||||||
expect((events[0]?.data as { choices?: Array<{ delta?: { content?: string } }> })?.choices?.[0]?.delta?.content).toBe(
|
expect(
|
||||||
'Hi'
|
(events[0]?.data as { choices?: Array<{ delta?: { content?: string } }> })?.choices?.[0]
|
||||||
|
?.delta?.content
|
||||||
|
).toBe('Hi');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('accepts standalone carriage-return line endings', () => {
|
||||||
|
const parser = new SSEParser({ throwOnMalformedJson: true });
|
||||||
|
const events = parser.parse(
|
||||||
|
['event: message', 'data: {"choices":[{"delta":{"content":"Legacy"}}]}', '', ''].join('\r')
|
||||||
);
|
);
|
||||||
|
|
||||||
|
expect(events).toHaveLength(1);
|
||||||
|
expect(
|
||||||
|
(events[0]?.data as { choices?: Array<{ delta?: { content?: string } }> })?.choices?.[0]
|
||||||
|
?.delta?.content
|
||||||
|
).toBe('Legacy');
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
@@ -0,0 +1,72 @@
|
|||||||
|
import { describe, expect, it } from 'bun:test';
|
||||||
|
import { EventEmitter } from 'events';
|
||||||
|
import { attachDisconnectAbortHandlers } from '../../../src/proxy/server/messages-route';
|
||||||
|
|
||||||
|
class FakeSocket extends EventEmitter {
|
||||||
|
destroyed = false;
|
||||||
|
}
|
||||||
|
|
||||||
|
class FakeRequest extends EventEmitter {
|
||||||
|
destroyed = false;
|
||||||
|
socket = new FakeSocket();
|
||||||
|
}
|
||||||
|
|
||||||
|
class FakeResponse extends EventEmitter {
|
||||||
|
destroyed = false;
|
||||||
|
writableEnded = false;
|
||||||
|
socket = new FakeSocket();
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('attachDisconnectAbortHandlers', () => {
|
||||||
|
it('cleans up registered listeners after the request completes', () => {
|
||||||
|
const req = new FakeRequest();
|
||||||
|
const res = new FakeResponse();
|
||||||
|
const controller = new AbortController();
|
||||||
|
|
||||||
|
const cleanup = attachDisconnectAbortHandlers(
|
||||||
|
req as never,
|
||||||
|
res as never,
|
||||||
|
controller,
|
||||||
|
() => {}
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(req.listenerCount('aborted')).toBe(1);
|
||||||
|
expect(req.listenerCount('close')).toBe(1);
|
||||||
|
expect(req.socket.listenerCount('close')).toBe(1);
|
||||||
|
expect(res.listenerCount('close')).toBe(1);
|
||||||
|
expect(res.socket.listenerCount('close')).toBe(1);
|
||||||
|
|
||||||
|
cleanup();
|
||||||
|
|
||||||
|
expect(req.listenerCount('aborted')).toBe(0);
|
||||||
|
expect(req.listenerCount('close')).toBe(0);
|
||||||
|
expect(req.socket.listenerCount('close')).toBe(0);
|
||||||
|
expect(res.listenerCount('close')).toBe(0);
|
||||||
|
expect(res.socket.listenerCount('close')).toBe(0);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('aborts at most once when disconnect signals race each other', () => {
|
||||||
|
const req = new FakeRequest();
|
||||||
|
const res = new FakeResponse();
|
||||||
|
const controller = new AbortController();
|
||||||
|
let disconnectCount = 0;
|
||||||
|
|
||||||
|
const cleanup = attachDisconnectAbortHandlers(
|
||||||
|
req as never,
|
||||||
|
res as never,
|
||||||
|
controller,
|
||||||
|
() => {
|
||||||
|
disconnectCount += 1;
|
||||||
|
}
|
||||||
|
);
|
||||||
|
|
||||||
|
req.emit('close');
|
||||||
|
req.socket.emit('close');
|
||||||
|
res.emit('close');
|
||||||
|
|
||||||
|
expect(controller.signal.aborted).toBe(true);
|
||||||
|
expect(disconnectCount).toBe(1);
|
||||||
|
|
||||||
|
cleanup();
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -0,0 +1,70 @@
|
|||||||
|
import { describe, expect, it } from 'bun:test';
|
||||||
|
import { createAnthropicProxyResponse } from '../../../../src/proxy/transformers/sse-stream-transformer';
|
||||||
|
|
||||||
|
describe('proxy SSE stream transformer', () => {
|
||||||
|
it('converts OpenAI JSON into Anthropic message JSON', async () => {
|
||||||
|
const response = new Response(
|
||||||
|
JSON.stringify({
|
||||||
|
id: 'chatcmpl_1',
|
||||||
|
model: 'claude-sonnet-4.5',
|
||||||
|
choices: [
|
||||||
|
{
|
||||||
|
index: 0,
|
||||||
|
message: {
|
||||||
|
role: 'assistant',
|
||||||
|
content: 'Here is the result.',
|
||||||
|
reasoning_content: 'Need to call the tool first.',
|
||||||
|
tool_calls: [
|
||||||
|
{
|
||||||
|
id: 'toolu_2',
|
||||||
|
type: 'function',
|
||||||
|
function: { name: 'search', arguments: '{"q":"proxy"}' },
|
||||||
|
},
|
||||||
|
],
|
||||||
|
},
|
||||||
|
finish_reason: 'tool_calls',
|
||||||
|
},
|
||||||
|
],
|
||||||
|
usage: { prompt_tokens: 12, completion_tokens: 4, total_tokens: 16 },
|
||||||
|
}),
|
||||||
|
{
|
||||||
|
status: 200,
|
||||||
|
headers: { 'Content-Type': 'application/json' },
|
||||||
|
}
|
||||||
|
);
|
||||||
|
|
||||||
|
const transformed = await createAnthropicProxyResponse(response);
|
||||||
|
const body = (await transformed.json()) as {
|
||||||
|
type: string;
|
||||||
|
stop_reason: string;
|
||||||
|
content: Array<{ type: string; thinking?: string; name?: string }>;
|
||||||
|
};
|
||||||
|
|
||||||
|
expect(body.type).toBe('message');
|
||||||
|
expect(body.stop_reason).toBe('tool_use');
|
||||||
|
expect(body.content.map((block) => block.type)).toEqual(['thinking', 'text', 'tool_use']);
|
||||||
|
expect(body.content[0]?.thinking).toContain('Need to call the tool first');
|
||||||
|
expect(body.content[2]?.name).toBe('search');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('converts OpenAI SSE chunks into Anthropic SSE events', async () => {
|
||||||
|
const openAISse = [
|
||||||
|
'data: {"id":"chatcmpl_2","object":"chat.completion.chunk","created":1,"model":"claude-sonnet-4.5","choices":[{"index":0,"delta":{"role":"assistant","content":"Hello"},"finish_reason":null}]}\n\n',
|
||||||
|
'data: {"id":"chatcmpl_2","object":"chat.completion.chunk","created":1,"model":"claude-sonnet-4.5","choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":5,"completion_tokens":1,"total_tokens":6}}\n\n',
|
||||||
|
'data: [DONE]\n\n',
|
||||||
|
].join('');
|
||||||
|
|
||||||
|
const transformed = await createAnthropicProxyResponse(
|
||||||
|
new Response(openAISse, {
|
||||||
|
status: 200,
|
||||||
|
headers: { 'Content-Type': 'text/event-stream' },
|
||||||
|
})
|
||||||
|
);
|
||||||
|
|
||||||
|
const body = await transformed.text();
|
||||||
|
expect(body).toContain('event: message_start');
|
||||||
|
expect(body).toContain('event: content_block_start');
|
||||||
|
expect(body).toContain('"type":"text_delta"');
|
||||||
|
expect(body).toContain('event: message_stop');
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in new issue
Block a user