"""Unit tests for :class:`omnigent_client._sessions.SessionsNamespace`. Mocks at the HTTP transport boundary via :class:`httpx.MockTransport`, using real types (``Session``, ``SessionEventInput``, the typed ``ServerStreamEvent`` union) throughout — per the project testing guide, MagicMock is reserved for cases where a real type cannot be constructed, and the SDK boundary types are all easy to construct. What each test claims to prove (and what failure indicates): * ``test_create_*``: that ``create()`` issues the right multipart request shape and round-trips the typed :class:`Session`. Failure means the request body is wrong (server rejects the upload) or the response decoder drops a field (e.g. ``runner_id``). * ``test_get_*``: same for ``get()``. * ``test_set_reasoning_effort_*``: that mutable reasoning-effort metadata is PATCHed through the sessions API rather than staying as a client-only cache. * ``test_post_event_*`` / ``test_interrupt_*``: that ``post_event()`` posts to the correct URL with the correct body, and that ``interrupt()`` sends the wire literal that the server's ``_INTERRUPT_TYPE`` matches. Failure means the cancel path is silently broken. * ``test_stream_*``: that the SSE parser yields typed :data:`ServerStreamEvent` instances and that malformed/unknown payloads are skipped without aborting iteration. Failure means a schema drift between server and SDK silently drops events. * ``test_*_404``: that the namespace propagates :class:`OmnigentError` for non-2xx responses; failure means errors are silently swallowed. """ from __future__ import annotations import json from collections.abc import Callable, Iterable from typing import Any import httpx import pytest from omnigent_client._errors import OmnigentError from omnigent_client._sessions import ( Session, SessionsNamespace, ) from omnigent.server.schemas import ( CompletedEvent, OutputTextDeltaEvent, SessionInputConsumedEvent, SessionStatusEvent, ) # ── Helpers ─────────────────────────────────────────────────────────── def _make_namespace( handler: Callable[[httpx.Request], httpx.Response], ) -> tuple[SessionsNamespace, httpx.AsyncClient]: """ Build a :class:`SessionsNamespace` wired to a mock HTTP transport. :param handler: Callable invoked for every request. Receives the :class:`httpx.Request` and returns an :class:`httpx.Response`. :returns: The namespace and the underlying client (caller closes the client in a teardown). """ transport = httpx.MockTransport(handler) client = httpx.AsyncClient(transport=transport, base_url="http://srv") return SessionsNamespace(client, "http://srv"), client def _format_sse_lines(events: Iterable[tuple[str, dict[str, Any] | str]]) -> bytes: """ Render a sequence of (event_type, payload) into SSE wire bytes. Mirrors the server's ``_format_sse`` helper. The payload is JSON- encoded unless it's a string sentinel like ``"[DONE]"``, in which case it's emitted verbatim. :param events: Pairs of ``(event_type, payload)``. Payload is a dict for normal events or a string for the ``[DONE]`` sentinel (in which case ``event_type`` is ignored — the server emits ``data: [DONE]`` without an ``event:`` line). :returns: SSE-framed bytes ready to feed into a mocked response. """ parts: list[str] = [] for event_type, payload in events: if isinstance(payload, str): # Terminal/[DONE] sentinel — no event: line. parts.append(f"data: {payload}\n\n") else: parts.append(f"event: {event_type}\ndata: {json.dumps(payload)}\n\n") return "".join(parts).encode("utf-8") def _session_response_body( session_id: str = "conv_abc", agent_id: str = "ag_abc", status: str = "running", items: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """ Build a minimal :class:`SessionResponse` JSON dict. :param session_id: Session id, e.g. ``"conv_abc"``. :param agent_id: Bound agent id, e.g. ``"ag_abc"``. :param status: Session status, e.g. ``"running"``. :param items: Committed items list, defaulting to empty. :returns: A dict matching the server's ``SessionResponse`` shape. """ return { "id": session_id, "agent_id": agent_id, "status": status, "created_at": 1700000000, "items": items if items is not None else [], } # ── create() ────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_create_posts_bundle_and_returns_typed_session() -> None: captured: dict[str, Any] = {} calls: list[str] = [] def handler(request: httpx.Request) -> httpx.Response: calls.append(f"{request.method} {request.url.path}") if request.method == "POST": captured["url"] = str(request.url) captured["method"] = request.method captured["content_type"] = request.headers["content-type"] captured["body"] = request.content return httpx.Response(201, json={"session_id": "conv_abc"}) if request.method == "GET": assert str(request.url) == "http://srv/v1/sessions/conv_abc" return httpx.Response( 200, json=_session_response_body(status="idle") | {"runner_id": "runner_local_test"}, ) raise AssertionError(f"unexpected request: {request.method} {request.url}") ns, client = _make_namespace(handler) try: session = await ns.create( b"bundle-bytes", filename="agent.tar.gz", title="debug title", labels={"env": "test"}, reasoning_effort="high", ) finally: await client.aclose() # Wire shape: POST /v1/sessions with metadata + bundle parts, # then GET /v1/sessions/{id} because create returns only # {"session_id": ...}. assert calls == ["POST /v1/sessions", "GET /v1/sessions/conv_abc"] assert captured["method"] == "POST" assert captured["url"] == "http://srv/v1/sessions" assert str(captured["content_type"]).startswith("multipart/form-data; boundary=") body = bytes(captured["body"]) assert b'name="metadata"' in body assert ( b'{"title": "debug title", "labels": {"env": "test"}, "reasoning_effort": "high"}' in body ) assert b'name="bundle"; filename="agent.tar.gz"' in body assert b"bundle-bytes" in body # Response is a real Session (not a dict), proving from_dict ran. assert isinstance(session, Session) assert session.id == "conv_abc" assert session.agent_id == "ag_abc" assert session.status == "idle" assert session.runner_id == "runner_local_test" @pytest.mark.asyncio async def test_create_with_empty_metadata_sends_json_object() -> None: captured: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: if request.method == "POST": captured["body"] = request.content return httpx.Response(201, json={"session_id": "conv_abc"}) return httpx.Response(200, json=_session_response_body(items=[])) ns, client = _make_namespace(handler) try: session = await ns.create(b"bundle-bytes") finally: await client.aclose() body = bytes(captured["body"]) assert b'name="metadata"' in body assert b"{}" in body assert session.items == [] @pytest.mark.asyncio async def test_create_404_raises_omnigent_error() -> None: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 404, json={"error": {"code": "not_found", "message": "no such agent"}}, ) ns, client = _make_namespace(handler) try: with pytest.raises(OmnigentError) as exc_info: await ns.create(b"bundle-bytes") finally: await client.aclose() # Server-supplied message must propagate verbatim — proves the # SDK isn't swallowing the body or substituting a generic string. assert "no such agent" in str(exc_info.value) # ── get() ───────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_get_returns_typed_session() -> None: def handler(request: httpx.Request) -> httpx.Response: assert str(request.url) == "http://srv/v1/sessions/conv_abc" return httpx.Response( 200, json=_session_response_body( items=[{"id": "msg_1", "type": "message"}], ), ) ns, client = _make_namespace(handler) try: session = await ns.get("conv_abc") finally: await client.aclose() assert isinstance(session, Session) # items round-trip as raw dicts (heterogeneous, intentionally # un-modeled per the namespace docstring). assert session.items == [{"id": "msg_1", "type": "message"}] @pytest.mark.asyncio async def test_get_parses_agent_name() -> None: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 200, json=_session_response_body() | {"agent_name": "claude-native-ui"}, ) ns, client = _make_namespace(handler) try: session = await ns.get("conv_abc") finally: await client.aclose() # Proves Session.from_dict parses `agent_name` through — the REPL # uses it to refresh the displayed agent after an in-place switch. # A silent drop here would leave it at the None default and the # toolbar stuck on the launch-time agent. assert session.agent_name == "claude-native-ui" @pytest.mark.asyncio async def test_get_agent_name_defaults_to_none_when_omitted() -> None: def handler(request: httpx.Request) -> httpx.Response: # Old servers omit agent_name from the payload entirely. return httpx.Response(200, json=_session_response_body()) ns, client = _make_namespace(handler) try: session = await ns.get("conv_abc") finally: await client.aclose() # None (not "" or a KeyError) is the contract for old servers — # the REPL's hydrate keeps its launch-time name on None. assert session.agent_name is None # ── bind_runner() ──────────────────────────────────────────────────── @pytest.mark.asyncio async def test_bind_runner_patches_runner_id_and_returns_snapshot() -> None: captured: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: captured["url"] = str(request.url) captured["method"] = request.method captured["body"] = json.loads(request.content.decode()) return httpx.Response( 200, json=_session_response_body(status="idle") | {"runner_id": "runner_local_test"}, ) ns, client = _make_namespace(handler) try: session = await ns.bind_runner( "conv_abc", runner_id="runner_local_test", ) finally: await client.aclose() assert captured == { "url": "http://srv/v1/sessions/conv_abc", "method": "PATCH", "body": {"runner_id": "runner_local_test"}, } assert session.runner_id == "runner_local_test" @pytest.mark.asyncio async def test_bind_runner_400_raises_omnigent_error() -> None: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 400, json={"error": {"code": "invalid_input", "message": "runner is offline"}}, ) ns, client = _make_namespace(handler) try: with pytest.raises(OmnigentError) as exc_info: await ns.bind_runner("conv_abc", runner_id="runner_offline") finally: await client.aclose() assert "runner is offline" in str(exc_info.value) # ── set_reasoning_effort() ─────────────────────────────────────────── @pytest.mark.asyncio async def test_set_reasoning_effort_patches_metadata_and_returns_snapshot() -> None: captured: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: captured["url"] = str(request.url) captured["method"] = request.method captured["body"] = json.loads(request.content.decode()) return httpx.Response( 200, json=_session_response_body(status="idle") | {"reasoning_effort": "high"}, ) ns, client = _make_namespace(handler) try: session = await ns.set_reasoning_effort( "conv_abc", reasoning_effort="high", ) finally: await client.aclose() assert captured == { "url": "http://srv/v1/sessions/conv_abc", "method": "PATCH", "body": {"reasoning_effort": "high"}, } assert session.reasoning_effort == "high" @pytest.mark.asyncio async def test_set_reasoning_effort_none_sends_clear_alias() -> None: captured: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = json.loads(request.content.decode()) return httpx.Response(200, json=_session_response_body(status="idle")) ns, client = _make_namespace(handler) try: session = await ns.set_reasoning_effort( "conv_abc", reasoning_effort=None, ) finally: await client.aclose() assert captured["body"] == {"reasoning_effort": "default"} assert session.reasoning_effort is None # ── set_archived() ─────────────────────────────────────────────────── @pytest.mark.asyncio async def test_set_archived_patches_and_returns_snapshot() -> None: captured: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: captured["url"] = str(request.url) captured["method"] = request.method captured["body"] = json.loads(request.content.decode()) return httpx.Response( 200, json=_session_response_body(status="idle") | {"archived": True}, ) ns, client = _make_namespace(handler) try: session = await ns.set_archived("conv_abc", archived=True) finally: await client.aclose() # The wire call must be a PATCH carrying exactly {"archived": true}; # a wrong path/method/body would fail to archive server-side. assert captured == { "url": "http://srv/v1/sessions/conv_abc", "method": "PATCH", "body": {"archived": True}, } # Proves Session.from_dict parses the new `archived` field through — # if it didn't, this would be the False default and pass silently. assert session.archived is True @pytest.mark.asyncio async def test_set_archived_unarchive_sends_false() -> None: captured: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = json.loads(request.content.decode()) return httpx.Response( 200, json=_session_response_body(status="idle") | {"archived": False}, ) ns, client = _make_namespace(handler) try: session = await ns.set_archived("conv_abc", archived=False) finally: await client.aclose() # Unarchive sends the explicit false (not an omitted/clear alias) so # the server flips the flag back rather than leaving it unchanged. assert captured["body"] == {"archived": False} assert session.archived is False # ── post_event / interrupt ──────────────────────────────────────────── @pytest.mark.asyncio async def test_post_event_posts_body_to_events_url() -> None: captured: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: captured["url"] = str(request.url) captured["body"] = json.loads(request.content.decode()) return httpx.Response(202, json={"queued": True, "item_id": "ci_123"}) ns, client = _make_namespace(handler) event = {"type": "message", "data": {"role": "user", "content": []}} try: ack = await ns.post_event("conv_abc", event) finally: await client.aclose() assert captured["url"] == "http://srv/v1/sessions/conv_abc/events" assert captured["body"] == event assert ack == {"queued": True, "item_id": "ci_123"} @pytest.mark.asyncio async def test_resolve_elicitation_posts_result_to_resolve_url() -> None: """ ``resolve_elicitation`` POSTs the bare MCP ``ElicitationResult`` body to the elicitation's dedicated resolve URL — the elicitation id rides in the URL path, not the body (URL-based elicitation). Asserting the exact URL guards against the verdict regressing back to a generic ``approval`` event on ``/events``. """ captured: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: captured["url"] = str(request.url) captured["body"] = json.loads(request.content.decode()) return httpx.Response(202, json={"queued": False}) ns, client = _make_namespace(handler) try: ack = await ns.resolve_elicitation("conv_abc", "elicit_xyz", {"action": "accept"}) finally: await client.aclose() assert captured["url"] == ("http://srv/v1/sessions/conv_abc/elicitations/elicit_xyz/resolve") assert captured["body"] == {"action": "accept"} assert ack == {"queued": False} @pytest.mark.asyncio async def test_interrupt_posts_interrupt_event_literal() -> None: captured: dict[str, Any] = {} def handler(request: httpx.Request) -> httpx.Response: captured["body"] = json.loads(request.content.decode()) return httpx.Response(202, json={"queued": False}) ns, client = _make_namespace(handler) try: await ns.interrupt("conv_abc") finally: await client.aclose() # The literal "interrupt" must match the server's _INTERRUPT_TYPE # constant. If this drifts the server treats it as an unknown # event type and returns 400 — the test catches the drift here. assert captured["body"] == {"type": "interrupt", "data": {}} @pytest.mark.asyncio async def test_post_event_404_raises() -> None: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 404, json={"error": {"code": "not_found", "message": "no session"}}, ) ns, client = _make_namespace(handler) try: with pytest.raises(OmnigentError): await ns.post_event("conv_x", {"type": "message", "data": {}}) finally: await client.aclose() # ── stream() ────────────────────────────────────────────────────────── def _completed_response_dict( response_id: str = "resp_1", status: str = "completed", ) -> dict[str, Any]: """ Build a minimal :class:`ResponseObject` JSON dict for use in the typed terminal event payload. :param response_id: Response id, e.g. ``"resp_1"``. :param status: Status string, e.g. ``"completed"``. :returns: Dict matching the server's ``ResponseObject`` shape's required fields. """ return { "id": response_id, "status": status, "model": "test-model", "created_at": 1700000000, } @pytest.mark.asyncio async def test_stream_yields_typed_events_in_order() -> None: payloads: list[tuple[str, dict[str, Any] | str]] = [ ( "session.status", { "type": "session.status", "conversation_id": "conv_abc", "status": "running", }, ), ( "response.output_text.delta", {"type": "response.output_text.delta", "delta": "Hello "}, ), ( "response.output_text.delta", {"type": "response.output_text.delta", "delta": "world"}, ), ( "response.completed", { "type": "response.completed", "response": _completed_response_dict(), }, ), ("done", "[DONE]"), ] def handler(request: httpx.Request) -> httpx.Response: assert str(request.url) == "http://srv/v1/sessions/conv_abc/stream" return httpx.Response( 200, content=_format_sse_lines(payloads), headers={"content-type": "text/event-stream"}, ) ns, client = _make_namespace(handler) try: events = [event async for event in ns.stream("conv_abc")] finally: await client.aclose() # 4 well-formed events; [DONE] terminates without yielding. # If 3, the terminal event was lost — likely the [DONE] handling # regressed and short-circuited the previous event. # If 5, [DONE] was yielded as an event — adapter would have raised # but the caller would still see a count mismatch. assert len(events) == 4 # Real types — proves the TypeAdapter dispatched on the # discriminator field. If isinstance fails here, the typed-union # validation regressed and downstream consumers would break. assert isinstance(events[0], SessionStatusEvent) assert events[0].status == "running" assert isinstance(events[1], OutputTextDeltaEvent) assert events[1].delta == "Hello " assert isinstance(events[3], CompletedEvent) assert events[3].response.id == "resp_1" @pytest.mark.asyncio async def test_stream_skips_malformed_and_unknown_events() -> None: payloads: list[tuple[str, dict[str, Any] | str]] = [ # Unknown discriminator — should be logged and skipped. ( "made.up.event", {"type": "made.up.event", "data": "ignored"}, ), # Well-formed event after the bad one — must still be yielded. ( "response.output_text.delta", {"type": "response.output_text.delta", "delta": "ok"}, ), ("done", "[DONE]"), ] # Inject a raw malformed line directly between the two real # events to also cover the JSON-decode failure path. raw = b"event: made.up.event\ndata: not-json{{{\n\n" + _format_sse_lines(payloads) def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 200, content=raw, headers={"content-type": "text/event-stream"}, ) ns, client = _make_namespace(handler) try: events = [event async for event in ns.stream("conv_abc")] finally: await client.aclose() # Exactly one event survives — the well-formed # OutputTextDeltaEvent. If 0, the adapter is too strict and a # malformed event is killing the iteration. If 2+, an unknown # event leaked through, indicating the discriminator validation # was bypassed. assert len(events) == 1 assert isinstance(events[0], OutputTextDeltaEvent) assert events[0].delta == "ok" @pytest.mark.asyncio async def test_stream_404_raises_before_first_yield() -> None: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 404, json={"error": {"code": "not_found", "message": "no session"}}, ) ns, client = _make_namespace(handler) try: with pytest.raises(OmnigentError): async for _ in ns.stream("conv_missing"): pytest.fail("Should have raised before yielding any event") finally: await client.aclose() # ── full session lifecycle ──────────────────────────────────────────── @pytest.mark.asyncio async def test_sdk_full_session_lifecycle_with_reconnect() -> None: """ Drive the SDK through the 4-step lifecycle a real client follows. 1. Create a new session. 2. Subscribe to the live stream. 3. Send a message; the stream emits :class:`SessionInputConsumedEvent` echoing the user input and an :class:`OutputTextDeltaEvent` + :class:`CompletedEvent` with the assistant's reply. 4. Disconnect from the stream. GET the session snapshot to confirm the assistant text persisted. Subscribe again, send a follow-up, see the new response. After the second turn the snapshot must include BOTH assistant messages in order (history persistence invariant). Asserts the SDK / server contract surfaces the live stream and the durable snapshot as the same data viewed differently: everything observed live during turn 1 is still visible after turn 2 via GET, plus turn 2's response on top. Production breakage that causes this test to fail: * ``SessionsNamespace.create`` drops the returned session id, so subsequent calls target the wrong path. Caught by the MockTransport's URL assertions. * ``SessionsNamespace.stream`` short-circuits before terminal event or fails to dispatch on the discriminator (events arrive as raw dicts, not typed). Caught by the ``isinstance`` checks on observed events. * ``SessionsNamespace.get`` returns a session whose ``items`` don't preserve turn-1 history after turn 2 completes. Caught by the final cross-stream / snapshot equality assertion. """ # State the mock server holds across the test. The handler # mutates these to simulate server-side persistence. session_id = "conv_lifecycle" agent_id = "ag_lifecycle" # Items committed to history (in the server's view). turn 1 sets # the user message + assistant reply; turn 2 appends another pair. history: list[dict[str, Any]] = [] # Pop-one queue of canned SSE payload sequences, one per stream # subscribe. After step 3 the SDK disconnects and resubscribes, # so we need two SSE scripts. sse_scripts: list[list[tuple[str, dict[str, Any] | str]]] = [] # The sequence of POSTs the SDK issued, for assertion. posted_events: list[dict[str, Any]] = [] def handler(request: httpx.Request) -> httpx.Response: """ Stand in for the full ``/v1/sessions`` surface. Dispatches on method + path: * POST /v1/sessions → create the session, seed history. * GET /v1/sessions/{id} → return snapshot of ``history``. * GET /v1/sessions/{id}/stream → consume one sse_scripts entry. * POST /v1/sessions/{id}/events → record the event, mutate history with the user message immediately so the GET in step 4 matches what the live stream's ``session.input. consumed`` echo reported. :param request: The incoming :class:`httpx.Request`. :returns: A typed :class:`httpx.Response`. """ url = str(request.url) method = request.method if method == "POST" and url == "http://srv/v1/sessions": return httpx.Response(201, json={"session_id": session_id}) if method == "GET" and url == f"http://srv/v1/sessions/{session_id}": return httpx.Response( 200, json=_session_response_body( session_id=session_id, agent_id=agent_id, status="idle", items=list(history), ), ) if method == "GET" and url == f"http://srv/v1/sessions/{session_id}/stream": assert sse_scripts, ( "stream subscription opened but no scripted SSE " "sequence remains — SDK opened more streams than the " "test prepared. The test specifies exactly two " "subscribe calls (one per turn)." ) return httpx.Response( 200, content=_format_sse_lines(sse_scripts.pop(0)), headers={"content-type": "text/event-stream"}, ) if method == "POST" and url == f"http://srv/v1/sessions/{session_id}/events": body = json.loads(request.content.decode()) posted_events.append(body) # Persist the user input synchronously so the subsequent # GET sees it — matches the real server, which writes the # queued item before responding 202. if body.get("type") == "message": history.append({"type": "message", "data": body["data"]}) return httpx.Response(202, json={"queued": True}) raise AssertionError( f"Unexpected request: {method} {url} (body={request.content!r})", ) # ── Script the SSE bodies for both turns ────────────────────── # # Turn 1: session.input.consumed (the server echoes the user # input as it materializes into history), then a single delta, # then the terminal response.completed. turn1_consumed_payload: dict[str, Any] = { "type": "session.input.consumed", "data": { "item_id": "ci_t1", "type": "message", "data": { "role": "user", "content": [{"type": "input_text", "text": "Hello 1"}], }, }, } turn1_assistant_text = "Hi there from turn 1" sse_scripts.append( [ ( "session.input.consumed", turn1_consumed_payload, ), ( "response.output_text.delta", { "type": "response.output_text.delta", "delta": turn1_assistant_text, }, ), ( "response.completed", { "type": "response.completed", "response": _completed_response_dict(response_id="resp_t1"), }, ), ("done", "[DONE]"), ], ) turn2_assistant_text = "Reply to turn 2" sse_scripts.append( [ ( "session.input.consumed", { "type": "session.input.consumed", "data": { "item_id": "ci_t2", "type": "message", "data": { "role": "user", "content": [{"type": "input_text", "text": "Hello 2"}], }, }, }, ), ( "response.output_text.delta", { "type": "response.output_text.delta", "delta": turn2_assistant_text, }, ), ( "response.completed", { "type": "response.completed", "response": _completed_response_dict(response_id="resp_t2"), }, ), ("done", "[DONE]"), ], ) ns, client = _make_namespace(handler) try: # ── Step 1: Create ──────────────────────────────────────── session = await ns.create(b"bundle-bytes") # The SDK retains the durable id used by every subsequent # call; if create() dropped it, the URL assertions in the # handler would fire on the next request. assert session.id == session_id # ── Steps 2 + 3: Subscribe, send, observe turn 1 ───────── # # Open the stream FIRST (the new pub-sub model drops events # published before any subscriber connects, so subscribe # must precede post_event for the test to be valid). stream_iter_1 = ns.stream(session_id).__aiter__() # Issue the user message after the subscriber is connected. # In production this is racy, but with MockTransport the # entire stream body is prebuffered so ordering of the # handler dispatch doesn't matter — what matters for this # test is asserting the SDK's behavior end-to-end. await ns.post_event( session_id, { "type": "message", "data": { "role": "user", "content": [{"type": "input_text", "text": "Hello 1"}], }, }, ) turn1_events = [] async for event in stream_iter_1: turn1_events.append(event) # Append the assistant reply to the server's history at the # point the turn terminates. The real server persists it via # ``conv_store.append`` after the workflow's # response.completed fires — same observable result. history.append( { "type": "message", "data": { "role": "assistant", "content": [ {"type": "output_text", "text": turn1_assistant_text}, ], }, }, ) # Turn 1 stream invariants. If any of these fail, the SDK # isn't surfacing the live shape the spec advertises. consumed_events_1 = [e for e in turn1_events if isinstance(e, SessionInputConsumedEvent)] assert len(consumed_events_1) == 1, ( f"Turn 1 emitted {len(consumed_events_1)} " f"session.input.consumed events, expected 1. The event " f"is the SDK's signal that the user input was accepted " f"into history; missing it means callers can't tell " f"when their input is live." ) assert consumed_events_1[0].data.data["content"][0]["text"] == "Hello 1", ( "The consumed event must echo the exact user text the " "SDK posted. Mismatch indicates the SessionInputConsumed " "decoder dropped fields from the nested data payload." ) # Concatenated delta text equals the assistant's full reply. turn1_text = "".join(e.delta for e in turn1_events if isinstance(e, OutputTextDeltaEvent)) assert turn1_text == turn1_assistant_text, ( f"Turn 1 delta-text concat = {turn1_text!r}; expected " f"{turn1_assistant_text!r}. A mismatch indicates either " f"a missed delta or duplicate yielding." ) # Stream observed exactly one terminal event in turn 1. completed_1 = [e for e in turn1_events if isinstance(e, CompletedEvent)] assert len(completed_1) == 1 assert completed_1[0].response.id == "resp_t1" # ── Step 4a: Disconnect (implicit — the async-for exited) ── # # The SDK's stream() generator's finally would have closed # the underlying httpx response; we've left the loop, so # there's no further reading. # ── Step 4b: GET snapshot mid-flight ────────────────────── snapshot_after_turn_1 = await ns.get(session_id) # History contains the user message + the assistant reply # from turn 1. The exact two-item set is what step 4's final # equality check will compare against post-turn-2. post_turn_1_items = list(snapshot_after_turn_1.items) assert len(post_turn_1_items) == 2, ( f"Snapshot after turn 1 has {len(post_turn_1_items)} " f"items, expected 2 (user + assistant). If 0/1, the " f"SDK is dropping history fields from the snapshot " f"decode; if 3+, server-side dedup regressed." ) # The assistant text recorded in history matches the live # stream's delta concat — proves the live + durable views # are the same data. assistant_item = post_turn_1_items[1] assert assistant_item["data"]["content"][0]["text"] == turn1_assistant_text, ( "The assistant text returned by GET must match the live " "stream's delta concat. If they diverge, the SDK's view " "of history is inconsistent with what it just observed." ) # ── Step 4c: Resubscribe + second turn ──────────────────── stream_iter_2 = ns.stream(session_id).__aiter__() await ns.post_event( session_id, { "type": "message", "data": { "role": "user", "content": [{"type": "input_text", "text": "Hello 2"}], }, }, ) turn2_events = [] async for event in stream_iter_2: turn2_events.append(event) history.append( { "type": "message", "data": { "role": "assistant", "content": [ {"type": "output_text", "text": turn2_assistant_text}, ], }, }, ) turn2_text = "".join(e.delta for e in turn2_events if isinstance(e, OutputTextDeltaEvent)) assert turn2_text == turn2_assistant_text, ( f"Turn 2 delta-text concat = {turn2_text!r}; expected " f"{turn2_assistant_text!r}. A mismatch indicates the " f"second subscribe didn't get a fresh stream cleanly." ) # ── Step 4d: Final GET — both turns persist in order ────── final_snapshot = await ns.get(session_id) final_items = list(final_snapshot.items) # 4 items: user1, assistant1, user2, assistant2. assert len(final_items) == 4, ( f"Final snapshot has {len(final_items)} items, " f"expected 4. If <4, an item from one of the turns " f"vanished; if >4, the SDK is duplicating items " f"across the reconnect." ) # Order check — the assistant message from turn 1 must # appear BEFORE turn 2's items. This is the history- # persistence invariant the spec promises: clients can # always reconstruct full session state via GET, even # across stream disconnects. assert final_items[1]["data"]["content"][0]["text"] == turn1_assistant_text assert final_items[3]["data"]["content"][0]["text"] == turn2_assistant_text # ── Cross-check: stream-1 view ⊆ final history ──────────── # The user-facing invariant for step 4's final assertion: # everything the client saw live during turn 1 still appears # in the post-turn-2 snapshot. The assistant text from # turn 1 is the strongest evidence — its presence proves # durable replay is consistent with what was streamed live. live_turn_1_text = "".join( e.delta for e in turn1_events if isinstance(e, OutputTextDeltaEvent) ) snapshot_texts = [ item["data"]["content"][0]["text"] for item in final_items if item.get("type") == "message" and item["data"].get("role") == "assistant" ] assert live_turn_1_text in snapshot_texts, ( f"Turn 1 assistant text {live_turn_1_text!r} (observed " f"live) is missing from final snapshot's assistant " f"messages {snapshot_texts!r}. History persistence " f"across reconnect is broken — the SDK or server is " f"losing turn-1 state when turn 2 runs." ) # ── Final assertions on the post-record ─────────────────── assert len(posted_events) == 2, ( f"Expected exactly 2 user posts (one per turn); " f"got {len(posted_events)}. The SDK should not be " f"retrying or duplicating user messages." ) finally: await client.aclose() # ── Fork ──────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_fork_posts_correct_url_and_parses_response() -> None: """``fork()`` POSTs to ``/v1/sessions/{id}/fork`` and returns the raw dict. Verifies the URL path, request body, and that the response dict is returned verbatim. A wrong URL means the server never receives the request; a wrong body means the server rejects it. """ captured_request: httpx.Request | None = None def _handler(request: httpx.Request) -> httpx.Response: nonlocal captured_request captured_request = request return httpx.Response( status_code=201, json={ "id": "conv_fork", "agent_id": "ag_cloned", "status": "idle", "created_at": 1234, "title": "Fork of original", "items": [], }, ) ns, client = _make_namespace(_handler) try: result = await ns.fork("conv_src", title="Fork of original") finally: await client.aclose() # Verify the request was sent to the correct URL. assert captured_request is not None assert captured_request.url.path == "/v1/sessions/conv_src/fork" assert captured_request.method == "POST" # Verify the body included the title. body = json.loads(captured_request.content) assert body == {"title": "Fork of original"}, f"Expected body with title, got {body}" # Verify the response is parsed correctly. assert result["id"] == "conv_fork" assert result["agent_id"] == "ag_cloned" assert result["status"] == "idle" assert result["title"] == "Fork of original" @pytest.mark.asyncio async def test_fork_omits_title_when_none() -> None: """``fork()`` sends an empty body when no title is provided. The server should derive a default title. If the SDK sends ``{"title": null}`` instead of ``{}``, the server might reject it or override the default-title logic incorrectly. """ captured_request: httpx.Request | None = None def _handler(request: httpx.Request) -> httpx.Response: nonlocal captured_request captured_request = request return httpx.Response( status_code=201, json={ "id": "conv_fork", "agent_id": "ag_cloned", "status": "idle", "created_at": 1234, "title": None, "items": [], }, ) ns, client = _make_namespace(_handler) try: await ns.fork("conv_src") finally: await client.aclose() assert captured_request is not None body = json.loads(captured_request.content) # No title key in the body when None is passed. assert "title" not in body, f"Expected empty body (no title key), got {body}" @pytest.mark.asyncio async def test_fork_404_raises() -> None: """``fork()`` raises ``OmnigentError`` when the source session is missing. Failure to raise means the SDK is swallowing server errors. """ def _handler(request: httpx.Request) -> httpx.Response: return httpx.Response( status_code=404, json={"error": {"message": "Session not found", "code": "not_found"}}, ) ns, client = _make_namespace(_handler) try: with pytest.raises(OmnigentError): await ns.fork("conv_nonexistent") finally: await client.aclose() # ── child_sessions_tree() / subtree_busy() ─────────────────────────── def _tree_handler( tree: dict[str, list[dict[str, Any]]], ) -> Callable[[httpx.Request], httpx.Response]: """Serve ``GET …/{id}/child_sessions`` from an in-memory parent→children map. The recursion helper queries each node's children with a fresh request, so the map keys are parent ids and the values are the ``ChildSessionSummary`` rows that parent returns. Unknown parents return an empty page. """ def handler(request: httpx.Request) -> httpx.Response: parts = request.url.path.split("/") # …/v1/sessions//child_sessions assert parts[-1] == "child_sessions" parent_id = parts[-2] return httpx.Response(200, json={"data": tree.get(parent_id, [])}) return handler def _child(sid: str, **fields: Any) -> dict[str, Any]: return {"id": sid, **fields} @pytest.mark.asyncio async def test_child_sessions_tree_recurses_and_tags_parent() -> None: """The tree helper walks every level and stamps each row with its parent. Failure means the SDK rollup (and the CLI tree it now feeds) loses grandchildren or mis-attaches the hierarchy. """ tree = { "root": [_child("a"), _child("b")], "a": [_child("a1")], "a1": [_child("a1x")], "b": [], } ns, client = _make_namespace(_tree_handler(tree)) try: nodes = await ns.child_sessions_tree("root") finally: await client.aclose() by_id = {n["id"]: n for n in nodes} assert set(by_id) == {"a", "b", "a1", "a1x"} # root itself excluded assert by_id["a"]["parent_id"] == "root" assert by_id["a1"]["parent_id"] == "a" assert by_id["a1x"]["parent_id"] == "a1" @pytest.mark.asyncio async def test_child_sessions_tree_respects_max_depth() -> None: """``max_depth`` caps descent — depth 1 returns direct children only.""" tree = { "root": [_child("a")], "a": [_child("a1")], } ns, client = _make_namespace(_tree_handler(tree)) try: nodes = await ns.child_sessions_tree("root", max_depth=1) finally: await client.aclose() assert [n["id"] for n in nodes] == ["a"] # a1 is one level too deep @pytest.mark.asyncio async def test_child_sessions_tree_cycle_guard() -> None: """A child pointing back at an ancestor is visited once, not forever.""" tree = { "root": [_child("a")], "a": [_child("root"), _child("a1")], # 'root' is a back-edge } ns, client = _make_namespace(_tree_handler(tree)) try: nodes = await ns.child_sessions_tree("root") finally: await client.aclose() # 'root' is the seed (already seen) so the back-edge is dropped; 'a1' stays. assert [n["id"] for n in nodes] == ["a", "a1"] @pytest.mark.asyncio async def test_subtree_busy_true_when_deep_descendant_busy() -> None: """A busy grandchild makes the whole subtree read busy. This is the rollup #444 asks for: the parent's own status is idle but a descendant is still working. """ tree = { "root": [_child("a", busy=False, current_task_status="completed")], "a": [_child("a1", busy=True, current_task_status=None)], } ns, client = _make_namespace(_tree_handler(tree)) try: assert await ns.subtree_busy("root") is True finally: await client.aclose() @pytest.mark.asyncio async def test_subtree_busy_false_when_all_terminal() -> None: """All descendants settled → subtree not busy (safe to inject 'your turn').""" tree = { "root": [ _child("a", busy=False, current_task_status="completed"), _child("b", busy=False, current_task_status="failed"), ], "a": [_child("a1", busy=False, current_task_status="cancelled")], } ns, client = _make_namespace(_tree_handler(tree)) try: assert await ns.subtree_busy("root") is False finally: await client.aclose() @pytest.mark.asyncio async def test_subtree_busy_false_when_no_children() -> None: ns, client = _make_namespace(_tree_handler({"root": []})) try: assert await ns.subtree_busy("root") is False finally: await client.aclose() @pytest.mark.asyncio async def test_subtree_busy_counts_awaiting_input_as_busy() -> None: """A descendant parked on an elicitation keeps the subtree busy (web parity).""" tree = { "root": [ _child("a", busy=False, current_task_status="completed", pending_elicitations_count=1) ], } ns, client = _make_namespace(_tree_handler(tree)) try: assert await ns.subtree_busy("root") is True finally: await client.aclose()