diff --git a/libs/sdk-py/langgraph_sdk/_sync/stream.py b/libs/sdk-py/langgraph_sdk/_sync/stream.py index 8e4415258..3b338d99d 100644 --- a/libs/sdk-py/langgraph_sdk/_sync/stream.py +++ b/libs/sdk-py/langgraph_sdk/_sync/stream.py @@ -12,11 +12,13 @@ Sync mirror of `libs/sdk-py/langgraph_sdk/_async/stream.py`. from __future__ import annotations import contextlib +import queue import threading from collections.abc import Iterator, Mapping from dataclasses import dataclass from typing import Any, Literal, TypedDict +from langchain_core.language_models.chat_model_stream import ChatModelStream from langchain_protocol import Event, SubscribeParams from langgraph_sdk._sync.http import SyncHttpClient @@ -56,6 +58,42 @@ _ALL_CHANNELS: list[str] = [ ] +def _exact_namespace_params( + channels: list[str], + namespace: list[str], +) -> SubscribeParams: + return { + "channels": channels, + "namespaces": [list(namespace)], + "depth": 0, + } + + +def _event_namespace(params_field: Any) -> list[str]: + if not isinstance(params_field, dict): + return [] + namespace = params_field.get("namespace") or [] + return list(namespace) if isinstance(namespace, list) else [] + + +def _message_event_id(data: dict[str, Any]) -> str | None: + message_id = data.get("id") or data.get("message_id") + return str(message_id) if message_id is not None else None + + +def _message_route_key(data: dict[str, Any], fallback: str | None = None) -> str: + metadata = data.get("metadata") if isinstance(data.get("metadata"), dict) else {} + run_id = metadata.get("run_id") if metadata else None + message_id = _message_event_id(data) + if run_id is not None: + return f"run:{run_id}" + if message_id is not None: + return f"message:{message_id}" + if fallback is not None: + return f"message:{fallback}" + return "__single__" + + class _BlockingResult: def __init__(self) -> None: self._event = threading.Event() @@ -209,6 +247,398 @@ class _SyncValuesProjection: self._thread._unregister_subscription(sub.id) +class _SyncMessagesProjection: + """Typed projection for root-scope `thread.messages`. + + Iterating yields one `ChatModelStream` per message-start event. Each + stream is fully dispatched before being yielded so that `str(message.text)` + works immediately inside a `for` loop. + """ + + def __init__( + self, thread: SyncThreadStream, namespace: list[str] | None = None + ) -> None: + self._thread = thread + self._namespace = list(namespace or []) + + def __iter__(self) -> Iterator[ChatModelStream]: + return self._messages_iter() + + def _messages_iter(self) -> Iterator[ChatModelStream]: + if self._thread._transport is None: + raise RuntimeError("SyncThreadStream not entered — use `with`.") + root_inbox = self._thread._root_messages_inbox if not self._namespace else None + if root_inbox is not None: + yield from _drain_messages_inbox(root_inbox, self._namespace, self._thread) + return + params = _exact_namespace_params(["messages"], self._namespace) + sub = self._thread._register_subscription(params) + active: dict[str, ChatModelStream] = {} + try: + self._thread._reconcile_stream(params) + self._thread._ensure_fanout_running() + while True: + item = sub.queue.get() + if item is None: + return + params_field = item.get("params") or {} + if _event_namespace(params_field) != self._namespace: + continue + data = ( + params_field.get("data") if isinstance(params_field, dict) else None + ) + if not isinstance(data, dict): + continue + event_type = data.get("event") + if event_type == "message-start": + message_id = _message_event_id(data) + key = _message_route_key(data, fallback=message_id) + metadata = ( + data.get("metadata") + if isinstance(data.get("metadata"), dict) + else {} + ) + stream = ChatModelStream( + namespace=list(self._namespace), + node=metadata.get("langgraph_node") if metadata else None, + message_id=message_id, + ) + active[key] = stream + self._thread._register_active_message_stream(stream) + stream.dispatch(data) + # Pre-dispatch all remaining events for this message so the + # caller can access str(message.text) inside a for loop. + while not stream._done: + next_item = sub.queue.get() + if next_item is None: + sub.queue.put(None) + break + next_params = next_item.get("params") or {} + next_data = ( + next_params.get("data") + if isinstance(next_params, dict) + else None + ) + if not isinstance(next_data, dict): + continue + next_event_type = next_data.get("event") + next_key = _message_route_key(next_data) + target = active.get(next_key) + if target is not None: + target.dispatch(next_data) + if next_event_type in ("message-finish", "error"): + self._thread._unregister_active_message_stream(target) + for rk, cand in list(active.items()): + if cand is target: + del active[rk] + yield stream + else: + key = _message_route_key(data) + stream = active.get(key) + if stream is None: + # No active stream matches this event's key. Drop rather + # than silently misroute to the only remaining stream. + continue + stream.dispatch(data) + if event_type in ("message-finish", "error"): + self._thread._unregister_active_message_stream(stream) + for route_key, candidate in list(active.items()): + if candidate is stream: + del active[route_key] + finally: + for s in active.values(): + self._thread._unregister_active_message_stream(s) + self._thread._unregister_subscription(sub.id) + + +def _drain_messages_inbox( + inbox: queue.Queue[Event | None], + namespace: list[str], + thread: SyncThreadStream, +) -> Iterator[ChatModelStream]: + """Drain a pre-filled inbox of messages events, yielding one stream per message. + + Mirrors the pre-dispatch pattern in `_SyncMessagesProjection._messages_iter`: + each `message-start` triggers an inner loop that reads ahead until the stream + is `_done` before yielding, so callers can do `str(stream.text)` immediately. + """ + active: dict[str, ChatModelStream] = {} + try: + while True: + item = inbox.get() + if item is None: + return + params_field = item.get("params") or {} + data = params_field.get("data") if isinstance(params_field, dict) else None + if not isinstance(data, dict): + continue + event_type = data.get("event") + if event_type == "message-start": + message_id = _message_event_id(data) + key = _message_route_key(data, fallback=message_id) + metadata = ( + data.get("metadata") + if isinstance(data.get("metadata"), dict) + else {} + ) + stream = ChatModelStream( + namespace=list(namespace), + node=metadata.get("langgraph_node") if metadata else None, + message_id=message_id, + ) + active[key] = stream + thread._register_active_message_stream(stream) + stream.dispatch(data) + # Pre-dispatch all remaining events for this message so the + # caller can access str(stream.text) immediately inside a for loop. + while not stream._done: + next_item = inbox.get() + if next_item is None: + inbox.put(None) + break + next_params = next_item.get("params") or {} + next_data = ( + next_params.get("data") + if isinstance(next_params, dict) + else None + ) + if not isinstance(next_data, dict): + continue + next_event_type = next_data.get("event") + next_key = _message_route_key(next_data) + target = active.get(next_key) + if target is not None: + target.dispatch(next_data) + if next_event_type in ("message-finish", "error"): + thread._unregister_active_message_stream(target) + for rk, cand in list(active.items()): + if cand is target: + del active[rk] + yield stream + else: + key = _message_route_key(data) + stream = active.get(key) + if stream is None: + # No active stream matches this event's key. Drop rather + # than silently misroute to the only remaining stream. + continue + stream.dispatch(data) + if event_type in ("message-finish", "error"): + thread._unregister_active_message_stream(stream) + for route_key, candidate in list(active.items()): + if candidate is stream: + del active[route_key] + finally: + for s in active.values(): + thread._unregister_active_message_stream(s) + + +class SyncToolCallHandle: + """Sync handle for one root-scope tool call.""" + + def __init__( + self, + *, + tool_call_id: str, + name: str, + input: Any = None, + namespace: list[str] | None = None, + max_queue_size: int = 1024, + ) -> None: + self.tool_call_id = tool_call_id + self.name = name + self.input = input + self.namespace = list(namespace or []) + self.done = False + self.error: BaseException | None = None + self._result: _BlockingResult = _BlockingResult() + self._deltas: queue.Queue[str | None] = queue.Queue(maxsize=max_queue_size) + self._deltas_consumed: bool = False + + @property + def output(self) -> Any: + """Block until the tool call completes and return its output.""" + return self._result.result() + + @property + def deltas(self) -> Iterator[str]: + """Iterate over tool output deltas emitted before the terminal event. + + Raises: + RuntimeError: if called more than once — the underlying queue is + single-consumer and cannot be fanned out safely. + """ + if self._deltas_consumed: + raise RuntimeError( + "SyncToolCallHandle.deltas can only be iterated by a single consumer." + ) + self._deltas_consumed = True + return self._delta_iter() + + def _delta_iter(self) -> Iterator[str]: + while True: + item = self._deltas.get() + if item is None: + return + yield item + + def _push_delta(self, delta: str) -> None: + if self.done: + return + self._deltas.put_nowait(delta) + + def _finish(self, output: Any) -> None: + if self.done: + return + self.done = True + self._result.set_result(output) + self._deltas.put_nowait(None) + + def _fail(self, err: BaseException) -> None: + if self.done: + return + self.done = True + self.error = err + self._result.set_exception(err) + self._deltas.put_nowait(None) + + +class _SyncToolCallsProjection: + """Typed projection for root-scope `thread.tool_calls`.""" + + def __init__( + self, thread: SyncThreadStream, namespace: list[str] | None = None + ) -> None: + self._thread = thread + self._namespace = list(namespace or []) + + def __iter__(self) -> Iterator[SyncToolCallHandle]: + return self._tool_calls_iter() + + def _tool_calls_iter(self) -> Iterator[SyncToolCallHandle]: + if self._thread._transport is None: + raise RuntimeError("SyncThreadStream not entered — use `with`.") + params = _exact_namespace_params(["tools"], self._namespace) + sub = self._thread._register_subscription(params) + active: dict[str, SyncToolCallHandle] = {} + try: + self._thread._reconcile_stream(params) + self._thread._ensure_fanout_running() + while True: + item = sub.queue.get() + if item is None: + return + params_field = item.get("params") or {} + if _event_namespace(params_field) != self._namespace: + continue + data = ( + params_field.get("data") if isinstance(params_field, dict) else None + ) + if not isinstance(data, dict): + continue + event_type = data.get("event") + tool_call_id = data.get("tool_call_id") + if not isinstance(tool_call_id, str): + continue + if event_type == "tool-started": + tool_name = data.get("tool_name") + if not isinstance(tool_name, str): + tool_name = "" + handle = SyncToolCallHandle( + tool_call_id=tool_call_id, + name=tool_name, + input=data.get("input"), + namespace=list(self._namespace), + ) + active[tool_call_id] = handle + self._thread._register_active_tool_call(handle) + # Pre-dispatch events until this tool call completes so that + # `call.output` is resolved when the caller receives the handle. + while not handle.done: + next_item = sub.queue.get() + if next_item is None: + sub.queue.put(None) + break + next_params = next_item.get("params") or {} + if _event_namespace(next_params) != self._namespace: + continue + next_data = ( + next_params.get("data") + if isinstance(next_params, dict) + else None + ) + if not isinstance(next_data, dict): + continue + next_event_type = next_data.get("event") + next_tcid = next_data.get("tool_call_id") + if not isinstance(next_tcid, str): + continue + if next_event_type == "tool-output-delta": + h = active.get(next_tcid) + delta = next_data.get("delta") + if h is not None and isinstance(delta, str): + h._push_delta(delta) + elif next_event_type == "tool-finished": + h = active.pop(next_tcid, None) + if h is not None: + self._thread._unregister_active_tool_call(h) + h._finish(next_data.get("output")) + elif next_event_type == "tool-error": + h = active.pop(next_tcid, None) + if h is not None: + self._thread._unregister_active_tool_call(h) + message = next_data.get("message") + h._fail( + RuntimeError( + str(message) if message else "Tool call errored" + ) + ) + yield handle + elif event_type == "tool-output-delta": + h = active.get(tool_call_id) + delta = data.get("delta") + if h is not None and isinstance(delta, str): + h._push_delta(delta) + elif event_type == "tool-finished": + h = active.pop(tool_call_id, None) + if h is not None: + self._thread._unregister_active_tool_call(h) + h._finish(data.get("output")) + elif event_type == "tool-error": + h = active.pop(tool_call_id, None) + if h is not None: + self._thread._unregister_active_tool_call(h) + message = data.get("message") + h._fail( + RuntimeError( + str(message) if message else "Tool call errored" + ) + ) + finally: + # Read terminal error from _run_done if it is already resolved. + # We do NOT block here: callers who need a terminal observation + # should access `thread.output` directly. Blocking in iterator + # teardown would stall every early break or exception exit for + # up to the full wait timeout (previously 1 s). + run_done = self._thread._run_done + terminal_err: BaseException | None = None + if run_done is not None and run_done.done(): + try: + terminal = run_done.result() + terminal_err = terminal.error + except Exception: + pass + err: BaseException = ( + terminal_err + if terminal_err is not None + else RuntimeError("Tool call stream closed before terminal tool event.") + ) + for h in active.values(): + self._thread._unregister_active_tool_call(h) + h._fail(err) + self._thread._unregister_subscription(sub.id) + + class SyncThreadStream: """Synchronous context manager for one thread's v3 streaming session. @@ -243,8 +673,13 @@ class SyncThreadStream: self._lifecycle_watcher_handle: SyncEventStreamHandle | None = None self._run_seen: bool = False self._run_done: _BlockingResult | None = None + self._active_message_streams: set[ChatModelStream] = set() + self._active_tool_calls: set[SyncToolCallHandle] = set() + self._root_messages_inbox: queue.Queue[Event | None] | None = None self.run = SyncRunModule(self) self.values = _SyncValuesProjection(self) + self.messages = _SyncMessagesProjection(self, namespace=[]) + self.tool_calls = _SyncToolCallsProjection(self, namespace=[]) def __enter__(self) -> SyncThreadStream: if self._closed: @@ -314,6 +749,8 @@ class SyncThreadStream: if thread is not None and thread.is_alive(): with contextlib.suppress(RuntimeError): thread.join(timeout=1.0) + self._fail_active_message_streams(RuntimeError("SyncThreadStream closed")) + self._fail_active_tool_calls(RuntimeError("SyncThreadStream closed")) if self._transport is not None: self._transport.close() @@ -339,6 +776,33 @@ class SyncThreadStream: raise RuntimeError("SyncThreadStream not entered — use `with`.") self._controller.reconcile_stream(candidate_filter) + def _activate_root_messages_inbox(self) -> queue.Queue[Event | None]: + if self._root_messages_inbox is None: + self._root_messages_inbox = queue.Queue() + return self._root_messages_inbox + + def _register_active_message_stream(self, stream: ChatModelStream) -> None: + self._active_message_streams.add(stream) + + def _unregister_active_message_stream(self, stream: ChatModelStream) -> None: + self._active_message_streams.discard(stream) + + def _fail_active_message_streams(self, err: BaseException) -> None: + for stream in list(self._active_message_streams): + stream.fail(err) + self._active_message_streams.clear() + + def _register_active_tool_call(self, handle: SyncToolCallHandle) -> None: + self._active_tool_calls.add(handle) + + def _unregister_active_tool_call(self, handle: SyncToolCallHandle) -> None: + self._active_tool_calls.discard(handle) + + def _fail_active_tool_calls(self, err: BaseException) -> None: + for handle in list(self._active_tool_calls): + handle._fail(err) + self._active_tool_calls.clear() + def subscribe( self, channels: list[str], @@ -498,6 +962,8 @@ class SyncThreadStream: error = RuntimeError( f"Run errored: {error_msg}" if error_msg else "Run errored" ) + self._fail_active_message_streams(error) + self._fail_active_tool_calls(error) run_done.set_result(_RunTerminal(status="errored", error=error)) else: run_done.set_result(_RunTerminal(status="completed")) diff --git a/libs/sdk-py/tests/streaming/_events.py b/libs/sdk-py/tests/streaming/_events.py index 1ef35f7c4..201dda5db 100644 --- a/libs/sdk-py/tests/streaming/_events.py +++ b/libs/sdk-py/tests/streaming/_events.py @@ -77,9 +77,12 @@ def message_start_event( *, message_id: str = "msg-1", role: str = "ai", - run_id: str = "run-1", + run_id: str | None = None, node: str = "agent", ) -> dict[str, Any]: + metadata: dict[str, Any] = {"langgraph_node": node} + if run_id is not None: + metadata["run_id"] = run_id return _base( seq, "messages", @@ -88,7 +91,7 @@ def message_start_event( "event": "message-start", "id": message_id, "role": role, - "metadata": {"run_id": run_id, "langgraph_node": node}, + "metadata": metadata, }, ) diff --git a/libs/sdk-py/tests/streaming/test_sync_projections.py b/libs/sdk-py/tests/streaming/test_sync_projections.py index 0325ba793..09979bad2 100644 --- a/libs/sdk-py/tests/streaming/test_sync_projections.py +++ b/libs/sdk-py/tests/streaming/test_sync_projections.py @@ -2,11 +2,29 @@ from __future__ import annotations +from typing import cast + import httpx +import pytest +from langchain_core.language_models.chat_model_stream import ChatModelStream +from langchain_protocol import Event from langgraph_sdk._sync.http import SyncHttpClient from langgraph_sdk._sync.threads import SyncThreadsClient -from streaming._events import lifecycle_completed_event +from streaming._events import ( + lifecycle_completed_event, + lifecycle_errored_event, + lifecycle_started_event, + message_error_event, + message_finish_event, + message_start_event, + message_text_delta_event, + message_text_finish_event, + tool_error_event, + tool_finished_event, + tool_output_delta_event, + tool_started_event, +) from streaming._sync_fake_server import SyncFakeServer @@ -23,3 +41,407 @@ def test_sync_values_first_yield_is_rest_state_and_output_returns_final_state(): assert first == {"answer": 42} assert output == {"answer": 42} + + +def test_sync_messages_yield_chat_model_stream(): + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + message_start_event(seq=1, message_id="msg-1"), + message_text_delta_event(seq=2, text="hi", message_id="msg-1"), + message_text_finish_event(seq=3, text="hi", message_id="msg-1"), + message_finish_event(seq=4, message_id="msg-1"), + lifecycle_completed_event(seq=5), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + messages = list(thread.messages) + + assert len(messages) == 1 + assert isinstance(messages[0], ChatModelStream) + assert messages[0].message_id == "msg-1" + assert str(messages[0].text) == "hi" + assert messages[0].output.id == "msg-1" + + +def test_sync_tool_calls_yield_handle_deltas_and_output(): + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + tool_started_event(seq=1, tool_call_id="call-1", tool_name="search"), + tool_output_delta_event(seq=2, tool_call_id="call-1", delta="a"), + tool_output_delta_event(seq=3, tool_call_id="call-1", delta="b"), + tool_finished_event(seq=4, tool_call_id="call-1", output={"ok": True}), + lifecycle_completed_event(seq=5), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + calls = list(thread.tool_calls) + + assert [call.tool_call_id for call in calls] == ["call-1"] + assert list(calls[0].deltas) == ["a", "b"] + assert calls[0].output == {"ok": True} + + +# --------------------------------------------------------------------------- +# Task 10.6 — messages pre-dispatch inner loop must filter by namespace +# --------------------------------------------------------------------------- + + +def test_sync_messages_ignores_nested_namespace_for_root_projection(): + """Root projection must not yield messages from child namespaces.""" + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + message_start_event(seq=1, namespace=["child:1"], message_id="nested"), + message_text_delta_event(seq=2, namespace=["child:1"], text="nested"), + message_finish_event(seq=3, namespace=["child:1"]), + lifecycle_completed_event(seq=4), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + messages = list(thread.messages) + + assert messages == [] + + +# --------------------------------------------------------------------------- +# Task 10.7 — comprehensive sync messages projection tests +# --------------------------------------------------------------------------- + + +def test_sync_messages_multiple_messages_are_distinct_streams(): + """Two sequential messages produce two separate ChatModelStream objects.""" + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + message_start_event(seq=1, message_id="msg-1"), + message_text_delta_event(seq=2, text="one", message_id="msg-1"), + message_text_finish_event(seq=3, text="one", message_id="msg-1"), + message_finish_event(seq=4, message_id="msg-1"), + message_start_event(seq=5, message_id="msg-2"), + message_text_delta_event(seq=6, text="two", message_id="msg-2"), + message_text_finish_event(seq=7, text="two", message_id="msg-2"), + message_finish_event(seq=8, message_id="msg-2"), + lifecycle_completed_event(seq=9), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + streams = list(thread.messages) + + assert [s.message_id for s in streams] == ["msg-1", "msg-2"] + assert [str(s.text) for s in streams] == ["one", "two"] + + +def test_sync_messages_error_event_fails_active_stream(): + """A messages `error` event marks the stream as failed.""" + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + message_start_event(seq=1, message_id="msg-1"), + message_error_event(seq=2, message="model failed", message_id="msg-1"), + lifecycle_completed_event(seq=3), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + streams = list(thread.messages) + + assert len(streams) == 1 + with pytest.raises(RuntimeError, match="model failed"): + _ = streams[0].output + + +# --------------------------------------------------------------------------- +# Task 10.7 — comprehensive sync tool_calls projection tests +# --------------------------------------------------------------------------- + + +def test_sync_tool_calls_multiple_concurrent_calls_route_by_id(): + """Two interleaved tool calls each resolve to the correct output.""" + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + tool_started_event(seq=1, tool_call_id="call-a", tool_name="alpha"), + tool_output_delta_event(seq=2, tool_call_id="call-a", delta="a1"), + tool_finished_event(seq=3, tool_call_id="call-a", output="A"), + tool_started_event(seq=4, tool_call_id="call-b", tool_name="beta"), + tool_output_delta_event(seq=5, tool_call_id="call-b", delta="b1"), + tool_finished_event(seq=6, tool_call_id="call-b", output="B"), + lifecycle_completed_event(seq=7), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + calls = list(thread.tool_calls) + + by_id = {call.tool_call_id: call for call in calls} + assert set(by_id) == {"call-a", "call-b"} + assert list(by_id["call-a"].deltas) == ["a1"] + assert list(by_id["call-b"].deltas) == ["b1"] + assert by_id["call-a"].output == "A" + assert by_id["call-b"].output == "B" + + +def test_sync_tool_calls_ignores_nested_namespace_for_root_projection(): + """Root tool_calls projection must not yield handles from child namespaces.""" + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + tool_started_event(seq=1, namespace=["child:1"], tool_call_id="nested"), + tool_finished_event(seq=2, namespace=["child:1"], tool_call_id="nested"), + lifecycle_completed_event(seq=3), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + calls = list(thread.tool_calls) + + assert calls == [] + + +def test_sync_tool_calls_error_event_fails_output(): + """A `tool-error` event fails the handle's output property.""" + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + tool_started_event(seq=1, tool_call_id="call-1"), + tool_output_delta_event(seq=2, tool_call_id="call-1", delta="before"), + tool_error_event(seq=3, tool_call_id="call-1", message="boom"), + lifecycle_completed_event(seq=4), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + calls = list(thread.tool_calls) + + assert len(calls) == 1 + assert list(calls[0].deltas) == ["before"] + with pytest.raises(RuntimeError, match="boom"): + _ = calls[0].output + + +# --------------------------------------------------------------------------- +# Task 10.10 — run-error propagates to active tool-call handle +# --------------------------------------------------------------------------- + + +def test_sync_tool_calls_run_error_fails_active_handle(): + """A lifecycle errored event propagates its message to the active handle.""" + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + tool_started_event(seq=1, tool_call_id="call-1"), + tool_output_delta_event(seq=2, tool_call_id="call-1", delta="partial"), + lifecycle_errored_event(seq=3, error="run failed"), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + calls = list(thread.tool_calls) + + assert len(calls) == 1 + with pytest.raises(RuntimeError, match="Run errored: run failed"): + _ = calls[0].output + + +# --------------------------------------------------------------------------- +# Task 10.9 — _drain_messages_inbox must pre-dispatch before yielding +# --------------------------------------------------------------------------- + + +def test_sync_drain_messages_inbox_pre_dispatches_before_yield(): + """When draining the root inbox, str(message.text) must work immediately on yield.""" + fake = SyncFakeServer() + fake.script([lifecycle_completed_event(seq=10)]) + fake.set_state({}) + collected_texts: list[str] = [] + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + # Activate the root inbox and populate it directly (simulating what + # a sync _SubgraphsProjection would do). + inbox = thread._activate_root_messages_inbox() + inbox.put( + cast( + Event, + message_start_event(seq=1, message_id="msg-x"), + ) + ) + inbox.put( + cast( + Event, + message_text_delta_event(seq=2, text="hello", message_id="msg-x"), + ) + ) + inbox.put( + cast( + Event, + message_text_finish_event(seq=3, text="hello", message_id="msg-x"), + ) + ) + inbox.put(cast(Event, message_finish_event(seq=4, message_id="msg-x"))) + inbox.put(None) # EOF sentinel + + # Read text immediately inside the for loop — this requires pre-dispatch. + for stream in thread.messages: + collected_texts.append(str(stream.text)) + + assert collected_texts == ["hello"] + + +# --------------------------------------------------------------------------- +# Fix A — remove blocking 1s wait in tool_calls iterator finally +# --------------------------------------------------------------------------- + + +def test_sync_tool_calls_explicit_close_does_not_block_1s(): + """Closing the tool_calls iterator must return in <200ms even without a terminal event. + + The prior finally block called `run_done.result(timeout=1.0)` unconditionally, + causing a mandatory 1-second stall whenever the caller breaks out of the + iterator before a lifecycle terminal event arrives. + """ + fake = SyncFakeServer() + # Script has a started lifecycle and one tool, but NO terminal lifecycle event. + # If the blocking wait is present, the iterator's finally will stall for 1s. + fake.script( + [ + lifecycle_started_event(seq=0), + tool_started_event(seq=1, tool_call_id="call-1"), + ] + ) + import time + from collections.abc import Generator + from typing import cast + + from langgraph_sdk._sync.stream import SyncToolCallHandle + + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + gen = cast( + Generator[SyncToolCallHandle, None, None], + thread.tool_calls._tool_calls_iter(), + ) + _handle = next(gen) # receive the one tool-started handle + start = time.monotonic() + # Explicitly close the generator — must not stall 1s. + gen.close() + elapsed = time.monotonic() - start + + assert elapsed < 0.2, f"tool_calls close() took {elapsed:.3f}s (expected <0.2s)" + + +# --------------------------------------------------------------------------- +# Fix B — drop len(active)==1 silent message routing fallback +# --------------------------------------------------------------------------- + + +def test_sync_messages_orphan_delta_without_matching_key_is_dropped(): + """A delta whose message_id doesn't match any active stream must be dropped. + + The old code fell back to routing the event to the only active stream when + `len(active) == 1`, causing orphan/mismatched deltas to silently corrupt + an unrelated stream's content. + """ + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=0), + message_start_event(seq=1, message_id="msg-A"), + # Delta with a mismatched message_id — must be dropped, not routed to msg-A. + message_text_delta_event(seq=2, text="orphan", message_id="msg-UNKNOWN"), + message_text_delta_event(seq=3, text="real", message_id="msg-A"), + message_finish_event(seq=4, message_id="msg-A"), + lifecycle_completed_event(seq=5), + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + threads = SyncThreadsClient(SyncHttpClient(raw)) + with threads.stream(thread_id="t-1", assistant_id="agent") as thread: + thread.run.start(input={}) + streams = list(thread.messages) + + assert len(streams) == 1 + # Only the correctly-keyed delta "real" must appear; "orphan" must be dropped. + assert str(streams[0].text) == "real" + + +# --------------------------------------------------------------------------- +# Fix C — bound SyncToolCallHandle._deltas via max_queue_size +# --------------------------------------------------------------------------- + + +def test_sync_tool_call_handle_deltas_queue_is_bounded(): + """SyncToolCallHandle._deltas must be a bounded queue. + + Unbounded queues allow producers to enqueue indefinitely, causing memory + growth when consumers are slow. + """ + from langgraph_sdk._sync.stream import SyncToolCallHandle + + handle_default = SyncToolCallHandle(tool_call_id="tc1", name="foo") + assert handle_default._deltas.maxsize > 0, ( + "default maxsize must be positive (bounded)" + ) + + handle_custom = SyncToolCallHandle(tool_call_id="tc2", name="bar", max_queue_size=8) + assert handle_custom._deltas.maxsize == 8 + + +# --------------------------------------------------------------------------- +# Fix D — enforce single consumer on SyncToolCallHandle.deltas +# --------------------------------------------------------------------------- + + +def test_sync_tool_call_handle_deltas_single_consumer_guard(): + """Accessing `handle.deltas` a second time must raise immediately. + + `_deltas` is a single-consumer queue; fanning out to multiple consumers + would cause each consumer to miss events already consumed by the other. + The property must raise before returning the iterator so the caller + sees the error even without iterating. + """ + from langgraph_sdk._sync.stream import SyncToolCallHandle + + handle = SyncToolCallHandle(tool_call_id="tc1", name="foo") + + # First access: fine — returns the iterator. + _iter_1 = handle.deltas + + # Second access: must raise immediately (before any iteration). + with pytest.raises(RuntimeError, match="single consumer"): + _ = handle.deltas