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

@@ -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';
+  }
 }

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();
+      }
+    });
+  });
 });