Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion src/acp/_sse.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,10 @@ def _decode_event(data_lines: list[str]) -> dict[str, Any] | None:
if not payload:
return None
try:
return json.loads(payload)
message = json.loads(payload)
except json.JSONDecodeError:
return None
return message if isinstance(message, dict) else None


def _append_field(line: str, data_lines: list[str]) -> None:
Expand All @@ -60,6 +61,7 @@ async def parse_sse_stream(chunks: AsyncIterator[bytes]) -> AsyncIterator[dict[s
"""Parse an SSE byte stream, yielding decoded JSON-RPC ``data:`` payloads.

Comments (lines starting with ``:``) and non-``data`` fields are ignored.
Invalid JSON and non-object JSON payloads are skipped.
Multi-line ``data:`` fields are concatenated with newlines per the spec. A
blank line dispatches the buffered event.
"""
Expand Down
19 changes: 19 additions & 0 deletions tests/http/test_http_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,25 @@ async def test_connection_scoped_sse_delivers_new_session_result() -> None:
await client.aclose()


@pytest.mark.asyncio
@pytest.mark.parametrize("payload", [b"[]", b'"text"', b"42"])
async def test_connection_scoped_sse_skips_non_object_json(payload: bytes) -> None:
server = FakeServer()
transport, client = _make_transport(server)
try:
await transport.send({"jsonrpc": "2.0", "id": 0, "method": "initialize", "params": {}})
await asyncio.wait_for(transport.receive(), timeout=1)
await transport.send({"jsonrpc": "2.0", "id": 1, "method": "session/new", "params": {}})
server.conn_stream.put_nowait(b"data: " + payload + b"\n\n")
response = {"jsonrpc": "2.0", "id": 1, "result": {"sessionId": "sess-1"}}
server.push_conn(response)

assert await asyncio.wait_for(transport.receive(), timeout=1) == response
finally:
await transport.close()
await client.aclose()


@pytest.mark.asyncio
async def test_session_scoped_sse_opens_after_new_session() -> None:
server = FakeServer()
Expand Down
8 changes: 8 additions & 0 deletions tests/http/test_sse.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,3 +72,11 @@ async def test_parse_skips_invalid_json() -> None:
stream = _aiter([b'data: not-json\n\ndata: {"id":1}\n\n'])
events = [event async for event in parse_sse_stream(stream)]
assert events == [{"id": 1}]


@pytest.mark.asyncio
@pytest.mark.parametrize("payload", [b"[]", b'[ {"id":2} ]', b'"text"', b"42", b"true", b"false", b"null"])
async def test_parse_skips_non_object_json(payload: bytes) -> None:
stream = _aiter([b"data: " + payload + b'\n\ndata: {"id":1}\n\n'])
events = [event async for event in parse_sse_stream(stream)]
assert events == [{"id": 1}]
Loading