fix(proxy): keep stream guards active through sse piping

This commit is contained in:
Tam Nhu Tran
2026-04-15 02:52:29 -04:00
parent a3407093d7
commit 841eeb497c
4 changed files with 36 additions and 16 deletions
-2
View File
@@ -221,8 +221,6 @@ export async function handleProxyMessagesRequest(
resolveOpenAIChatCompletionsUrl(upstream.route.profile.baseUrl), resolveOpenAIChatCompletionsUrl(upstream.route.profile.baseUrl),
buildFetchInit(upstream.route.profile, upstream.body, controller.signal, insecureDispatcher) buildFetchInit(upstream.route.profile, upstream.body, controller.signal, insecureDispatcher)
); );
clearTimeout(timeout);
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,
@@ -3,8 +3,8 @@ import { GlmtTransformer } from '../../glmt/glmt-transformer';
import { SSEParser } from '../../glmt/sse-parser'; import { SSEParser } from '../../glmt/sse-parser';
import type { OpenAIResponse, SSEEvent } from '../../glmt/pipeline'; import type { OpenAIResponse, SSEEvent } from '../../glmt/pipeline';
const JSON_TRANSLATION_ERROR_MESSAGE = 'Failed to translate Cursor JSON response'; const JSON_TRANSLATION_ERROR_MESSAGE = 'Failed to translate OpenAI-compatible JSON response';
const STREAM_TRANSLATION_ERROR_MESSAGE = 'Failed to translate Cursor SSE response'; const STREAM_TRANSLATION_ERROR_MESSAGE = 'Failed to translate OpenAI-compatible SSE response';
type ResponseHeaders = Headers | Record<string, string> | Array<[string, string]>; type ResponseHeaders = Headers | Record<string, string> | Array<[string, string]>;
@@ -103,7 +103,7 @@ async function createAnthropicErrorProxyResponse(response: Response): Promise<Re
: response.status >= 400 && response.status < 500 : response.status >= 400 && response.status < 500
? 'invalid_request_error' ? 'invalid_request_error'
: 'api_error'; : 'api_error';
let message = `Cursor request failed with status ${response.status}`; let message = `Upstream request failed with status ${response.status}`;
try { try {
const contentType = (response.headers.get('content-type') || '').toLowerCase(); const contentType = (response.headers.get('content-type') || '').toLowerCase();
@@ -145,7 +145,7 @@ async function createAnthropicJsonResponse(response: Response): Promise<Response
const anthropicResponse = new GlmtTransformer().transformResponse(openAIResponse); const anthropicResponse = new GlmtTransformer().transformResponse(openAIResponse);
if (isSyntheticTransformationFallback(anthropicResponse)) { if (isSyntheticTransformationFallback(anthropicResponse)) {
logTranslationError( logTranslationError(
'Cursor JSON translation produced synthetic fallback response', 'OpenAI-compatible JSON translation produced synthetic fallback response',
anthropicResponse anthropicResponse
); );
return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE); return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE);
@@ -156,7 +156,7 @@ async function createAnthropicJsonResponse(response: Response): Promise<Response
headers: { 'Content-Type': 'application/json' }, headers: { 'Content-Type': 'application/json' },
}); });
} catch (error) { } catch (error) {
logTranslationError('Cursor JSON translation failed', error); logTranslationError('OpenAI-compatible JSON translation failed', error);
return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE); return createAnthropicErrorResponse(502, 'api_error', JSON_TRANSLATION_ERROR_MESSAGE);
} }
} }
@@ -167,7 +167,7 @@ function createAnthropicStreamingResponse(response: Response): Response {
return createAnthropicErrorResponse( return createAnthropicErrorResponse(
502, 502,
'api_error', 'api_error',
'Cursor stream ended before a response body was available' 'Upstream stream ended before a response body was available'
); );
} }
@@ -209,7 +209,7 @@ function createAnthropicStreamingResponse(response: Response): Response {
} }
} }
} catch (error) { } catch (error) {
logTranslationError('Cursor SSE translation failed', error); logTranslationError('OpenAI-compatible SSE translation failed', error);
controller.enqueue( controller.enqueue(
encoder.encode( encoder.encode(
formatSseEvent( formatSseEvent(
@@ -140,7 +140,7 @@ describe('openai proxy message edge cases', () => {
type: 'error', type: 'error',
error: { error: {
type: 'api_error', type: 'api_error',
message: 'Failed to translate Cursor JSON response', message: 'Failed to translate OpenAI-compatible JSON response',
}, },
}); });
}); });
@@ -202,6 +202,28 @@ describe('openai proxy message edge cases', () => {
}); });
}); });
it('emits an SSE error if the upstream stalls after response headers are sent', async () => {
process.env.CCS_OPENAI_PROXY_REQUEST_TIMEOUT_MS = '50';
await startProxyWithHandler((_req, res) => {
res.writeHead(200, { 'Content-Type': 'text/event-stream' });
res.write(
'data: {"id":"chatcmpl_1","model":"hf-model","choices":[{"index":0,"delta":{"role":"assistant","content":"partial"}}]}\n\n'
);
});
const response = await requestProxy({
model: 'hf-model',
stream: true,
messages: [{ role: 'user', content: 'hello' }],
});
expect(response.status).toBe(200);
const body = await response.text();
expect(body).toContain('event: message_start');
expect(body).toContain('event: error');
expect(body).toContain('"message":"Failed to translate OpenAI-compatible SSE response"');
});
it('aborts the upstream request when the client disconnects mid-flight', async () => { it('aborts the upstream request when the client disconnects mid-flight', async () => {
await startProxyWithHandler(() => {}); await startProxyWithHandler(() => {});
@@ -281,7 +281,7 @@ describe('createAnthropicProxyResponse', () => {
}; };
expect(body.type).toBe('error'); expect(body.type).toBe('error');
expect(body.error?.type).toBe('api_error'); expect(body.error?.type).toBe('api_error');
expect(body.error?.message).toBe('Failed to translate Cursor JSON response'); expect(body.error?.message).toBe('Failed to translate OpenAI-compatible JSON response');
}); });
it('returns 502 when Cursor response is missing choices', async () => { it('returns 502 when Cursor response is missing choices', async () => {
@@ -305,7 +305,7 @@ describe('createAnthropicProxyResponse', () => {
}; };
expect(body.type).toBe('error'); expect(body.type).toBe('error');
expect(body.error?.type).toBe('api_error'); expect(body.error?.type).toBe('api_error');
expect(body.error?.message).toBe('Failed to translate Cursor JSON response'); expect(body.error?.message).toBe('Failed to translate OpenAI-compatible JSON response');
}); });
it('returns 502 when Cursor response has empty choices', async () => { it('returns 502 when Cursor response has empty choices', async () => {
@@ -330,7 +330,7 @@ describe('createAnthropicProxyResponse', () => {
}; };
expect(body.type).toBe('error'); expect(body.type).toBe('error');
expect(body.error?.type).toBe('api_error'); expect(body.error?.type).toBe('api_error');
expect(body.error?.message).toBe('Failed to translate Cursor JSON response'); expect(body.error?.message).toBe('Failed to translate OpenAI-compatible JSON response');
}); });
it('returns Anthropic error envelopes for non-OK upstream JSON errors', async () => { it('returns Anthropic error envelopes for non-OK upstream JSON errors', async () => {
@@ -382,7 +382,7 @@ describe('createAnthropicProxyResponse', () => {
}; };
expect(body.type).toBe('error'); expect(body.type).toBe('error');
expect(body.error?.type).toBe('api_error'); expect(body.error?.type).toBe('api_error');
expect(body.error?.message).toBe('Failed to translate Cursor JSON response'); expect(body.error?.message).toBe('Failed to translate OpenAI-compatible JSON response');
}); });
it('converts OpenAI SSE chunks into Anthropic SSE events', async () => { it('converts OpenAI SSE chunks into Anthropic SSE events', async () => {
@@ -420,7 +420,7 @@ describe('createAnthropicProxyResponse', () => {
expect(body).toContain('event: error'); expect(body).toContain('event: error');
expect(body).toContain('"type":"error"'); expect(body).toContain('"type":"error"');
expect(body).toContain('"error":{"type":"api_error"'); expect(body).toContain('"error":{"type":"api_error"');
expect(body).toContain('Failed to translate Cursor SSE response'); expect(body).toContain('Failed to translate OpenAI-compatible SSE response');
}); });
it('emits Anthropic-style error events when SSE JSON is malformed', async () => { it('emits Anthropic-style error events when SSE JSON is malformed', async () => {
@@ -435,6 +435,6 @@ describe('createAnthropicProxyResponse', () => {
expect(body).toContain('event: error'); expect(body).toContain('event: error');
expect(body).toContain('"type":"error"'); expect(body).toContain('"type":"error"');
expect(body).toContain('"error":{"type":"api_error"'); expect(body).toContain('"error":{"type":"api_error"');
expect(body).toContain('Failed to translate Cursor SSE response'); expect(body).toContain('Failed to translate OpenAI-compatible SSE response');
}); });
}); });