diff --git a/src/cursor/cursor-executor.ts b/src/cursor/cursor-executor.ts index 0919c7f4..06b9e267 100644 --- a/src/cursor/cursor-executor.ts +++ b/src/cursor/cursor-executor.ts @@ -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, }, ], diff --git a/src/cursor/cursor-protobuf-decoder.ts b/src/cursor/cursor-protobuf-decoder.ts index 7345ab82..7a45d005 100644 --- a/src/cursor/cursor-protobuf-decoder.ts +++ b/src/cursor/cursor-protobuf-decoder.ts @@ -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) { diff --git a/src/cursor/cursor-protobuf-schema.ts b/src/cursor/cursor-protobuf-schema.ts index 4dbc32a0..255c6f0a 100644 --- a/src/cursor/cursor-protobuf-schema.ts +++ b/src/cursor/cursor-protobuf-schema.ts @@ -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, diff --git a/src/cursor/cursor-translator.ts b/src/cursor/cursor-translator.ts index cd2d7c0f..451c089c 100644 --- a/src/cursor/cursor-translator.ts +++ b/src/cursor/cursor-translator.ts @@ -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`); } } diff --git a/tests/unit/cursor/cursor-protobuf.test.ts b/tests/unit/cursor/cursor-protobuf.test.ts index ca4f5207..85bc2e3c 100644 --- a/tests/unit/cursor/cursor-protobuf.test.ts +++ b/tests/unit/cursor/cursor-protobuf.test.ts @@ -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; + } + }); }); });