feat(streaming): resume thread subscriptions after disconnects (#36)

Author: moonyue-w

CHANGELOG.md

@@ -5,6 +5,11 @@ existing `0.1.0` release; earlier development prereleases are not listed.
 
 ## [Unreleased]
 
+### Added
+
+- 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.
+
 ## [0.13.0]
 
 ### Added

README.md

@@ -492,6 +492,38 @@ with client.sessions.events.stream(
 
 Reuse one `session_id` for the whole conversation, and reuse one idempotency key across HTTP retries of the same logical message. `event_deltas` opts into incremental events for the listed types; those previews are not deduplicated, and the final event repeats the complete content, so render deltas but treat the final event as the source of truth. An idle status does not by itself mean success — the session may be waiting for a confirmation or have reached its budget, so check `stop_reason` and the final reply.
 
+
+### Thread event resumption
+
+Forward and Managed also support automatic resumption for a single Thread. Use the
+same retry and cancellation behavior as Session streams, with a thread-scoped
+`Last-Event-ID` cursor:
+
+```python
+# Forward and Managed use the same call shape.
+with client.sessions.threads.events.resumable_stream(
+    thread_id, session_id=session_id, last_event_id=after_id,
+) as stream:
+    for event in stream:
+        print(event.type)
+
+# AsyncForward / AsyncManaged
+async with async_client.sessions.threads.events.resumable_stream(
+    thread_id, session_id=session_id, last_event_id=after_id,
+) as stream:
+    async for event in stream:
+        print(event.type)
+```
+
+`session.thread_status_idle` ends only the current turn, so the subscription stays
+open. `session.thread_status_terminated`, `session.status_terminated`, or
+`session.deleted` ends it normally. A non-retryable error (including an invalid
+cursor or missing thread) is returned to the caller; the SDK never clears the
+cursor to silently start over. The existing one-shot Thread stream remains
+available. See the [Forward Thread stream](https://docs.qoder.com/cloud-agents/api/forward/sessions/stream-thread-events)
+and [Managed Thread stream](https://docs.qoder.com/cloud-agents/api/sessions/threads/events/stream)
+contracts for cursor validity.
+
 ## Resuming a stream
 
 The low-level `stream()` method exposes `stream.last_event_id` but does not reconnect. Use the handwritten `resumable_stream()` helper to reconnect automatically after EOF, timeouts, transient transport failures, and retryable HTTP statuses. It checkpoints only complete events and sends the latest checkpoint as `Last-Event-ID`; close it with a context manager when finished.

docs/api/reference.md

@@ -1186,8 +1186,12 @@ class ResumableStream(Generic[T], Iterator[T])
 #### \_\_init\_\_
 
 ```python
-def __init__(*, open_stream: Callable[[str | None], Stream[T]],
-             retryable: Retryable, last_event_id: str | None) -> None
+def __init__(
+        *,
+        open_stream: Callable[[str | None], Stream[T]],
+        retryable: Retryable,
+        last_event_id: str | None,
+        is_terminal: Callable[[object], bool] = _is_terminal_event) -> None
 ```
 
 [[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/common/_resumable_streaming.py)
@@ -1258,8 +1262,12 @@ class AsyncResumableStream(Generic[T], AsyncIterator[T])
 #### \_\_init\_\_
 
 ```python
-def __init__(*, open_stream: Callable[[str | None], Awaitable[AsyncStream[T]]],
-             retryable: Retryable, last_event_id: str | None) -> None
+def __init__(
+        *,
+        open_stream: Callable[[str | None], Awaitable[AsyncStream[T]]],
+        retryable: Retryable,
+        last_event_id: str | None,
+        is_terminal: Callable[[object], bool] = _is_terminal_event) -> None
 ```
 
 [[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/common/_resumable_streaming.py)
@@ -8799,6 +8807,27 @@ def stream(
 
 GET /sessions/{session_id}/threads/{thread_id}/stream.
 
+<a id="qca.forward.resources.sessions.threads.events.Events.resumable_stream"></a>
+
+#### resumable\_stream
+
+```python
+def resumable_stream(
+    thread_id: str,
+    *,
+    session_id: str,
+    last_event_id: Union[str, None, NotGiven] = NOT_GIVEN,
+    extra_headers: Dict[str, str] | None = None,
+    extra_query: Dict[str, Any] | None = None,
+    extra_body: Dict[str, Any] | None = None,
+    timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN
+) -> ResumableStream[SessionEvent]
+```
+
+[[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/forward/resources/sessions/threads/events.py)
+
+Continuously reconnect GET /sessions/{session_id}/threads/{thread_id}/stream.
+
 <a id="qca.forward.resources.sessions.threads.events.AsyncEvents"></a>
 
 ## AsyncEvents
@@ -8853,6 +8882,27 @@ async def stream(
 
 GET /sessions/{session_id}/threads/{thread_id}/stream.
 
+<a id="qca.forward.resources.sessions.threads.events.AsyncEvents.resumable_stream"></a>
+
+#### resumable\_stream
+
+```python
+def resumable_stream(
+    thread_id: str,
+    *,
+    session_id: str,
+    last_event_id: Union[str, None, NotGiven] = NOT_GIVEN,
+    extra_headers: Dict[str, str] | None = None,
+    extra_query: Dict[str, Any] | None = None,
+    extra_body: Dict[str, Any] | None = None,
+    timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN
+) -> AsyncResumableStream[SessionEvent]
+```
+
+[[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/forward/resources/sessions/threads/events.py)
+
+Continuously reconnect GET /sessions/{session_id}/threads/{thread_id}/stream.
+
 <a id="qca.forward.resources.sessions.threads.threads"></a>
 
 # qca.forward.resources.sessions.threads.threads
@@ -25171,6 +25221,7 @@ def stream(
     workspace_id: Union[str, None, NotGiven] = NOT_GIVEN,
     event_deltas: Union[List[Literal["agent.message", "agent.thinking"]], None,
                         NotGiven] = NOT_GIVEN,
+    last_event_id: Union[str, None, NotGiven] = NOT_GIVEN,
     betas: Union[List[str], None, NotGiven] = NOT_GIVEN,
     extra_headers: Dict[str, str] | None = None,
     extra_query: Dict[str, Any] | None = None,
@@ -25183,6 +25234,31 @@ def stream(
 
 GET /sessions/{session_id}/threads/{thread_id}/stream.
 
+<a id="qca.managed.resources.sessions.threads.events.Events.resumable_stream"></a>
+
+#### resumable\_stream
+
+```python
+def resumable_stream(
+    thread_id: str,
+    *,
+    session_id: str,
+    workspace_id: Union[str, None, NotGiven] = NOT_GIVEN,
+    event_deltas: Union[List[Literal["agent.message", "agent.thinking"]], None,
+                        NotGiven] = NOT_GIVEN,
+    betas: Union[List[str], None, NotGiven] = NOT_GIVEN,
+    last_event_id: Union[str, None, NotGiven] = NOT_GIVEN,
+    extra_headers: Dict[str, str] | None = None,
+    extra_query: Dict[str, Any] | None = None,
+    extra_body: Dict[str, Any] | None = None,
+    timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN
+) -> ResumableStream[StreamSessionThreadEventsUnion]
+```
+
+[[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/managed/resources/sessions/threads/events.py)
+
+Continuously reconnect GET /sessions/{session_id}/threads/{thread_id}/stream.
+
 <a id="qca.managed.resources.sessions.threads.events.AsyncEvents"></a>
 
 ## AsyncEvents
@@ -25231,6 +25307,7 @@ async def stream(
     workspace_id: Union[str, None, NotGiven] = NOT_GIVEN,
     event_deltas: Union[List[Literal["agent.message", "agent.thinking"]], None,
                         NotGiven] = NOT_GIVEN,
+    last_event_id: Union[str, None, NotGiven] = NOT_GIVEN,
     betas: Union[List[str], None, NotGiven] = NOT_GIVEN,
     extra_headers: Dict[str, str] | None = None,
     extra_query: Dict[str, Any] | None = None,
@@ -25243,6 +25320,31 @@ async def stream(
 
 GET /sessions/{session_id}/threads/{thread_id}/stream.
 
+<a id="qca.managed.resources.sessions.threads.events.AsyncEvents.resumable_stream"></a>
+
+#### resumable\_stream
+
+```python
+def resumable_stream(
+    thread_id: str,
+    *,
+    session_id: str,
+    workspace_id: Union[str, None, NotGiven] = NOT_GIVEN,
+    event_deltas: Union[List[Literal["agent.message", "agent.thinking"]], None,
+                        NotGiven] = NOT_GIVEN,
+    betas: Union[List[str], None, NotGiven] = NOT_GIVEN,
+    last_event_id: Union[str, None, NotGiven] = NOT_GIVEN,
+    extra_headers: Dict[str, str] | None = None,
+    extra_query: Dict[str, Any] | None = None,
+    extra_body: Dict[str, Any] | None = None,
+    timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN
+) -> AsyncResumableStream[StreamSessionThreadEventsUnion]
+```
+
+[[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/managed/resources/sessions/threads/events.py)
+
+Continuously reconnect GET /sessions/{session_id}/threads/{thread_id}/stream.
+
 <a id="qca.managed.resources.sessions.threads.threads"></a>
 
 # qca.managed.resources.sessions.threads.threads
@@ -35945,6 +36047,10 @@ class SessionThreadEventStreamParams(TypedDict)
 
 [[view_source]](https://github.com/QoderAI/qoder-cloud-agents-sdk-python/blob/main/src/qca/managed/types/session_thread_event_stream_params.py)
 
+<a id="qca.managed.types.session_thread_event_stream_params.SessionThreadEventStreamParams.last_event_id"></a>
+
+#### last\_event\_id
+
 <a id="qca.managed.types.session_thread_event_stream_params.SessionThreadEventStreamParams.workspace_id"></a>
 
 #### workspace\_id

src/qca/common/_resumable_streaming.py

@@ -11,6 +11,7 @@
 
 from ._exceptions import APIConnectionError, APIStatusError
 from ._streaming import AsyncStream, Stream
+from ._types import NotGiven
 
 T = TypeVar("T")
 
@@ -26,14 +27,31 @@ def _is_terminal_event(event: object) -> bool:
     return getattr(event, "type", None) in _TERMINAL_EVENT_TYPES
 
 
+def _resumable_options(
+    last_event_id: str | None | NotGiven, extra_headers: dict[str, str] | None
+) -> tuple[str | None, dict[str, str]]:
+    cursor = None if isinstance(last_event_id, NotGiven) else last_event_id
+    headers = dict(extra_headers or {})
+    for name in tuple(headers):
+        if name.lower() == "last-event-id":
+            cursor = headers.pop(name)
+    return cursor, headers
+
+
+def _is_terminal_thread_event(event: object) -> bool:
+    return _is_terminal_event(event) or getattr(event, "type", None) == "session.thread_status_terminated"
+
+
 class ResumableStream(Generic[T], Iterator[T]):
     def __init__(
         self,
         *,
         open_stream: Callable[[str | None], Stream[T]],
         retryable: Retryable,
         last_event_id: str | None,
+        is_terminal: Callable[[object], bool] = _is_terminal_event,
     ) -> None:
+        self._is_terminal = is_terminal
         self._open_stream = open_stream
         self._retryable = retryable
         self._last_event_id = last_event_id
@@ -65,7 +83,7 @@ def _iter_events(self) -> Iterator[T]:
                         return
                     if stream.last_event_id:
                         self._last_event_id = stream.last_event_id
-                    terminal = _is_terminal_event(event)
+                    terminal = self._is_terminal(event)
                     yield event
                     if terminal:
                         return
@@ -114,7 +132,9 @@ def __init__(
         open_stream: Callable[[str | None], Awaitable[AsyncStream[T]]],
         retryable: Retryable,
         last_event_id: str | None,
+        is_terminal: Callable[[object], bool] = _is_terminal_event,
     ) -> None:
+        self._is_terminal = is_terminal
         self._open_stream = open_stream
         self._retryable = retryable
         self._last_event_id = last_event_id
@@ -190,7 +210,7 @@ async def _iter_events(self) -> AsyncIterator[T]:
                         raise RuntimeError("stream read was interrupted without closing")
                     if stream.last_event_id:
                         self._last_event_id = stream.last_event_id
-                    terminal = _is_terminal_event(event)
+                    terminal = self._is_terminal(event)
                     yield event
                     if terminal:
                         return

src/qca/forward/resources/sessions/events.py

@@ -5,7 +5,7 @@
 import httpx
 
 from qca.common._resource import AsyncAPIResource, SyncAPIResource
-from qca.common._resumable_streaming import AsyncResumableStream, ResumableStream
+from qca.common._resumable_streaming import AsyncResumableStream, ResumableStream, _resumable_options
 from qca.common._streaming import AsyncStream, Stream
 from qca.common._types import NOT_GIVEN, NotGiven
 from qca.common._utils import make_request_options, path_template
@@ -17,17 +17,6 @@
 __all__ = ["Events", "AsyncEvents"]
 
 
-def _resumable_options(
-    last_event_id: Union[str, None, NotGiven], extra_headers: Dict[str, str] | None
-) -> tuple[str | None, Dict[str, str]]:
-    cursor = None if isinstance(last_event_id, NotGiven) else last_event_id
-    headers = dict(extra_headers or {})
-    for name in tuple(headers):
-        if name.lower() == "last-event-id":
-            cursor = headers.pop(name)
-    return cursor, headers
-
-
 class Events(SyncAPIResource):
     def list(
         self,