Add sync StreamMux, SubgraphRunStream, and chat model stream enhancements

Adds AsyncStreamMux sync counterpart, exports SubgraphRunStream, improves
EventLog future resolution safety, and expands run_stream with sync graph
run support. Includes comprehensive test updates across event log, mux,
reducers, and run stream modules.

libs/langgraph/langgraph/stream/__init__.py

@@ -1,6 +1,7 @@
 """Stream protocol types and infrastructure for LangGraph."""
 
 from langgraph.stream._convert import STREAM_V2_MODES, convert_to_protocol_event
+from langgraph.stream._event_log import EventLog
 from langgraph.stream._mux import AsyncStreamMux, StreamMux
 from langgraph.stream._types import (
     InterruptPayload,
@@ -12,6 +13,7 @@ from langgraph.stream.run_stream import (
     AsyncGraphRunStream,
     AsyncSubgraphRunStream,
     GraphRunStream,
+    SubgraphRunStream,
     create_async_graph_run_stream,
     create_graph_run_stream,
 )
@@ -24,17 +26,19 @@ from langgraph.stream.transformers import (
 
 __all__ = [
     "STREAM_V2_MODES",
+    "AsyncStreamMux",
     "AsyncChatModelStream",
     "AsyncGraphRunStream",
-    "AsyncStreamMux",
     "AsyncSubgraphRunStream",
     "ChatModelStream",
+    "EventLog",
     "GraphRunStream",
     "InterruptPayload",
     "MessagesTransformer",
     "ProtocolEvent",
     "StreamChannel",
     "StreamMux",
+    "SubgraphRunStream",
     "StreamTransformer",
     "StreamingHandler",
     "ValuesTransformer",

libs/langgraph/langgraph/stream/_convert.py

@@ -7,6 +7,7 @@ original payload.
 
 from __future__ import annotations
 
+import time
 from typing import Any
 
 from langgraph.stream._types import ProtocolEvent, _ProtocolEventParams
@@ -41,17 +42,23 @@ def convert_to_protocol_event(
     The ``seq`` field is left as ``0`` here; the :class:`StreamMux` is
     the sole seq assigner and overwrites it inside ``push()``.
 
-    Args:
-        ns: Namespace tuple from the ``StreamChunk``.
-        mode: Stream mode string (``"values"``, ``"updates"``, etc.).
-        payload: The raw payload from the stream.
-        node: Optional node name for provenance.
+    Parameters
+    ----------
+    ns:
+        Namespace tuple from the ``StreamChunk``.
+    mode:
+        Stream mode string (``"values"``, ``"updates"``, etc.).
+    payload:
+        The raw payload from the stream.
+    node:
+        Optional node name for provenance.
     """
     if mode not in _SUPPORTED_MODES:
         return None
 
     params: _ProtocolEventParams = {
         "namespace": list(ns),
+        "timestamp": _now_ms(),
         "data": payload,
     }
     if node is not None:
@@ -64,4 +71,9 @@ def convert_to_protocol_event(
     )
 
 
+def _now_ms() -> int:
+    """Current time in milliseconds since epoch."""
+    return int(time.time() * 1000)
+
+
 __all__ = ["STREAM_V2_MODES", "convert_to_protocol_event"]

libs/langgraph/langgraph/stream/_event_log.py

@@ -0,0 +1,136 @@
+"""Replayable append-only event buffer for StreamingHandler.
+
+``EventLog`` stores protocol events in an ordered list and supports
+multiple independent async iterators, each with their own cursor
+offset.  Subscribers that join mid-stream replay from a given offset
+without losing earlier events.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+def _resolve_future(fut: asyncio.Future[None]) -> None:
+    """Set a future's result if it hasn't already completed or been cancelled.
+
+    Runs on the event loop thread (scheduled via ``call_soon_threadsafe``)
+    so that the ``done()`` check and ``set_result`` are atomic with
+    respect to cancellation.
+    """
+    if not fut.done():
+        fut.set_result(None)
+
+
+class EventLog(Generic[T]):
+    """Append-only event buffer with cursor-based async iteration.
+
+    Multiple consumers can subscribe independently and each will see
+    every event from their starting offset onward.
+    """
+
+    __slots__ = ("_items", "_closed", "_error", "_waiters", "_lock")
+
+    def __init__(self) -> None:
+        self._items: list[T] = []
+        self._closed = False
+        self._error: BaseException | None = None
+        self._waiters: list[asyncio.Future[None]] = []
+        self._lock = threading.Lock()
+
+    # -- Producer API -------------------------------------------------------
+
+    def append(self, item: T) -> None:
+        """Append an event and wake all waiting consumers."""
+        with self._lock:
+            if self._closed:
+                raise RuntimeError("EventLog is closed")
+            self._items.append(item)
+            self._wake_all()
+
+    def close(self) -> None:
+        """Mark the log as complete.  Iterators will end gracefully."""
+        with self._lock:
+            self._closed = True
+            self._wake_all()
+
+    def fail(self, error: BaseException) -> None:
+        """Mark the log as failed.  Iterators will raise *error*."""
+        with self._lock:
+            self._error = error
+            self._closed = True
+            self._wake_all()
+
+    # -- Consumer API -------------------------------------------------------
+
+    def __aiter__(self) -> _Cursor[T]:
+        """Return a fresh cursor from the beginning of the log."""
+        return _Cursor(self)
+
+    # -- Inspection ---------------------------------------------------------
+
+    def __len__(self) -> int:
+        return len(self._items)
+
+    def __getitem__(self, index: int) -> T:
+        return self._items[index]
+
+    @property
+    def closed(self) -> bool:
+        return self._closed
+
+    # -- Internal -----------------------------------------------------------
+
+    def _wake_all(self) -> None:
+        for fut in self._waiters:
+            try:
+                fut.get_loop().call_soon_threadsafe(_resolve_future, fut)
+            except RuntimeError:
+                # Loop already closed — ignore.
+                pass
+        self._waiters.clear()
+
+
+class _Cursor(Generic[T]):
+    """An independent async iterator over an :class:`EventLog`."""
+
+    __slots__ = ("_log", "_offset")
+
+    def __init__(self, log: EventLog[T]) -> None:
+        self._log = log
+        self._offset = 0
+
+    def __aiter__(self) -> _Cursor[T]:
+        return self
+
+    async def __anext__(self) -> T:
+        while True:
+            with self._log._lock:
+                if self._offset < len(self._log._items):
+                    item = self._log._items[self._offset]
+                    self._offset += 1
+                    return item
+                if self._log._error is not None:
+                    raise self._log._error
+                if self._log._closed:
+                    raise StopAsyncIteration
+                # Nothing available yet — register a waiter
+                fut: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+                self._log._waiters.append(fut)
+            # Wait outside the lock
+            try:
+                await fut
+            except asyncio.CancelledError:
+                with self._log._lock:
+                    try:
+                        self._log._waiters.remove(fut)
+                    except ValueError:
+                        pass  # Already removed by _wake_all
+                raise
+
+
+__all__ = ["EventLog"]

libs/langgraph/langgraph/stream/_mux.py

@@ -1,37 +1,40 @@
 """Central event dispatcher with transformer pipeline for StreamingHandler.
 
-``StreamMux`` is the synchronous core: it holds the main
-event log (a plain list), tracks discovered namespaces for subgraph stream
+``StreamMux`` is the sync-safe core: it holds the main
+:class:`EventLog`, tracks discovered namespaces for subgraph stream
 creation, and pipes every event through the registered
 :class:`StreamTransformer` pipeline before appending it to the log.
 
-``AsyncStreamMux`` extends ``StreamMux`` with async consumer APIs
-(output futures, async event subscriptions, subgraph discovery).
+``AsyncStreamMux`` extends the base with async subscription endpoints
+(output futures, namespace waiters, filtered event iteration).
 """
 
 from __future__ import annotations
 
 import asyncio
+import time
 from collections.abc import AsyncIterator
 from typing import Any
 
+from langgraph.stream._event_log import EventLog
 from langgraph.stream._types import InterruptPayload, ProtocolEvent, StreamTransformer
 from langgraph.stream.stream_channel import StreamChannel, is_stream_channel
 
 
 class StreamMux:
-    """Synchronous event dispatcher for the StreamingHandler infrastructure.
+    """Sync-safe event dispatcher for the StreamingHandler infrastructure.
 
     The mux owns the main event log, applies the transformer pipeline to
     every incoming event, and tracks namespace discovery and latest values.
 
-    For async consumer APIs (output futures, async event subscriptions,
-    subgraph discovery) use :class:`AsyncStreamMux`.
+    For async subscription endpoints (output futures, namespace waiters,
+    filtered event iteration), use :class:`AsyncStreamMux`.
     """
 
     def __init__(self, transformers: list[StreamTransformer] | None = None) -> None:
-        self._event_log: list[ProtocolEvent] = []
+        self._event_log: EventLog[ProtocolEvent] = EventLog()
         self._transformers: list[StreamTransformer] = list(transformers or [])
+        self._channels: list[StreamChannel[Any]] = []
         self._current_namespace: list[str] = []
         self._next_emit_seq: int = 0
 
@@ -73,6 +76,7 @@ class StreamMux:
             top_segment = ns[0]
             if top_segment not in self._discovered_ns:
                 self._discovered_ns[top_segment] = True
+                self._on_ns_discovered(top_segment)
 
         # Track values
         if event["method"] == "values":
@@ -109,23 +113,41 @@ class StreamMux:
             self._event_log.append(event)
 
     def close(self, output: Any = None) -> None:
-        """Close the mux and finalize all transformers."""
+        """Close the mux, finalizing transformers and the event log."""
         if self._closed:
             return
         self._closed = True
 
+        # Finalize transformers (optional method)
         for transformer in self._transformers:
-            transformer.finalize()
+            if hasattr(transformer, "finalize"):
+                transformer.finalize()
+
+        # Close wired channels
+        for channel in self._channels:
+            channel._close()
+
+        # Close the event log
+        self._event_log.close()
 
     def fail(self, error: BaseException) -> None:
-        """Fail the mux and propagate the error to all consumers."""
+        """Fail the mux, propagating the error to transformers and channels."""
         if self._closed:
             return
         self._closed = True
         self._error = error
 
+        # Fail transformers (optional method)
         for transformer in self._transformers:
-            transformer.fail(error)
+            if hasattr(transformer, "fail"):
+                transformer.fail(error)
+
+        # Fail wired channels
+        for channel in self._channels:
+            channel._fail(error)
+
+        # Fail the event log
+        self._event_log.fail(error)
 
     # -- Inspection ---------------------------------------------------------
 
@@ -138,7 +160,7 @@ class StreamMux:
         return list(self._interrupts)
 
     @property
-    def event_log(self) -> list[ProtocolEvent]:
+    def event_log(self) -> EventLog[ProtocolEvent]:
         return self._event_log
 
     def get_latest_values(self, ns: list[str] | None = None) -> Any:
@@ -147,28 +169,38 @@ class StreamMux:
 
     # -- Internal -----------------------------------------------------------
 
+    def _on_ns_discovered(self, segment: str) -> None:
+        """Hook called when a new top-level namespace segment is discovered.
+
+        The base implementation is a no-op.  :class:`AsyncStreamMux`
+        overrides this to wake namespace waiters.
+        """
+
     def register_transformer(self, transformer: StreamTransformer) -> None:
         """Register a new transformer and replay all buffered events through it.
 
         This is the safe way to add a late-arriving transformer after the mux
         has already started processing events.  The sequence is:
 
-        1. Snapshot the current log length.
+        1. Snapshot the current log length (no await → no gap possible in
+           asyncio's cooperative threading model).
         2. Append the transformer so future ``push()`` calls reach it.
         3. Replay events ``[0, snapshot)`` through the transformer.
         4. If the mux is already closed, call ``finalize()`` immediately so
            the transformer's log/channel terminates cleanly.
 
-        No namespace filtering is applied — all buffered events are
-        replayed.  Transformers that need namespace filtering should do
-        so inside their ``process()`` implementation.
+        ``process()`` is only called for events whose namespace starts with
+        any prefix — callers that need namespace filtering should do so inside
+        their ``process()`` implementation, or wrap this call with their own
+        filtering logic.
         """
         snapshot = len(self._event_log)
         self._transformers.append(transformer)
         for i in range(snapshot):
             transformer.process(self._event_log[i])
         if self._closed:
-            transformer.finalize()
+            if hasattr(transformer, "finalize"):
+                transformer.finalize()
 
     def wire_channels(self, projection: Any) -> None:
         """Scan *projection* for :class:`StreamChannel` instances and wire them.
@@ -191,6 +223,8 @@ class StreamMux:
         for _key, value in items.items():
             if is_stream_channel(value):
                 channel: StreamChannel[Any] = value
+                self._channels.append(channel)
+
                 def _make_forwarder(ch: StreamChannel[Any]) -> Any:
                     def _forward(item: Any) -> None:
                         if self._closed:
@@ -206,6 +240,7 @@ class StreamMux:
                                 method=ch.channel_name,
                                 params={
                                     "namespace": list(self._current_namespace),
+                                    "timestamp": int(time.time() * 1000),
                                     "data": item,
                                 },
                             )
@@ -217,42 +252,31 @@ class StreamMux:
                 channel._wire(_make_forwarder(channel))
 
 
-# ---------------------------------------------------------------------------
-# AsyncStreamMux — async consumer APIs on top of the sync core
-# ---------------------------------------------------------------------------
-
-
 class AsyncStreamMux(StreamMux):
     """Async extension of :class:`StreamMux`.
 
-    Adds output futures, async event subscriptions, and subgraph
-    discovery on top of the synchronous producer/transformer core.
+    Adds output futures, namespace waiters, and async subscription
+    endpoints (``subscribe_events``, ``subscribe_subgraphs``,
+    ``get_output_future``).
     """
 
     def __init__(self, transformers: list[StreamTransformer] | None = None) -> None:
-        super().__init__(transformers=transformers)
-        # Notification event — set on every push/close/fail to wake async consumers
-        self._notify: asyncio.Event = asyncio.Event()
+        super().__init__(transformers)
         # Waiters for new namespace discovery
         self._ns_waiters: list[asyncio.Future[None]] = []
         # Output promise tracking
         self._output_futures: dict[str, asyncio.Future[Any]] = {}
 
-    # -- Producer overrides (extend to resolve async primitives) -------------
-
-    def push(self, event: ProtocolEvent) -> None:
-        # Peek at namespace before super().push() so we can detect new
-        # discoveries and wake waiters.
-        ns = event["params"].get("namespace", [])
-        is_new_ns = bool(ns) and ns[0] not in self._discovered_ns
-        super().push(event)
-        if is_new_ns and ns[0] in self._discovered_ns:
-            self._wake_ns_waiters()
-        self._notify.set()
+    # -- Producer API overrides ---------------------------------------------
 
     def close(self, output: Any = None) -> None:
+        """Close the mux, resolving all output futures."""
+        if self._closed:
+            return
+
+        # Let the base class finalize transformers, channels, and event log
         super().close(output)
-        self._notify.set()
+
         # Resolve output futures
         for ns_key, fut in self._output_futures.items():
             if not fut.done():
@@ -261,12 +285,18 @@ class AsyncStreamMux(StreamMux):
                     fut.get_loop().call_soon_threadsafe(fut.set_result, value)
                 except RuntimeError:
                     pass
+
         # Wake namespace waiters
         self._wake_ns_waiters()
 
     def fail(self, error: BaseException) -> None:
+        """Fail the mux, rejecting all output futures."""
+        if self._closed:
+            return
+
+        # Let the base class fail transformers, channels, and event log
         super().fail(error)
-        self._notify.set()
+
         # Reject output futures
         for fut in self._output_futures.values():
             if not fut.done():
@@ -274,38 +304,25 @@ class AsyncStreamMux(StreamMux):
                     fut.get_loop().call_soon_threadsafe(fut.set_exception, error)
                 except RuntimeError:
                     pass
+
         # Wake namespace waiters
         self._wake_ns_waiters()
 
-    # -- Async consumer API -------------------------------------------------
+    # -- Consumer API -------------------------------------------------------
 
-    async def subscribe_events(
-        self, path: list[str] | None = None, offset: int = 0
+    def subscribe_events(
+        self, path: list[str] | None = None
     ) -> AsyncIterator[ProtocolEvent]:
-        """Async iterate over events matching *path*.
+        """Return an async iterator over events matching *path*.
 
         If *path* is ``None`` or empty, all events are yielded.
         Otherwise, only events whose namespace starts with *path*
         are yielded.
-
-        Uses the list + ``asyncio.Event`` notification pattern: poll
-        the event log, yield what's new, await the notify event for more.
         """
-        cursor = offset
-        while True:
-            while cursor < len(self._event_log):
-                event = self._event_log[cursor]
-                cursor += 1
-                if not path or _ns_starts_with(
-                    event["params"].get("namespace", []), path
-                ):
-                    yield event
-            if self._closed:
-                if self._error is not None:
-                    raise self._error
-                return
-            self._notify.clear()
-            await self._notify.wait()
+        cursor = aiter(self._event_log)
+        if not path:
+            return cursor
+        return _FilteredEventIterator(cursor, path)
 
     async def subscribe_subgraphs(
         self, path: list[str] | None = None, offset: int = 0
@@ -359,6 +376,10 @@ class AsyncStreamMux(StreamMux):
 
     # -- Internal -----------------------------------------------------------
 
+    def _on_ns_discovered(self, segment: str) -> None:
+        """Wake namespace waiters when a new namespace is discovered."""
+        self._wake_ns_waiters()
+
     def _wake_ns_waiters(self) -> None:
         for fut in self._ns_waiters:
             if not fut.done():
@@ -369,6 +390,25 @@ class AsyncStreamMux(StreamMux):
         self._ns_waiters.clear()
 
 
+class _FilteredEventIterator:
+    """Async iterator that filters events by namespace prefix."""
+
+    __slots__ = ("_cursor", "_path")
+
+    def __init__(self, cursor: AsyncIterator[ProtocolEvent], path: list[str]) -> None:
+        self._cursor = cursor
+        self._path = path
+
+    def __aiter__(self) -> _FilteredEventIterator:
+        return self
+
+    async def __anext__(self) -> ProtocolEvent:
+        while True:
+            event = await self._cursor.__anext__()
+            ns = event["params"].get("namespace", [])
+            if _ns_starts_with(ns, self._path):
+                return event
+
 
 def _ns_key(ns: list[str] | tuple[str, ...]) -> str:
     """Convert a namespace list to a hashable key."""

libs/langgraph/langgraph/stream/_types.py

@@ -6,8 +6,7 @@ in-process-only types needed by the LangGraph streaming infrastructure.
 
 from __future__ import annotations
 
-from abc import ABC, abstractmethod
-from typing import Any
+from typing import Any, Protocol, runtime_checkable
 
 # ---------------------------------------------------------------------------
 # Re-exports from langchain-protocol (CDDL-derived)
@@ -56,6 +55,7 @@ class _ProtocolEventParams(TypedDict):
     """Payload envelope for a :class:`ProtocolEvent`."""
 
     namespace: Namespace
+    timestamp: int
     node: NotRequired[str]
     data: Any
 
@@ -74,7 +74,8 @@ class ProtocolEvent(TypedDict):
     params: _ProtocolEventParams
 
 
-class StreamTransformer(ABC):
+@runtime_checkable
+class StreamTransformer(Protocol):
     """Extension point for custom stream projections.
 
     Implementations are registered with ``StreamingHandler`` and receive every
@@ -86,7 +87,6 @@ class StreamTransformer(ABC):
 
     """
 
-    @abstractmethod
     def init(self) -> Any:
         """Return the initial projection value.
 
@@ -96,7 +96,6 @@ class StreamTransformer(ABC):
         """
         ...
 
-    @abstractmethod
     def process(self, event: ProtocolEvent) -> bool:
         """Process an event.
 
@@ -111,6 +110,7 @@ class StreamTransformer(ABC):
         Optional — the mux auto-closes any :class:`StreamChannel` instances,
         so transformers that only use channels can omit this.
         """
+        ...
 
     def fail(self, err: BaseException) -> None:
         """Called once when the run fails.
@@ -118,6 +118,7 @@ class StreamTransformer(ABC):
         Optional — the mux auto-fails any :class:`StreamChannel` instances,
         so transformers that only use channels can omit this.
         """
+        ...
 
 
 class InterruptPayload(TypedDict):

libs/langgraph/langgraph/stream/chat_model_stream.py

@@ -13,11 +13,69 @@ or ``full = await msg.text``).
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Generator
+from collections.abc import Callable, Generator, Iterator
 from typing import Any
 
 from langgraph.stream._types import UsageInfo
 
+
+# ---------------------------------------------------------------------------
+# Sync dual projection — iterable of deltas, str() for accumulated text
+# ---------------------------------------------------------------------------
+
+
+class _SyncDualProjection:
+    """Pump-driven sync iterable of string deltas.
+
+    Iterating yields incremental text fragments as the pump delivers
+    new ``content-block-delta`` events.  Calling ``str()`` drains the
+    pump and returns the full accumulated string.
+
+    This is the sync counterpart of :class:`_DualProjection` (the async
+    variant used by ``AsyncChatModelStream``).
+    """
+
+    __slots__ = ("_stream", "_attr", "_pump_one")
+
+    def __init__(
+        self,
+        stream: ChatModelStream,
+        attr: str,
+        pump_one: Callable[[], bool],
+    ) -> None:
+        self._stream = stream
+        self._attr = attr
+        self._pump_one = pump_one
+
+    def __iter__(self) -> Iterator[str]:
+        prev_len = 0
+        while True:
+            cur = getattr(self._stream, self._attr)
+            if len(cur) > prev_len:
+                yield cur[prev_len:]
+                prev_len = len(cur)
+            if self._stream._done:
+                return
+            if not self._pump_one():
+                # Source exhausted — yield any remaining
+                cur = getattr(self._stream, self._attr)
+                if len(cur) > prev_len:
+                    yield cur[prev_len:]
+                return
+
+    def __str__(self) -> str:
+        while not self._stream._done:
+            if not self._pump_one():
+                break
+        return getattr(self._stream, self._attr)
+
+    def __repr__(self) -> str:
+        return repr(getattr(self._stream, self._attr))
+
+    def __bool__(self) -> bool:
+        return bool(getattr(self._stream, self._attr))
+
+
 # ---------------------------------------------------------------------------
 # Sync variant
 # ---------------------------------------------------------------------------
@@ -56,21 +114,52 @@ class ChatModelStream:
         self._usage_value: UsageInfo | None = None
         self._done = False
 
+        # Optional pump for sync streaming (set via _bind_pump)
+        self._pump_one: Callable[[], bool] | None = None
+
+    # -- Pump binding (called by GraphRunStream) ---------------------------
+
+    def _bind_pump(self, pump_one: Callable[[], bool]) -> None:
+        """Bind a pump function for sync token-by-token streaming.
+
+        When bound, ``.text`` and ``.reasoning`` return
+        :class:`_SyncDualProjection` instances that drive the pump and
+        yield deltas as the LLM produces tokens.
+        """
+        self._pump_one = pump_one
+
     # -- Public projections ------------------------------------------------
 
     @property
-    def text(self) -> str:
-        """Accumulated text content."""
+    def text(self) -> str | _SyncDualProjection:
+        """Text content.
+
+        When a pump is bound (sync streaming), returns a
+        :class:`_SyncDualProjection` — iterable of deltas,
+        ``str()`` for the full accumulated text.  Otherwise returns
+        the accumulated text string directly.
+        """
+        if self._pump_one is not None and not self._done:
+            return _SyncDualProjection(self, "_text_acc", self._pump_one)
         return self._text_acc
 
     @property
-    def reasoning(self) -> str:
-        """Accumulated reasoning content."""
+    def reasoning(self) -> str | _SyncDualProjection:
+        """Reasoning content.
+
+        Same dual behavior as :attr:`text`.
+        """
+        if self._pump_one is not None and not self._done:
+            return _SyncDualProjection(self, "_reasoning_acc", self._pump_one)
         return self._reasoning_acc
 
     @property
     def usage(self) -> UsageInfo | None:
         """Usage info, available after the message finishes."""
+        if self._pump_one is not None and not self._done:
+            while not self._done:
+                if not self._pump_one():
+                    break
         return self._usage_value
 
     @property
@@ -130,22 +219,22 @@ class ChatModelStream:
 
 
 # ---------------------------------------------------------------------------
-# Dual-projection helpers — sync data container + async notification layer
+# Async dual-projection helpers
 # ---------------------------------------------------------------------------
 
 
 class _DualProjection:
-    """Sync data container for incremental deltas and a final value.
+    """Async iterable of deltas that is also awaitable for the final value.
 
-    Stores deltas as they arrive and tracks the final accumulated value.
-    No async primitives — see :class:`_AsyncDualProjection` for the
-    async-iterable + awaitable extension.
+    When iterated, yields delta values (e.g. text fragments) as they arrive.
+    When awaited, returns the accumulated final value (e.g. full text string).
     """
 
     def __init__(self) -> None:
         self._deltas: list[Any] = []
         self._done = False
         self._error: BaseException | None = None
+        self._waiters: list[asyncio.Future[None]] = []
         self._final_value: Any = None
         self._final_set = False
 
@@ -154,48 +243,33 @@ class _DualProjection:
     def _push(self, delta: Any) -> None:
         """Add a new delta value."""
         self._deltas.append(delta)
+        self._wake()
 
     def _finish(self, accumulated: Any) -> None:
         """Set the final accumulated value and mark as done."""
         self._final_value = accumulated
         self._final_set = True
         self._done = True
+        self._wake()
 
     def _fail(self, error: BaseException) -> None:
         self._error = error
         self._done = True
+        self._wake()
 
-
-class _AsyncDualProjection(_DualProjection):
-    """Async extension of :class:`_DualProjection`.
-
-    Async iterable of deltas that is also awaitable for the final value.
-    Uses an ``asyncio.Event`` to notify async consumers when new data
-    arrives — the same pattern as :class:`AsyncStreamMux`.
-    """
-
-    def __init__(self) -> None:
-        super().__init__()
-        self._notify: asyncio.Event = asyncio.Event()
-
-    # -- Producer overrides (extend to notify) -----------------------------
-
-    def _push(self, delta: Any) -> None:
-        super()._push(delta)
-        self._notify.set()
-
-    def _finish(self, accumulated: Any) -> None:
-        super()._finish(accumulated)
-        self._notify.set()
-
-    def _fail(self, error: BaseException) -> None:
-        super()._fail(error)
-        self._notify.set()
+    def _wake(self) -> None:
+        for fut in self._waiters:
+            if not fut.done():
+                try:
+                    fut.get_loop().call_soon_threadsafe(fut.set_result, None)
+                except RuntimeError:
+                    pass
+        self._waiters.clear()
 
     # -- Async iterable (yields deltas) ------------------------------------
 
-    def __aiter__(self) -> _AsyncDualProjectionIterator:
-        return _AsyncDualProjectionIterator(self)
+    def __aiter__(self) -> _DualProjectionIterator:
+        return _DualProjectionIterator(self)
 
     # -- Awaitable (returns final value) -----------------------------------
 
@@ -206,23 +280,25 @@ class _AsyncDualProjection(_DualProjection):
         while not self._final_set:
             if self._error is not None:
                 raise self._error
-            self._notify.clear()
-            await self._notify.wait()
+            loop = asyncio.get_running_loop()
+            fut: asyncio.Future[None] = loop.create_future()
+            self._waiters.append(fut)
+            await fut
         if self._error is not None:
             raise self._error
         return self._final_value
 
 
-class _AsyncDualProjectionIterator:
-    """Async iterator over an :class:`_AsyncDualProjection`'s deltas."""
+class _DualProjectionIterator:
+    """Async iterator over a :class:`_DualProjection`'s deltas."""
 
     __slots__ = ("_proj", "_offset")
 
-    def __init__(self, proj: _AsyncDualProjection) -> None:
+    def __init__(self, proj: _DualProjection) -> None:
         self._proj = proj
         self._offset = 0
 
-    def __aiter__(self) -> _AsyncDualProjectionIterator:
+    def __aiter__(self) -> _DualProjectionIterator:
         return self
 
     async def __anext__(self) -> Any:
@@ -235,8 +311,10 @@ class _AsyncDualProjectionIterator:
                 raise self._proj._error
             if self._proj._done:
                 raise StopAsyncIteration
-            self._proj._notify.clear()
-            await self._proj._notify.wait()
+            loop = asyncio.get_running_loop()
+            fut: asyncio.Future[None] = loop.create_future()
+            self._proj._waiters.append(fut)
+            await fut
 
 
 # ---------------------------------------------------------------------------
@@ -268,24 +346,24 @@ class AsyncChatModelStream(ChatModelStream):
         message_id: str | None = None,
     ) -> None:
         super().__init__(namespace=namespace, node=node, message_id=message_id)
-        self._text_proj = _AsyncDualProjection()
-        self._reasoning_proj = _AsyncDualProjection()
-        self._usage_proj = _AsyncDualProjection()
+        self._text_proj = _DualProjection()
+        self._reasoning_proj = _DualProjection()
+        self._usage_proj = _DualProjection()
 
     # -- Public projections (override sync properties) ---------------------
 
     @property
-    def text(self) -> _AsyncDualProjection:
+    def text(self) -> _DualProjection:
         """Text content — async iterable of deltas, awaitable for full text."""
         return self._text_proj
 
     @property
-    def reasoning(self) -> _AsyncDualProjection:
+    def reasoning(self) -> _DualProjection:
         """Reasoning content — async iterable of deltas, awaitable for full text."""
         return self._reasoning_proj
 
     @property
-    def usage(self) -> _AsyncDualProjection:
+    def usage(self) -> _DualProjection:
         """Usage info — awaitable for :class:`UsageInfo`."""
         return self._usage_proj
 
@@ -321,4 +399,4 @@ class AsyncChatModelStream(ChatModelStream):
         self._usage_proj._fail(error)
 
 
-__all__ = ["AsyncChatModelStream", "ChatModelStream"]
+__all__ = ["AsyncChatModelStream", "ChatModelStream", "_SyncDualProjection"]

libs/langgraph/langgraph/stream/run_stream.py

@@ -2,13 +2,9 @@
 
 These are the top-level objects returned by
 ``StreamingHandler.stream()`` / ``StreamingHandler.astream()``.
-
-``AsyncGraphRunStream`` wraps an :class:`AsyncStreamMux` and exposes
-``.values``, ``.messages``, ``.subgraphs``, ``.output``, and
-``.messages_from()``.
-
-``GraphRunStream`` wraps a :class:`StreamMux` and exposes the sync
-equivalents: ``.values``, ``.messages``, and ``.output``.
+They wrap a :class:`StreamMux` and expose named
+projections (``.values``, ``.messages``, ``.subgraphs``, ``.output``)
+for ergonomic consumption.
 """
 
 from __future__ import annotations
@@ -18,6 +14,7 @@ from collections.abc import AsyncIterator, Callable, Iterator
 from typing import Any
 
 from langgraph.stream._convert import convert_to_protocol_event
+from langgraph.stream._event_log import EventLog
 from langgraph.stream._mux import AsyncStreamMux, StreamMux
 from langgraph.stream._types import InterruptPayload, ProtocolEvent, StreamTransformer
 from langgraph.stream.chat_model_stream import AsyncChatModelStream, ChatModelStream
@@ -43,23 +40,8 @@ class _ValuesProjection:
         self._ns = ns
         self._mapper = mapper
 
-    async def __aiter__(self) -> AsyncIterator[Any]:
-        log = self._values_transformer.values_log
-        cursor = 0
-        while True:
-            while cursor < len(log):
-                item = log[cursor]
-                cursor += 1
-                if item.get("namespace", []) == self._ns:
-                    data = item["data"]
-                    if data is not None and self._mapper is not None:
-                        yield self._mapper(data)
-                    else:
-                        yield data
-            if self._mux._closed:
-                return
-            self._mux._notify.clear()
-            await self._mux._notify.wait()
+    def __aiter__(self) -> AsyncIterator[Any]:
+        return _ValuesIterator(self._values_transformer, self._ns, self._mapper)
 
     def __await__(self) -> Any:
         return self._await_impl().__await__()
@@ -71,6 +53,33 @@ class _ValuesProjection:
         return value
 
 
+class _ValuesIterator:
+    """Filters the values log to events matching a namespace."""
+
+    def __init__(
+        self,
+        transformer: ValuesTransformer,
+        ns: list[str],
+        mapper: Callable[[Any], Any] | None = None,
+    ) -> None:
+        self._cursor = aiter(transformer.values_log)
+        self._ns = ns
+        self._mapper = mapper
+
+    def __aiter__(self) -> _ValuesIterator:
+        return self
+
+    async def __anext__(self) -> Any:
+        while True:
+            item = await self._cursor.__anext__()
+            item_ns = item.get("namespace", [])
+            if item_ns == self._ns:
+                data = item["data"]
+                if data is not None and self._mapper is not None:
+                    return self._mapper(data)
+                return data
+
+
 # ---------------------------------------------------------------------------
 # Messages projection
 # ---------------------------------------------------------------------------
@@ -79,21 +88,11 @@ class _ValuesProjection:
 class _MessagesProjection:
     """Async iterable of :class:`AsyncChatModelStream` instances."""
 
-    def __init__(self, mux: AsyncStreamMux, messages_transformer: MessagesTransformer) -> None:
-        self._mux = mux
+    def __init__(self, messages_transformer: MessagesTransformer) -> None:
         self._transformer = messages_transformer
 
-    async def __aiter__(self) -> AsyncIterator[AsyncChatModelStream]:
-        log = self._transformer.messages_log
-        cursor = 0
-        while True:
-            while cursor < len(log):
-                yield log[cursor]
-                cursor += 1
-            if self._mux._closed:
-                return
-            self._mux._notify.clear()
-            await self._mux._notify.wait()
+    def __aiter__(self) -> AsyncIterator[AsyncChatModelStream]:
+        return aiter(self._transformer.messages_log)
 
 
 # ---------------------------------------------------------------------------
@@ -187,7 +186,7 @@ class AsyncGraphRunStream:
     def messages(self) -> _MessagesProjection:
         """Async iterable of :class:`AsyncChatModelStream` instances."""
         t = self._find_transformer("messages")
-        return _MessagesProjection(self._mux, t)
+        return _MessagesProjection(t)
 
     def messages_from(self, node: str) -> _MessagesProjection:
         """Async iterable of messages from a specific node."""
@@ -197,7 +196,7 @@ class AsyncGraphRunStream:
             stream_cls=AsyncChatModelStream,
         )
         self._mux.register_transformer(filtered)
-        return _MessagesProjection(self._mux, filtered)
+        return _MessagesProjection(filtered)
 
     @property
     def subgraphs(self) -> _SubgraphsProjection:
@@ -356,16 +355,15 @@ async def create_async_graph_run_stream(
 
 
 class _PumpDrivenLog:
-    """Wraps a list so that iteration drives the sync pump.
+    """Wraps an ``EventLog`` so that iteration drives the sync pump.
 
-    Used by all :class:`GraphRunStream` projections (``__iter__``,
-    ``.values``, ``.messages``, ``.extensions``) so that iterating
-    any projection lazily consumes the source.
+    Used by :attr:`GraphRunStream.extensions` to make extension logs
+    iterable without requiring the caller to drain the stream first.
     """
 
     __slots__ = ("_log", "_pump_one")
 
-    def __init__(self, log: list, pump_one: Callable[[], bool]) -> None:
+    def __init__(self, log: EventLog, pump_one: Callable[[], bool]) -> None:
         self._log = log
         self._pump_one = pump_one
 
@@ -494,20 +492,91 @@ class GraphRunStream:
     def messages(self) -> Iterator[ChatModelStream]:
         """Sync iterable of :class:`ChatModelStream` instances.
 
-        Each yielded ``ChatModelStream`` is fully populated (``done=True``)
-        so that sync consumers can read ``.text``, ``.reasoning``, and
-        ``.usage`` immediately.
+        Each ``ChatModelStream`` is yielded as soon as the LLM begins
+        responding (on ``message-start``).  Its ``.text`` and
+        ``.reasoning`` properties are pump-driven
+        :class:`~langgraph.stream.chat_model_stream._SyncDualProjection`
+        instances that yield deltas as tokens arrive::
+
+            for msg in run.messages:
+                for delta in msg.text:
+                    print(delta, end="", flush=True)
+
+        If you don't need streaming, ``str(msg.text)`` pumps until
+        the message completes and returns the full text.
+
+        After each message is consumed, the pump advances through
+        non-message events (tool completions, values, etc.) so that
+        other transformer state is up-to-date before the next message
+        is yielded.  This means you can check
+        ``run.extensions["tools"]`` between messages and see inline
+        results.
         """
         t = self._find_transformer("messages")
         if t is None:
             return
-        for msg in _PumpDrivenLog(t.value, self._pump_one):
-            # Pump until this message is complete so sync consumers
-            # get a fully populated ChatModelStream.
-            while not msg.done:
+        log = t.value
+        for msg in _PumpDrivenLog(log, self._pump_one):
+            msg._bind_pump(self._pump_one)
+            yield msg
+            # Advance the pump past non-message events so other
+            # transformers have up-to-date state before the next
+            # message is yielded.
+            prev_count = len(log)
+            while len(log) == prev_count:
                 if not self._pump_one():
                     break
-            yield msg
+
+    # -- Subgraphs ----------------------------------------------------------
+
+    @property
+    def subgraphs(self) -> Iterator[SubgraphRunStream]:
+        """Sync iterable of :class:`SubgraphRunStream` for child graphs.
+
+        Namespaces are discovered lazily as events are pumped from the
+        source.  Each yielded stream has its own ``values``, ``messages``,
+        and ``output`` projections scoped to the child namespace.
+
+        After yielding a subgraph, the caller may consume its projections
+        (e.g. ``sub.values``), which pumps more events and can discover
+        new namespaces.  The loop re-checks for newly discovered
+        namespaces after each yield before attempting another pump.
+        """
+        yielded: set[str] = set()
+
+        while True:
+            # Yield any newly discovered namespaces.  Re-check after
+            # each yield because consuming a subgraph's projections
+            # can pump events that discover further namespaces.
+            found_new = False
+            for ns_segment in list(self._mux._discovered_ns):
+                if ns_segment in yielded:
+                    continue
+                found_new = True
+                yielded.add(ns_segment)
+                child_ns = self._ns + [ns_segment]
+                child_transformers: list[StreamTransformer] = [
+                    ValuesTransformer(),
+                    MessagesTransformer(namespace=child_ns),
+                ]
+                for t in child_transformers:
+                    t.init()
+                    self._mux.register_transformer(t)
+
+                yield SubgraphRunStream(
+                    mux=self._mux,
+                    namespace=child_ns,
+                    transformers=child_transformers,
+                    pump_one=self._pump_one,
+                    output_mapper=self._output_mapper,
+                )
+
+            if found_new:
+                continue  # re-check before pumping
+
+            # No new namespaces — pump one event
+            if not self._pump_one():
+                break
 
     # -- State --------------------------------------------------------------
 
@@ -529,13 +598,140 @@ class GraphRunStream:
             name = getattr(t, "name", None)
             value = getattr(t, "value", None)
             if name is not None and value is not None:
-                if isinstance(value, list):
+                if isinstance(value, EventLog):
                     result[name] = _PumpDrivenLog(value, self._pump_one)
                 else:
                     result[name] = value
         return result
 
 
+# ---------------------------------------------------------------------------
+# SubgraphRunStream — sync child stream
+# ---------------------------------------------------------------------------
+
+
+class SubgraphRunStream:
+    """Synchronous run stream for a child subgraph.
+
+    Shares the parent's :class:`StreamMux` and pump function.  Has its
+    own transformer set registered on the shared mux so that projections
+    (``values``, ``messages``, ``output``) are scoped to the child
+    namespace.
+
+    Adds ``.name`` and ``.index`` parsed from the last namespace segment
+    (e.g. ``"researcher:2"`` → ``name="researcher"``, ``index=2``).
+    """
+
+    def __init__(
+        self,
+        *,
+        mux: StreamMux,
+        namespace: list[str],
+        transformers: list[StreamTransformer],
+        pump_one: Callable[[], bool],
+        output_mapper: Callable[[Any], Any] | None = None,
+    ) -> None:
+        self._mux = mux
+        self._ns = namespace
+        self._transformers = transformers
+        self._pump_one = pump_one
+        self._output_mapper = output_mapper
+
+    # -- Identity -----------------------------------------------------------
+
+    @property
+    def name(self) -> str:
+        if self._ns:
+            segment = self._ns[-1]
+            return segment.split(":")[0] if ":" in segment else segment
+        return ""
+
+    @property
+    def index(self) -> int:
+        if self._ns:
+            segment = self._ns[-1]
+            if ":" in segment:
+                try:
+                    return int(segment.split(":")[-1])
+                except ValueError:
+                    pass
+        return 0
+
+    # -- Transformer lookup -------------------------------------------------
+
+    def _find_transformer(self, name: str) -> StreamTransformer | None:
+        for t in self._transformers:
+            if getattr(t, "name", None) == name:
+                return t
+        return None
+
+    # -- Helpers ------------------------------------------------------------
+
+    def _map(self, value: Any) -> Any:
+        if value is not None and self._output_mapper is not None:
+            return self._output_mapper(value)
+        return value
+
+    def _pump_all(self) -> None:
+        while self._pump_one():
+            pass
+
+    # -- Raw event iteration (sync) -----------------------------------------
+
+    def __iter__(self) -> Iterator[ProtocolEvent]:
+        for event in _PumpDrivenLog(self._mux.event_log, self._pump_one):
+            ns = event["params"].get("namespace", [])
+            if ns[: len(self._ns)] == self._ns:
+                yield event
+
+    # -- Named projections (sync) -------------------------------------------
+
+    @property
+    def output(self) -> Any:
+        """The final output state (blocking). Drains the source."""
+        self._pump_all()
+        return self._map(self._mux.get_latest_values(self._ns))
+
+    @property
+    def values(self) -> Iterator[Any]:
+        """Sync iterable of intermediate state snapshots."""
+        t = self._find_transformer("values")
+        if t is None:
+            return
+        for item in _PumpDrivenLog(t.value, self._pump_one):
+            if item.get("namespace", []) == self._ns:
+                yield self._map(item["data"])
+
+    @property
+    def messages(self) -> Iterator[ChatModelStream]:
+        """Sync iterable of :class:`ChatModelStream` instances.
+
+        Each ``ChatModelStream`` is yielded as soon as the LLM begins
+        responding.  See :attr:`GraphRunStream.messages` for usage.
+        """
+        t = self._find_transformer("messages")
+        if t is None:
+            return
+        log = t.value
+        for msg in _PumpDrivenLog(log, self._pump_one):
+            msg._bind_pump(self._pump_one)
+            yield msg
+            prev_count = len(log)
+            while len(log) == prev_count:
+                if not self._pump_one():
+                    break
+
+    # -- State --------------------------------------------------------------
+
+    @property
+    def interrupted(self) -> bool:
+        return self._mux.interrupted
+
+    @property
+    def interrupts(self) -> list[InterruptPayload]:
+        return self._mux.interrupts
+
+
 def create_graph_run_stream(
     source: Iterator[tuple[tuple[str, ...], str, Any]],
     *,
@@ -581,6 +777,7 @@ __all__ = [
     "AsyncGraphRunStream",
     "AsyncSubgraphRunStream",
     "GraphRunStream",
+    "SubgraphRunStream",
     "create_async_graph_run_stream",
     "create_graph_run_stream",
 ]

libs/langgraph/langgraph/stream/stream_channel.py

@@ -1,17 +1,24 @@
 """StreamChannel — typed push-based channel for StreamTransformer projections.
 
-A ``StreamChannel`` wraps a list and declares a protocol channel name.
-When the :class:`StreamMux` detects a ``StreamChannel`` in a transformer's
-``init()`` return, it wires every ``push()`` call to inject a
-:class:`ProtocolEvent` into the main event stream using the channel's
-name as the ``method``.
+A ``StreamChannel`` wraps an :class:`EventLog` and declares a protocol
+channel name.  When the :class:`StreamMux` detects a ``StreamChannel``
+in a transformer's ``init()`` return, it wires every ``push()`` call to
+inject a :class:`ProtocolEvent` into the main event stream using the
+channel's name as the ``method``.
+
+In-process consumers iterate the channel directly (it is an async
+iterable).  Remote SDK clients subscribe via
+``session.subscribe("custom:<channelName>")``.
+
 """
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import AsyncIterator, Callable
 from typing import Any, Generic, TypeVar
 
+from langgraph.stream._event_log import EventLog
+
 T = TypeVar("T")
 
 
@@ -20,26 +27,46 @@ class StreamChannel(Generic[T]):
 
     Transformer authors create a ``StreamChannel`` in ``init()`` and
     call ``push()`` inside ``process()`` to emit domain objects.  The
-    mux auto-wires pushes to protocol events.
+    mux auto-wires pushes to protocol events and auto-closes/fails the
+    channel on run completion.
     """
 
-    __slots__ = ("channel_name", "_items", "_on_push")
+    __slots__ = ("channel_name", "_log", "_on_push")
 
     def __init__(self, name: str) -> None:
         self.channel_name = name
-        self._items: list[T] = []
+        self._log: EventLog[T] = EventLog()
         self._on_push: Callable[[Any], None] | None = None
 
     def push(self, item: T) -> None:
-        """Push an item to the channel."""
-        self._items.append(item)
+        """Push an item to the channel.
+
+        If the mux has wired this channel, the push also injects a
+        protocol event into the main event stream.
+        """
+        self._log.append(item)
         if self._on_push is not None:
             self._on_push(item)
 
+    # -- Async iteration (in-process consumption) ---------------------------
+
+    def __aiter__(self) -> AsyncIterator[T]:
+        return aiter(self._log)
+
+    # -- Internal (called by the mux) ---------------------------------------
+
     def _wire(self, fn: Callable[[Any], None]) -> None:
         """Wire a callback invoked on every ``push()``.  Called by the mux."""
         self._on_push = fn
 
+    def _close(self) -> None:
+        """Close the underlying log.  Called by the mux on normal completion."""
+        self._log.close()
+
+    def _fail(self, err: BaseException) -> None:
+        """Fail the underlying log.  Called by the mux on failure."""
+        self._log.fail(err)
+
 
 def is_stream_channel(value: object) -> bool:
     """Check if *value* is a :class:`StreamChannel` instance."""

libs/langgraph/langgraph/stream/transformers.py

@@ -9,33 +9,36 @@ from __future__ import annotations
 
 from typing import Any
 
-from langgraph.stream._types import ProtocolEvent, StreamTransformer
+from langgraph.stream._event_log import EventLog
+from langgraph.stream._types import ProtocolEvent
 from langgraph.stream.chat_model_stream import ChatModelStream
 
 # Type alias for the stream class constructor signature
 _StreamCls = type[ChatModelStream]
 
 
-class ValuesTransformer(StreamTransformer):
-    """Extracts ``values`` events and populates a values log.
+class ValuesTransformer:
+    """Extracts ``values`` events and populates a values event log.
 
     Maintains the latest state per namespace and provides a separate
-    log that :class:`AsyncGraphRunStream` / :class:`GraphRunStream` uses for ``.values``
+    event log that :class:`AsyncGraphRunStream` / :class:`GraphRunStream` uses for ``.values``
     iteration.
+
+    Implements the :class:`StreamTransformer` protocol.
     """
 
     name = "values"
 
     def __init__(self) -> None:
-        self._values_log: list[dict[str, Any]] = []
+        self._values_log: EventLog[dict[str, Any]] = EventLog()
         self._latest: dict[str, Any] = {}
 
     @property
-    def value(self) -> list[dict[str, Any]]:
+    def value(self) -> EventLog[dict[str, Any]]:
         return self._values_log
 
     @property
-    def values_log(self) -> list[dict[str, Any]]:
+    def values_log(self) -> EventLog[dict[str, Any]]:
         return self._values_log
 
     def get_latest(self, ns_key: str = "") -> Any:
@@ -58,18 +61,20 @@ class ValuesTransformer(StreamTransformer):
         return True
 
     def finalize(self) -> None:
-        pass
+        self._values_log.close()
 
     def fail(self, err: BaseException) -> None:
-        pass
+        self._values_log.fail(err)
 
 
-class MessagesTransformer(StreamTransformer):
+class MessagesTransformer:
     """Groups ``messages`` events into :class:`ChatModelStream` instances.
 
     One ``ChatModelStream`` is created per ``message-start`` event.
     Content-block events are routed to the active stream until
     ``message-finish`` or ``message-error`` closes it.
+
+    Implements the :class:`StreamTransformer` protocol.
     """
 
     name = "messages"
@@ -86,17 +91,17 @@ class MessagesTransformer(StreamTransformer):
         self._stream_cls: _StreamCls = stream_cls or ChatModelStream
 
         # Message log for .messages iteration
-        self._messages_log: list[ChatModelStream] = []
+        self._messages_log: EventLog[ChatModelStream] = EventLog()
 
         # Current active stream per namespace key
         self._active: dict[str, ChatModelStream] = {}
 
     @property
-    def value(self) -> list[ChatModelStream]:
+    def value(self) -> EventLog[ChatModelStream]:
         return self._messages_log
 
     @property
-    def messages_log(self) -> list[ChatModelStream]:
+    def messages_log(self) -> EventLog[ChatModelStream]:
         return self._messages_log
 
     def init(self) -> Any:
@@ -155,15 +160,17 @@ class MessagesTransformer(StreamTransformer):
         return True
 
     def finalize(self) -> None:
-        # Finish any remaining active streams
+        # Close any remaining active streams
         for stream in self._active.values():
             stream._finish({"reason": "stop"})
         self._active.clear()
+        self._messages_log.close()
 
     def fail(self, err: BaseException) -> None:
         for stream in self._active.values():
             stream._fail(err)
         self._active.clear()
+        self._messages_log.fail(err)
 
 
 __all__ = [

libs/langgraph/tests/test_graph_run_stream.py

@@ -9,7 +9,7 @@ from typing_extensions import TypedDict
 from langgraph.config import get_stream_writer
 from langgraph.graph import END, START, MessagesState, StateGraph
 from langgraph.stream import AsyncChatModelStream, StreamingHandler
-from langgraph.stream._types import ProtocolEvent, StreamTransformer
+from langgraph.stream._types import ProtocolEvent
 from tests.fake_chat import FakeChatModel
 
 
@@ -379,7 +379,7 @@ async def test_subgraph_child_output():
 # ---------------------------------------------------------------------------
 
 
-class _CountTransformer(StreamTransformer):
+class _CountTransformer:
     """Counts events. Exposes count via .value for extensions."""
 
     name = "event_count"
@@ -427,73 +427,6 @@ def test_sync_custom_reducer_extensions():
     assert run.extensions["event_count"] == counter.value
 
 
-# ---------------------------------------------------------------------------
-# Double iteration over .values
-# ---------------------------------------------------------------------------
-
-
-@pytest.mark.anyio
-async def test_async_values_double_iteration():
-    """Iterating over run.values twice should yield the same snapshots both times."""
-    graph = make_simple_graph()
-    run = await StreamingHandler(graph).astream({"value": "x", "items": []})
-    await asyncio.sleep(0.1)
-
-    first = []
-    async for v in run.values:
-        first.append(v)
-
-    second = []
-    async for v in run.values:
-        second.append(v)
-
-    assert len(first) == 3
-    assert first == second
-
-
-def test_sync_values_double_iteration():
-    """Iterating over run.values twice should yield the same snapshots both times."""
-    graph = make_simple_graph()
-    run = StreamingHandler(graph).stream({"value": "x", "items": []})
-
-    first = list(run.values)
-    second = list(run.values)
-
-    assert len(first) == 3
-    assert first == second
-
-
-@pytest.mark.anyio
-async def test_async_raw_events_double_iteration():
-    """Iterating over the raw event stream twice should yield the same events."""
-    graph = make_simple_graph()
-    run = await StreamingHandler(graph).astream({"value": "x", "items": []})
-    await asyncio.sleep(0.1)
-
-    first = []
-    async for event in run:
-        first.append(event)
-
-    second = []
-    async for event in run:
-        second.append(event)
-
-    assert len(first) > 0
-    assert first == second
-
-
-def test_sync_raw_events_double_iteration():
-    """Iterating over the raw event stream twice should yield the same events."""
-    graph = make_simple_graph()
-    run = StreamingHandler(graph).stream({"value": "x", "items": []})
-
-    first = list(run)
-    second = list(run)
-
-    assert len(first) > 0
-    assert first == second
-
-
 # ---------------------------------------------------------------------------
 # Tool transformer via extensions
 # ---------------------------------------------------------------------------
@@ -507,13 +440,15 @@ class _ToolExecution:
         self.output = output
 
 
-class _ToolsTransformer(StreamTransformer):
+class _ToolsTransformer:
     """Groups tool-started/tool-finished custom events into _ToolExecution objects."""
 
     name = "tools"
 
     def __init__(self) -> None:
-        self._log: list[_ToolExecution] = []
+        from langgraph.stream._event_log import EventLog
+
+        self._log: EventLog[_ToolExecution] = EventLog()
         self._pending: dict[str, dict] = {}
         self.value = self._log
 
@@ -548,10 +483,10 @@ class _ToolsTransformer(StreamTransformer):
         return True
 
     def finalize(self) -> None:
-        pass
+        self._log.close()
 
     def fail(self, err: BaseException) -> None:
-        pass
+        self._log.fail(err)
 
 
 def _make_tool_graph():

libs/langgraph/tests/test_stream_convert.py

@@ -51,6 +51,13 @@ def test_namespace_passthrough():
     assert evt["params"]["namespace"] == ["agent", "0"]
 
 
+def test_timestamp_populated():
+    evt = convert_to_protocol_event((), "values", {})
+    assert evt is not None
+    assert isinstance(evt["params"]["timestamp"], int)
+    assert evt["params"]["timestamp"] > 0
+
+
 def test_unknown_mode_returns_none():
     assert convert_to_protocol_event((), "unknown_mode", {}) is None
 

libs/langgraph/tests/test_stream_event_log.py

@@ -0,0 +1,133 @@
+import asyncio
+
+import pytest
+
+from langgraph.stream._event_log import EventLog
+
+
+@pytest.mark.anyio
+async def test_push_and_iterate_in_order():
+    log = EventLog()
+    log.append("a")
+    log.append("b")
+    log.append("c")
+    log.close()
+    items = [item async for item in aiter(log)]
+    assert items == ["a", "b", "c"]
+
+
+@pytest.mark.anyio
+async def test_multiple_independent_cursors():
+    log = EventLog()
+    log.append("x")
+    log.append("y")
+    log.close()
+    items1 = [item async for item in aiter(log)]
+    items2 = [item async for item in aiter(log)]
+    assert items1 == ["x", "y"]
+    assert items2 == ["x", "y"]
+
+
+@pytest.mark.anyio
+async def test_close_ends_iteration():
+    log = EventLog()
+    log.close()
+    items = [item async for item in aiter(log)]
+    assert items == []
+
+
+@pytest.mark.anyio
+async def test_fail_raises_error():
+    log = EventLog()
+    log.fail(RuntimeError("boom"))
+    with pytest.raises(RuntimeError, match="boom"):
+        async for _ in aiter(log):
+            pass
+
+
+@pytest.mark.anyio
+async def test_concurrent_push_and_iterate():
+    log = EventLog()
+    received = []
+
+    async def consumer():
+        async for item in aiter(log):
+            received.append(item)
+
+    async def producer():
+        for i in range(5):
+            log.append(i)
+            await asyncio.sleep(0.01)
+        log.close()
+
+    await asyncio.gather(producer(), consumer())
+    assert received == [0, 1, 2, 3, 4]
+
+
+@pytest.mark.anyio
+async def test_items_before_cursor_visible():
+    log = EventLog()
+    log.append("a")
+    log.append("b")
+    cursor = aiter(log)
+    log.append("c")
+    log.close()
+    items = [item async for item in cursor]
+    assert items == ["a", "b", "c"]
+
+
+@pytest.mark.anyio
+async def test_empty_log_closed_yields_nothing():
+    log = EventLog()
+    log.close()
+    items = [item async for item in aiter(log)]
+    assert items == []
+
+
+@pytest.mark.anyio
+async def test_fail_mid_iteration():
+    """A cursor that has consumed some items should raise when fail() is called."""
+    log = EventLog()
+    received = []
+
+    async def consumer():
+        async for item in aiter(log):
+            received.append(item)
+
+    async def producer():
+        log.append("a")
+        log.append("b")
+        await asyncio.sleep(0.02)
+        log.fail(RuntimeError("mid-stream error"))
+
+    with pytest.raises(RuntimeError, match="mid-stream error"):
+        await asyncio.gather(producer(), consumer())
+    assert received == ["a", "b"]
+
+
+@pytest.mark.anyio
+async def test_abandoned_cursor_cleans_up_waiters():
+    """Abandoned async cursors should not leave stale futures in the
+    EventLog waiter list.
+
+    When a cursor's __anext__ is cancelled (e.g. consumer breaks out of
+    ``async for``), the Future it registered in ``_waiters`` should be
+    cleaned up.  Otherwise the list grows without bound until the next
+    append/close/fail triggers ``_wake_all()``.
+    """
+    log: EventLog[str] = EventLog()
+
+    for _ in range(10):
+        cursor = aiter(log)
+        task = asyncio.ensure_future(cursor.__anext__())
+        await asyncio.sleep(0)  # let task register its waiter
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+    assert len(log._waiters) == 0, (
+        f"Expected 0 waiters after abandoning 10 cursors, "
+        f"got {len(log._waiters)}.  Abandoned cursors leak futures."
+    )

libs/langgraph/tests/test_stream_mux.py

@@ -4,7 +4,7 @@ import pytest
 
 from langgraph.stream._convert import convert_to_protocol_event
 from langgraph.stream._mux import AsyncStreamMux, StreamMux
-from langgraph.stream._types import ProtocolEvent, StreamTransformer
+from langgraph.stream._types import ProtocolEvent
 from langgraph.stream.stream_channel import StreamChannel
 
 
@@ -14,7 +14,7 @@ def _event(mode: str, data: Any, ns: list[str] | None = None) -> ProtocolEvent:
     return ev
 
 
-class _MockTransformer(StreamTransformer):
+class _MockTransformer:
     def __init__(self, *, suppress: bool = False):
         self.calls: list[ProtocolEvent] = []
         self._suppress = suppress
@@ -243,7 +243,7 @@ async def test_channel_push_during_process_preserves_namespace():
     ``namespace: []`` instead of the original.
     """
 
-    class _ChannelTransformer(StreamTransformer):
+    class _ChannelTransformer:
         """Pushes to its channel whenever it sees a ``values`` event."""
 
         def __init__(self, name: str) -> None:

libs/langgraph/tests/test_stream_reducers.py

@@ -22,29 +22,38 @@ def _event(
 # -- ValuesTransformer ---------------------------------------------------------
 
 
-def test_values_captures_values_events():
+@pytest.mark.anyio
+async def test_values_captures_values_events():
     reducer = ValuesTransformer()
     reducer.init()
     reducer.process(_event("values", {"a": 1}))
     reducer.process(_event("values", {"b": 2}))
     reducer.finalize()
 
-    assert len(reducer.values_log) == 2
-    assert reducer.values_log[0]["data"] == {"a": 1}
-    assert reducer.values_log[1]["data"] == {"b": 2}
+    collected = []
+    async for item in reducer.values_log:
+        collected.append(item)
+    assert len(collected) == 2
+    assert collected[0]["data"] == {"a": 1}
+    assert collected[1]["data"] == {"b": 2}
 
 
-def test_values_ignores_other_modes():
+@pytest.mark.anyio
+async def test_values_ignores_other_modes():
     reducer = ValuesTransformer()
     reducer.init()
     reducer.process(_event("updates", {"x": 1}))
     reducer.process(_event("messages", {"event": "message-start"}))
     reducer.finalize()
 
-    assert len(reducer.values_log) == 0
+    collected = []
+    async for item in reducer.values_log:
+        collected.append(item)
+    assert len(collected) == 0
 
 
-def test_values_latest_per_namespace():
+@pytest.mark.anyio
+async def test_values_latest_per_namespace():
     reducer = ValuesTransformer()
     reducer.init()
     reducer.process(_event("values", {"v": 1}, ns=["child:0"]))
@@ -52,6 +61,15 @@ def test_values_latest_per_namespace():
     assert reducer.get_latest("child:0") == {"v": 2}
 
 
+@pytest.mark.anyio
+async def test_values_finalize_closes_log():
+    reducer = ValuesTransformer()
+    reducer.init()
+    reducer.process(_event("values", {"a": 1}))
+    reducer.finalize()
+    assert reducer.values_log.closed
+
+
 # -- MessagesTransformer -------------------------------------------------------
 
 
@@ -85,7 +103,8 @@ def _msg_finish(ns=None, node=None):
     )
 
 
-def test_messages_groups_lifecycle():
+@pytest.mark.anyio
+async def test_messages_groups_lifecycle():
     reducer = MessagesTransformer()
     reducer.init()
     reducer.process(_msg_start())
@@ -93,12 +112,16 @@ def test_messages_groups_lifecycle():
     reducer.process(_msg_finish())
     reducer.finalize()
 
-    assert len(reducer.messages_log) == 1
-    assert isinstance(reducer.messages_log[0], ChatModelStream)
-    assert reducer.messages_log[0].done
+    collected = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 1
+    assert isinstance(collected[0], ChatModelStream)
+    assert collected[0].done
 
 
-def test_messages_multiple_sequential():
+@pytest.mark.anyio
+async def test_messages_multiple_sequential():
     reducer = MessagesTransformer()
     reducer.init()
     reducer.process(_msg_start(message_id="m1"))
@@ -107,10 +130,14 @@ def test_messages_multiple_sequential():
     reducer.process(_msg_finish())
     reducer.finalize()
 
-    assert len(reducer.messages_log) == 2
+    collected = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 2
 
 
-def test_messages_namespace_filter():
+@pytest.mark.anyio
+async def test_messages_namespace_filter():
     reducer = MessagesTransformer(namespace=["root"])
     reducer.init()
     reducer.process(_msg_start(ns=["root"]))
@@ -119,10 +146,14 @@ def test_messages_namespace_filter():
     reducer.process(_msg_finish(ns=["other"]))
     reducer.finalize()
 
-    assert len(reducer.messages_log) == 1
+    collected = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 1
 
 
-def test_messages_node_filter():
+@pytest.mark.anyio
+async def test_messages_node_filter():
     reducer = MessagesTransformer(node_filter="agent")
     reducer.init()
     reducer.process(_msg_start(node="agent"))
@@ -131,10 +162,14 @@ def test_messages_node_filter():
     reducer.process(_msg_finish(node="tools"))
     reducer.finalize()
 
-    assert len(reducer.messages_log) == 1
+    collected = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 1
 
 
-def test_messages_error_event():
+@pytest.mark.anyio
+async def test_messages_error_event():
     """An error event should fail the active ChatModelStream."""
     reducer = MessagesTransformer()
     reducer.init()
@@ -145,17 +180,35 @@ def test_messages_error_event():
     )
     reducer.finalize()
 
-    assert len(reducer.messages_log) == 1
-    assert reducer.messages_log[0].done
+    collected: list[ChatModelStream] = []
+    async for stream in reducer.messages_log:
+        collected.append(stream)
+    assert len(collected) == 1
+    assert collected[0].done
 
 
-def test_messages_fail_propagates_to_active():
-    """transformer.fail() should mark active streams as done."""
+@pytest.mark.anyio
+async def test_messages_fail_propagates_to_active():
+    """transformer.fail() should propagate the error to any active streams."""
     reducer = MessagesTransformer()
     reducer.init()
     reducer.process(_msg_start())
     reducer.process(_content_delta("partial"))
     reducer.fail(RuntimeError("graph failed"))
 
-    assert len(reducer.messages_log) == 1
-    assert reducer.messages_log[0].done
+    # The messages log should be failed too
+    with pytest.raises(RuntimeError, match="graph failed"):
+        async for _ in reducer.messages_log:
+            pass
+
+
+@pytest.mark.anyio
+async def test_values_fail_propagates():
+    reducer = ValuesTransformer()
+    reducer.init()
+    reducer.process(_event("values", {"a": 1}))
+    reducer.fail(RuntimeError("graph failed"))
+
+    with pytest.raises(RuntimeError, match="graph failed"):
+        async for _ in reducer.values_log:
+            pass

libs/langgraph/tests/test_stream_run_stream.py

@@ -4,12 +4,13 @@ from typing import Any
 
 import pytest
 
-from langgraph.stream._mux import AsyncStreamMux, StreamMux
+from langgraph.stream._mux import AsyncStreamMux
 from langgraph.stream._types import ProtocolEvent
 from langgraph.stream.chat_model_stream import ChatModelStream
 from langgraph.stream.run_stream import (
     AsyncGraphRunStream,
     AsyncSubgraphRunStream,
+    SubgraphRunStream,
     create_async_graph_run_stream,
     create_graph_run_stream,
 )
@@ -290,10 +291,8 @@ def test_sync_messages():
     assert collected[0].done
 
 
-def test_sync_messages_text_accessible():
-    """Sync consumers should be able to read ChatModelStream text content
-    without an async event loop.
-    """
+def test_sync_messages_text_streaming():
+    """Sync consumers can iterate msg.text for deltas."""
     chunks = [
         ((), "messages", {"event": "message-start", "message_id": "m1"}),
         (
@@ -314,18 +313,29 @@ def test_sync_messages_text_accessible():
         ),
         ((), "messages", {"event": "message-finish", "reason": "stop"}),
     ]
-    run = create_graph_run_stream(_sync_source(chunks))
 
+    # Iterate deltas
+    run = create_graph_run_stream(_sync_source(chunks))
     for msg in run.messages:
-        assert isinstance(msg.text, str)
-        assert msg.text == "Hello world"
+        deltas = list(msg.text)
+        assert deltas == ["Hello", " world"]
         assert msg.done
 
+    # str() returns full text
+    run = create_graph_run_stream(_sync_source(chunks))
+    for msg in run.messages:
+        assert str(msg.text) == "Hello world"
 
-def test_sync_messages_content_populated_when_yielded():
-    """When sync run.messages yields a ChatModelStream, its content should
-    be fully populated (done=True) with all text accumulated.
-    """
+    # After message is done, .text returns plain str
+    run = create_graph_run_stream(_sync_source(chunks))
+    for msg in run.messages:
+        list(msg.text)  # exhaust deltas
+        assert isinstance(msg.text, str)
+        assert msg.text == "Hello world"
+
+
+def test_sync_messages_multiple():
+    """Multiple sync messages each stream their own deltas."""
     chunks = [
         ((), "messages", {"event": "message-start", "message_id": "m1"}),
         (
@@ -350,12 +360,10 @@ def test_sync_messages_content_populated_when_yielded():
     ]
     run = create_graph_run_stream(_sync_source(chunks))
 
-    messages = list(run.messages)
-    assert len(messages) == 2
-    assert messages[0].done
-    assert messages[0].text == "answer"
-    assert messages[1].done
-    assert messages[1].text == "second"
+    all_deltas = []
+    for msg in run.messages:
+        all_deltas.append(list(msg.text))
+    assert all_deltas == [["answer"], ["second"]]
 
 
 def test_sync_output_mapper():
@@ -498,11 +506,14 @@ def test_sync_lazy_interleaved_projections():
     assert next(vit) == {"v": 1}
     assert consumed == 1
 
-    # Pull first message — pumps until message-finish (item 3) so the
-    # ChatModelStream is fully populated before yielding.
+    # Pull first message — yielded on message-start (item 2).
+    # Consuming str(msg.text) drives the pump to message-finish (item 3).
     mit = iter(run.messages)
     msg = next(mit)
     assert isinstance(msg, ChatModelStream)
+    assert consumed == 2
+    assert not msg.done
+    str(msg.text)  # pump until message completes
     assert msg.done
     assert consumed == 3
 
@@ -608,3 +619,362 @@ async def test_subgraph_child_values_receive_post_discovery_events():
         f"Expected 2 child value snapshots but got {len(values)}: {values}.  "
         "Child transformer missed post-discovery events."
     )
+
+
+# ---------------------------------------------------------------------------
+# SubgraphRunStream — sync subgraph tests
+# ---------------------------------------------------------------------------
+
+
+def test_sync_subgraphs_discovery():
+    """Iterating .subgraphs should discover child namespaces and yield
+    SubgraphRunStream instances with correct name and index.
+    """
+    chunks = [
+        (("agent:0",), "values", {"v": 1}),
+        (("agent:1",), "values", {"v": 2}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    subs = list(run.subgraphs)
+    assert len(subs) == 2
+    assert all(isinstance(s, SubgraphRunStream) for s in subs)
+    assert subs[0].name == "agent"
+    assert subs[0].index == 0
+    assert subs[1].name == "agent"
+    assert subs[1].index == 1
+
+
+def test_sync_subgraph_name_no_index():
+    """Subgraph without a colon-delimited index should have index=0."""
+    chunks = [
+        (("planner",), "values", {"v": 1}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    subs = list(run.subgraphs)
+    assert len(subs) == 1
+    assert subs[0].name == "planner"
+    assert subs[0].index == 0
+
+
+def test_sync_subgraph_no_subgraphs():
+    """When all events are root-level, .subgraphs should yield nothing."""
+    chunks = [
+        ((), "values", {"v": 1}),
+        ((), "values", {"v": 2}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    subs = list(run.subgraphs)
+    assert subs == []
+
+
+def test_sync_subgraph_values():
+    """SubgraphRunStream.values should yield only values from the child namespace."""
+    chunks = [
+        (("child:0",), "values", {"v": 1}),
+        ((), "values", {"root": True}),
+        (("child:0",), "values", {"v": 2}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        vals = list(sub.values)
+        assert vals == [{"v": 1}, {"v": 2}]
+
+
+def test_sync_subgraph_values_multiple_children():
+    """Each child stream should only see its own values."""
+    chunks = [
+        (("a:0",), "values", {"who": "a0"}),
+        (("b:0",), "values", {"who": "b0"}),
+        (("a:0",), "values", {"who": "a0-2"}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    children: dict[str, list[Any]] = {}
+    for sub in run.subgraphs:
+        children[f"{sub.name}:{sub.index}"] = list(sub.values)
+
+    assert children["a:0"] == [{"who": "a0"}, {"who": "a0-2"}]
+    assert children["b:0"] == [{"who": "b0"}]
+
+
+def test_sync_subgraph_output():
+    """SubgraphRunStream.output should return the last values for the child."""
+    chunks = [
+        (("child:0",), "values", {"v": 1}),
+        (("child:0",), "values", {"v": 2}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        assert sub.output == {"v": 2}
+
+
+def test_sync_subgraph_output_with_mapper():
+    """Output mapper should apply to subgraph output."""
+    chunks = [
+        (("child:0",), "values", {"v": 42}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(
+        _sync_source(chunks), output_mapper=lambda x: {"mapped": x.get("v")}
+    )
+    for sub in run.subgraphs:
+        assert sub.output == {"mapped": 42}
+
+
+def test_sync_subgraph_values_with_mapper():
+    """Output mapper should apply to each yielded value snapshot."""
+    chunks = [
+        (("child:0",), "values", {"v": 1}),
+        (("child:0",), "values", {"v": 2}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(
+        _sync_source(chunks), output_mapper=lambda x: {"m": x.get("v")}
+    )
+    for sub in run.subgraphs:
+        vals = list(sub.values)
+        assert vals == [{"m": 1}, {"m": 2}]
+
+
+def test_sync_subgraph_messages():
+    """SubgraphRunStream.messages should yield fully populated ChatModelStream instances."""
+    chunks = [
+        (
+            ("agent:0",),
+            "messages",
+            {"event": "message-start", "message_id": "m1", "__node__": "agent"},
+        ),
+        (
+            ("agent:0",),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "hello"},
+                "__node__": "agent",
+            },
+        ),
+        (
+            ("agent:0",),
+            "messages",
+            {"event": "message-finish", "reason": "stop", "__node__": "agent"},
+        ),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        msgs = list(sub.messages)
+        assert len(msgs) == 1
+        assert isinstance(msgs[0], ChatModelStream)
+        assert msgs[0].done
+        assert msgs[0].text == "hello"
+
+
+def test_sync_subgraph_messages_isolated():
+    """Messages from different subgraphs should not leak between children."""
+    chunks = [
+        (
+            ("a:0",),
+            "messages",
+            {"event": "message-start", "message_id": "m-a", "__node__": "a"},
+        ),
+        (
+            ("a:0",),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "from-a"},
+                "__node__": "a",
+            },
+        ),
+        (
+            ("a:0",),
+            "messages",
+            {"event": "message-finish", "reason": "stop", "__node__": "a"},
+        ),
+        (
+            ("b:0",),
+            "messages",
+            {"event": "message-start", "message_id": "m-b", "__node__": "b"},
+        ),
+        (
+            ("b:0",),
+            "messages",
+            {
+                "event": "content-block-delta",
+                "content_block": {"type": "text", "text": "from-b"},
+                "__node__": "b",
+            },
+        ),
+        (
+            ("b:0",),
+            "messages",
+            {"event": "message-finish", "reason": "stop", "__node__": "b"},
+        ),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    msg_texts: dict[str, list[str]] = {}
+    for sub in run.subgraphs:
+        msg_texts[sub.name] = [str(m.text) for m in sub.messages]
+
+    assert msg_texts["a"] == ["from-a"]
+    assert msg_texts["b"] == ["from-b"]
+
+
+def test_sync_subgraph_raw_iter():
+    """Iterating a SubgraphRunStream directly should yield events scoped
+    to the child namespace.
+    """
+    chunks = [
+        (("child:0",), "values", {"v": 1}),
+        ((), "values", {"root": True}),
+        (("child:0",), "updates", {"node": "x"}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        events = list(sub)
+        methods = [e["method"] for e in events]
+        assert "values" in methods
+        assert "updates" in methods
+        # Root events should not appear
+        for e in events:
+            assert e["params"]["namespace"] == ["child:0"]
+
+
+def test_sync_subgraph_events_after_discovery():
+    """Events arriving after a namespace is first discovered should still
+    be visible in the child's values iteration.
+    """
+    chunks = [
+        (("child:0",), "values", {"v": 1}),  # triggers discovery
+        ((), "values", {"root": 1}),
+        (("child:0",), "values", {"v": 2}),  # after discovery
+        (("child:0",), "values", {"v": 3}),  # after discovery
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        vals = list(sub.values)
+        assert vals == [{"v": 1}, {"v": 2}, {"v": 3}]
+
+
+def test_sync_subgraph_lazy_pump():
+    """Subgraph iteration should pump the source lazily."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [
+            (("child:0",), "values", {"v": 1}),
+            (("child:0",), "values", {"v": 2}),
+            (("child:0",), "values", {"v": 3}),
+            ((), "values", {"done": True}),
+        ]:
+            consumed += 1
+            yield chunk
+
+    run = create_graph_run_stream(counting_source())
+    assert consumed == 0
+
+    for sub in run.subgraphs:
+        # Discovery pumped the first event
+        it = iter(sub.values)
+        v = next(it)
+        assert v == {"v": 1}
+        # Should not have consumed everything yet
+        assert consumed < 4
+        break  # don't exhaust subgraphs
+
+
+def test_sync_subgraph_interleave_parent_values():
+    """Parent values and subgraph values should both be accessible
+    when interleaving iteration.
+    """
+    chunks = [
+        ((), "values", {"root": 1}),
+        (("child:0",), "values", {"child": 1}),
+        ((), "values", {"root": 2}),
+        (("child:0",), "values", {"child": 2}),
+        ((), "values", {"root": 3}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+
+    # First drain parent values
+    root_vals = list(run.values)
+    assert root_vals == [{"root": 1}, {"root": 2}, {"root": 3}]
+
+    # Source is exhausted, but subgraph transformers were registered
+    # via replay — subgraph iteration should still see buffered events
+    # Note: subgraphs must be iterated while source is being pumped
+    # to discover namespaces. Since we drained via values, namespace
+    # "child:0" was already discovered. But subgraphs iteration also
+    # needs to pump — and the source is exhausted. Let's verify it
+    # yields the discovered child.
+    subs = list(run.subgraphs)
+    assert len(subs) == 1
+    assert subs[0].name == "child"
+    # The child transformer was registered via replay, so it saw the events
+    vals = list(subs[0].values)
+    assert vals == [{"child": 1}, {"child": 2}]
+
+
+def test_sync_subgraph_interrupted():
+    """Subgraph .interrupted should reflect the mux's interrupt state."""
+
+    class _FakeInterrupt:
+        def __init__(self, id: str):
+            self.id = id
+
+    chunks = [
+        (("child:0",), "values", {"__interrupt__": [_FakeInterrupt("i1")]}),
+        ((), "values", {"done": True}),
+    ]
+    run = create_graph_run_stream(_sync_source(chunks))
+    for sub in run.subgraphs:
+        # Pump to process the interrupt
+        _ = sub.output
+        assert sub.interrupted is True
+        assert len(sub.interrupts) == 1
+
+
+def test_sync_subgraph_source_error():
+    """If the source raises mid-stream, subgraphs that were already
+    discovered should still have their buffered data.
+    """
+
+    def bad_source():
+        yield (("child:0",), "values", {"v": 1})
+        yield (("child:0",), "values", {"v": 2})
+        raise ValueError("boom")
+
+    run = create_graph_run_stream(bad_source())
+    for sub in run.subgraphs:
+        vals = list(sub.values)
+        assert vals == [{"v": 1}, {"v": 2}]
+        assert run._mux._error is not None
+
+
+def test_sync_subgraph_output_drains_source():
+    """Accessing subgraph .output should drain the full source."""
+    consumed = 0
+
+    def counting_source():
+        nonlocal consumed
+        for chunk in [
+            (("child:0",), "values", {"v": 1}),
+            (("child:0",), "values", {"v": 2}),
+            ((), "values", {"done": True}),
+        ]:
+            consumed += 1
+            yield chunk
+
+    run = create_graph_run_stream(counting_source())
+    for sub in run.subgraphs:
+        result = sub.output
+        assert result == {"v": 2}
+        assert consumed == 3