From 7505c73243dbdfd3f81e094a803b0da69648f16a Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Tue, 2 Jun 2026 10:03:39 -0400 Subject: [PATCH] Revert "Terminate sync stream iterators on stream-end, not on root-terminal lifecycle" This reverts commit bbc71c398bd86832d44b36cd3f8ef4225179a0a9. --- .../langgraph_sdk/stream/sync_controller.py | 48 ++++++++++++++----- .../streaming/test_sync_shared_stream.py | 43 +---------------- 2 files changed, 37 insertions(+), 54 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/stream/sync_controller.py b/libs/sdk-py/langgraph_sdk/stream/sync_controller.py index d22ea7469..165c7bfdf 100644 --- a/libs/sdk-py/langgraph_sdk/stream/sync_controller.py +++ b/libs/sdk-py/langgraph_sdk/stream/sync_controller.py @@ -22,6 +22,31 @@ 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 @@ -149,14 +174,14 @@ class SyncStreamController: for sub in subscriptions: if matches_subscription(event, sub.params): sub.queue.put(event) - # 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). + # 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() except Exception: pass # transport drop — attempt reconnect below @@ -199,10 +224,9 @@ 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 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. + # 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`. 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 567c55796..456241384 100644 --- a/libs/sdk-py/tests/streaming/test_sync_shared_stream.py +++ b/libs/sdk-py/tests/streaming/test_sync_shared_stream.py @@ -8,11 +8,7 @@ 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 ( - lifecycle_completed_event, - lifecycle_started_event, - values_event, -) +from streaming._events import values_event from streaming._sync_fake_server import SyncFakeServer, SyncStreamScript @@ -89,40 +85,3 @@ 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