diff --git a/CHANGELOG.md b/CHANGELOG.md index efaf39b..4d7094e 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 17be1d9..143a9c6 100644 --- a/README.md +++ b/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. diff --git a/docs/api/reference.md b/docs/api/reference.md index 4bd5870..701bce4 100644 --- a/docs/api/reference.md +++ b/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. + + +#### 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. + ## AsyncEvents @@ -8853,6 +8882,27 @@ async def stream( GET /sessions/{session_id}/threads/{thread_id}/stream. + + +#### 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. + # 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. + + +#### 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. + ## 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. + + +#### 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. + # 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) + + +#### last\_event\_id + #### workspace\_id diff --git a/src/qca/common/_resumable_streaming.py b/src/qca/common/_resumable_streaming.py index eeb0ff4..c896caa 100644 --- a/src/qca/common/_resumable_streaming.py +++ b/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,6 +27,21 @@ 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, @@ -33,7 +49,9 @@ 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: + 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 diff --git a/src/qca/forward/resources/sessions/events.py b/src/qca/forward/resources/sessions/events.py index 242b83b..2d2ecc0 100644 --- a/src/qca/forward/resources/sessions/events.py +++ b/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, diff --git a/src/qca/forward/resources/sessions/threads/events.py b/src/qca/forward/resources/sessions/threads/events.py index 80e85a9..591fcab 100644 --- a/src/qca/forward/resources/sessions/threads/events.py +++ b/src/qca/forward/resources/sessions/threads/events.py @@ -5,6 +5,12 @@ import httpx from qca.common._resource import AsyncAPIResource, SyncAPIResource +from qca.common._resumable_streaming import ( + AsyncResumableStream, + ResumableStream, + _is_terminal_thread_event, + _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 @@ -69,6 +75,36 @@ def stream( ) return self._client.request("GET", _path, cast_to=SessionEvent, options=options, stream=True) + def resumable_stream( + self, + 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]: + """Continuously reconnect GET /sessions/{session_id}/threads/{thread_id}/stream.""" + cursor, headers = _resumable_options(last_event_id, extra_headers) + query = dict(extra_query or {}) + body = dict(extra_body or {}) + return ResumableStream( + open_stream=lambda next_cursor: self.stream( + thread_id, + session_id=session_id, + last_event_id=next_cursor, + extra_headers=headers, + extra_query=query, + extra_body=body, + timeout=timeout, + ), + retryable=self._client._retryable, + last_event_id=cursor, + is_terminal=_is_terminal_thread_event, + ) + class AsyncEvents(AsyncAPIResource): def list( @@ -126,3 +162,33 @@ async def stream( timeout=timeout, ) return await self._client.request("GET", _path, cast_to=SessionEvent, options=options, stream=True) + + def resumable_stream( + self, + 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]: + """Continuously reconnect GET /sessions/{session_id}/threads/{thread_id}/stream.""" + cursor, headers = _resumable_options(last_event_id, extra_headers) + query = dict(extra_query or {}) + body = dict(extra_body or {}) + return AsyncResumableStream( + open_stream=lambda next_cursor: self.stream( + thread_id, + session_id=session_id, + last_event_id=next_cursor, + extra_headers=headers, + extra_query=query, + extra_body=body, + timeout=timeout, + ), + retryable=self._client._retryable, + last_event_id=cursor, + is_terminal=_is_terminal_thread_event, + ) diff --git a/src/qca/managed/resources/sessions/events.py b/src/qca/managed/resources/sessions/events.py index bed97ed..7827f3d 100644 --- a/src/qca/managed/resources/sessions/events.py +++ b/src/qca/managed/resources/sessions/events.py @@ -6,7 +6,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 @@ -25,17 +25,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, diff --git a/src/qca/managed/resources/sessions/threads/events.py b/src/qca/managed/resources/sessions/threads/events.py index 9c5d316..44ff7c3 100644 --- a/src/qca/managed/resources/sessions/threads/events.py +++ b/src/qca/managed/resources/sessions/threads/events.py @@ -5,6 +5,12 @@ import httpx from qca.common._resource import AsyncAPIResource, SyncAPIResource +from qca.common._resumable_streaming import ( + AsyncResumableStream, + ResumableStream, + _is_terminal_thread_event, + _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 @@ -54,6 +60,7 @@ def stream( session_id: str, 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, @@ -67,7 +74,7 @@ def stream( options = make_request_options( body={}, query={"event_deltas[]": event_deltas}, - headers={"qoder-workspace-id": workspace_id, "x-qoder-beta": betas}, + headers={"qoder-workspace-id": workspace_id, "x-qoder-beta": betas, "Last-Event-ID": last_event_id}, extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, @@ -75,6 +82,44 @@ def stream( ) return self._client.request("GET", _path, cast_to=StreamSessionThreadEventsUnion, options=options, stream=True) + def resumable_stream( + self, + 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]: + """Continuously reconnect GET /sessions/{session_id}/threads/{thread_id}/stream.""" + cursor, headers = _resumable_options(last_event_id, extra_headers) + deltas = list(event_deltas) if isinstance(event_deltas, list) else event_deltas + beta_values = list(betas) if isinstance(betas, list) else betas + query = dict(extra_query or {}) + body = dict(extra_body or {}) + return ResumableStream( + open_stream=lambda next_cursor: self.stream( + thread_id, + session_id=session_id, + workspace_id=workspace_id, + event_deltas=deltas, + betas=beta_values, + last_event_id=next_cursor, + extra_headers=headers, + extra_query=query, + extra_body=body, + timeout=timeout, + ), + retryable=self._client._retryable, + last_event_id=cursor, + is_terminal=_is_terminal_thread_event, + ) + class AsyncEvents(AsyncAPIResource): def list( @@ -117,6 +162,7 @@ async def stream( session_id: str, 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, @@ -130,7 +176,7 @@ async def stream( options = make_request_options( body={}, query={"event_deltas[]": event_deltas}, - headers={"qoder-workspace-id": workspace_id, "x-qoder-beta": betas}, + headers={"qoder-workspace-id": workspace_id, "x-qoder-beta": betas, "Last-Event-ID": last_event_id}, extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, @@ -139,3 +185,41 @@ async def stream( return await self._client.request( "GET", _path, cast_to=StreamSessionThreadEventsUnion, options=options, stream=True ) + + def resumable_stream( + self, + 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]: + """Continuously reconnect GET /sessions/{session_id}/threads/{thread_id}/stream.""" + cursor, headers = _resumable_options(last_event_id, extra_headers) + deltas = list(event_deltas) if isinstance(event_deltas, list) else event_deltas + beta_values = list(betas) if isinstance(betas, list) else betas + query = dict(extra_query or {}) + body = dict(extra_body or {}) + return AsyncResumableStream( + open_stream=lambda next_cursor: self.stream( + thread_id, + session_id=session_id, + workspace_id=workspace_id, + event_deltas=deltas, + betas=beta_values, + last_event_id=next_cursor, + extra_headers=headers, + extra_query=query, + extra_body=body, + timeout=timeout, + ), + retryable=self._client._retryable, + last_event_id=cursor, + is_terminal=_is_terminal_thread_event, + ) diff --git a/src/qca/managed/types/session_thread_event_stream_params.py b/src/qca/managed/types/session_thread_event_stream_params.py index 65a7bb4..3c6f157 100644 --- a/src/qca/managed/types/session_thread_event_stream_params.py +++ b/src/qca/managed/types/session_thread_event_stream_params.py @@ -8,6 +8,7 @@ class SessionThreadEventStreamParams(TypedDict, total=False): + last_event_id: Optional[str] workspace_id: Optional[str] event_deltas: List[Literal["agent.message", "agent.thinking"]] betas: List[str] diff --git a/tests/_surface.py b/tests/_surface.py index ab5aa38..bf60dab 100644 --- a/tests/_surface.py +++ b/tests/_surface.py @@ -25,7 +25,11 @@ DECLARATION = re.compile(r"^(GET|POST|PUT|PATCH|DELETE) (/[^\s.]*)\.") CONTROL_PARAMS = ("extra_headers", "extra_query", "extra_body", "timeout") -HANDWRITTEN_RESOURCE_METHODS = {"sessions.events.resumable_stream", "realtime.connect"} +HANDWRITTEN_RESOURCE_METHODS = { + "sessions.events.resumable_stream", + "sessions.threads.events.resumable_stream", + "realtime.connect", +} CONTRACT_FIXTURE = json.loads((Path(__file__).parent / "fixtures" / "api-contracts.json").read_text()) CONTRACT_ROWS = CONTRACT_FIXTURE["contracts"] CONTRACTS = {row["id"]: row for row in CONTRACT_ROWS} diff --git a/tests/test_resource_surface.py b/tests/test_resource_surface.py index 7f6e98b..0c2de0a 100644 --- a/tests/test_resource_surface.py +++ b/tests/test_resource_surface.py @@ -125,7 +125,9 @@ async def test_sync_and_async_clients_expose_the_same_methods(mode): try: assert method_names(client) == method_names(async_client) assert {e.attribute for e in ENDPOINTS if e.mode == mode} | ( - HANDWRITTEN_RESOURCE_METHODS if mode == "forward" else {"sessions.events.resumable_stream"} + HANDWRITTEN_RESOURCE_METHODS + if mode == "forward" + else {"sessions.events.resumable_stream", "sessions.threads.events.resumable_stream"} ) == method_names(client) finally: await close_client(client, False) diff --git a/tests/test_resumable_streaming.py b/tests/test_resumable_streaming.py index 308ed34..5881fec 100644 --- a/tests/test_resumable_streaming.py +++ b/tests/test_resumable_streaming.py @@ -92,6 +92,12 @@ def make_client(mode: str, async_: bool, handler: Any): ) +def open_resumable(client, scope, **kwargs): + if scope == "thread": + return client.sessions.threads.events.resumable_stream("thread", session_id="session", **kwargs) + return client.sessions.events.resumable_stream("session", **kwargs) + + async def close_client(client: Any, async_: bool) -> None: if async_: await client.close() @@ -101,7 +107,8 @@ async def close_client(client: Any, async_: bool) -> None: @pytest.mark.parametrize("mode", ["forward", "managed"]) @pytest.mark.parametrize("async_", [False, True], ids=["sync", "async"]) -async def test_resumable_stream_reconnects_over_real_http(monkeypatch, mode, async_): +@pytest.mark.parametrize("scope", ["session", "thread"]) +async def test_resumable_stream_reconnects_over_real_http(monkeypatch, mode, async_, scope): monkeypatch.setattr("qca.common._resumable_streaming.random.uniform", lambda *_: 0) prefix = "forward" if mode == "forward" else "cloud" @@ -113,7 +120,7 @@ async def test_resumable_stream_reconnects_over_real_http(monkeypatch, mode, asy max_retries=0, http_client=http_client, ) - stream = client.sessions.events.resumable_stream("session") + stream = open_resumable(client, scope) try: if async_: events = [await anext(stream), await anext(stream)] @@ -127,7 +134,8 @@ async def test_resumable_stream_reconnects_over_real_http(monkeypatch, mode, asy assert [event.id for event in events] == ["evt-1", "evt-2"] assert [event.type for event in events] == ["event_delta", "session.status_terminated"] assert stream.last_event_id == "evt-2" - path = f"/api/v1/{prefix}/sessions/session/events/stream" + suffix = "threads/thread/stream" if scope == "thread" else "events/stream" + path = f"/api/v1/{prefix}/sessions/session/{suffix}" assert requests == [ {"path": path, "last_event_id": None}, {"path": path, "last_event_id": "evt-1"}, @@ -276,7 +284,8 @@ def handle(request): @pytest.mark.parametrize("mode", ["forward", "managed"]) @pytest.mark.parametrize("async_", [False, True], ids=["sync", "async"]) -async def test_resumable_stream_does_not_retry_409(mode, async_): +@pytest.mark.parametrize("scope", ["session", "thread"]) +async def test_resumable_stream_does_not_retry_409(mode, async_, scope): requests = [] def handle(request): @@ -288,7 +297,7 @@ def handle(request): ) client = make_client(mode, async_, handle) - stream = client.sessions.events.resumable_stream("session", last_event_id="invalid") + stream = open_resumable(client, scope, last_event_id="invalid") try: with pytest.raises(ConflictError): if async_: @@ -305,7 +314,8 @@ def handle(request): @pytest.mark.parametrize("mode", ["forward", "managed"]) -def test_sync_resumable_stream_close_interrupts_active_read(mode): +@pytest.mark.parametrize("scope", ["session", "thread"]) +def test_sync_resumable_stream_close_interrupts_active_read(mode, scope): class HangingBody(httpx.SyncByteStream): def __init__(self) -> None: self.started = threading.Event() @@ -324,7 +334,7 @@ def close(self) -> None: body = HangingBody() client = make_client(mode, False, lambda _: httpx.Response(200, stream=body)) - stream = client.sessions.events.resumable_stream("session") + stream = open_resumable(client, scope) try: with ThreadPoolExecutor(max_workers=1) as executor: pending = executor.submit(next, stream) @@ -338,7 +348,8 @@ def close(self) -> None: @pytest.mark.parametrize("mode", ["forward", "managed"]) -async def test_async_resumable_stream_close_interrupts_active_read_and_is_idempotent(mode): +@pytest.mark.parametrize("scope", ["session", "thread"]) +async def test_async_resumable_stream_close_interrupts_active_read_and_is_idempotent(mode, scope): class HangingBody(httpx.AsyncByteStream): def __init__(self) -> None: self.started = asyncio.Event() @@ -357,7 +368,7 @@ async def aclose(self) -> None: body = HangingBody() client = make_client(mode, True, lambda _: httpx.Response(200, stream=body)) - stream = client.sessions.events.resumable_stream("session") + stream = open_resumable(client, scope) pending = asyncio.create_task(anext(stream)) try: await asyncio.wait_for(body.started.wait(), timeout=1) @@ -372,7 +383,8 @@ async def aclose(self) -> None: @pytest.mark.parametrize("mode", ["forward", "managed"]) -async def test_async_resumable_stream_cancellation_closes_active_read(mode): +@pytest.mark.parametrize("scope", ["session", "thread"]) +async def test_async_resumable_stream_cancellation_closes_active_read(mode, scope): class HangingBody(httpx.AsyncByteStream): def __init__(self) -> None: self.started = asyncio.Event() @@ -388,7 +400,7 @@ async def aclose(self) -> None: body = HangingBody() client = make_client(mode, True, lambda _: httpx.Response(200, stream=body)) - stream = client.sessions.events.resumable_stream("session") + stream = open_resumable(client, scope) pending = asyncio.create_task(anext(stream)) try: await asyncio.wait_for(body.started.wait(), timeout=1) @@ -404,7 +416,8 @@ async def aclose(self) -> None: @pytest.mark.parametrize("mode", ["forward", "managed"]) @pytest.mark.parametrize("async_", [False, True], ids=["sync", "async"]) @pytest.mark.parametrize("terminal_type", ["session.status_terminated", "session.deleted"]) -async def test_terminal_event_stops_reconnection(mode, async_, terminal_type): +@pytest.mark.parametrize("scope", ["session", "thread"]) +async def test_terminal_event_stops_reconnection(mode, async_, terminal_type, scope): requests = [] def handle(request): @@ -413,7 +426,7 @@ def handle(request): return httpx.Response(200, content=body, headers={"content-type": "text/event-stream"}) client = make_client(mode, async_, handle) - stream = client.sessions.events.resumable_stream("session") + stream = open_resumable(client, scope) try: if async_: event = await anext(stream) @@ -570,3 +583,112 @@ def record_backoff(_lower, upper): with pytest.raises(StopAsyncIteration): await anext(stream) assert backoffs == expected_backoffs + + +@pytest.mark.parametrize("mode", ["forward", "managed"]) +@pytest.mark.parametrize("async_", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("ending", ["eof", "transport"]) +async def test_thread_resumption_keeps_cursor_options_and_stops_at_termination(monkeypatch, mode, async_, ending): + monkeypatch.setattr("qca.common._resumable_streaming.random.uniform", lambda *_: 0) + requests = [] + bodies = [] + + def handle(request): + requests.append(request) + assert request.url.path.endswith("/sessions/session/threads/thread/stream") + assert not request.url.query + assert request.headers["last-event-id"] == ("override" if len(requests) == 1 else "checkpoint") + assert request.headers["x-test"] == "kept" + assert request.extensions["timeout"]["read"] == 7 + if mode == "managed": + assert request.headers["qoder-workspace-id"] == "workspace" + assert request.headers["x-qoder-beta"] == "beta-one" + first = b'id: checkpoint\nevent: session.thread_status_idle\ndata: {"id":"payload-id","type":"session.thread_status_idle"}\n\nid: partial\ndata: {"id":"not-yielded"}' + last = b'id: terminal\nevent: session.thread_status_terminated\ndata: {"type":"session.thread_status_terminated"}\n\n' + body = ScriptedBody( + first if len(requests) == 1 else last, disconnect=len(requests) == 1 and ending == "transport" + ) + bodies.append(body) + return httpx.Response(200, stream=body, headers={"content-type": "text/event-stream"}) + + client = make_client(mode, async_, handle) + headers = {"last-event-id": "override", "x-test": "kept"} + betas = ["beta-one"] + kwargs = {"workspace_id": "workspace", "betas": betas} if mode == "managed" else {} + stream = open_resumable(client, "thread", last_event_id="initial", extra_headers=headers, timeout=7, **kwargs) + headers["x-test"] = "changed" + betas.append("changed") + try: + events = [event async for event in stream] if async_ else list(stream) + assert [event.type for event in events] == ["session.thread_status_idle", "session.thread_status_terminated"] + assert stream.last_event_id == "terminal" + assert len(requests) == 2 + assert all(body.closed for body in bodies) + finally: + if async_: + await stream.close() + else: + stream.close() + await close_client(client, async_) + + +@pytest.mark.parametrize("mode", ["forward", "managed"]) +@pytest.mark.parametrize("async_", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize("status", [400, 401, 403, 404, 429, 503]) +async def test_thread_resumption_http_policy(monkeypatch, mode, async_, status): + from qca import APIStatusError + + monkeypatch.setattr("qca.common._resumable_streaming.random.uniform", lambda *_: 0) + calls = [] + + def handle(request): + calls.append(request) + assert request.headers["last-event-id"] == "initial" + if len(calls) == 1: + return httpx.Response(status, json={"error": {"message": "test failure"}}) + return httpx.Response( + 200, + content=b'event: session.thread_status_terminated\ndata: {"type":"session.thread_status_terminated"}\n\n', + ) + + client = make_client(mode, async_, handle) + stream = open_resumable(client, "thread", last_event_id="initial") + try: + if status in (429, 503): + events = [event async for event in stream] if async_ else list(stream) + assert len(events) == 1 + assert len(calls) == 2 + else: + with pytest.raises(APIStatusError) as exc: + if async_: + await anext(stream) + else: + next(stream) + assert exc.value.status_code == status + assert len(calls) == 1 + finally: + if async_: + await stream.close() + else: + stream.close() + await close_client(client, async_) + + +@pytest.mark.parametrize("mode", ["forward", "managed"]) +@pytest.mark.parametrize("async_", [False, True], ids=["sync", "async"]) +async def test_child_termination_does_not_stop_session_stream(mode, async_): + content = ( + b'event: session.thread_status_terminated\ndata: {"type":"session.thread_status_terminated"}\n\n' + b'event: session.deleted\ndata: {"type":"session.deleted"}\n\n' + ) + client = make_client(mode, async_, lambda _: httpx.Response(200, content=content)) + stream = open_resumable(client, "session") + try: + events = [event async for event in stream] if async_ else list(stream) + assert [event.type for event in events] == ["session.thread_status_terminated", "session.deleted"] + finally: + if async_: + await stream.close() + else: + stream.close() + await close_client(client, async_)