From 6e38d7b4d2dbeaa59e41df0232b77dd4ec8ef7b0 Mon Sep 17 00:00:00 2001
From: yixi ning <ningyixi0@gmail.com>
Date: Sun, 18 Oct 2026 09:12:30 +0000
Subject: [PATCH] fix(http): skip non-object JSON in SSE streams

---
 src/acp/_sse.py                |  4 +++-
 tests/http/test_http_client.py | 19 +++++++++++++++++++
 tests/http/test_sse.py         |  8 ++++++++
 3 files changed, 30 insertions(+), 1 deletion(-)

diff --git a/src/acp/_sse.py b/src/acp/_sse.py
index c3b5a7f3..73f868c9 100644
--- a/src/acp/_sse.py
+++ b/src/acp/_sse.py
@@ -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:
@@ -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.
     """
diff --git a/tests/http/test_http_client.py b/tests/http/test_http_client.py
index 2cad6df3..daaa1f6a 100644
--- a/tests/http/test_http_client.py
+++ b/tests/http/test_http_client.py
@@ -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()
diff --git a/tests/http/test_sse.py b/tests/http/test_sse.py
index 03658292..cc1e1704 100644
--- a/tests/http/test_sse.py
+++ b/tests/http/test_sse.py
@@ -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}]