From 545c8b9515d5afeda17976f418e22037cc9acef9 Mon Sep 17 00:00:00 2001 From: "Kai (Tam Nhu) Tran" <61256810+kaitranntt@users.noreply.github.com> Date: Sat, 7 Feb 2026 04:09:30 -0500 Subject: [PATCH] fix(cliproxy): guard against empty upstream SSE responses in agy profile (#489) * fix(cliproxy): guard against empty upstream SSE responses in agy profile When CLIProxyAPIPlus drops unsigned thinking blocks during sub-agent execution, the response stream can contain no content_block_start or message_delta events. This causes Claude Code CLI to crash with "No assistant message found". Add empty response detection in the tool sanitization proxy's streaming handler. Both the pipe-through and SSE-processing paths now track whether meaningful content was received. If upstream sent data but no content blocks on a 200 OK, a synthetic minimal valid SSE response is injected to prevent the client crash and surface a clear error message. Closes #350 * fix(cliproxy): improve empty response detection and add tests Address code review findings: - Remove message_delta from content detection (lifecycle event, not content); only content_block_start indicates actual content - Add try-catch in end handlers to handle client disconnects gracefully - Add 3 integration tests: empty stream injection, normal stream passthrough, 4xx/5xx non-injection * fix(cliproxy): avoid duplicate message_start in synthetic response Track whether upstream already sent a message_start event. When injecting the synthetic error response, omit message_start if upstream already sent one, preventing duplicate events in the SSE stream. Addresses PR review feedback from ccs-reviewer[bot]. * test(cliproxy): add SSE processing path and real failure mode tests Address code review observations: - Add test exercising SSE processing path (sanitized tool names) for empty response detection, ensuring both code paths are covered - Add test mirroring real failure mode where upstream sends only message_start then ends abruptly (no message_delta/message_stop) - Document duplicate message_start assumption with inline comment --- src/cliproxy/tool-sanitization-proxy.ts | 103 ++++++++- ...ool-sanitization-proxy-integration.test.ts | 204 ++++++++++++++++++ 2 files changed, 299 insertions(+), 8 deletions(-) diff --git a/src/cliproxy/tool-sanitization-proxy.ts b/src/cliproxy/tool-sanitization-proxy.ts index 78ddecdb..76887a7b 100644 --- a/src/cliproxy/tool-sanitization-proxy.ts +++ b/src/cliproxy/tool-sanitization-proxy.ts @@ -439,10 +439,45 @@ export class ToolSanitizationProxy { (upstreamRes) => { clientRes.writeHead(upstreamRes.statusCode || 200, upstreamRes.headers); - // If no changes were made, just pipe through + // Track whether upstream emitted any content (guards against empty proxy responses) + let hasReceivedContent = false; + let hasReceivedData = false; + let hasReceivedMessageStart = false; + const isSuccessResponse = + (upstreamRes.statusCode || 200) >= 200 && (upstreamRes.statusCode || 200) < 300; + + // If no changes were made, intercept to detect empty responses. + // In the real failure case (issue #350), upstream sends message_start but + // NOT message_delta/message_stop, so the synthetic response completes the + // stream. If upstream DID send message_stop, Claude Code treats the second + // synthetic block as additional content in the same conversation turn. if (!mapper.hasChanges()) { - upstreamRes.pipe(clientRes); - upstreamRes.on('end', () => resolve()); + upstreamRes.on('data', (chunk: Buffer) => { + hasReceivedData = true; + const chunkStr = chunk.toString('utf8'); + if (chunkStr.includes('"content_block_start"')) { + hasReceivedContent = true; + } + if (chunkStr.includes('"message_start"')) { + hasReceivedMessageStart = true; + } + clientRes.write(chunk); + }); + upstreamRes.on('end', () => { + try { + if (!hasReceivedContent && isSuccessResponse && hasReceivedData) { + this.writeLog( + 'warn', + '[tool-sanitization-proxy] Empty response detected from upstream (no content blocks). Injecting synthetic response to prevent client crash.' + ); + clientRes.write(this.buildSyntheticErrorResponse(hasReceivedMessageStart)); + } + clientRes.end(); + } catch { + // Client may have disconnected — safe to ignore + } + resolve(); + }); upstreamRes.on('error', reject); return; } @@ -451,6 +486,7 @@ export class ToolSanitizationProxy { let buffer = ''; upstreamRes.on('data', (chunk: Buffer) => { + hasReceivedData = true; buffer += chunk.toString('utf8'); // Process complete SSE events @@ -460,18 +496,42 @@ export class ToolSanitizationProxy { for (const event of events) { if (!event.trim()) continue; + if (event.includes('"content_block_start"')) { + hasReceivedContent = true; + } + if (event.includes('"message_start"')) { + hasReceivedMessageStart = true; + } + const processedEvent = this.processSSEEvent(event, mapper); clientRes.write(processedEvent + '\n\n'); } }); upstreamRes.on('end', () => { - // Process any remaining buffer - if (buffer.trim()) { - const processedEvent = this.processSSEEvent(buffer, mapper); - clientRes.write(processedEvent + '\n\n'); + try { + // Process any remaining buffer + if (buffer.trim()) { + if (buffer.includes('"content_block_start"')) { + hasReceivedContent = true; + } + const processedEvent = this.processSSEEvent(buffer, mapper); + clientRes.write(processedEvent + '\n\n'); + } + + // Safety net: if upstream sent data but no content blocks, inject synthetic response + if (!hasReceivedContent && isSuccessResponse && hasReceivedData) { + this.writeLog( + 'warn', + '[tool-sanitization-proxy] Empty response detected from upstream (no content blocks). Injecting synthetic response to prevent client crash.' + ); + clientRes.write(this.buildSyntheticErrorResponse(hasReceivedMessageStart)); + } + + clientRes.end(); + } catch { + // Client may have disconnected — safe to ignore } - clientRes.end(); resolve(); }); @@ -542,4 +602,31 @@ export class ToolSanitizationProxy { return processedLines.join('\n'); } + + /** + * Build a synthetic minimal SSE response when upstream returns empty content. + * Prevents Claude Code from crashing with "No assistant message found". + * Omits message_start if upstream already sent one to avoid duplicate events. + */ + private buildSyntheticErrorResponse(upstreamSentMessageStart = false): string { + const events: string[] = []; + + // Only include message_start if upstream didn't already send one + if (!upstreamSentMessageStart) { + const msgId = `msg_synthetic_${Date.now()}`; + events.push( + `event: message_start\ndata: {"type":"message_start","message":{"id":"${msgId}","type":"message","role":"assistant","content":[],"model":"unknown","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":0,"output_tokens":0}}}` + ); + } + + events.push( + `event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, + `event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"[Proxy Error] The upstream API returned an empty response. This typically occurs when the proxy drops unsigned thinking blocks during sub-agent execution. Please retry the request."}}`, + `event: content_block_stop\ndata: {"type":"content_block_stop","index":0}`, + `event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"input_tokens":0,"output_tokens":0}}`, + `event: message_stop\ndata: {"type":"message_stop"}` + ); + + return events.join('\n\n') + '\n\n'; + } } diff --git a/tests/unit/cliproxy/tool-sanitization-proxy-integration.test.ts b/tests/unit/cliproxy/tool-sanitization-proxy-integration.test.ts index 038fc13c..f0055f31 100644 --- a/tests/unit/cliproxy/tool-sanitization-proxy-integration.test.ts +++ b/tests/unit/cliproxy/tool-sanitization-proxy-integration.test.ts @@ -490,4 +490,208 @@ describe('ToolSanitizationProxy Integration', () => { } }); }); + + describe('Empty Response Safety Net (Issue #350)', () => { + it('injects synthetic response when streaming 200 has no content blocks', async () => { + const proxy = new ToolSanitizationProxy({ + upstreamBaseUrl: `http://127.0.0.1:${mockUpstreamPort}`, + }); + const port = await proxy.start(); + + // Upstream sends message_start + message_delta + message_stop but NO content blocks + mockResponse = { + status: 200, + stream: true, + body: [ + 'event: message_start\n', + 'data: {"type":"message_start","message":{"id":"msg_test","type":"message","role":"assistant","content":[],"model":"test","stop_reason":null}}\n\n', + 'event: message_delta\n', + 'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":10,"output_tokens":0}}\n\n', + 'event: message_stop\n', + 'data: {"type":"message_stop"}\n\n', + ], + }; + + try { + const response = await fetch(`http://127.0.0.1:${port}/v1/messages`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + stream: true, + tools: [{ name: 'valid_tool' }], + }), + }); + + const text = await response.text(); + // Should contain the synthetic error message + expect(text).toContain('[Proxy Error]'); + expect(text).toContain('content_block_start'); + expect(text).toContain('content_block_delta'); + expect(text).toContain('content_block_stop'); + // Should NOT have duplicate message_start (upstream already sent one) + const messageStartCount = (text.match(/"type":"message_start"/g) || []).length; + expect(messageStartCount).toBe(1); // Only the upstream one, not a synthetic duplicate + } finally { + proxy.stop(); + } + }); + + it('does NOT inject synthetic response when content blocks exist', async () => { + const proxy = new ToolSanitizationProxy({ + upstreamBaseUrl: `http://127.0.0.1:${mockUpstreamPort}`, + }); + const port = await proxy.start(); + + mockResponse = { + status: 200, + stream: true, + body: [ + 'event: message_start\n', + 'data: {"type":"message_start","message":{"id":"msg_ok","type":"message","role":"assistant","content":[],"model":"test"}}\n\n', + 'event: content_block_start\n', + 'data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}\n\n', + 'event: content_block_delta\n', + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}\n\n', + 'event: content_block_stop\n', + 'data: {"type":"content_block_stop","index":0}\n\n', + 'event: message_delta\n', + 'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}\n\n', + 'event: message_stop\n', + 'data: {"type":"message_stop"}\n\n', + ], + }; + + try { + const response = await fetch(`http://127.0.0.1:${port}/v1/messages`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + stream: true, + tools: [{ name: 'valid_tool' }], + }), + }); + + const text = await response.text(); + expect(text).toContain('Hello'); + expect(text).not.toContain('[Proxy Error]'); + } finally { + proxy.stop(); + } + }); + + it('does NOT inject synthetic response for 4xx/5xx errors', async () => { + const proxy = new ToolSanitizationProxy({ + upstreamBaseUrl: `http://127.0.0.1:${mockUpstreamPort}`, + }); + const port = await proxy.start(); + + // 429 rate limit with SSE body but no content blocks + mockResponse = { + status: 429, + stream: true, + body: [ + 'event: error\n', + 'data: {"type":"error","error":{"type":"rate_limit_error","message":"Too many requests"}}\n\n', + ], + }; + + try { + const response = await fetch(`http://127.0.0.1:${port}/v1/messages`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + stream: true, + tools: [{ name: 'valid_tool' }], + }), + }); + + expect(response.status).toBe(429); + const text = await response.text(); + expect(text).not.toContain('[Proxy Error]'); + expect(text).toContain('rate_limit_error'); + } finally { + proxy.stop(); + } + }); + + it('injects synthetic response via SSE processing path (with sanitized tool names)', async () => { + const proxy = new ToolSanitizationProxy({ + upstreamBaseUrl: `http://127.0.0.1:${mockUpstreamPort}`, + }); + const port = await proxy.start(); + + // Upstream returns empty content — no content_block_start events + // Using sanitized tool name ('foo__bar__bar' → 'foo__bar') triggers the SSE processing path + mockResponse = { + status: 200, + stream: true, + body: [ + 'event: message_start\n', + 'data: {"type":"message_start","message":{"id":"msg_sse","type":"message","role":"assistant","content":[],"model":"test","stop_reason":null}}\n\n', + 'event: message_delta\n', + 'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"input_tokens":10,"output_tokens":0}}\n\n', + 'event: message_stop\n', + 'data: {"type":"message_stop"}\n\n', + ], + }; + + try { + const response = await fetch(`http://127.0.0.1:${port}/v1/messages`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + stream: true, + tools: [{ name: 'foo__bar__bar' }], // Triggers sanitization → SSE processing path + }), + }); + + const text = await response.text(); + expect(text).toContain('[Proxy Error]'); + expect(text).toContain('content_block_start'); + const messageStartCount = (text.match(/"type":"message_start"/g) || []).length; + expect(messageStartCount).toBe(1); + } finally { + proxy.stop(); + } + }); + + it('injects synthetic response when upstream sends only message_start (real failure mode)', async () => { + const proxy = new ToolSanitizationProxy({ + upstreamBaseUrl: `http://127.0.0.1:${mockUpstreamPort}`, + }); + const port = await proxy.start(); + + // Real failure: upstream sends message_start then abruptly ends — no message_delta or message_stop + mockResponse = { + status: 200, + stream: true, + body: [ + 'event: message_start\n', + 'data: {"type":"message_start","message":{"id":"msg_abrupt","type":"message","role":"assistant","content":[],"model":"test","stop_reason":null}}\n\n', + ], + }; + + try { + const response = await fetch(`http://127.0.0.1:${port}/v1/messages`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + stream: true, + tools: [{ name: 'valid_tool' }], + }), + }); + + const text = await response.text(); + expect(text).toContain('[Proxy Error]'); + expect(text).toContain('content_block_start'); + expect(text).toContain('message_delta'); + expect(text).toContain('message_stop'); + // Only 1 message_start — upstream's original + const messageStartCount = (text.match(/"type":"message_start"/g) || []).length; + expect(messageStartCount).toBe(1); + } finally { + proxy.stop(); + } + }); + }); });