From a3407093d70bef44874cdf26a06436e0d287c1a7 Mon Sep 17 00:00:00 2001 From: Tam Nhu Tran Date: Wed, 15 Apr 2026 01:44:01 -0400 Subject: [PATCH] refactor(proxy): internalize sse translation and cleanup handlers --- src/cursor/cursor-anthropic-response.ts | 253 +---------------- src/glmt/sse-parser.ts | 2 +- src/proxy/server/messages-route.ts | 85 ++++-- .../transformers/sse-stream-transformer.ts | 254 +++++++++++++++++- tests/unit/glmt/sse-parser.test.ts | 18 +- tests/unit/proxy/messages-route.test.ts | 72 +++++ .../sse-stream-transformer.test.ts | 70 +++++ 7 files changed, 476 insertions(+), 278 deletions(-) create mode 100644 tests/unit/proxy/messages-route.test.ts create mode 100644 tests/unit/proxy/transformers/sse-stream-transformer.test.ts diff --git a/src/cursor/cursor-anthropic-response.ts b/src/cursor/cursor-anthropic-response.ts index 6575fd14..c5c5a0e2 100644 --- a/src/cursor/cursor-anthropic-response.ts +++ b/src/cursor/cursor-anthropic-response.ts @@ -1,249 +1,4 @@ -import { DeltaAccumulator } from '../glmt/delta-accumulator'; -import { GlmtTransformer } from '../glmt/glmt-transformer'; -import { SSEParser } from '../glmt/sse-parser'; -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 | 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 { - 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 { - 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({ - 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 { - 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 { + createAnthropicErrorResponse, + createAnthropicProxyResponse, +} from '../proxy/transformers/sse-stream-transformer'; diff --git a/src/glmt/sse-parser.ts b/src/glmt/sse-parser.ts index c161a7df..d6fff5d6 100644 --- a/src/glmt/sse-parser.ts +++ b/src/glmt/sse-parser.ts @@ -49,7 +49,7 @@ export class SSEParser { * @returns Array of parsed events */ 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) if (this.buffer.length > this.maxBufferSize) { diff --git a/src/proxy/server/messages-route.ts b/src/proxy/server/messages-route.ts index ae0d33d9..13c329e8 100644 --- a/src/proxy/server/messages-route.ts +++ b/src/proxy/server/messages-route.ts @@ -109,6 +109,61 @@ function formatTimeoutDuration(timeoutMs: number): string { 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( req: http.IncomingMessage, res: http.ServerResponse, @@ -145,8 +200,11 @@ export async function handleProxyMessagesRequest( const controller = new AbortController(); timeoutMs = getRequestTimeoutMs(); const timeout = setTimeout(() => controller.abort(), timeoutMs); - const abortOnDisconnect = (source: string) => { - if (!controller.signal.aborted && !res.writableEnded) { + const cleanupDisconnectHandlers = attachDisconnectAbortHandlers( + req, + res, + controller, + (source) => { logger.info( 'request.disconnect', 'Aborting upstream request after local client disconnect', @@ -155,25 +213,8 @@ export async function handleProxyMessagesRequest( 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 { const upstreamResponse = await fetch( @@ -181,7 +222,7 @@ export async function handleProxyMessagesRequest( buildFetchInit(upstream.route.profile, upstream.body, controller.signal, insecureDispatcher) ); clearTimeout(timeout); - clearInterval(disconnectPoll); + cleanupDisconnectHandlers(); logger.info('response.received', 'Received upstream response', { profileName: profile.profileName, routedProfileName: upstream.route.profile.profileName, @@ -191,7 +232,7 @@ export async function handleProxyMessagesRequest( await pipeWebResponseToNode(response, res); } finally { clearTimeout(timeout); - clearInterval(disconnectPoll); + cleanupDisconnectHandlers(); } } catch (error) { const message = error instanceof Error ? error.message : 'Unknown proxy error'; diff --git a/src/proxy/transformers/sse-stream-transformer.ts b/src/proxy/transformers/sse-stream-transformer.ts index 53774c21..0211e211 100644 --- a/src/proxy/transformers/sse-stream-transformer.ts +++ b/src/proxy/transformers/sse-stream-transformer.ts @@ -1,7 +1,253 @@ -import { - createAnthropicErrorResponse, - createAnthropicProxyResponse, -} from '../../cursor/cursor-anthropic-response'; +import { DeltaAccumulator } from '../../glmt/delta-accumulator'; +import { GlmtTransformer } from '../../glmt/glmt-transformer'; +import { SSEParser } from '../../glmt/sse-parser'; +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 | 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 { + 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 { + 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({ + 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 { + 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 { async transform(response: Response): Promise { diff --git a/tests/unit/glmt/sse-parser.test.ts b/tests/unit/glmt/sse-parser.test.ts index 10f14177..740fc0e1 100644 --- a/tests/unit/glmt/sse-parser.test.ts +++ b/tests/unit/glmt/sse-parser.test.ts @@ -28,8 +28,22 @@ describe('SSEParser', () => { const events = parser.parse('\n\n'); expect(events).toHaveLength(1); - expect((events[0]?.data as { choices?: Array<{ delta?: { content?: string } }> })?.choices?.[0]?.delta?.content).toBe( - 'Hi' + expect( + (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'); }); }); diff --git a/tests/unit/proxy/messages-route.test.ts b/tests/unit/proxy/messages-route.test.ts new file mode 100644 index 00000000..3ac40437 --- /dev/null +++ b/tests/unit/proxy/messages-route.test.ts @@ -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(); + }); +}); diff --git a/tests/unit/proxy/transformers/sse-stream-transformer.test.ts b/tests/unit/proxy/transformers/sse-stream-transformer.test.ts new file mode 100644 index 00000000..9faebabc --- /dev/null +++ b/tests/unit/proxy/transformers/sse-stream-transformer.test.ts @@ -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'); + }); +});