diff --git a/src/providers/sse.ts b/src/providers/sse.ts index 2ccb87d..53364fb 100644 --- a/src/providers/sse.ts +++ b/src/providers/sse.ts @@ -176,27 +176,35 @@ export function parseAnthropicSseLine( const payload = trimmed.slice(5).trim(); if (state.currentEvent === 'message_start') { + let parsed: unknown; try { - const parsed = JSON.parse(payload) as { - message?: { model?: string }; - }; - if (parsed.message?.model) { - return { kind: 'model', model: parsed.message.model }; - } + parsed = JSON.parse(payload); } catch { - // Skip malformed JSON + throw new Error('Malformed Anthropic SSE data: invalid JSON'); + } + + if (!parsed || typeof parsed !== 'object') return null; + + const data = parsed as { message?: { model?: string } }; + if (data.message?.model) { + return { kind: 'model', model: data.message.model }; } return null; } if (state.currentEvent === 'content_block_delta') { + let parsed: unknown; try { - const parsed = JSON.parse(payload) as { delta?: { text?: string } }; - if (parsed.delta?.text) { - return { kind: 'text', text: parsed.delta.text }; - } + parsed = JSON.parse(payload); } catch { - // Skip malformed JSON + throw new Error('Malformed Anthropic SSE data: invalid JSON'); + } + + if (!parsed || typeof parsed !== 'object') return null; + + const data = parsed as { delta?: { text?: string } }; + if (data.delta?.text) { + return { kind: 'text', text: data.delta.text }; } return null; } diff --git a/tests/stream-sse.test.mjs b/tests/stream-sse.test.mjs index 6b6b6f6..c123b4f 100644 --- a/tests/stream-sse.test.mjs +++ b/tests/stream-sse.test.mjs @@ -269,6 +269,36 @@ test('parseAnthropicSseLine extracts model from message_start', () => { assert.deepEqual(result, { kind: 'model', model: 'claude-sonnet-4' }); }); +test('parseAnthropicSseLine rejects malformed message_start JSON', () => { + const state = { currentEvent: '' }; + parseAnthropicSseLine('event: message_start', state); + assert.throws( + () => parseAnthropicSseLine('data: {not valid JSON}', state), + /Malformed Anthropic SSE data: invalid JSON/, + ); +}); + +test('parseAnthropicSseLine rejects malformed content_block_delta JSON', () => { + const state = { currentEvent: '' }; + parseAnthropicSseLine('event: content_block_delta', state); + assert.throws( + () => parseAnthropicSseLine('data: {not valid JSON}', state), + /Malformed Anthropic SSE data: invalid JSON/, + ); +}); + +test('parseAnthropicSseLine ignores null message_start JSON', () => { + const state = { currentEvent: '' }; + parseAnthropicSseLine('event: message_start', state); + assert.equal(parseAnthropicSseLine('data: null', state), null); +}); + +test('parseAnthropicSseLine ignores null content_block_delta JSON', () => { + const state = { currentEvent: '' }; + parseAnthropicSseLine('event: content_block_delta', state); + assert.equal(parseAnthropicSseLine('data: null', state), null); +}); + test('parseAnthropicSseLine returns SSE_STREAM_END on message_stop', () => { const state = { currentEvent: '' }; parseAnthropicSseLine('event: message_stop', state); @@ -303,6 +333,48 @@ test('Anthropic completeStream reassembles event/data split across network chunk }, ); }); + +test('Anthropic completeStream propagates malformed JSON and releases the response stream', async () => { + let cancelled = false; + let malformedSent = false; + let closeTimer; + let sourceController; + const response = new Response( + new ReadableStream({ + start(controller) { + sourceController = controller; + controller.enqueue(new TextEncoder().encode('event: content_block_delta\n')); + }, + pull(stream) { + if (malformedSent) return; + malformedSent = true; + stream.enqueue(new TextEncoder().encode('data: {"delta":{"text":"before"}}\n')); + stream.enqueue(new TextEncoder().encode('data: {not valid JSON}\n')); + closeTimer = setTimeout(() => { + if (!cancelled) sourceController.close(); + }, 100); + }, + cancel() { + cancelled = true; + if (closeTimer) clearTimeout(closeTimer); + }, + }), + { status: 200 }, + ); + + await withMockedFetch( + async () => response, + async () => { + const iterator = new AnthropicProvider().completeStream(anthropicParams)[Symbol.asyncIterator](); + assert.deepEqual(await iterator.next(), { done: false, value: { kind: 'text', text: 'before' } }); + await assert.rejects(() => iterator.next(), /Malformed Anthropic SSE data: invalid JSON/); + + assert.equal(cancelled, true); + assert.equal(response.body.locked, false); + await iterator.return?.().catch(() => {}); + }, + ); +}); test('OpenAI completeStream emits reasoning progressively and keeps it separate from visible content', async () => { const provider = new OpenAICompatibleProvider(); const responses = [