From 0407aeb6481f81ac10d2359b8d5f274d07b8e3a1 Mon Sep 17 00:00:00 2001
From: adhavan18 <sk.tamiladhavan@gmail.com>
Date: Mon, 14 Sep 2026 18:22:39 +0530
Subject: [PATCH] fix(decoders): close the response on a jsonl decode or
 transport error

__decode__ and AsyncJSONLDecoder.__decode__ only closed the underlying
http_response when the generator ran to completion. a json decode error,
or a transport error from the underlying byte stream, propagated straight
out of the generator and left the connection open, since close() was never
called on any non-normal exit path.

wrapped both generator bodies in try/finally calling close()/await close(),
matching how the SSE stream decoder already handles this.

verified with two new tests that construct a real httpx2 stream (not a bare
Response(200), which starts pre-closed and would make the test a no-op) and
confirm response.is_closed is True after a decode error - both fail on
unfixed main and pass with the fix.
---
 src/anthropic/_decoders/jsonl.py | 74 ++++++++++++++++++--------------
 tests/decoders/test_jsonl.py     | 21 +++++++++
 2 files changed, 63 insertions(+), 32 deletions(-)

diff --git a/src/anthropic/_decoders/jsonl.py b/src/anthropic/_decoders/jsonl.py
index d36a59791..ab0b9b4b7 100644
--- a/src/anthropic/_decoders/jsonl.py
+++ b/src/anthropic/_decoders/jsonl.py
@@ -42,22 +42,27 @@ def close(self) -> None:
 
     def __decode__(self) -> Iterator[_T]:
         buf = b""
-        for chunk in self._raw_iterator:
-            for line in chunk.splitlines(keepends=True):
-                buf += line
-                if buf.endswith((b"\r", b"\n", b"\r\n")):
-                    yield construct_type_unchecked(
-                        value=json.loads(buf),
-                        type_=self._line_type,
-                    )
-                    buf = b""
-
-        # flush
-        if buf:
-            yield construct_type_unchecked(
-                value=json.loads(buf),
-                type_=self._line_type,
-            )
+        try:
+            for chunk in self._raw_iterator:
+                for line in chunk.splitlines(keepends=True):
+                    buf += line
+                    if buf.endswith((b"\r", b"\n", b"\r\n")):
+                        yield construct_type_unchecked(
+                            value=json.loads(buf),
+                            type_=self._line_type,
+                        )
+                        buf = b""
+
+            # flush
+            if buf:
+                yield construct_type_unchecked(
+                    value=json.loads(buf),
+                    type_=self._line_type,
+                )
+        finally:
+            # Ensure the response is closed even if the consumer doesn't read all
+            # data, or a JSON decoding or transport-read error is raised above.
+            self.close()
 
     def __next__(self) -> _T:
         return self._iterator.__next__()
@@ -98,22 +103,27 @@ async def close(self) -> None:
 
     async def __decode__(self) -> AsyncIterator[_T]:
         buf = b""
-        async for chunk in self._raw_iterator:
-            for line in chunk.splitlines(keepends=True):
-                buf += line
-                if buf.endswith((b"\r", b"\n", b"\r\n")):
-                    yield construct_type_unchecked(
-                        value=json.loads(buf),
-                        type_=self._line_type,
-                    )
-                    buf = b""
-
-        # flush
-        if buf:
-            yield construct_type_unchecked(
-                value=json.loads(buf),
-                type_=self._line_type,
-            )
+        try:
+            async for chunk in self._raw_iterator:
+                for line in chunk.splitlines(keepends=True):
+                    buf += line
+                    if buf.endswith((b"\r", b"\n", b"\r\n")):
+                        yield construct_type_unchecked(
+                            value=json.loads(buf),
+                            type_=self._line_type,
+                        )
+                        buf = b""
+
+            # flush
+            if buf:
+                yield construct_type_unchecked(
+                    value=json.loads(buf),
+                    type_=self._line_type,
+                )
+        finally:
+            # Ensure the response is closed even if the consumer doesn't read all
+            # data, or a JSON decoding or transport-read error is raised above.
+            await self.close()
 
     async def __anext__(self) -> _T:
         return await self._iterator.__anext__()
diff --git a/tests/decoders/test_jsonl.py b/tests/decoders/test_jsonl.py
index 9ab8a6f27..f01c1046d 100644
--- a/tests/decoders/test_jsonl.py
+++ b/tests/decoders/test_jsonl.py
@@ -60,6 +60,27 @@ def body() -> Iterator[bytes]:
     assert await iter_next(iterator) == {"content": "известни"}
 
 
+def test_sync_closes_response_on_decode_error() -> None:
+    response = httpx2.Response(200, stream=httpx2.ByteStream(b"invalid json\n"))
+    decoder = JSONLDecoder(line_type=object, raw_iterator=response.iter_bytes(), http_response=response)
+
+    with pytest.raises(Exception):
+        list(decoder)
+
+    assert response.is_closed
+
+
+async def test_async_closes_response_on_decode_error() -> None:
+    response = httpx2.Response(200, stream=httpx2.ByteStream(b"invalid json\n"))
+    decoder = AsyncJSONLDecoder(line_type=object, raw_iterator=response.aiter_bytes(), http_response=response)
+
+    with pytest.raises(Exception):
+        async for _ in decoder:
+            pass
+
+    assert response.is_closed
+
+
 async def to_aiter(iter: Iterator[bytes]) -> AsyncIterator[bytes]:
     for chunk in iter:
         yield chunk