fix(streaming): interrupt blocked synchronous SSE reads

CHANGELOG.md

@@ -10,6 +10,10 @@ existing `0.1.0` release; earlier development prereleases are not listed.
 - Add automatic Thread SSE resumption for Forward and Managed, preserving the latest event cursor and request options across retryable disconnects; thread termination stops the subscription while idle events keep it open.
 - Add a typed `Last-Event-ID` parameter to Managed Thread streams.
 
+### Fixed
+
+- Closing a synchronous Session or Thread SSE subscription from another thread now wakes a blocked HTTP/1.x socket read, including HTTPS, instead of waiting for more data or a read timeout. Fully consumed connections remain reusable; HTTP/2 and transports without an exposed socket retain their own close behavior.
+
 ## [0.13.0]
 
 ### Added

README.md

@@ -549,6 +549,12 @@ async with client.sessions.events.resumable_stream(session_id) as stream:
 
 The helper retries until closed, cancelled, or it receives `session.status_terminated` / `session.deleted`. It uses jittered exponential backoff and the same HTTP retry classification as the client (in particular, `409` is not retried). It does not query event history, discard an invalid cursor, or deduplicate event IDs because multiple preview deltas may share one ID.
 
+With the default synchronous HTTP/1.x transport, `close()` also wakes a body read
+blocked in another thread, including over HTTPS. HTTP/2 connections are shared,
+so their sockets are not interrupted; HTTP/2 and custom transports without an
+exposed socket retain their transport's close behavior. Async callers can cancel
+the consuming task or await `close()` to interrupt an active read.
+
 Events are also readable after the fact through `client.sessions.events.list(session_id)`, which paginates like any other list method.
 
 ## Handling errors

docs/api/reference.md

@@ -1409,6 +1409,46 @@ def last_event_id() -> str | None
 
 [[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/common/_streaming.py)
 
+<a id="qca.common._streaming._InterruptibleSyncBody"></a>
+
+## \_InterruptibleSyncBody
+
+```python
+class _InterruptibleSyncBody(httpx.SyncByteStream)
+```
+
+[[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/common/_streaming.py)
+
+<a id="qca.common._streaming._InterruptibleSyncBody.__init__"></a>
+
+#### \_\_init\_\_
+
+```python
+def __init__(body: httpx.SyncByteStream, connection: socket.socket) -> None
+```
+
+[[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/common/_streaming.py)
+
+<a id="qca.common._streaming._InterruptibleSyncBody.__iter__"></a>
+
+#### \_\_iter\_\_
+
+```python
+def __iter__() -> Iterator[bytes]
+```
+
+[[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/common/_streaming.py)
+
+<a id="qca.common._streaming._InterruptibleSyncBody.close"></a>
+
+#### close
+
+```python
+def close() -> None
+```
+
+[[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/common/_streaming.py)
+
 <a id="qca.common._streaming.Stream"></a>
 
 ## Stream

src/qca/common/_streaming.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import json
+import socket
+import threading
 from collections.abc import AsyncIterator, Iterator
 from typing import Any, Generic, TypeVar
 
@@ -72,9 +74,52 @@ def _parse(self, frame: tuple[str | None, str]) -> T:
         return result
 
 
+class _InterruptibleSyncBody(httpx.SyncByteStream):
+    def __init__(self, body: httpx.SyncByteStream, connection: socket.socket) -> None:
+        self._body = body
+        self._connection = connection
+        self._lock = threading.Lock()
+        self._reading = False
+        self._closed = False
+
+    def __iter__(self) -> Iterator[bytes]:
+        with self._lock:
+            if self._closed:
+                return
+            self._reading = True
+        try:
+            yield from self._body
+        finally:
+            with self._lock:
+                self._reading = False
+
+    def close(self) -> None:
+        # Interrupt before releasing the response to the pool; a later shutdown
+        # could otherwise affect a different request reusing the connection.
+        with self._lock:
+            if self._closed:
+                return
+            self._closed = True
+            if self._reading:
+                try:
+                    self._connection.shutdown(socket.SHUT_RDWR)
+                except OSError:
+                    pass
+            self._body.close()
+
+
 class Stream(BaseStream[T], Iterator[T]):
     def __init__(self, response: httpx.Response, cast_to: Any, *, strict: bool = False) -> None:
         super().__init__(response, cast_to, strict=strict)
+        if not response.is_closed and response.extensions.get("http_version") in (b"HTTP/1.0", b"HTTP/1.1"):
+            network_stream = response.extensions.get("network_stream")
+            get_extra_info = getattr(network_stream, "get_extra_info", None)
+            try:
+                connection = get_extra_info("socket") if callable(get_extra_info) else None
+            except (OSError, NotImplementedError):
+                connection = None
+            if isinstance(connection, socket.socket) and isinstance(response.stream, httpx.SyncByteStream):
+                response.stream = _InterruptibleSyncBody(response.stream, connection)
         self._iterator = self._iter_events()
 
     def __next__(self) -> T:

tests/test_sync_stream_cancellation.py

@@ -0,0 +1,138 @@
+import threading
+from concurrent.futures import ThreadPoolExecutor, TimeoutError
+from contextlib import contextmanager
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+from qca import Forward, Managed
+from qca.common._streaming import Stream
+
+
+@contextmanager
+def local_server(*, hanging):
+    release = threading.Event()
+    requests = []
+
+    class Handler(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
+        def do_GET(self):
+            requests.append((self.path, self.client_address))
+            sse = self.path.endswith("/stream")
+            body = b'data: {"type":"agent.message"}\n\n' if sse else b'{"data":[],"has_more":false}'
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream" if sse else "application/json")
+            if sse and hanging:
+                self.send_header("Transfer-Encoding", "chunked")
+                self.end_headers()
+                self.wfile.flush()
+                release.wait(timeout=10)
+                self.close_connection = True
+            else:
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+                self.wfile.flush()
+
+        def log_message(self, *args):
+            pass
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{server.server_port}", requests, release
+    finally:
+        release.set()
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+        assert not thread.is_alive()
+
+
+@pytest.mark.parametrize("mode", ["forward", "managed"])
+@pytest.mark.parametrize("scope", ["session", "thread"])
+def test_sync_close_wakes_real_http_read_without_reconnecting(mode, scope):
+    reading = threading.Event()
+
+    def trace(event, info):
+        if event == "http11.receive_response_body.started":
+            reading.set()
+
+    def request_hook(request):
+        request.extensions["trace"] = trace
+
+    with local_server(hanging=True) as (base_url, requests, release):
+        http_client = httpx.Client(trust_env=False, event_hooks={"request": [request_hook]})
+        client = (Forward if mode == "forward" else Managed)(
+            pat="test", base_url=base_url, http_client=http_client, timeout=3, max_retries=0
+        )
+        stream = (
+            client.sessions.events.resumable_stream("session")
+            if scope == "session"
+            else client.sessions.threads.events.resumable_stream("thread", session_id="session")
+        )
+        executor = ThreadPoolExecutor(max_workers=1)
+        pending = executor.submit(next, stream)
+        try:
+            assert reading.wait(timeout=2)
+            with pytest.raises(TimeoutError):
+                pending.result(timeout=0.1)
+            stream.close()
+            stream.close()
+            with pytest.raises(StopIteration):
+                pending.result(timeout=1)
+            # A cancelled SSE connection must not poison or close the shared client.
+            assert client.models.list().data == []
+            assert len([path for path, _ in requests if path.endswith("/stream")]) == 1
+        finally:
+            release.set()
+            stream.close()
+            executor.shutdown(wait=True)
+            client.close()
+
+
+def test_fully_consumed_stream_can_reuse_connection_after_repeated_close():
+    with local_server(hanging=False) as (base_url, requests, _):
+        with httpx.Client(base_url=base_url, trust_env=False) as client:
+            response = client.send(client.build_request("GET", "/stream"), stream=True)
+            stream = Stream(response, dict)
+            assert list(stream) == [{"type": "agent.message"}]
+            stream.close()
+            assert client.get("/models").status_code == 200
+            stream.close()
+            assert client.get("/models").status_code == 200
+            assert len({address for _, address in requests}) == 1
+
+
+@pytest.mark.parametrize("version", [b"HTTP/2", None])
+def test_close_does_not_inspect_shared_or_unknown_protocol_socket(version):
+    def get_extra_info(name):
+        pytest.fail("must not access a shared or unknown protocol socket")
+
+    response = httpx.Response(
+        200,
+        stream=httpx.ByteStream(b""),
+        extensions={"http_version": version, "network_stream": SimpleNamespace(get_extra_info=get_extra_info)},
+    )
+    Stream(response, dict).close()
+    assert response.is_closed
+
+
+@pytest.mark.parametrize("unsupported", [None, OSError, NotImplementedError])
+def test_custom_transport_without_socket_still_closes(unsupported):
+    def get_extra_info(name):
+        if unsupported:
+            raise unsupported("socket metadata is unavailable")
+        return None
+
+    response = httpx.Response(
+        200,
+        stream=httpx.ByteStream(b""),
+        extensions={"http_version": b"HTTP/1.1", "network_stream": SimpleNamespace(get_extra_info=get_extra_info)},
+    )
+    Stream(response, dict).close()
+    assert response.is_closed