fix(sse): propagate malformed Anthropic JSON

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;
   }

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 = [