mirror of
https://github.com/tiennm99/ccs.git
synced 2026-09-02 18:18:43 +00:00
fix(proxy): keep stream guards active through sse piping
This commit is contained in:
@@ -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');
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user