feat(streaming): add resumable session event streams

Author: moonyue-w

README.md

@@ -158,21 +158,29 @@ Reuse one `session_id` for the whole conversation, and reuse one idempotency key
 
 ## Resuming a stream
 
-The SDK does not reconnect a dropped stream on its own. Persist `stream.last_event_id` and pass it back as `last_event_id` on the next call; the server replays from there, so you never have to resend a message it already accepted.
+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.
 
 ```python
-from qca import APIConnectionError
+with client.sessions.events.resumable_stream(
+    session_id,
+    last_event_id=saved_event_id,
+    event_deltas=["agent.message"],
+) as stream:
+    for event in stream:
+        saved_event_id = stream.last_event_id
+        print(event.type)
+```
 
-last_event_id = None
-try:
-    with client.sessions.events.stream(session_id, last_event_id=last_event_id) as stream:
-        for event in stream:
-            last_event_id = stream.last_event_id
-            print(event.type)
-except APIConnectionError:
-    pass  # reconnect with the last_event_id recorded above
+The async form is a native async iterator and context manager; unlike `stream()`, constructing it does not require `await`:
+
+```python
+async with client.sessions.events.resumable_stream(session_id) as stream:
+    async for event in stream:
+        print(event.type)
 ```
 
+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.
+
 Events are also readable after the fact through `client.sessions.events.list(session_id)`, which paginates like any other list method.
 
 ## Handling errors

src/qca/common/_resumable_streaming.py

@@ -0,0 +1,244 @@
+from __future__ import annotations
+
+import random
+import threading
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
+from time import monotonic
+from typing import Generic, TypeVar
+
+import anyio
+import httpx
+
+from ._exceptions import APIConnectionError, APIStatusError
+from ._streaming import AsyncStream, Stream
+
+T = TypeVar("T")
+
+_INITIAL_BACKOFF = 0.5
+_MAX_BACKOFF = 10.0
+_HEALTHY_RESET_AFTER = 5.0
+_TERMINAL_EVENT_TYPES = frozenset({"session.status_terminated", "session.deleted"})
+
+Retryable = Callable[[httpx.Request, httpx.Response | None], bool]
+
+
+def _is_terminal_event(event: object) -> bool:
+    return getattr(event, "type", None) in _TERMINAL_EVENT_TYPES
+
+
+class ResumableStream(Generic[T], Iterator[T]):
+    def __init__(
+        self,
+        *,
+        open_stream: Callable[[str | None], Stream[T]],
+        retryable: Retryable,
+        last_event_id: str | None,
+    ) -> None:
+        self._open_stream = open_stream
+        self._retryable = retryable
+        self._last_event_id = last_event_id
+        self._backoff = _INITIAL_BACKOFF
+        self._closed = threading.Event()
+        self._stream: Stream[T] | None = None
+        self._iterator = self._iter_events()
+
+    @property
+    def last_event_id(self) -> str | None:
+        return self._last_event_id
+
+    def __next__(self) -> T:
+        return next(self._iterator)
+
+    def _iter_events(self) -> Iterator[T]:
+        while not self._closed.is_set():
+            connected_at: float | None = None
+            connected_for: float | None = None
+            stream: Stream[T] | None = None
+            try:
+                stream = self._open_stream(self._last_event_id)
+                self._stream = stream
+                connected_at = monotonic()
+                if self._closed.is_set():
+                    return
+                for event in stream:
+                    if self._closed.is_set():
+                        return
+                    if stream.last_event_id:
+                        self._last_event_id = stream.last_event_id
+                    terminal = _is_terminal_event(event)
+                    yield event
+                    if terminal:
+                        return
+            except APIConnectionError as exc:
+                if self._closed.is_set():
+                    return
+                if not self._retryable(exc.request, None):
+                    raise
+            except APIStatusError as exc:
+                if self._closed.is_set():
+                    return
+                if not self._retryable(exc.request, exc.response):
+                    raise
+            finally:
+                if connected_at is not None:
+                    connected_for = monotonic() - connected_at
+                if stream is not None:
+                    stream.close()
+                self._stream = None
+
+            if self._closed.is_set():
+                return
+            if connected_for is not None and connected_for >= _HEALTHY_RESET_AFTER:
+                self._backoff = _INITIAL_BACKOFF
+            delay = random.uniform(self._backoff / 2, self._backoff)
+            self._backoff = min(self._backoff * 2, _MAX_BACKOFF)
+            if self._closed.wait(delay):
+                return
+
+    def close(self) -> None:
+        self._closed.set()
+        if self._stream is not None:
+            self._stream.close()
+
+    def __enter__(self) -> ResumableStream[T]:
+        return self
+
+    def __exit__(self, *_: object) -> None:
+        self.close()
+
+
+class AsyncResumableStream(Generic[T], AsyncIterator[T]):
+    def __init__(
+        self,
+        *,
+        open_stream: Callable[[str | None], Awaitable[AsyncStream[T]]],
+        retryable: Retryable,
+        last_event_id: str | None,
+    ) -> None:
+        self._open_stream = open_stream
+        self._retryable = retryable
+        self._last_event_id = last_event_id
+        self._backoff = _INITIAL_BACKOFF
+        self._closed = False
+        self._close_event: anyio.Event | None = None
+        self._cancel_scope: anyio.CancelScope | None = None
+        self._stream: AsyncStream[T] | None = None
+        self._iterator = self._iter_events()
+
+    @property
+    def last_event_id(self) -> str | None:
+        return self._last_event_id
+
+    async def __anext__(self) -> T:
+        return await self._iterator.__anext__()
+
+    async def _open_next_stream(self) -> AsyncStream[T] | None:
+        stream: AsyncStream[T] | None = None
+        with anyio.CancelScope() as cancel_scope:
+            self._cancel_scope = cancel_scope
+            if self._closed:
+                cancel_scope.cancel()
+            try:
+                stream = await self._open_stream(self._last_event_id)
+            finally:
+                if self._cancel_scope is cancel_scope:
+                    self._cancel_scope = None
+        return stream
+
+    async def _next_event(self, stream: AsyncStream[T]) -> tuple[bool, T | None]:
+        event: T | None = None
+        exhausted = False
+        with anyio.CancelScope() as cancel_scope:
+            self._cancel_scope = cancel_scope
+            if self._closed:
+                cancel_scope.cancel()
+            try:
+                event = await stream.__anext__()
+            except StopAsyncIteration:
+                exhausted = True
+            finally:
+                if self._cancel_scope is cancel_scope:
+                    self._cancel_scope = None
+        return exhausted, event
+
+    async def _close_stream(self, stream: AsyncStream[T]) -> None:
+        if self._stream is not stream:
+            return
+        self._stream = None
+        await stream.close()
+
+    async def _iter_events(self) -> AsyncIterator[T]:
+        while not self._closed:
+            connected_at: float | None = None
+            connected_for: float | None = None
+            stream: AsyncStream[T] | None = None
+            try:
+                stream = await self._open_next_stream()
+                if stream is None:
+                    return
+                self._stream = stream
+                connected_at = monotonic()
+                if self._closed:
+                    return
+                while not self._closed:
+                    exhausted, event = await self._next_event(stream)
+                    if self._closed:
+                        return
+                    if exhausted:
+                        break
+                    if event is None:
+                        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)
+                    yield event
+                    if terminal:
+                        return
+            except APIConnectionError as exc:
+                if self._closed:
+                    return
+                if not self._retryable(exc.request, None):
+                    raise
+            except APIStatusError as exc:
+                if self._closed:
+                    return
+                if not self._retryable(exc.request, exc.response):
+                    raise
+            finally:
+                if connected_at is not None:
+                    connected_for = monotonic() - connected_at
+                if stream is not None:
+                    with anyio.CancelScope(shield=True):
+                        await self._close_stream(stream)
+
+            if self._closed:
+                return
+            if connected_for is not None and connected_for >= _HEALTHY_RESET_AFTER:
+                self._backoff = _INITIAL_BACKOFF
+            delay = random.uniform(self._backoff / 2, self._backoff)
+            self._backoff = min(self._backoff * 2, _MAX_BACKOFF)
+            close_event = anyio.Event()
+            self._close_event = close_event
+            if self._closed:
+                close_event.set()
+            with anyio.move_on_after(delay):
+                await close_event.wait()
+            self._close_event = None
+
+    async def close(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if self._close_event is not None:
+            self._close_event.set()
+        if self._cancel_scope is not None:
+            self._cancel_scope.cancel()
+        if self._stream is not None:
+            with anyio.CancelScope(shield=True):
+                await self._close_stream(self._stream)
+
+    async def __aenter__(self) -> AsyncResumableStream[T]:
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        await self.close()

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

@@ -5,6 +5,7 @@
 import httpx
 
 from qca.common._resource import AsyncAPIResource, SyncAPIResource
+from qca.common._resumable_streaming import AsyncResumableStream, ResumableStream
 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
@@ -16,6 +17,17 @@
 __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,
@@ -110,6 +122,40 @@ def stream(
         )
         return self._client.request("GET", _path, cast_to=SessionEvent, options=options, stream=True)
 
+    def resumable_stream(
+        self,
+        session_id: str,
+        *,
+        event_deltas: Union[List[str], None, NotGiven] = NOT_GIVEN,
+        include_tool_calls: Union[bool, None, NotGiven] = NOT_GIVEN,
+        include_thinking: Union[bool, 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[SessionEvent]:
+        """Continuously reconnect GET /sessions/{session_id}/events/stream."""
+        cursor, headers = _resumable_options(last_event_id, extra_headers)
+        deltas = list(event_deltas) if isinstance(event_deltas, list) else event_deltas
+        query = dict(extra_query or {})
+        body = dict(extra_body or {})
+        return ResumableStream(
+            open_stream=lambda next_cursor: self.stream(
+                session_id,
+                event_deltas=deltas,
+                include_tool_calls=include_tool_calls,
+                include_thinking=include_thinking,
+                last_event_id=next_cursor,
+                extra_headers=headers,
+                extra_query=query,
+                extra_body=body,
+                timeout=timeout,
+            ),
+            retryable=self._client._retryable,
+            last_event_id=cursor,
+        )
+
 
 class AsyncEvents(AsyncAPIResource):
     def list(
@@ -206,3 +252,37 @@ async def stream(
             timeout=timeout,
         )
         return await self._client.request("GET", _path, cast_to=SessionEvent, options=options, stream=True)
+
+    def resumable_stream(
+        self,
+        session_id: str,
+        *,
+        event_deltas: Union[List[str], None, NotGiven] = NOT_GIVEN,
+        include_tool_calls: Union[bool, None, NotGiven] = NOT_GIVEN,
+        include_thinking: Union[bool, 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[SessionEvent]:
+        """Continuously reconnect GET /sessions/{session_id}/events/stream."""
+        cursor, headers = _resumable_options(last_event_id, extra_headers)
+        deltas = list(event_deltas) if isinstance(event_deltas, list) else event_deltas
+        query = dict(extra_query or {})
+        body = dict(extra_body or {})
+        return AsyncResumableStream(
+            open_stream=lambda next_cursor: self.stream(
+                session_id,
+                event_deltas=deltas,
+                include_tool_calls=include_tool_calls,
+                include_thinking=include_thinking,
+                last_event_id=next_cursor,
+                extra_headers=headers,
+                extra_query=query,
+                extra_body=body,
+                timeout=timeout,
+            ),
+            retryable=self._client._retryable,
+            last_event_id=cursor,
+        )