diff --git a/src/cursor/cursor-executor.ts b/src/cursor/cursor-executor.ts index 3a4093b4..6153c24c 100644 --- a/src/cursor/cursor-executor.ts +++ b/src/cursor/cursor-executor.ts @@ -4,13 +4,12 @@ */ import * as crypto from 'crypto'; -import * as zlib from 'zlib'; import type { IncomingHttpHeaders } from 'http'; import { generateCursorBody, extractTextFromResponse } from './cursor-protobuf.js'; import { buildCursorRequest } from './cursor-translator.js'; import type { CursorTool, CursorCredentials } from './cursor-protobuf-schema.js'; -import { COMPRESS_FLAG } from './cursor-protobuf-schema.js'; +import { StreamingFrameParser, decompressPayload } from './cursor-stream-parser.js'; /** Executor parameters */ interface ExecutorParams { @@ -57,42 +56,6 @@ async function getHttp2() { } } -/** - * Decompress payload if needed - * NOTE: Uses synchronous gzip for single-request CLI tool. Async not warranted for small payloads. - */ -function decompressPayload(payload: Buffer, flags: number): Buffer { - // Check if payload is JSON error - if (payload.length > 10 && payload[0] === 0x7b && payload[1] === 0x22) { - try { - const text = payload.toString('utf-8'); - if (text.startsWith('{"error"')) { - return payload; - } - } catch (err) { - if (process.env.CCS_DEBUG) { - console.error('[cursor] JSON error detection failed:', err); - } - } - } - - if ( - flags === COMPRESS_FLAG.GZIP || - flags === COMPRESS_FLAG.GZIP_ALT || - flags === COMPRESS_FLAG.GZIP_BOTH - ) { - try { - return zlib.gunzipSync(payload); - } catch (err) { - if (process.env.CCS_DEBUG) { - console.error('[cursor] gzip decompression failed:', err); - } - return Buffer.alloc(0); - } - } - return payload; -} - /** * Create error response from JSON error */ @@ -342,6 +305,20 @@ export class CursorExecutor { const transformedBody = this.transformRequest(model, body, stream, credentials); try { + // Streaming requests use incremental HTTP/2 → SSE pipeline + if (stream) { + const response = await this.executeStreaming( + url, + headers, + transformedBody, + model, + body, + signal + ); + return { response, url, headers, transformedBody: body }; + } + + // Non-streaming: buffer entire response then transform to JSON const http2 = await getHttp2(); const response = http2 ? await this.makeHttp2Request(url, headers, transformedBody, signal) @@ -365,11 +342,7 @@ export class CursorExecutor { return { response: errorResponse, url, headers, transformedBody: body }; } - const transformedResponse = - stream === true - ? this.transformProtobufToSSE(response.body, model, body) - : this.transformProtobufToJSON(response.body, model, body); - + const transformedResponse = this.transformProtobufToJSON(response.body, model, body); return { response: transformedResponse, url, headers, transformedBody: body }; } catch (error) { const errorResponse = new Response( @@ -389,6 +362,270 @@ export class CursorExecutor { } } + /** + * Execute a streaming request, piping HTTP/2 data events through + * StreamingFrameParser for incremental SSE output. + * Falls back to buffered transformProtobufToSSE when http2 is unavailable. + */ + private async executeStreaming( + url: string, + headers: Record, + body: Uint8Array, + model: string, + requestBody: ExecutorParams['body'], + signal?: AbortSignal + ): Promise { + const http2 = await getHttp2(); + if (!http2) { + const response = await this.makeFetchRequest(url, headers, body, signal); + return this.transformProtobufToSSE(response.body, model, requestBody); + } + + const responseId = `chatcmpl-cursor-${Date.now()}`; + const created = Math.floor(Date.now() / 1000); + + return new Promise((resolve, reject) => { + const urlObj = new URL(url); + const client = http2.connect(`https://${urlObj.host}`); + + client.on('error', (err) => { + client.close(); + reject(err); + }); + + const req = client.request({ + ':method': 'POST', + ':path': urlObj.pathname, + ':authority': urlObj.host, + ':scheme': 'https', + ...headers, + }); + + req.on('response', (hdrs) => { + const status = Number(hdrs[':status']) || 500; + + if (status !== 200) { + const errorChunks: Buffer[] = []; + req.on('data', (c: Buffer) => errorChunks.push(c)); + req.on('end', () => { + client.close(); + const errorText = Buffer.concat(errorChunks).toString(); + resolve( + new Response( + JSON.stringify({ + error: { + message: `[${status}]: ${errorText}`, + type: 'invalid_request_error', + code: '', + }, + }), + { status, headers: { 'Content-Type': 'application/json' } } + ) + ); + }); + return; + } + + // Status 200: set up incremental streaming pipeline + const parser = new StreamingFrameParser(); + const enc = new TextEncoder(); + const toolCallsMap = new Map< + string, + { + id: string; + type: string; + function: { name: string; arguments: string }; + isLast: boolean; + index: number; + } + >(); + let chunkCount = 0; + let toolCallCount = 0; + let streamClosed = false; + + const readable = new ReadableStream({ + start(controller) { + const emitSSE = (data: string) => { + if (streamClosed) return; + controller.enqueue(enc.encode(`data: ${data}\n\n`)); + }; + + const closeStream = () => { + if (streamClosed) return; + streamClosed = true; + try { + controller.close(); + } catch { + /* already closed */ + } + client.close(); + }; + + const buildChunk = (delta: Record, finishReason: string | null) => + JSON.stringify({ + id: responseId, + object: 'chat.completion.chunk', + created, + model, + choices: [{ index: 0, delta, finish_reason: finishReason }], + }); + + req.on('data', (chunk: Buffer) => { + if (streamClosed) return; + for (const frame of parser.push(chunk)) { + if (frame.type === 'error') { + emitSSE( + JSON.stringify({ + error: { message: frame.message, type: frame.errorType, code: '' }, + }) + ); + emitSSE('[DONE]'); + closeStream(); + return; + } + + if (frame.type === 'toolCall') { + const tc = frame.toolCall; + + // Emit role chunk on first content + if (chunkCount === 0) { + emitSSE(buildChunk({ role: 'assistant', content: '' }, null)); + chunkCount++; + } + + if (toolCallsMap.has(tc.id)) { + const existing = toolCallsMap.get(tc.id); + if (!existing) continue; + existing.function.arguments += tc.function.arguments; + existing.isLast = tc.isLast; + if (tc.function.arguments) { + emitSSE( + buildChunk( + { + tool_calls: [ + { + index: existing.index, + id: tc.id, + type: 'function', + function: { + name: tc.function.name, + arguments: tc.function.arguments, + }, + }, + ], + }, + null + ) + ); + chunkCount++; + } + } else { + const idx = toolCallCount++; + toolCallsMap.set(tc.id, { ...tc, index: idx }); + emitSSE( + buildChunk( + { + tool_calls: [ + { + index: idx, + id: tc.id, + type: 'function', + function: { + name: tc.function.name, + arguments: tc.function.arguments, + }, + }, + ], + }, + null + ) + ); + chunkCount++; + } + } + + if (frame.type === 'text') { + const delta = + chunkCount === 0 && toolCallCount === 0 + ? { role: 'assistant', content: frame.text } + : { content: frame.text }; + emitSSE(buildChunk(delta, null)); + chunkCount++; + } + } + }); + + req.on('end', () => { + if (streamClosed) return; + if (chunkCount === 0 && toolCallCount === 0) { + emitSSE(buildChunk({ role: 'assistant', content: '' }, null)); + } + emitSSE( + JSON.stringify({ + id: responseId, + object: 'chat.completion.chunk', + created, + model, + choices: [ + { + index: 0, + delta: {}, + finish_reason: toolCallCount > 0 ? 'tool_calls' : 'stop', + }, + ], + usage: { prompt_tokens: 0, completion_tokens: 0, total_tokens: 0 }, + }) + ); + emitSSE('[DONE]'); + closeStream(); + }); + + req.on('error', (err) => { + if (!streamClosed) { + try { + controller.error(err); + } catch { + /* already closed */ + } + } + client.close(); + }); + + if (signal) { + const onAbort = () => { + req.close(); + closeStream(); + }; + signal.addEventListener('abort', onAbort, { once: true }); + const cleanup = () => signal.removeEventListener('abort', onAbort); + req.on('end', cleanup); + req.on('error', cleanup); + } + }, + }); + + resolve( + new Response(readable, { + status: 200, + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + }, + }) + ); + }); + + req.on('error', (err) => { + client.close(); + reject(err); + }); + + req.write(body); + req.end(); + }); + } + /** * Parse protobuf buffer into frames and extract text/toolcalls. * Shared logic between JSON and SSE transformers. @@ -590,11 +827,8 @@ export class CursorExecutor { }); } + /** @deprecated Use executeStreaming() for real-time SSE. Retained for fetch fallback. */ transformProtobufToSSE(buffer: Buffer, model: string, _body: ExecutorParams['body']): Response { - // TODO(#531): Implement true streaming — currently buffers entire response before transforming. - // This should pipe HTTP/2 data events through a TransformStream for incremental SSE output. - // See: https://github.com/kaitranntt/ccs/issues/531 - // NOTE: Chunk boundary splits may emit duplicate SSE messages if a frame spans multiple chunks. const responseId = `chatcmpl-cursor-${Date.now()}`; const created = Math.floor(Date.now() / 1000); diff --git a/src/cursor/cursor-protobuf-decoder.ts b/src/cursor/cursor-protobuf-decoder.ts index 7a45d005..1dde8dd6 100644 --- a/src/cursor/cursor-protobuf-decoder.ts +++ b/src/cursor/cursor-protobuf-decoder.ts @@ -160,8 +160,8 @@ function extractToolCall(toolCallData: Uint8Array): { let isLast = false; // Extract tool call ID - if (toolCall.has(FIELD.TOOL_ID)) { - const idField = toolCall.get(FIELD.TOOL_ID); + if (toolCall.has(FIELD.ToolCall.ID)) { + const idField = toolCall.get(FIELD.ToolCall.ID); if (idField && idField[0]) { const fullId = new TextDecoder().decode(idField[0].value as Uint8Array); toolCallId = fullId.split('\n')[0]; // Take first line @@ -169,44 +169,44 @@ function extractToolCall(toolCallData: Uint8Array): { } // Extract tool name - if (toolCall.has(FIELD.TOOL_NAME)) { - const nameField = toolCall.get(FIELD.TOOL_NAME); + if (toolCall.has(FIELD.ToolCall.NAME)) { + const nameField = toolCall.get(FIELD.ToolCall.NAME); if (nameField && nameField[0]) { toolName = new TextDecoder().decode(nameField[0].value as Uint8Array); } } // Extract is_last flag - if (toolCall.has(FIELD.TOOL_IS_LAST)) { - const lastField = toolCall.get(FIELD.TOOL_IS_LAST); + if (toolCall.has(FIELD.ToolCall.IS_LAST)) { + const lastField = toolCall.get(FIELD.ToolCall.IS_LAST); if (lastField && lastField[0]) { isLast = (lastField[0].value as number) !== 0; } } // Extract MCP params - nested real tool info - if (toolCall.has(FIELD.TOOL_MCP_PARAMS)) { + if (toolCall.has(FIELD.ToolCall.MCP_PARAMS)) { try { - const mcpField = toolCall.get(FIELD.TOOL_MCP_PARAMS); + const mcpField = toolCall.get(FIELD.ToolCall.MCP_PARAMS); if (!mcpField || !mcpField[0]) return null; const mcpParams = decodeMessage(mcpField[0].value as Uint8Array); - if (mcpParams.has(FIELD.MCP_TOOLS_LIST)) { - const toolsList = mcpParams.get(FIELD.MCP_TOOLS_LIST); + if (mcpParams.has(FIELD.McpParams.TOOLS_LIST)) { + const toolsList = mcpParams.get(FIELD.McpParams.TOOLS_LIST); if (!toolsList || !toolsList[0]) return null; const tool = decodeMessage(toolsList[0].value as Uint8Array); - if (tool.has(FIELD.MCP_NESTED_NAME)) { - const nestedName = tool.get(FIELD.MCP_NESTED_NAME); + if (tool.has(FIELD.McpNested.NAME)) { + const nestedName = tool.get(FIELD.McpNested.NAME); if (nestedName && nestedName[0]) { toolName = new TextDecoder().decode(nestedName[0].value as Uint8Array); } } - if (tool.has(FIELD.MCP_NESTED_PARAMS)) { - const nestedParams = tool.get(FIELD.MCP_NESTED_PARAMS); + if (tool.has(FIELD.McpNested.PARAMS)) { + const nestedParams = tool.get(FIELD.McpNested.PARAMS); if (nestedParams && nestedParams[0]) { rawArgs = new TextDecoder().decode(nestedParams[0].value as Uint8Array); } @@ -221,8 +221,8 @@ function extractToolCall(toolCallData: Uint8Array): { } // Fallback to raw_args - if (!rawArgs && toolCall.has(FIELD.TOOL_RAW_ARGS)) { - const rawArgsField = toolCall.get(FIELD.TOOL_RAW_ARGS); + if (!rawArgs && toolCall.has(FIELD.ToolCall.RAW_ARGS)) { + const rawArgsField = toolCall.get(FIELD.ToolCall.RAW_ARGS); if (rawArgsField && rawArgsField[0]) { rawArgs = new TextDecoder().decode(rawArgsField[0].value as Uint8Array); } @@ -255,21 +255,21 @@ function extractTextAndThinking(responseData: Uint8Array): { let thinking: string | null = null; // Extract text - if (nested.has(FIELD.RESPONSE_TEXT)) { - const textField = nested.get(FIELD.RESPONSE_TEXT); + if (nested.has(FIELD.ChatResponse.TEXT)) { + const textField = nested.get(FIELD.ChatResponse.TEXT); if (textField && textField[0]) { text = new TextDecoder().decode(textField[0].value as Uint8Array); } } // Extract thinking - if (nested.has(FIELD.THINKING)) { + if (nested.has(FIELD.ChatResponse.THINKING)) { try { - const thinkingField = nested.get(FIELD.THINKING); + const thinkingField = nested.get(FIELD.ChatResponse.THINKING); if (thinkingField && thinkingField[0]) { const thinkingMsg = decodeMessage(thinkingField[0].value as Uint8Array); - if (thinkingMsg.has(FIELD.THINKING_TEXT)) { - const thinkingTextField = thinkingMsg.get(FIELD.THINKING_TEXT); + if (thinkingMsg.has(FIELD.Thinking.TEXT)) { + const thinkingTextField = thinkingMsg.get(FIELD.Thinking.TEXT); if (thinkingTextField && thinkingTextField[0]) { thinking = new TextDecoder().decode(thinkingTextField[0].value as Uint8Array); } @@ -304,8 +304,8 @@ export function extractTextFromResponse(payload: Uint8Array): { const fields = decodeMessage(payload); // Field 1: ClientSideToolV2Call - if (fields.has(FIELD.TOOL_CALL)) { - const toolCallField = fields.get(FIELD.TOOL_CALL); + if (fields.has(FIELD.Response.TOOL_CALL)) { + const toolCallField = fields.get(FIELD.Response.TOOL_CALL); if (toolCallField && toolCallField[0]) { const toolCall = extractToolCall(toolCallField[0].value as Uint8Array); if (toolCall) { @@ -315,8 +315,8 @@ export function extractTextFromResponse(payload: Uint8Array): { } // Field 2: StreamUnifiedChatResponse - if (fields.has(FIELD.RESPONSE)) { - const responseField = fields.get(FIELD.RESPONSE); + if (fields.has(FIELD.Response.RESPONSE)) { + const responseField = fields.get(FIELD.Response.RESPONSE); if (responseField && responseField[0]) { const { text, thinking } = extractTextAndThinking(responseField[0].value as Uint8Array); diff --git a/src/cursor/cursor-protobuf-encoder.ts b/src/cursor/cursor-protobuf-encoder.ts index eb1133c9..f0dffb02 100644 --- a/src/cursor/cursor-protobuf-encoder.ts +++ b/src/cursor/cursor-protobuf-encoder.ts @@ -84,10 +84,10 @@ export function encodeToolResult(toolResult: CursorToolResult): Uint8Array { const rawArgs = toolResult.raw_args || '{}'; return concatArrays( - encodeField(FIELD.TOOL_RESULT_CALL_ID, WIRE_TYPE.LEN, toolCallId), - encodeField(FIELD.TOOL_RESULT_NAME, WIRE_TYPE.LEN, toolName), - encodeField(FIELD.TOOL_RESULT_INDEX, WIRE_TYPE.VARINT, toolIndex), - encodeField(FIELD.TOOL_RESULT_RAW_ARGS, WIRE_TYPE.LEN, rawArgs) + encodeField(FIELD.ToolResult.CALL_ID, WIRE_TYPE.LEN, toolCallId), + encodeField(FIELD.ToolResult.NAME, WIRE_TYPE.LEN, toolName), + encodeField(FIELD.ToolResult.INDEX, WIRE_TYPE.VARINT, toolIndex), + encodeField(FIELD.ToolResult.RAW_ARGS, WIRE_TYPE.LEN, rawArgs) ); } @@ -103,22 +103,22 @@ export function encodeMessage( toolResults: CursorToolResult[] ): Uint8Array { return concatArrays( - encodeField(FIELD.MSG_CONTENT, WIRE_TYPE.LEN, content), - encodeField(FIELD.MSG_ROLE, WIRE_TYPE.VARINT, role), - encodeField(FIELD.MSG_ID, WIRE_TYPE.LEN, messageId), + encodeField(FIELD.Message.CONTENT, WIRE_TYPE.LEN, content), + encodeField(FIELD.Message.ROLE, WIRE_TYPE.VARINT, role), + encodeField(FIELD.Message.ID, WIRE_TYPE.LEN, messageId), ...(toolResults.length > 0 ? toolResults.map((tr) => - encodeField(FIELD.MSG_TOOL_RESULTS, WIRE_TYPE.LEN, encodeToolResult(tr)) + encodeField(FIELD.Message.TOOL_RESULTS, WIRE_TYPE.LEN, encodeToolResult(tr)) ) : []), - encodeField(FIELD.MSG_IS_AGENTIC, WIRE_TYPE.VARINT, hasTools ? 1 : 0), + encodeField(FIELD.Message.IS_AGENTIC, WIRE_TYPE.VARINT, hasTools ? 1 : 0), encodeField( - FIELD.MSG_UNIFIED_MODE, + FIELD.Message.UNIFIED_MODE, WIRE_TYPE.VARINT, hasTools ? UNIFIED_MODE.AGENT : UNIFIED_MODE.CHAT ), ...(isLast && hasTools - ? [encodeField(FIELD.MSG_SUPPORTED_TOOLS, WIRE_TYPE.LEN, encodeVarint(1))] + ? [encodeField(FIELD.Message.SUPPORTED_TOOLS, WIRE_TYPE.LEN, encodeVarint(1))] : []) ); } @@ -127,7 +127,7 @@ export function encodeMessage( * Encode instruction text */ export function encodeInstruction(text: string): Uint8Array { - return text ? encodeField(FIELD.INSTRUCTION_TEXT, WIRE_TYPE.LEN, text) : new Uint8Array(0); + return text ? encodeField(FIELD.Instruction.TEXT, WIRE_TYPE.LEN, text) : new Uint8Array(0); } /** @@ -135,8 +135,8 @@ export function encodeInstruction(text: string): Uint8Array { */ export function encodeModel(modelName: string): Uint8Array { return concatArrays( - encodeField(FIELD.MODEL_NAME, WIRE_TYPE.LEN, modelName), - encodeField(FIELD.MODEL_EMPTY, WIRE_TYPE.LEN, new Uint8Array(0)) + encodeField(FIELD.Model.NAME, WIRE_TYPE.LEN, modelName), + encodeField(FIELD.Model.EMPTY, WIRE_TYPE.LEN, new Uint8Array(0)) ); } @@ -145,16 +145,16 @@ export function encodeModel(modelName: string): Uint8Array { */ export function encodeCursorSetting(): Uint8Array { const unknown6 = concatArrays( - encodeField(FIELD.SETTING6_FIELD_1, WIRE_TYPE.LEN, new Uint8Array(0)), - encodeField(FIELD.SETTING6_FIELD_2, WIRE_TYPE.LEN, new Uint8Array(0)) + encodeField(FIELD.Setting.U6_FIELD_1, WIRE_TYPE.LEN, new Uint8Array(0)), + encodeField(FIELD.Setting.U6_FIELD_2, WIRE_TYPE.LEN, new Uint8Array(0)) ); return concatArrays( - encodeField(FIELD.SETTING_PATH, WIRE_TYPE.LEN, 'cursor\\aisettings'), - encodeField(FIELD.SETTING_UNKNOWN_3, WIRE_TYPE.LEN, new Uint8Array(0)), - encodeField(FIELD.SETTING_UNKNOWN_6, WIRE_TYPE.LEN, unknown6), - encodeField(FIELD.SETTING_UNKNOWN_8, WIRE_TYPE.VARINT, 1), - encodeField(FIELD.SETTING_UNKNOWN_9, WIRE_TYPE.VARINT, 1) + encodeField(FIELD.Setting.PATH, WIRE_TYPE.LEN, 'cursor\\aisettings'), + encodeField(FIELD.Setting.UNKNOWN_3, WIRE_TYPE.LEN, new Uint8Array(0)), + encodeField(FIELD.Setting.UNKNOWN_6, WIRE_TYPE.LEN, unknown6), + encodeField(FIELD.Setting.UNKNOWN_8, WIRE_TYPE.VARINT, 1), + encodeField(FIELD.Setting.UNKNOWN_9, WIRE_TYPE.VARINT, 1) ); } @@ -163,11 +163,11 @@ export function encodeCursorSetting(): Uint8Array { */ export function encodeMetadata(): Uint8Array { return concatArrays( - encodeField(FIELD.META_PLATFORM, WIRE_TYPE.LEN, process.platform || 'linux'), - encodeField(FIELD.META_ARCH, WIRE_TYPE.LEN, process.arch || 'x64'), - encodeField(FIELD.META_VERSION, WIRE_TYPE.LEN, process.version || 'v20.0.0'), - encodeField(FIELD.META_CWD, WIRE_TYPE.LEN, process.cwd() || '/'), - encodeField(FIELD.META_TIMESTAMP, WIRE_TYPE.LEN, new Date().toISOString()) + encodeField(FIELD.Metadata.PLATFORM, WIRE_TYPE.LEN, process.platform || 'linux'), + encodeField(FIELD.Metadata.ARCH, WIRE_TYPE.LEN, process.arch || 'x64'), + encodeField(FIELD.Metadata.VERSION, WIRE_TYPE.LEN, process.version || 'v20.0.0'), + encodeField(FIELD.Metadata.CWD, WIRE_TYPE.LEN, process.cwd() || '/'), + encodeField(FIELD.Metadata.TIMESTAMP, WIRE_TYPE.LEN, new Date().toISOString()) ); } @@ -176,9 +176,9 @@ export function encodeMetadata(): Uint8Array { */ export function encodeMessageId(messageId: string, role: RoleType, summaryId?: string): Uint8Array { return concatArrays( - encodeField(FIELD.MSGID_ID, WIRE_TYPE.LEN, messageId), - ...(summaryId ? [encodeField(FIELD.MSGID_SUMMARY, WIRE_TYPE.LEN, summaryId)] : []), - encodeField(FIELD.MSGID_ROLE, WIRE_TYPE.VARINT, role) + encodeField(FIELD.MessageId.ID, WIRE_TYPE.LEN, messageId), + ...(summaryId ? [encodeField(FIELD.MessageId.SUMMARY, WIRE_TYPE.LEN, summaryId)] : []), + encodeField(FIELD.MessageId.ROLE, WIRE_TYPE.VARINT, role) ); } @@ -191,12 +191,12 @@ export function encodeMcpTool(tool: CursorTool): Uint8Array { const inputSchema = tool.function?.parameters || tool.input_schema || {}; return concatArrays( - ...(toolName ? [encodeField(FIELD.MCP_TOOL_NAME, WIRE_TYPE.LEN, toolName)] : []), - ...(toolDesc ? [encodeField(FIELD.MCP_TOOL_DESC, WIRE_TYPE.LEN, toolDesc)] : []), + ...(toolName ? [encodeField(FIELD.McpTool.NAME, WIRE_TYPE.LEN, toolName)] : []), + ...(toolDesc ? [encodeField(FIELD.McpTool.DESC, WIRE_TYPE.LEN, toolDesc)] : []), ...(Object.keys(inputSchema).length > 0 - ? [encodeField(FIELD.MCP_TOOL_PARAMS, WIRE_TYPE.LEN, JSON.stringify(inputSchema))] + ? [encodeField(FIELD.McpTool.PARAMS, WIRE_TYPE.LEN, JSON.stringify(inputSchema))] : []), - encodeField(FIELD.MCP_TOOL_SERVER, WIRE_TYPE.LEN, 'custom') + encodeField(FIELD.McpTool.SERVER, WIRE_TYPE.LEN, 'custom') ); } diff --git a/src/cursor/cursor-protobuf-schema.ts b/src/cursor/cursor-protobuf-schema.ts index 255c6f0a..01a78fdf 100644 --- a/src/cursor/cursor-protobuf-schema.ts +++ b/src/cursor/cursor-protobuf-schema.ts @@ -30,113 +30,102 @@ export const THINKING_LEVEL = { HIGH: 2, } as const; -/** Field numbers for all protobuf messages */ +/** Field numbers namespaced by protobuf message type */ export const FIELD = { - // ===== StreamUnifiedChatRequestWithTools (top level) ===== - REQUEST: 1, + /** StreamUnifiedChatRequestWithTools (top level) */ + Request: { REQUEST: 1 }, - // ===== StreamUnifiedChatRequest ===== - MESSAGES: 1, - UNKNOWN_2: 2, - INSTRUCTION: 3, - UNKNOWN_4: 4, - MODEL: 5, - WEB_TOOL: 8, - UNKNOWN_13: 13, - CURSOR_SETTING: 15, - UNKNOWN_19: 19, - CONVERSATION_ID: 23, - METADATA: 26, - IS_AGENTIC: 27, - SUPPORTED_TOOLS: 29, - MESSAGE_IDS: 30, - MCP_TOOLS: 34, - LARGE_CONTEXT: 35, - UNKNOWN_38: 38, - UNIFIED_MODE: 46, - UNKNOWN_47: 47, - SHOULD_DISABLE_TOOLS: 48, - THINKING_LEVEL: 49, - UNKNOWN_51: 51, - UNKNOWN_53: 53, - UNIFIED_MODE_NAME: 54, + /** StreamUnifiedChatRequest */ + Chat: { + MESSAGES: 1, + UNKNOWN_2: 2, + INSTRUCTION: 3, + UNKNOWN_4: 4, + MODEL: 5, + WEB_TOOL: 8, + UNKNOWN_13: 13, + CURSOR_SETTING: 15, + UNKNOWN_19: 19, + CONVERSATION_ID: 23, + METADATA: 26, + IS_AGENTIC: 27, + SUPPORTED_TOOLS: 29, + MESSAGE_IDS: 30, + MCP_TOOLS: 34, + LARGE_CONTEXT: 35, + UNKNOWN_38: 38, + UNIFIED_MODE: 46, + UNKNOWN_47: 47, + SHOULD_DISABLE_TOOLS: 48, + THINKING_LEVEL: 49, + UNKNOWN_51: 51, + UNKNOWN_53: 53, + UNIFIED_MODE_NAME: 54, + }, - // ===== ConversationMessage ===== - MSG_CONTENT: 1, - MSG_ROLE: 2, - MSG_ID: 13, - MSG_TOOL_RESULTS: 18, - MSG_IS_AGENTIC: 29, - MSG_UNIFIED_MODE: 47, - MSG_SUPPORTED_TOOLS: 51, + /** ConversationMessage */ + Message: { + CONTENT: 1, + ROLE: 2, + ID: 13, + TOOL_RESULTS: 18, + IS_AGENTIC: 29, + UNIFIED_MODE: 47, + SUPPORTED_TOOLS: 51, + }, - // ===== ConversationMessage.ToolResult ===== - TOOL_RESULT_CALL_ID: 1, - TOOL_RESULT_NAME: 2, - TOOL_RESULT_INDEX: 3, - TOOL_RESULT_RAW_ARGS: 5, - TOOL_RESULT_RESULT: 8, // Reserved for future tool result parsing + /** ConversationMessage.ToolResult */ + ToolResult: { + CALL_ID: 1, + NAME: 2, + INDEX: 3, + RAW_ARGS: 5, + RESULT: 8, + }, - // ===== Model ===== - MODEL_NAME: 1, - MODEL_EMPTY: 4, + /** Model */ + Model: { NAME: 1, EMPTY: 4 }, - // ===== Instruction ===== - INSTRUCTION_TEXT: 1, + /** Instruction */ + Instruction: { TEXT: 1 }, - // ===== CursorSetting ===== - SETTING_PATH: 1, - SETTING_UNKNOWN_3: 3, - SETTING_UNKNOWN_6: 6, - SETTING_UNKNOWN_8: 8, - SETTING_UNKNOWN_9: 9, + /** CursorSetting (U6_FIELD_* are sub-fields of Unknown6 nested message) */ + Setting: { + PATH: 1, + UNKNOWN_3: 3, + UNKNOWN_6: 6, + UNKNOWN_8: 8, + UNKNOWN_9: 9, + U6_FIELD_1: 1, + U6_FIELD_2: 2, + }, - // ===== CursorSetting.Unknown6 ===== - SETTING6_FIELD_1: 1, - SETTING6_FIELD_2: 2, + /** Metadata */ + Metadata: { PLATFORM: 1, ARCH: 2, VERSION: 3, CWD: 4, TIMESTAMP: 5 }, - // ===== Metadata ===== - META_PLATFORM: 1, - META_ARCH: 2, - META_VERSION: 3, - META_CWD: 4, - META_TIMESTAMP: 5, + /** MessageId */ + MessageId: { ID: 1, SUMMARY: 2, ROLE: 3 }, - // ===== MessageId ===== - MSGID_ID: 1, - MSGID_SUMMARY: 2, - MSGID_ROLE: 3, + /** MCPTool */ + McpTool: { NAME: 1, DESC: 2, PARAMS: 3, SERVER: 4 }, - // ===== MCPTool ===== - MCP_TOOL_NAME: 1, - MCP_TOOL_DESC: 2, - MCP_TOOL_PARAMS: 3, - MCP_TOOL_SERVER: 4, + /** StreamUnifiedChatResponseWithTools (response top-level) */ + Response: { TOOL_CALL: 1, RESPONSE: 2 }, - // ===== StreamUnifiedChatResponseWithTools (response) ===== - TOOL_CALL: 1, - RESPONSE: 2, + /** ClientSideToolV2Call */ + ToolCall: { ID: 3, NAME: 9, RAW_ARGS: 10, IS_LAST: 11, MCP_PARAMS: 27 }, - // ===== ClientSideToolV2Call ===== - TOOL_ID: 3, - TOOL_NAME: 9, - TOOL_RAW_ARGS: 10, - TOOL_IS_LAST: 11, - TOOL_MCP_PARAMS: 27, + /** MCPParams */ + McpParams: { TOOLS_LIST: 1 }, - // ===== MCPParams ===== - MCP_TOOLS_LIST: 1, + /** MCPParams.Tool (nested) */ + McpNested: { NAME: 1, PARAMS: 3 }, - // ===== MCPParams.Tool (nested) ===== - MCP_NESTED_NAME: 1, - MCP_NESTED_PARAMS: 3, + /** StreamUnifiedChatResponse */ + ChatResponse: { TEXT: 1, THINKING: 25 }, - // ===== StreamUnifiedChatResponse ===== - RESPONSE_TEXT: 1, - THINKING: 25, - - // ===== Thinking ===== - THINKING_TEXT: 1, + /** Thinking */ + Thinking: { TEXT: 1 }, } as const; /** Type definitions */ @@ -144,7 +133,6 @@ export type WireType = (typeof WIRE_TYPE)[keyof typeof WIRE_TYPE]; export type RoleType = (typeof ROLE)[keyof typeof ROLE]; export type UnifiedModeType = (typeof UNIFIED_MODE)[keyof typeof UNIFIED_MODE]; export type ThinkingLevelType = (typeof THINKING_LEVEL)[keyof typeof THINKING_LEVEL]; -export type FieldNumber = (typeof FIELD)[keyof typeof FIELD]; /** Cursor credentials structure */ export interface CursorCredentials { diff --git a/src/cursor/cursor-protobuf.ts b/src/cursor/cursor-protobuf.ts index 6e0ee06a..7579cc42 100644 --- a/src/cursor/cursor-protobuf.ts +++ b/src/cursor/cursor-protobuf.ts @@ -82,55 +82,55 @@ export function encodeRequest( // Build arrays for messages and tools const messageFields = formattedMessages.map((fm) => encodeField( - FIELD.MESSAGES, + FIELD.Chat.MESSAGES, WIRE_TYPE.LEN, encodeMessage(fm.content, fm.role, fm.messageId, fm.isLast, fm.hasTools, fm.toolResults) ) ); const messageIdFields = messageIds.map((mid) => - encodeField(FIELD.MESSAGE_IDS, WIRE_TYPE.LEN, encodeMessageId(mid.messageId, mid.role)) + encodeField(FIELD.Chat.MESSAGE_IDS, WIRE_TYPE.LEN, encodeMessageId(mid.messageId, mid.role)) ); const toolFields = tools?.length > 0 - ? tools.map((tool) => encodeField(FIELD.MCP_TOOLS, WIRE_TYPE.LEN, encodeMcpTool(tool))) + ? tools.map((tool) => encodeField(FIELD.Chat.MCP_TOOLS, WIRE_TYPE.LEN, encodeMcpTool(tool))) : []; const supportedToolsField = isAgentic - ? [encodeField(FIELD.SUPPORTED_TOOLS, WIRE_TYPE.LEN, encodeVarint(1))] + ? [encodeField(FIELD.Chat.SUPPORTED_TOOLS, WIRE_TYPE.LEN, encodeVarint(1))] : []; // Concatenate all parts const parts: Uint8Array[] = [ ...messageFields, - encodeField(FIELD.UNKNOWN_2, WIRE_TYPE.VARINT, 1), - encodeField(FIELD.INSTRUCTION, WIRE_TYPE.LEN, encodeInstruction('')), - encodeField(FIELD.UNKNOWN_4, WIRE_TYPE.VARINT, 1), - encodeField(FIELD.MODEL, WIRE_TYPE.LEN, encodeModel(modelName)), - encodeField(FIELD.WEB_TOOL, WIRE_TYPE.LEN, ''), - encodeField(FIELD.UNKNOWN_13, WIRE_TYPE.VARINT, 1), - encodeField(FIELD.CURSOR_SETTING, WIRE_TYPE.LEN, encodeCursorSetting()), - encodeField(FIELD.UNKNOWN_19, WIRE_TYPE.VARINT, 1), - encodeField(FIELD.CONVERSATION_ID, WIRE_TYPE.LEN, randomUUID()), - encodeField(FIELD.METADATA, WIRE_TYPE.LEN, encodeMetadata()), - encodeField(FIELD.IS_AGENTIC, WIRE_TYPE.VARINT, isAgentic ? 1 : 0), + encodeField(FIELD.Chat.UNKNOWN_2, WIRE_TYPE.VARINT, 1), + encodeField(FIELD.Chat.INSTRUCTION, WIRE_TYPE.LEN, encodeInstruction('')), + encodeField(FIELD.Chat.UNKNOWN_4, WIRE_TYPE.VARINT, 1), + encodeField(FIELD.Chat.MODEL, WIRE_TYPE.LEN, encodeModel(modelName)), + encodeField(FIELD.Chat.WEB_TOOL, WIRE_TYPE.LEN, ''), + encodeField(FIELD.Chat.UNKNOWN_13, WIRE_TYPE.VARINT, 1), + encodeField(FIELD.Chat.CURSOR_SETTING, WIRE_TYPE.LEN, encodeCursorSetting()), + encodeField(FIELD.Chat.UNKNOWN_19, WIRE_TYPE.VARINT, 1), + encodeField(FIELD.Chat.CONVERSATION_ID, WIRE_TYPE.LEN, randomUUID()), + encodeField(FIELD.Chat.METADATA, WIRE_TYPE.LEN, encodeMetadata()), + encodeField(FIELD.Chat.IS_AGENTIC, WIRE_TYPE.VARINT, isAgentic ? 1 : 0), ...supportedToolsField, ...messageIdFields, ...toolFields, - encodeField(FIELD.LARGE_CONTEXT, WIRE_TYPE.VARINT, 0), - encodeField(FIELD.UNKNOWN_38, WIRE_TYPE.VARINT, 0), + encodeField(FIELD.Chat.LARGE_CONTEXT, WIRE_TYPE.VARINT, 0), + encodeField(FIELD.Chat.UNKNOWN_38, WIRE_TYPE.VARINT, 0), encodeField( - FIELD.UNIFIED_MODE, + FIELD.Chat.UNIFIED_MODE, WIRE_TYPE.VARINT, isAgentic ? UNIFIED_MODE.AGENT : UNIFIED_MODE.CHAT ), - encodeField(FIELD.UNKNOWN_47, WIRE_TYPE.LEN, ''), - encodeField(FIELD.SHOULD_DISABLE_TOOLS, WIRE_TYPE.VARINT, isAgentic ? 0 : 1), - encodeField(FIELD.THINKING_LEVEL, WIRE_TYPE.VARINT, thinkingLevel), - encodeField(FIELD.UNKNOWN_51, WIRE_TYPE.VARINT, 0), - encodeField(FIELD.UNKNOWN_53, WIRE_TYPE.VARINT, 1), - encodeField(FIELD.UNIFIED_MODE_NAME, WIRE_TYPE.LEN, isAgentic ? 'Agent' : 'Ask'), + encodeField(FIELD.Chat.UNKNOWN_47, WIRE_TYPE.LEN, ''), + encodeField(FIELD.Chat.SHOULD_DISABLE_TOOLS, WIRE_TYPE.VARINT, isAgentic ? 0 : 1), + encodeField(FIELD.Chat.THINKING_LEVEL, WIRE_TYPE.VARINT, thinkingLevel), + encodeField(FIELD.Chat.UNKNOWN_51, WIRE_TYPE.VARINT, 0), + encodeField(FIELD.Chat.UNKNOWN_53, WIRE_TYPE.VARINT, 1), + encodeField(FIELD.Chat.UNIFIED_MODE_NAME, WIRE_TYPE.LEN, isAgentic ? 'Agent' : 'Ask'), ]; return concatArrays(...parts); @@ -146,7 +146,7 @@ export function buildChatRequest( reasoningEffort: string | null = null ): Uint8Array { return encodeField( - FIELD.REQUEST, + FIELD.Request.REQUEST, WIRE_TYPE.LEN, encodeRequest(messages, modelName, tools, reasoningEffort) ); diff --git a/src/cursor/cursor-stream-parser.ts b/src/cursor/cursor-stream-parser.ts new file mode 100644 index 00000000..7fa06091 --- /dev/null +++ b/src/cursor/cursor-stream-parser.ts @@ -0,0 +1,136 @@ +/** + * Cursor Streaming Frame Parser + * Incrementally parses ConnectRPC frames from arbitrary chunk boundaries + */ + +import * as zlib from 'zlib'; +import { COMPRESS_FLAG } from './cursor-protobuf-schema.js'; +import { extractTextFromResponse } from './cursor-protobuf-decoder.js'; + +/** Frame parsing result types */ +export type FrameResult = + | { type: 'error'; message: string; status: number; errorType: string } + | { type: 'text'; text: string } + | { type: 'thinking'; text: string } + | { + type: 'toolCall'; + toolCall: { + id: string; + type: string; + function: { name: string; arguments: string }; + isLast: boolean; + }; + }; + +/** + * Decompress payload if gzip-compressed. + * Skips decompression for JSON error payloads. + * NOTE: Uses synchronous gzip for single-request CLI tool. Async not warranted for small payloads. + */ +export function decompressPayload(payload: Buffer, flags: number): Buffer { + if (payload.length > 10 && payload[0] === 0x7b && payload[1] === 0x22) { + try { + const text = payload.toString('utf-8'); + if (text.startsWith('{"error"')) return payload; + } catch { + // Not JSON, continue + } + } + + if ( + flags === COMPRESS_FLAG.GZIP || + flags === COMPRESS_FLAG.GZIP_ALT || + flags === COMPRESS_FLAG.GZIP_BOTH + ) { + try { + return zlib.gunzipSync(payload); + } catch (err) { + if (process.env.CCS_DEBUG) { + console.error('[cursor] gzip decompression failed:', err); + } + return Buffer.alloc(0); + } + } + return payload; +} + +/** + * Incrementally parses ConnectRPC frames from arbitrary chunk boundaries. + * + * Usage: + * const parser = new StreamingFrameParser(); + * req.on('data', (chunk) => { + * for (const frame of parser.push(chunk)) { handle(frame); } + * }); + */ +export class StreamingFrameParser { + private buffer: Buffer = Buffer.alloc(0); + + /** Feed a new chunk. Returns zero or more parsed frames. */ + push(chunk: Buffer): FrameResult[] { + this.buffer = Buffer.concat([this.buffer, chunk]); + const results: FrameResult[] = []; + + while (this.buffer.length >= 5) { + const length = this.buffer.readUInt32BE(1); + const frameSize = 5 + length; + if (this.buffer.length < frameSize) break; + + const flags = this.buffer[0]; + let payload = this.buffer.slice(5, frameSize); + this.buffer = this.buffer.slice(frameSize); + + payload = decompressPayload(payload, flags); + + // Check for JSON error + try { + const text = payload.toString('utf-8'); + if (text.startsWith('{') && text.includes('"error"')) { + const json = JSON.parse(text); + const msg = + json?.error?.details?.[0]?.debug?.details?.title || + json?.error?.details?.[0]?.debug?.details?.detail || + json?.error?.message || + 'API Error'; + const isRateLimit = json?.error?.code === 'resource_exhausted'; + results.push({ + type: 'error', + message: msg, + status: isRateLimit ? 429 : 400, + errorType: isRateLimit ? 'rate_limit_error' : 'api_error', + }); + return results; + } + } catch { + // Not JSON, continue to protobuf parsing + } + + const result = extractTextFromResponse(new Uint8Array(payload)); + + if (result.error) { + const errorLower = result.error.toLowerCase(); + const isRateLimit = + errorLower.includes('rate limit') || + errorLower.includes('resource_exhausted') || + errorLower.includes('too many requests'); + results.push({ + type: 'error', + message: result.error, + status: isRateLimit ? 429 : 400, + errorType: isRateLimit ? 'rate_limit_error' : 'server_error', + }); + return results; + } + + if (result.toolCall) results.push({ type: 'toolCall', toolCall: result.toolCall }); + if (result.text) results.push({ type: 'text', text: result.text }); + if (result.thinking) results.push({ type: 'thinking', text: result.thinking }); + } + + return results; + } + + hasPartial(): boolean { + return this.buffer.length > 0; + } +} diff --git a/tests/unit/cursor/cursor-protobuf.test.ts b/tests/unit/cursor/cursor-protobuf.test.ts index 7d74495d..0ea16a7c 100644 --- a/tests/unit/cursor/cursor-protobuf.test.ts +++ b/tests/unit/cursor/cursor-protobuf.test.ts @@ -19,6 +19,7 @@ import { buildCursorRequest } from '../../../src/cursor/cursor-translator'; import { generateCursorBody } from '../../../src/cursor/cursor-protobuf'; import { CursorExecutor } from '../../../src/cursor/cursor-executor'; import { WIRE_TYPE, FIELD } from '../../../src/cursor/cursor-protobuf-schema'; +import { StreamingFrameParser, decompressPayload } from '../../../src/cursor/cursor-stream-parser'; describe('Protobuf Encoding/Decoding', () => { describe('encodeVarint / decodeVarint round-trip', () => { @@ -372,11 +373,11 @@ describe('Request Encoding', () => { // Create two simple frames const frame1 = wrapConnectRPCFrame( - encodeField(FIELD.RESPONSE_TEXT, WIRE_TYPE.LEN, 'Frame 1'), + encodeField(FIELD.ChatResponse.TEXT, WIRE_TYPE.LEN, 'Frame 1'), false ); const frame2 = wrapConnectRPCFrame( - encodeField(FIELD.RESPONSE_TEXT, WIRE_TYPE.LEN, ' Frame 2'), + encodeField(FIELD.ChatResponse.TEXT, WIRE_TYPE.LEN, ' Frame 2'), false ); @@ -497,8 +498,8 @@ describe('CursorExecutor', () => { it('should handle basic text response', async () => { // Create minimal protobuf response with text const textContent = 'Hello, world!'; - const responseField = encodeField(FIELD.RESPONSE_TEXT, WIRE_TYPE.LEN, textContent); - const responseMsg = encodeField(FIELD.RESPONSE, WIRE_TYPE.LEN, responseField); + const responseField = encodeField(FIELD.ChatResponse.TEXT, WIRE_TYPE.LEN, textContent); + const responseMsg = encodeField(FIELD.Response.RESPONSE, WIRE_TYPE.LEN, responseField); const frame = wrapConnectRPCFrame(responseMsg, false); const result = executor.transformProtobufToJSON(Buffer.from(frame), 'gpt-4', { @@ -536,8 +537,8 @@ describe('CursorExecutor', () => { 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 responseField = encodeField(FIELD.ChatResponse.TEXT, WIRE_TYPE.LEN, textContent); + const responseMsg = encodeField(FIELD.Response.RESPONSE, WIRE_TYPE.LEN, responseField); const frame = wrapConnectRPCFrame(responseMsg, false); const result = executor.transformProtobufToSSE(Buffer.from(frame), 'gpt-4', { @@ -655,3 +656,188 @@ describe('CursorExecutor', () => { }); }); }); + +/** + * Helper: build a ConnectRPC frame from raw payload bytes + */ +function buildFrame(payload: Uint8Array, flags = 0): Buffer { + const header = Buffer.alloc(5); + header[0] = flags; + header.writeUInt32BE(payload.length, 1); + return Buffer.concat([header, Buffer.from(payload)]); +} + +/** + * Helper: build a protobuf text response frame + */ +function buildTextFrame(text: string): Buffer { + const responseField = encodeField(FIELD.ChatResponse.TEXT, WIRE_TYPE.LEN, text); + const responseMsg = encodeField(FIELD.Response.RESPONSE, WIRE_TYPE.LEN, responseField); + return buildFrame(responseMsg); +} + +describe('StreamingFrameParser', () => { + it('should parse a complete single frame', () => { + const parser = new StreamingFrameParser(); + const frame = buildTextFrame('Hello'); + const results = parser.push(frame); + + expect(results.length).toBe(1); + expect(results[0].type).toBe('text'); + if (results[0].type === 'text') { + expect(results[0].text).toBe('Hello'); + } + expect(parser.hasPartial()).toBe(false); + }); + + it('should buffer partial frame header (< 5 bytes)', () => { + const parser = new StreamingFrameParser(); + const frame = buildTextFrame('Hi'); + + // Send only first 3 bytes (partial header) + const results1 = parser.push(frame.slice(0, 3)); + expect(results1.length).toBe(0); + expect(parser.hasPartial()).toBe(true); + + // Send rest of the frame + const results2 = parser.push(frame.slice(3)); + expect(results2.length).toBe(1); + expect(results2[0].type).toBe('text'); + expect(parser.hasPartial()).toBe(false); + }); + + it('should buffer partial frame payload', () => { + const parser = new StreamingFrameParser(); + const frame = buildTextFrame('Hello world'); + + // Send header + partial payload + const splitPoint = 8; // past the 5-byte header but not full payload + const results1 = parser.push(frame.slice(0, splitPoint)); + expect(results1.length).toBe(0); + expect(parser.hasPartial()).toBe(true); + + // Complete the frame + const results2 = parser.push(frame.slice(splitPoint)); + expect(results2.length).toBe(1); + expect(results2[0].type).toBe('text'); + if (results2[0].type === 'text') { + expect(results2[0].text).toBe('Hello world'); + } + }); + + it('should parse multiple frames from single chunk', () => { + const parser = new StreamingFrameParser(); + const frame1 = buildTextFrame('First'); + const frame2 = buildTextFrame('Second'); + const combined = Buffer.concat([frame1, frame2]); + + const results = parser.push(combined); + expect(results.length).toBe(2); + expect(results[0].type).toBe('text'); + expect(results[1].type).toBe('text'); + if (results[0].type === 'text' && results[1].type === 'text') { + expect(results[0].text).toBe('First'); + expect(results[1].text).toBe('Second'); + } + }); + + it('should handle frame split across two chunks', () => { + const parser = new StreamingFrameParser(); + const frame1 = buildTextFrame('AAA'); + const frame2 = buildTextFrame('BBB'); + const combined = Buffer.concat([frame1, frame2]); + + // Split in the middle of frame2 + const splitPoint = frame1.length + 3; + const results1 = parser.push(combined.slice(0, splitPoint)); + expect(results1.length).toBe(1); // frame1 complete + expect(results1[0].type).toBe('text'); + + const results2 = parser.push(combined.slice(splitPoint)); + expect(results2.length).toBe(1); // frame2 complete + expect(results2[0].type).toBe('text'); + if (results2[0].type === 'text') { + expect(results2[0].text).toBe('BBB'); + } + }); + + it('should detect JSON error mid-stream', () => { + const parser = new StreamingFrameParser(); + + // First: a valid text frame + const textFrame = buildTextFrame('Before error'); + const results1 = parser.push(textFrame); + expect(results1.length).toBe(1); + + // Then: a JSON error frame + const errorJson = JSON.stringify({ + error: { code: 'resource_exhausted', message: 'Rate limit' }, + }); + const errorFrame = buildFrame(new TextEncoder().encode(errorJson)); + const results2 = parser.push(errorFrame); + + expect(results2.length).toBe(1); + expect(results2[0].type).toBe('error'); + if (results2[0].type === 'error') { + expect(results2[0].status).toBe(429); + expect(results2[0].errorType).toBe('rate_limit_error'); + } + }); + + it('should report hasPartial() correctly', () => { + const parser = new StreamingFrameParser(); + expect(parser.hasPartial()).toBe(false); + + // Push partial data + parser.push(Buffer.from([0x00, 0x00])); + expect(parser.hasPartial()).toBe(true); + + // Push enough to complete the frame (empty payload) + // header: flags=0, length=0 → 5 bytes total + const emptyFrame = Buffer.alloc(5); + emptyFrame[0] = 0; + emptyFrame.writeUInt32BE(0, 1); + + const parser2 = new StreamingFrameParser(); + parser2.push(emptyFrame); + expect(parser2.hasPartial()).toBe(false); + }); +}); + +describe('decompressPayload', () => { + it('should pass through uncompressed payload', () => { + const payload = Buffer.from('raw protobuf data'); + const result = decompressPayload(payload, 0x00); + expect(result).toEqual(payload); + }); + + it('should skip decompression for JSON error payloads', () => { + const errorPayload = Buffer.from('{"error":"something went wrong"}'); + const result = decompressPayload(errorPayload, 0x01); // GZIP flag but JSON error + expect(result).toEqual(errorPayload); + }); + + it('should return empty buffer on invalid gzip data', () => { + const invalidGzip = Buffer.from([0x01, 0x02, 0x03, 0x04, 0x05]); + const result = decompressPayload(invalidGzip, 0x01); + expect(result.length).toBe(0); + }); + + it('should decompress valid gzip payload', () => { + const zlib = require('zlib'); + const original = Buffer.from('Hello compressed world'); + const compressed = zlib.gzipSync(original); + + const result = decompressPayload(compressed, 0x01); + expect(result.toString()).toBe('Hello compressed world'); + }); + + it('should handle GZIP_ALT and GZIP_BOTH flags', () => { + const zlib = require('zlib'); + const original = Buffer.from('test data'); + const compressed = zlib.gzipSync(original); + + expect(decompressPayload(compressed, 0x02).toString()).toBe('test data'); + expect(decompressPayload(compressed, 0x03).toString()).toBe('test data'); + }); +});