fix(streaming): don't dispatch empty SSE keepalive comment frames

The SSE decoder's empty-block guard included last_event_id, which is sticky
across events per the SSE spec. Once any event carried an id, every subsequent
comment-only block (e.g. the server's ":\n\n" keepalive, sent every 15s on an
idle stream) fell through the guard and dispatched an empty-data event. The
typed Stream wrapper then calls .json() on it unconditionally, raising
JSONDecodeError and killing the stream.

This made idle browser telemetry streams crash after ~15s. Drop last_event_id
from the guard so dispatch depends only on the current block's event/data/retry
fields. Event-typed empty-data frames still dispatch (unchanged). Adds a
regression test for a keepalive comment following an id-bearing event.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rgarcia

src/kernel/_streaming.py

@@ -251,7 +251,12 @@ def decode(self, line: str) -> ServerSentEvent | None:
         # See: https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation  # noqa: E501
 
         if not line:
-            if not self._event and not self._data and not self._last_event_id and self._retry is None:
+            # Whether to dispatch depends only on what was set in the *current* block. last_event_id
+            # is sticky across events (per the SSE spec, it is intentionally not reset below), so it
+            # must not be part of this check -- otherwise, once any event carries an id, every
+            # subsequent comment-only block (e.g. a ``:\n\n`` keepalive) would dispatch an empty
+            # event, which then fails to JSON-decode in the typed Stream wrapper.
+            if not self._event and not self._data and self._retry is None:
                 return None
 
             sse = ServerSentEvent(

tests/test_streaming.py

@@ -26,6 +26,32 @@ def body() -> Iterator[bytes]:
     await assert_empty_iter(iterator)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
+async def test_keepalive_comment_after_event_with_id(sync: bool, client: Kernel, async_client: AsyncKernel) -> None:
+    # A ``:`` comment frame (the server's SSE keepalive) that arrives after an event which set an
+    # id must be ignored, not dispatched as an empty event. last_event_id is sticky, so this is a
+    # regression guard against it leaking an undecodable empty frame into the typed stream.
+    def body() -> Iterator[bytes]:
+        yield b"id: 1\n"
+        yield b'data: {"foo":true}\n'
+        yield b"\n"
+        yield b":\n"
+        yield b"\n"
+        yield b'data: {"bar":false}\n'
+        yield b"\n"
+
+    iterator = make_event_iterator(content=body(), sync=sync, client=client, async_client=async_client)
+
+    sse = await iter_next(iterator)
+    assert sse.json() == {"foo": True}
+
+    sse = await iter_next(iterator)
+    assert sse.json() == {"bar": False}
+
+    await assert_empty_iter(iterator)
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("sync", [True, False], ids=["sync", "async"])
 async def test_data_missing_event(sync: bool, client: Kernel, async_client: AsyncKernel) -> None: