From bbc71c398bd86832d44b36cd3f8ef4225179a0a9 Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Mon, 1 Jun 2026 22:40:57 -0400 Subject: [PATCH] Terminate sync stream iterators on stream-end, not on root-terminal lifecycle The sync stream controller pushed the terminal None sentinel into every subscription queue the instant it saw a root-namespace completed/failed lifecycle event. Any event the server emits after that (e.g. a direct-child subgraph lifecycle/tasks event that feeds thread.subgraphs) was then queued behind the None and never consumed, because iterators return on the first None. This surfaced as thread.subgraphs yielding no direct-child handles for a factory/deep-agent graph on the sync client while the async client worked. Align the sync controller with the async controller: terminate projection iterators when the shared stream ends (or on interrupt, signaled by the thread-stream), not on a mid-stream root-terminal event. Adds a regression test. --- .../langgraph_sdk/stream/sync_controller.py | 48 +++++-------------- .../streaming/test_sync_shared_stream.py | 43 ++++++++++++++++- 2 files changed, 54 insertions(+), 37 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/stream/sync_controller.py b/libs/sdk-py/langgraph_sdk/stream/sync_controller.py index 165c7bfdf..d22ea7469 100644 --- a/libs/sdk-py/langgraph_sdk/stream/sync_controller.py +++ b/libs/sdk-py/langgraph_sdk/stream/sync_controller.py @@ -22,31 +22,6 @@ from langgraph_sdk.stream.transport import ( _logger = logging.getLogger(__name__) -_ROOT_TERMINAL_LIFECYCLE_EVENTS = frozenset({"completed", "failed"}) - - -def _is_root_terminal_lifecycle(event: Any) -> bool: - """Return True for a root-namespace lifecycle event marking run end. - - Matches the wire shape ``{method: "lifecycle", params: {namespace: [], - data: {event: "completed" | "failed"}}}``. Subgraph lifecycle events - (non-empty namespace) do not terminate the parent run. - """ - if not isinstance(event, dict): - return False - if event.get("method") != "lifecycle": - return False - params = event.get("params") or {} - if not isinstance(params, dict): - return False - if params.get("namespace") or []: - return False - data = params.get("data") or {} - if not isinstance(data, dict): - return False - return data.get("event") in _ROOT_TERMINAL_LIFECYCLE_EVENTS - - @dataclass class _SyncSubscription: id: int @@ -174,14 +149,14 @@ class SyncStreamController: for sub in subscriptions: if matches_subscription(event, sub.params): sub.queue.put(event) - # Root-terminal lifecycle: push `None` into all sub - # queues so projection iterators exit when the run - # ends naturally. Terminal is processed in seq order - # on the shared SSE, so in-flight values/tools/ - # messages events for this run are already queued - # before None. - if _is_root_terminal_lifecycle(event): - self.signal_paused() + # Do NOT terminate iterators on the root-terminal + # lifecycle event: the server may still emit trailing + # events for this run after `completed`/`failed` (e.g. + # direct-child subgraph `tasks`/`lifecycle` events that + # feed `thread.subgraphs`). Pushing `None` here orphans + # those behind the sentinel. Iterators are terminated + # when the shared stream ends (below) or on interrupt + # (`signal_paused`, driven by the thread-stream). except Exception: pass # transport drop — attempt reconnect below @@ -224,9 +199,10 @@ class SyncStreamController: filters = [dict(sub.params) for sub in self._subscriptions.values()] if extra is not None: filters.append(dict(extra)) - # Always include lifecycle in the shared SSE filter so `_fanout` - # sees root-terminal events in seq order with the projection - # events. See `_is_root_terminal_lifecycle`. + # Always include lifecycle in the shared SSE filter so the + # thread-stream observes run started/completed/failed and interrupt + # events (driving interrupt wake-up and run-done resolution) even when + # no projection subscribed to the lifecycle channel. filters.append({"channels": ["lifecycle"]}) return compute_union_filter(filters) diff --git a/libs/sdk-py/tests/streaming/test_sync_shared_stream.py b/libs/sdk-py/tests/streaming/test_sync_shared_stream.py index 456241384..567c55796 100644 --- a/libs/sdk-py/tests/streaming/test_sync_shared_stream.py +++ b/libs/sdk-py/tests/streaming/test_sync_shared_stream.py @@ -8,7 +8,11 @@ from langgraph_sdk._sync.http import SyncHttpClient from langgraph_sdk._sync.threads import SyncThreadsClient from langgraph_sdk.stream.sync_controller import SyncStreamController from langgraph_sdk.stream.transport.sync_http import SyncProtocolSseTransport -from streaming._events import values_event +from streaming._events import ( + lifecycle_completed_event, + lifecycle_started_event, + values_event, +) from streaming._sync_fake_server import SyncFakeServer, SyncStreamScript @@ -85,3 +89,40 @@ def test_sync_shared_stream_reconnects_with_since_after_transport_drop(): assert second["seq"] == 2 assert end is None assert fake.stream_request_bodies[1]["since"] == 1 + + +def test_sync_controller_delivers_child_events_after_root_terminal(): + """A direct-child event arriving after the root ``completed`` lifecycle is + still delivered to subscribers. + + Regression: the controller used to push the terminal ``None`` sentinel the + instant it saw the root-terminal lifecycle event, orphaning any trailing + child-namespace events behind it (the events that feed ``thread.subgraphs``). + Iterators now terminate on stream-end, mirroring the async controller. + """ + fake = SyncFakeServer() + fake.script( + [ + lifecycle_started_event(seq=1), + lifecycle_completed_event(seq=2), # root-terminal + lifecycle_started_event( + seq=3, namespace=["sub:0"] + ), # child, after terminal + ] + ) + with httpx.Client(transport=fake.transport, base_url="http://test") as raw: + transport = SyncProtocolSseTransport(client=raw, thread_id="t-1") + controller = SyncStreamController(transport) + sub = controller.register_subscription({"channels": ["lifecycle"]}) + controller.reconcile_stream({"channels": ["lifecycle"]}) + controller.ensure_fanout_running() + + received = [] + while True: + item = sub.queue.get(timeout=1) + if item is None: + break + received.append(item) + controller.close() + + assert lifecycle_started_event(seq=3, namespace=["sub:0"]) in received