From 0c8ed9e52d14568087543791c56b55c6a9ce3db3 Mon Sep 17 00:00:00 2001 From: 404-Page-Found <139850808+404-Page-Found@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:15:20 +1000 Subject: [PATCH 1/6] fix(sse): propagate malformed Anthropic JSON --- src/providers/sse.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/providers/sse.ts b/src/providers/sse.ts index 2ccb87d..16f183e 100644 --- a/src/providers/sse.ts +++ b/src/providers/sse.ts @@ -184,7 +184,7 @@ export function parseAnthropicSseLine( return { kind: 'model', model: parsed.message.model }; } } catch { - // Skip malformed JSON + throw new Error('Malformed Anthropic SSE data: invalid JSON'); } return null; } @@ -196,7 +196,7 @@ export function parseAnthropicSseLine( return { kind: 'text', text: parsed.delta.text }; } } catch { - // Skip malformed JSON + throw new Error('Malformed Anthropic SSE data: invalid JSON'); } return null; } From be7c05d62d769c451cb094b97293b44a4593b432 Mon Sep 17 00:00:00 2001 From: 404-Page-Found <139850808+404-Page-Found@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:15:24 +1000 Subject: [PATCH 2/6] test(sse): cover malformed Anthropic JSON frames --- tests/stream-sse.test.mjs | 50 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/tests/stream-sse.test.mjs b/tests/stream-sse.test.mjs index 6b6b6f6..5abce92 100644 --- a/tests/stream-sse.test.mjs +++ b/tests/stream-sse.test.mjs @@ -269,6 +269,24 @@ 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 returns SSE_STREAM_END on message_stop', () => { const state = { currentEvent: '' }; parseAnthropicSseLine('event: message_stop', state); @@ -303,6 +321,38 @@ 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; + const response = new Response( + new ReadableStream({ + start(stream) { + stream.enqueue(new TextEncoder().encode('event: content_block_delta\n')); + }, + pull(stream) { + stream.enqueue(new TextEncoder().encode('data: {"delta":{"text":"before"}}\n')); + stream.enqueue(new TextEncoder().encode('data: {not valid JSON}\n')); + stream.close(); + }, + cancel() { + cancelled = true; + }, + }), + { 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); + }, + ); +}); test('OpenAI completeStream emits reasoning progressively and keeps it separate from visible content', async () => { const provider = new OpenAICompatibleProvider(); const responses = [ From 310e67ac6c4e4b8119203f74f98b6d31e75ac1bb Mon Sep 17 00:00:00 2001 From: 404-Page-Found <139850808+404-Page-Found@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:15:57 +1000 Subject: [PATCH 3/6] test(sse): keep response readable for cancellation regression --- tests/stream-sse.test.mjs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/stream-sse.test.mjs b/tests/stream-sse.test.mjs index 5abce92..c4b9446 100644 --- a/tests/stream-sse.test.mjs +++ b/tests/stream-sse.test.mjs @@ -324,18 +324,25 @@ 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; const response = new Response( new ReadableStream({ start(stream) { stream.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')); - stream.close(); + closeTimer = setTimeout(() => { + if (!cancelled) response.body?.getReader; + }, 100); }, cancel() { cancelled = true; + if (closeTimer) clearTimeout(closeTimer); }, }), { status: 200 }, @@ -350,6 +357,7 @@ test('Anthropic completeStream propagates malformed JSON and releases the respon assert.equal(cancelled, true); assert.equal(response.body.locked, false); + await iterator.return?.().catch(() => {}); }, ); }); From bf1a1b219dd114693b801ef9874fa6c8dd50f08c Mon Sep 17 00:00:00 2001 From: 404-Page-Found <139850808+404-Page-Found@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:16:05 +1000 Subject: [PATCH 4/6] test(sse): preserve cancellation coverage --- tests/stream-sse.test.mjs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/stream-sse.test.mjs b/tests/stream-sse.test.mjs index c4b9446..f277b0d 100644 --- a/tests/stream-sse.test.mjs +++ b/tests/stream-sse.test.mjs @@ -326,10 +326,12 @@ test('Anthropic completeStream propagates malformed JSON and releases the respon let cancelled = false; let malformedSent = false; let closeTimer; + let sourceController; const response = new Response( new ReadableStream({ - start(stream) { - stream.enqueue(new TextEncoder().encode('event: content_block_delta\n')); + start(controller) { + sourceController = controller; + controller.enqueue(new TextEncoder().encode('event: content_block_delta\n')); }, pull(stream) { if (malformedSent) return; @@ -337,7 +339,7 @@ test('Anthropic completeStream propagates malformed JSON and releases the respon stream.enqueue(new TextEncoder().encode('data: {"delta":{"text":"before"}}\n')); stream.enqueue(new TextEncoder().encode('data: {not valid JSON}\n')); closeTimer = setTimeout(() => { - if (!cancelled) response.body?.getReader; + if (!cancelled) sourceController.close(); }, 100); }, cancel() { From ca2dbcd55ac8f61d43bab13268ae7f7d4ead8f12 Mon Sep 17 00:00:00 2001 From: 404-Page-Found <139850808+404-Page-Found@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:23:22 +1000 Subject: [PATCH 5/6] fix(sse): guard non-object Anthropic JSON payloads --- src/providers/sse.ts | 28 ++++++++++++++++++---------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/src/providers/sse.ts b/src/providers/sse.ts index 16f183e..53364fb 100644 --- a/src/providers/sse.ts +++ b/src/providers/sse.ts @@ -176,28 +176,36 @@ 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 { 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 { 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; } From 23cf8f122975e31c89254380095a90b07098aeb5 Mon Sep 17 00:00:00 2001 From: 404-Page-Found <139850808+404-Page-Found@users.noreply.github.com> Date: Thu, 24 Sep 2026 11:23:29 +1000 Subject: [PATCH 6/6] test(sse): ignore null Anthropic JSON payloads --- tests/stream-sse.test.mjs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/stream-sse.test.mjs b/tests/stream-sse.test.mjs index f277b0d..c123b4f 100644 --- a/tests/stream-sse.test.mjs +++ b/tests/stream-sse.test.mjs @@ -287,6 +287,18 @@ test('parseAnthropicSseLine rejects malformed content_block_delta 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);