Revert "Terminate sync stream iterators on stream-end, not on root-terminal lifecycle"

This reverts commit bbc71c398b.

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)
 

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