mirror of
https://github.com/tiennm99/ccs.git
synced 2026-10-15 00:17:09 +00:00
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
This commit is contained in:
Kai (Tam Nhu) Tran authored and GitHub committed
1 parent
984e007468
commit
545c8b9515
2 files changed
+299
-8
No files matched your search
@@ -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();
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user