Remove StreamChannel close/fail no-ops and channel tracking from mux

libs/langgraph/langgraph/stream/_mux.py

@@ -32,7 +32,6 @@ class StreamMux:
     def __init__(self, transformers: list[StreamTransformer] | None = None) -> None:
         self._event_log: list[ProtocolEvent] = []
         self._transformers: list[StreamTransformer] = list(transformers or [])
-        self._channels: list[StreamChannel[Any]] = []
         self._current_namespace: list[str] = []
         self._next_emit_seq: int = 0
 
@@ -115,14 +114,9 @@ class StreamMux:
             return
         self._closed = True
 
-        # Finalize transformers
         for transformer in self._transformers:
             transformer.finalize()
 
-        # Close wired channels
-        for channel in self._channels:
-            channel._close()
-
     def fail(self, error: BaseException) -> None:
         """Fail the mux and propagate the error to all consumers."""
         if self._closed:
@@ -130,14 +124,9 @@ class StreamMux:
         self._closed = True
         self._error = error
 
-        # Fail transformers
         for transformer in self._transformers:
             transformer.fail(error)
 
-        # Fail wired channels
-        for channel in self._channels:
-            channel._fail(error)
-
     # -- Inspection ---------------------------------------------------------
 
     @property
@@ -202,8 +191,6 @@ 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:

libs/langgraph/langgraph/stream/stream_channel.py

@@ -5,10 +5,6 @@ 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.  Remote SDK clients
-subscribe via ``session.subscribe("custom:<channelName>")``.
-
 """
 
 from __future__ import annotations
@@ -24,8 +20,7 @@ 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 and auto-closes/fails the
-    channel on run completion.
+    mux auto-wires pushes to protocol events.
     """
 
     __slots__ = ("channel_name", "_items", "_on_push")
@@ -41,20 +36,10 @@ class StreamChannel(Generic[T]):
         if self._on_push is not None:
             self._on_push(item)
 
-    # -- 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:
-        """No-op for compatibility.  Called by the mux on normal completion."""
-        pass
-
-    def _fail(self, err: BaseException) -> None:
-        """No-op for compatibility.  Called by the mux on failure."""
-        pass
-
 
 def is_stream_channel(value: object) -> bool:
     """Check if *value* is a :class:`StreamChannel` instance."""