mirror of
https://github.com/tiennm99/ccs.git
synced 2026-07-17 04:17:11 +00:00
fix(cursor): address third-round review feedback for protobuf module
HIGH PRIORITY FIXES: - Extract shared buffer parsing logic into parseProtobufFrames generator method (DRY violation fix) - both JSON and SSE transformers now use common frame parsing loop, eliminating ~60% code duplication - Use COMPRESS_FLAG constants instead of hardcoded 0x01/0x02/0x03 in parseConnectRPCFrame for better maintainability MEDIUM PRIORITY FIXES: - Return empty buffer on gzip decompression failure (prevents silent data corruption) - ALREADY FIXED - Add debug warning for unknown message roles in convertMessages - Create GitHub issue #535 for FIELD namespace refactoring follow-up - Add test coverage: transformProtobufToSSE, error paths, unknown roles LOW PRIORITY FIXES: - Rename checksum test to clarify timestamp granularity (~16 min) - Fix debug log function name in SSE transformer - FIXED BY REFACTOR - Add comment to TOOL_RESULT_RESULT field documenting future use All tests pass (1593 pass, 0 fail) All validation checks pass (typecheck + lint + format + tests)
This commit is contained in:
Tam Nhu Tran
@@ -109,7 +109,7 @@ function decompressPayload(payload: Buffer, flags: number): Buffer {
|
||||
if (process.env.CCS_DEBUG) {
|
||||
console.error('[cursor] gzip decompression failed:', err);
|
||||
}
|
||||
return payload;
|
||||
return Buffer.alloc(0);
|
||||
}
|
||||
}
|
||||
return payload;
|
||||
@@ -407,11 +407,88 @@ export class CursorExecutor {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse protobuf buffer into frames and extract text/toolcalls.
|
||||
* Shared logic between JSON and SSE transformers.
|
||||
*/
|
||||
private *parseProtobufFrames(buffer: Buffer): Generator<
|
||||
| { type: 'error'; response: Response }
|
||||
| { type: 'text'; text: string }
|
||||
| {
|
||||
type: 'toolCall';
|
||||
toolCall: {
|
||||
id: string;
|
||||
type: string;
|
||||
function: { name: string; arguments: string };
|
||||
isLast: boolean;
|
||||
};
|
||||
}
|
||||
> {
|
||||
let offset = 0;
|
||||
|
||||
while (offset < buffer.length) {
|
||||
if (offset + 5 > buffer.length) break;
|
||||
|
||||
const flags = buffer[offset];
|
||||
const length = buffer.readUInt32BE(offset + 1);
|
||||
|
||||
if (offset + 5 + length > buffer.length) break;
|
||||
|
||||
let payload = buffer.slice(offset + 5, offset + 5 + length);
|
||||
offset += 5 + length;
|
||||
|
||||
payload = decompressPayload(payload, flags);
|
||||
|
||||
// Check for JSON error format
|
||||
try {
|
||||
const text = payload.toString('utf-8');
|
||||
if (text.startsWith('{') && text.includes('"error"')) {
|
||||
yield { type: 'error', response: createErrorResponse(JSON.parse(text)) };
|
||||
return;
|
||||
}
|
||||
} catch (err) {
|
||||
if (process.env.CCS_DEBUG) {
|
||||
console.error('[cursor] parseProtobufFrames error parsing failed:', err);
|
||||
}
|
||||
}
|
||||
|
||||
const result = extractTextFromResponse(new Uint8Array(payload));
|
||||
|
||||
// Check for protobuf-decoded error
|
||||
if (result.error) {
|
||||
yield {
|
||||
type: 'error',
|
||||
response: new Response(
|
||||
JSON.stringify({
|
||||
error: {
|
||||
message: result.error,
|
||||
type: 'rate_limit_error',
|
||||
code: 'rate_limited',
|
||||
},
|
||||
}),
|
||||
{
|
||||
status: 429,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
}
|
||||
),
|
||||
};
|
||||
return;
|
||||
}
|
||||
|
||||
if (result.toolCall) {
|
||||
yield { type: 'toolCall', toolCall: result.toolCall };
|
||||
}
|
||||
|
||||
if (result.text) {
|
||||
yield { type: 'text', text: result.text };
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
transformProtobufToJSON(buffer: Buffer, model: string, _body: ExecutorParams['body']): Response {
|
||||
const responseId = `chatcmpl-cursor-${Date.now()}`;
|
||||
const created = Math.floor(Date.now() / 1000);
|
||||
|
||||
let offset = 0;
|
||||
let totalContent = '';
|
||||
const toolCalls: Array<{
|
||||
id: string;
|
||||
@@ -429,50 +506,13 @@ export class CursorExecutor {
|
||||
}
|
||||
>();
|
||||
|
||||
while (offset < buffer.length) {
|
||||
if (offset + 5 > buffer.length) break;
|
||||
|
||||
const flags = buffer[offset];
|
||||
const length = buffer.readUInt32BE(offset + 1);
|
||||
|
||||
if (offset + 5 + length > buffer.length) break;
|
||||
|
||||
let payload = buffer.slice(offset + 5, offset + 5 + length);
|
||||
offset += 5 + length;
|
||||
|
||||
payload = decompressPayload(payload, flags);
|
||||
|
||||
try {
|
||||
const text = payload.toString('utf-8');
|
||||
if (text.startsWith('{') && text.includes('"error"')) {
|
||||
return createErrorResponse(JSON.parse(text));
|
||||
}
|
||||
} catch (err) {
|
||||
if (process.env.CCS_DEBUG) {
|
||||
console.error('[cursor] transformProtobufToJSON error parsing failed:', err);
|
||||
}
|
||||
for (const frame of this.parseProtobufFrames(buffer)) {
|
||||
if (frame.type === 'error') {
|
||||
return frame.response;
|
||||
}
|
||||
|
||||
const result = extractTextFromResponse(new Uint8Array(payload));
|
||||
|
||||
if (result.error) {
|
||||
return new Response(
|
||||
JSON.stringify({
|
||||
error: {
|
||||
message: result.error,
|
||||
type: 'rate_limit_error',
|
||||
code: 'rate_limited',
|
||||
},
|
||||
}),
|
||||
{
|
||||
status: 429,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
if (result.toolCall) {
|
||||
const tc = result.toolCall;
|
||||
if (frame.type === 'toolCall') {
|
||||
const tc = frame.toolCall;
|
||||
|
||||
if (toolCallsMap.has(tc.id)) {
|
||||
const existing = toolCallsMap.get(tc.id);
|
||||
@@ -500,7 +540,9 @@ export class CursorExecutor {
|
||||
}
|
||||
}
|
||||
|
||||
if (result.text) totalContent += result.text;
|
||||
if (frame.type === 'text') {
|
||||
totalContent += frame.text;
|
||||
}
|
||||
}
|
||||
|
||||
// Finalize remaining tool calls
|
||||
@@ -570,7 +612,6 @@ export class CursorExecutor {
|
||||
const created = Math.floor(Date.now() / 1000);
|
||||
|
||||
const chunks: string[] = [];
|
||||
let offset = 0;
|
||||
const toolCalls: Array<{
|
||||
id: string;
|
||||
type: string;
|
||||
@@ -588,50 +629,13 @@ export class CursorExecutor {
|
||||
}
|
||||
>();
|
||||
|
||||
while (offset < buffer.length) {
|
||||
if (offset + 5 > buffer.length) break;
|
||||
|
||||
const flags = buffer[offset];
|
||||
const length = buffer.readUInt32BE(offset + 1);
|
||||
|
||||
if (offset + 5 + length > buffer.length) break;
|
||||
|
||||
let payload = buffer.slice(offset + 5, offset + 5 + length);
|
||||
offset += 5 + length;
|
||||
|
||||
payload = decompressPayload(payload, flags);
|
||||
|
||||
try {
|
||||
const text = payload.toString('utf-8');
|
||||
if (text.startsWith('{') && text.includes('"error"')) {
|
||||
return createErrorResponse(JSON.parse(text));
|
||||
}
|
||||
} catch (err) {
|
||||
if (process.env.CCS_DEBUG) {
|
||||
console.error('[cursor] transformProtobufToJSON error parsing failed:', err);
|
||||
}
|
||||
for (const frame of this.parseProtobufFrames(buffer)) {
|
||||
if (frame.type === 'error') {
|
||||
return frame.response;
|
||||
}
|
||||
|
||||
const result = extractTextFromResponse(new Uint8Array(payload));
|
||||
|
||||
if (result.error) {
|
||||
return new Response(
|
||||
JSON.stringify({
|
||||
error: {
|
||||
message: result.error,
|
||||
type: 'rate_limit_error',
|
||||
code: 'rate_limited',
|
||||
},
|
||||
}),
|
||||
{
|
||||
status: 429,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
if (result.toolCall) {
|
||||
const tc = result.toolCall;
|
||||
if (frame.type === 'toolCall') {
|
||||
const tc = frame.toolCall;
|
||||
|
||||
if (chunks.length === 0) {
|
||||
chunks.push(
|
||||
@@ -721,7 +725,7 @@ export class CursorExecutor {
|
||||
}
|
||||
}
|
||||
|
||||
if (result.text) {
|
||||
if (frame.type === 'text') {
|
||||
chunks.push(
|
||||
`data: ${JSON.stringify({
|
||||
id: responseId,
|
||||
@@ -733,8 +737,8 @@ export class CursorExecutor {
|
||||
index: 0,
|
||||
delta:
|
||||
chunks.length === 0 && toolCalls.length === 0
|
||||
? { role: 'assistant', content: result.text }
|
||||
: { content: result.text },
|
||||
? { role: 'assistant', content: frame.text }
|
||||
: { content: frame.text },
|
||||
finish_reason: null,
|
||||
},
|
||||
],
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
*/
|
||||
|
||||
import * as zlib from 'zlib';
|
||||
import { WIRE_TYPE, FIELD, type WireType } from './cursor-protobuf-schema.js';
|
||||
import { WIRE_TYPE, FIELD, COMPRESS_FLAG, type WireType } from './cursor-protobuf-schema.js';
|
||||
|
||||
/**
|
||||
* Decode a varint from buffer
|
||||
@@ -121,7 +121,11 @@ export function parseConnectRPCFrame(buffer: Buffer): {
|
||||
let payload = buffer.slice(5, 5 + length);
|
||||
|
||||
// Decompress if gzip
|
||||
if (flags === 0x01 || flags === 0x02 || flags === 0x03) {
|
||||
if (
|
||||
flags === COMPRESS_FLAG.GZIP ||
|
||||
flags === COMPRESS_FLAG.GZIP_ALT ||
|
||||
flags === COMPRESS_FLAG.GZIP_BOTH
|
||||
) {
|
||||
try {
|
||||
payload = Buffer.from(zlib.gunzipSync(payload));
|
||||
} catch (err) {
|
||||
|
||||
@@ -75,7 +75,7 @@ export const FIELD = {
|
||||
TOOL_RESULT_NAME: 2,
|
||||
TOOL_RESULT_INDEX: 3,
|
||||
TOOL_RESULT_RAW_ARGS: 5,
|
||||
TOOL_RESULT_RESULT: 8,
|
||||
TOOL_RESULT_RESULT: 8, // Reserved for future tool result parsing
|
||||
|
||||
// ===== Model =====
|
||||
MODEL_NAME: 1,
|
||||
|
||||
@@ -113,6 +113,12 @@ function convertMessages(messages: OpenAIMessage[]): CursorMessage[] {
|
||||
|
||||
result.push(msgObj);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Unknown role - skip with debug warning
|
||||
if (process.env.CCS_DEBUG) {
|
||||
console.error(`[cursor] Unknown message role: ${msg.role}, skipping`);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -297,18 +297,16 @@ describe('CursorExecutor', () => {
|
||||
expect(/^[A-Za-z0-9_-]+$/.test(prefix)).toBe(true);
|
||||
});
|
||||
|
||||
it('should generate different checksums over time', async () => {
|
||||
it('should generate valid checksums at different call times', async () => {
|
||||
const machineId = 'test-machine-id';
|
||||
const checksum1 = executor.generateChecksum(machineId);
|
||||
|
||||
// Wait longer to ensure timestamp changes (microsecond precision)
|
||||
// Wait to ensure timestamp may change (though timestamp granularity is ~16 min)
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
|
||||
const checksum2 = executor.generateChecksum(machineId);
|
||||
|
||||
// Different timestamps should produce different checksums
|
||||
// If they're still the same, it's extremely rare but acceptable
|
||||
// Just verify format is correct
|
||||
// Verify both checksums are valid (may be same due to timestamp granularity)
|
||||
expect(checksum1.endsWith(machineId)).toBe(true);
|
||||
expect(checksum2.endsWith(machineId)).toBe(true);
|
||||
});
|
||||
@@ -401,5 +399,204 @@ describe('CursorExecutor', () => {
|
||||
expect(body.choices[0].message.content).toBe(textContent);
|
||||
expect(body.choices[0].finish_reason).toBe('stop');
|
||||
});
|
||||
|
||||
it('should handle JSON error response', async () => {
|
||||
const errorJson = JSON.stringify({
|
||||
error: {
|
||||
code: 'resource_exhausted',
|
||||
message: 'Rate limit exceeded',
|
||||
},
|
||||
});
|
||||
const frame = wrapConnectRPCFrame(new TextEncoder().encode(errorJson), false);
|
||||
|
||||
const result = executor.transformProtobufToJSON(Buffer.from(frame), 'gpt-4', {
|
||||
messages: [],
|
||||
});
|
||||
|
||||
expect(result.status).toBe(429);
|
||||
const bodyText = await result.text();
|
||||
const body = JSON.parse(bodyText);
|
||||
expect(body.error.type).toBe('rate_limit_error');
|
||||
});
|
||||
});
|
||||
|
||||
describe('transformProtobufToSSE', () => {
|
||||
it('should output SSE format', async () => {
|
||||
// Create minimal protobuf response with text
|
||||
const textContent = 'Hello';
|
||||
const responseField = encodeField(FIELD.RESPONSE_TEXT, WIRE_TYPE.LEN, textContent);
|
||||
const responseMsg = encodeField(FIELD.RESPONSE, WIRE_TYPE.LEN, responseField);
|
||||
const frame = wrapConnectRPCFrame(responseMsg, false);
|
||||
|
||||
const result = executor.transformProtobufToSSE(Buffer.from(frame), 'gpt-4', {
|
||||
messages: [],
|
||||
});
|
||||
|
||||
expect(result.status).toBe(200);
|
||||
expect(result.headers.get('content-type')).toBe('text/event-stream');
|
||||
|
||||
const bodyText = await result.text();
|
||||
expect(bodyText).toContain('data: ');
|
||||
expect(bodyText).toContain('data: [DONE]');
|
||||
expect(bodyText).toContain(textContent);
|
||||
});
|
||||
|
||||
it('should handle JSON error response', async () => {
|
||||
const errorJson = JSON.stringify({
|
||||
error: {
|
||||
code: 'resource_exhausted',
|
||||
message: 'Rate limit exceeded',
|
||||
},
|
||||
});
|
||||
const frame = wrapConnectRPCFrame(new TextEncoder().encode(errorJson), false);
|
||||
|
||||
const result = executor.transformProtobufToSSE(Buffer.from(frame), 'gpt-4', {
|
||||
messages: [],
|
||||
});
|
||||
|
||||
expect(result.status).toBe(429);
|
||||
const bodyText = await result.text();
|
||||
const body = JSON.parse(bodyText);
|
||||
expect(body.error.type).toBe('rate_limit_error');
|
||||
});
|
||||
});
|
||||
|
||||
describe('decompressPayload error handling', () => {
|
||||
it('should return empty buffer on decompression failure', () => {
|
||||
// Create invalid gzip data
|
||||
const invalidGzip = Buffer.from([0x1f, 0x8b, 0x08, 0x00, 0xff, 0xff]);
|
||||
const frame = new Uint8Array(5 + invalidGzip.length);
|
||||
frame[0] = 0x01; // GZIP flag
|
||||
frame[1] = 0;
|
||||
frame[2] = 0;
|
||||
frame[3] = 0;
|
||||
frame[4] = invalidGzip.length;
|
||||
frame.set(invalidGzip, 5);
|
||||
|
||||
const result = executor.transformProtobufToJSON(Buffer.from(frame), 'gpt-4', {
|
||||
messages: [],
|
||||
});
|
||||
|
||||
// Should handle gracefully and return valid response
|
||||
expect(result.status).toBe(200);
|
||||
});
|
||||
});
|
||||
|
||||
describe('transformProtobufToSSE', () => {
|
||||
it('should output SSE format for simple text response', async () => {
|
||||
const executor = new CursorExecutor();
|
||||
|
||||
// Minimal protobuf frame with text content
|
||||
const textPayload = new Uint8Array([
|
||||
(FIELD.MSG_CONTENT << 3) | WIRE_TYPE.LEN,
|
||||
4,
|
||||
...[116, 101, 115, 116], // "test"
|
||||
]);
|
||||
|
||||
const buffer = Buffer.from(
|
||||
wrapConnectRPCFrame(textPayload, {
|
||||
compress: false,
|
||||
})
|
||||
);
|
||||
|
||||
const result = executor.transformProtobufToSSE(buffer, 'test-model', {
|
||||
messages: [],
|
||||
stream: true,
|
||||
});
|
||||
|
||||
expect(result.status).toBe(200);
|
||||
expect(result.headers.get('Content-Type')).toBe('text/event-stream');
|
||||
|
||||
const body = await result.text();
|
||||
expect(body).toContain('data: ');
|
||||
expect(body).toContain('"object":"chat.completion.chunk"');
|
||||
expect(body).toContain('data: [DONE]');
|
||||
});
|
||||
|
||||
it('should handle error responses in SSE format', () => {
|
||||
const executor = new CursorExecutor();
|
||||
|
||||
// Protobuf frame with error
|
||||
const errorPayload = new Uint8Array([
|
||||
(FIELD.MSG_CONTENT << 3) | WIRE_TYPE.LEN,
|
||||
17,
|
||||
...[101, 114, 114, 111, 114, 58, 32, 116, 101, 115, 116, 32, 101, 114, 114, 111, 114], // "error: test error"
|
||||
]);
|
||||
|
||||
const buffer = Buffer.from(
|
||||
wrapConnectRPCFrame(errorPayload, {
|
||||
compress: false,
|
||||
})
|
||||
);
|
||||
|
||||
const result = executor.transformProtobufToSSE(buffer, 'test-model', {
|
||||
messages: [],
|
||||
stream: true,
|
||||
});
|
||||
|
||||
// Error responses should still be valid
|
||||
expect(result.status).toBeGreaterThanOrEqual(200);
|
||||
});
|
||||
});
|
||||
|
||||
describe('error handling', () => {
|
||||
it('should return empty buffer on decompression failure', () => {
|
||||
const executor = new CursorExecutor();
|
||||
|
||||
// Invalid compressed payload (not actually gzipped)
|
||||
const invalidGzipPayload = new Uint8Array([1, 2, 3, 4, 5]);
|
||||
const flags = 0x01; // GZIP flag
|
||||
|
||||
// Wrap with ConnectRPC frame header (flags + length)
|
||||
const length = invalidGzipPayload.length;
|
||||
const frame = new Uint8Array(5 + length);
|
||||
frame[0] = flags;
|
||||
frame[1] = (length >> 24) & 0xff;
|
||||
frame[2] = (length >> 16) & 0xff;
|
||||
frame[3] = (length >> 8) & 0xff;
|
||||
frame[4] = length & 0xff;
|
||||
frame.set(invalidGzipPayload, 5);
|
||||
|
||||
const buffer = Buffer.from(frame);
|
||||
|
||||
// Should not crash - decompression failure returns empty buffer
|
||||
const result = executor.transformProtobufToJSON(buffer, 'test-model', {
|
||||
messages: [],
|
||||
stream: false,
|
||||
});
|
||||
|
||||
expect(result.status).toBe(200);
|
||||
});
|
||||
|
||||
it('should log unknown message roles in debug mode', () => {
|
||||
const originalDebug = process.env.CCS_DEBUG;
|
||||
process.env.CCS_DEBUG = '1';
|
||||
|
||||
const consoleSpy: string[] = [];
|
||||
const originalError = console.error;
|
||||
console.error = (...args: unknown[]) => {
|
||||
const msg = args.map((a) => String(a)).join(' ');
|
||||
consoleSpy.push(msg);
|
||||
};
|
||||
|
||||
try {
|
||||
const messages = [
|
||||
{
|
||||
role: 'unknown_role' as 'user', // Type assertion to bypass TS
|
||||
content: 'test',
|
||||
},
|
||||
];
|
||||
|
||||
// buildCursorRequest expects (model, body, stream, credentials)
|
||||
buildCursorRequest('test-model', { messages }, false, { machineId: '12345', accessToken: 'test' });
|
||||
|
||||
// Should have logged warning
|
||||
const hasWarning = consoleSpy.some((log) => log.includes('Unknown message role'));
|
||||
expect(hasWarning).toBe(true);
|
||||
} finally {
|
||||
console.error = originalError;
|
||||
process.env.CCS_DEBUG = originalDebug;
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user