From 28cf5ed78dee80d1dfa7da1389af067d515532c5 Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Thu, 16 Apr 2026 09:32:47 -0400 Subject: [PATCH] =?UTF-8?q?Unify=20EventLog=20=E2=80=94=20remove=20sync/as?= =?UTF-8?q?ync=20split=20from=20transformer=20API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merge EventLog and AsyncEventLog into a single class with a _bind() mechanism. EventLog starts unbound; the StreamMux calls _bind(is_async) after transformer registration so only the correct iteration protocol is available. This removes the is_async parameter from EventLog, StreamChannel, and all transformer constructors — transformers just create EventLog() and never need to know whether they run in sync or async context. --- libs/langgraph/langgraph/stream/__init__.py | 3 +- libs/langgraph/langgraph/stream/_event_log.py | 137 +++++++++++------- libs/langgraph/langgraph/stream/_mux.py | 31 ++-- libs/langgraph/langgraph/stream/run_stream.py | 9 +- .../langgraph/stream/stream_channel.py | 28 ++-- .../langgraph/stream/streaming_handler.py | 4 +- .../langgraph/stream/transformers.py | 14 +- .../langgraph/tests/test_streaming_handler.py | 72 +++++++-- 8 files changed, 176 insertions(+), 122 deletions(-) diff --git a/libs/langgraph/langgraph/stream/__init__.py b/libs/langgraph/langgraph/stream/__init__.py index 479c669d0..386deeb1a 100644 --- a/libs/langgraph/langgraph/stream/__init__.py +++ b/libs/langgraph/langgraph/stream/__init__.py @@ -4,14 +4,13 @@ Provides a ``StreamingHandler`` that wraps a compiled graph and exposes ergonomic streaming projections through a transformer pipeline. """ -from langgraph.stream._event_log import AsyncEventLog, EventLog +from langgraph.stream._event_log import EventLog from langgraph.stream._types import ProtocolEvent, StreamTransformer from langgraph.stream.run_stream import AsyncGraphRunStream, GraphRunStream from langgraph.stream.stream_channel import StreamChannel from langgraph.stream.streaming_handler import StreamingHandler __all__ = [ - "AsyncEventLog", "AsyncGraphRunStream", "EventLog", "GraphRunStream", diff --git a/libs/langgraph/langgraph/stream/_event_log.py b/libs/langgraph/langgraph/stream/_event_log.py index 6fa774067..39c110073 100644 --- a/libs/langgraph/langgraph/stream/_event_log.py +++ b/libs/langgraph/langgraph/stream/_event_log.py @@ -8,16 +8,24 @@ from typing import Generic, TypeVar T = TypeVar("T") -class _EventLogBase(Generic[T]): - """Shared producer API for sync and async event logs. +class EventLog(Generic[T]): + """Append-only buffer that supports multiple independent consumers. - Append-only buffer that supports multiple independent consumers. - Subclasses provide the iteration protocol (sync or async). + Starts unbound — neither ``__iter__`` nor ``__aiter__`` is available + until the ``StreamMux`` calls ``_bind(is_async)``. After binding, + only the matching iteration protocol works; the other raises + ``TypeError``. - Producer API (thread-safe): + Producer API (thread-safe, works before and after binding): push(item) — append an item, notify all waiting cursors close() — mark the log as done fail(err) — mark the log as errored + + Sync iteration is pull-based: when a cursor catches up it calls + ``_request_more`` to drive the graph forward. + + Async iteration uses ``asyncio.Future`` objects — the producer + wakes cursors via ``loop.call_soon_threadsafe``. """ def __init__(self) -> None: @@ -26,6 +34,35 @@ class _EventLogBase(Generic[T]): self._error: BaseException | None = None self._lock = threading.Lock() + # Binding state — None means unbound. + self._is_async: bool | None = None + + # Sync pull callback (set by the run stream, not by bind). + self._request_more: Callable[[], bool] | None = None + + # Async waiters (allocated on bind). + self._async_waiters: list[asyncio.Future[None]] | None = None + + # ------------------------------------------------------------------ + # Binding + # ------------------------------------------------------------------ + + def _bind(self, *, is_async: bool) -> None: + """Bind this log to sync or async mode. + + Called by the ``StreamMux`` after transformer registration. + Must be called exactly once before any iteration. + """ + if self._is_async is not None: + raise RuntimeError("EventLog is already bound") + self._is_async = is_async + if is_async: + self._async_waiters = [] + + # ------------------------------------------------------------------ + # Producer API (thread-safe, mode-agnostic) + # ------------------------------------------------------------------ + def push(self, item: T) -> None: """Append *item* and wake all waiting cursors.""" with self._lock: @@ -47,31 +84,39 @@ class _EventLogBase(Generic[T]): self._closed = True self._notify() + # ------------------------------------------------------------------ + # Notification + # ------------------------------------------------------------------ + def _notify(self) -> None: - """Wake waiting consumers. Overridden by subclasses.""" + """Wake async waiters if bound to async mode.""" + waiters = self._async_waiters + if not waiters: + return + self._async_waiters = [] + for fut in waiters: + if not fut.done(): + try: + fut.get_loop().call_soon_threadsafe(fut.set_result, None) + except RuntimeError: + # Event loop already closed — nothing to notify. + pass - -class EventLog(_EventLogBase[T]): - """Sync event log with pull-based iteration. - - Each call to ``__iter__`` creates a new cursor starting from the - beginning. When a cursor catches up to the buffer and the log is - not yet closed, it calls ``_request_more`` to pull more data from - the producer (typically the graph iterator via the run stream). - - If no ``_request_more`` callback is set, the cursor returns - immediately when it reaches the end of the buffer — this is the - behavior used in unit tests where items are pushed before iteration. - - Use ``AsyncEventLog`` for async consumers. - """ - - def __init__(self) -> None: - super().__init__() - self._request_more: Callable[[], bool] | None = None + # ------------------------------------------------------------------ + # Sync iteration (pull-based) + # ------------------------------------------------------------------ def __iter__(self) -> Iterator[T]: """Return a new independent sync cursor over the log.""" + if self._is_async is None: + raise TypeError( + "EventLog has not been bound yet. " + "Register the transformer with a StreamMux first." + ) + if self._is_async: + raise TypeError( + "This EventLog is bound to async mode — use 'async for' instead." + ) return self._sync_cursor() def _sync_cursor(self) -> Iterator[T]: @@ -95,40 +140,19 @@ class EventLog(_EventLogBase[T]): # No producer callback and not closed — buffer is complete. return - -class AsyncEventLog(_EventLogBase[T]): - """Async event log with multi-cursor iteration. - - Each call to ``__aiter__`` creates a new cursor starting from the - beginning. Cursors await ``asyncio.Future`` objects when they - catch up to the producer. - - The producer (``push``/``close``/``fail``) is safe to call from - any thread — async waiters are notified via - ``loop.call_soon_threadsafe``. - - Use ``EventLog`` for sync consumers. - """ - - def __init__(self) -> None: - super().__init__() - self._async_waiters: list[asyncio.Future[None]] = [] - - def _notify(self) -> None: - waiters = self._async_waiters - if not waiters: - return - self._async_waiters = [] - for fut in waiters: - if not fut.done(): - try: - fut.get_loop().call_soon_threadsafe(fut.set_result, None) - except RuntimeError: - # Event loop already closed — nothing to notify. - pass + # ------------------------------------------------------------------ + # Async iteration + # ------------------------------------------------------------------ def __aiter__(self) -> AsyncIterator[T]: """Return a new independent async cursor over the log.""" + if self._is_async is None: + raise TypeError( + "EventLog has not been bound yet. " + "Register the transformer with a StreamMux first." + ) + if not self._is_async: + raise TypeError("This EventLog is bound to sync mode — use 'for' instead.") return self._async_cursor() async def _async_cursor(self) -> AsyncIterator[T]: @@ -144,5 +168,6 @@ class AsyncEventLog(_EventLogBase[T]): else: loop = asyncio.get_running_loop() fut: asyncio.Future[None] = loop.create_future() + assert self._async_waiters is not None self._async_waiters.append(fut) await fut diff --git a/libs/langgraph/langgraph/stream/_mux.py b/libs/langgraph/langgraph/stream/_mux.py index 71a987fcf..54666924c 100644 --- a/libs/langgraph/langgraph/stream/_mux.py +++ b/libs/langgraph/langgraph/stream/_mux.py @@ -4,7 +4,7 @@ import time from collections.abc import Callable from typing import Any -from langgraph.stream._event_log import AsyncEventLog, EventLog, _EventLogBase +from langgraph.stream._event_log import EventLog from langgraph.stream._types import ProtocolEvent, StreamTransformer from langgraph.stream.stream_channel import StreamChannel @@ -18,15 +18,15 @@ class StreamMux: into the main log. Pass ``is_async=True`` when the mux will be consumed via async - iteration (``handler.astream()``). This creates ``AsyncEventLog`` - instances instead of ``EventLog`` instances. + iteration (``handler.astream()``). All ``EventLog`` and + ``StreamChannel`` instances discovered during ``register()`` are + automatically bound to the matching mode. """ def __init__(self, *, is_async: bool = False) -> None: self._is_async = is_async - self._events: _EventLogBase[ProtocolEvent] = ( - AsyncEventLog() if is_async else EventLog() - ) + self._events: EventLog[ProtocolEvent] = EventLog() + self._events._bind(is_async=is_async) self._transformers: list[StreamTransformer] = [] self._channels: list[StreamChannel[Any]] = [] self._seq = 0 @@ -35,8 +35,8 @@ class StreamMux: """Register a transformer and return its projection dict. Calls ``transformer.init()``, stores the transformer for event - processing, and returns the projection. StreamChannels in the - projection are auto-wired. + processing, binds any ``EventLog`` or ``StreamChannel`` instances + in the projection, and returns the projection. """ projection = transformer.init() if not isinstance(projection, dict): @@ -45,7 +45,7 @@ class StreamMux: f"got {type(projection).__name__}" ) self._transformers.append(transformer) - self._wire_channels(projection) + self._bind_and_wire(projection) return projection def push(self, event: ProtocolEvent) -> None: @@ -107,17 +107,14 @@ class StreamMux: self._events.fail(err) # ------------------------------------------------------------------ - # StreamChannel auto-wiring + # Binding and StreamChannel auto-wiring # ------------------------------------------------------------------ - def _wire_channels(self, projection: dict[str, Any]) -> None: - """Find StreamChannel instances in *projection* and wire them.""" + def _bind_and_wire(self, projection: dict[str, Any]) -> None: + """Bind and wire EventLog / StreamChannel instances in *projection*.""" for value in projection.values(): if isinstance(value, StreamChannel): - # Ensure the channel's log matches the mux's mode. - if value._is_async != self._is_async: - value._is_async = self._is_async - value._log = AsyncEventLog() if self._is_async else EventLog() + value._bind(is_async=self._is_async) self._channels.append(value) channel_name = value.name @@ -128,6 +125,8 @@ class StreamMux: return _forward value._wire(_make_forward(channel_name)) + elif isinstance(value, EventLog): + value._bind(is_async=self._is_async) def _forward(self, channel_name: str, item: Any) -> None: """Inject a ProtocolEvent for a StreamChannel push. diff --git a/libs/langgraph/langgraph/stream/run_stream.py b/libs/langgraph/langgraph/stream/run_stream.py index 4aab11a99..4b15e545a 100644 --- a/libs/langgraph/langgraph/stream/run_stream.py +++ b/libs/langgraph/langgraph/stream/run_stream.py @@ -5,7 +5,7 @@ from collections.abc import AsyncIterator, Iterator from typing import Any from langgraph.stream._convert import convert_to_protocol_event -from langgraph.stream._event_log import AsyncEventLog, EventLog +from langgraph.stream._event_log import EventLog from langgraph.stream._mux import StreamMux from langgraph.stream._types import ProtocolEvent from langgraph.stream.stream_channel import StreamChannel @@ -46,12 +46,11 @@ class GraphRunStream: def _wire_request_more(self, mux: StreamMux, extensions: dict[str, Any]) -> None: """Set _request_more on all sync EventLogs so iteration drives the graph.""" - if isinstance(mux._events, EventLog): - mux._events._request_more = self._pump_next + mux._events._request_more = self._pump_next for value in extensions.values(): if isinstance(value, EventLog): value._request_more = self._pump_next - elif isinstance(value, StreamChannel) and isinstance(value._log, EventLog): + elif isinstance(value, StreamChannel): value._log._request_more = self._pump_next def _pump_next(self) -> bool: @@ -101,7 +100,6 @@ class GraphRunStream: def __iter__(self) -> Iterator[ProtocolEvent]: """Iterate all protocol events from the mux's main event log.""" - assert isinstance(self._mux._events, EventLog) return iter(self._mux._events) @@ -168,5 +166,4 @@ class AsyncGraphRunStream: def __aiter__(self) -> AsyncIterator[ProtocolEvent]: """Iterate all protocol events from the mux's main event log.""" - assert isinstance(self._mux._events, AsyncEventLog) return self._mux._events.__aiter__() diff --git a/libs/langgraph/langgraph/stream/stream_channel.py b/libs/langgraph/langgraph/stream/stream_channel.py index 45e552bd3..d1800da60 100644 --- a/libs/langgraph/langgraph/stream/stream_channel.py +++ b/libs/langgraph/langgraph/stream/stream_channel.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import AsyncIterator, Callable, Iterator from typing import Generic, TypeVar -from langgraph.stream._event_log import AsyncEventLog, EventLog, _EventLogBase +from langgraph.stream._event_log import EventLog T = TypeVar("T") @@ -12,25 +12,32 @@ class StreamChannel(Generic[T]): """A named projection channel with optional protocol auto-forwarding. Wraps an event log and declares a protocol channel name. When the - `StreamMux` detects a `StreamChannel` in a transformer's ``init()`` + ``StreamMux`` detects a ``StreamChannel`` in a transformer's ``init()`` return value, it automatically wires every ``push()`` to inject a - `ProtocolEvent` into the main event stream using the channel's name + ``ProtocolEvent`` into the main event stream using the channel's name as the ``method``. In-process consumers iterate the channel directly (``for item in ch`` or ``async for item in ch``). Remote SDK clients subscribe via ``session.subscribe("custom:")``. + Like ``EventLog``, a ``StreamChannel`` starts unbound. The mux + calls ``_bind(is_async)`` during registration so the correct + iteration protocol is available by the time user code sees it. + Lifecycle (``_close`` / ``_fail``) is managed by the mux — transformers using only StreamChannels don't need ``finalize`` / ``fail`` hooks. """ - def __init__(self, name: str, *, is_async: bool = False) -> None: + def __init__(self, name: str) -> None: self.name = name - self._is_async = is_async - self._log: _EventLogBase[T] = AsyncEventLog() if is_async else EventLog() + self._log: EventLog[T] = EventLog() self._wire_fn: Callable[[T], None] | None = None + def _bind(self, *, is_async: bool) -> None: + """Bind the underlying event log to sync or async mode.""" + self._log._bind(is_async=is_async) + def push(self, item: T) -> None: """Append *item* to the log and auto-forward if wired.""" self._log.push(item) @@ -58,16 +65,7 @@ class StreamChannel(Generic[T]): # ------------------------------------------------------------------ def __iter__(self) -> Iterator[T]: - if not isinstance(self._log, EventLog): - raise RuntimeError( - "Cannot use sync iteration on an async StreamChannel. " - "Use 'async for' instead." - ) return iter(self._log) def __aiter__(self) -> AsyncIterator[T]: - if not isinstance(self._log, AsyncEventLog): - raise RuntimeError( - "Cannot use async iteration on a sync StreamChannel. Use 'for' instead." - ) return self._log.__aiter__() diff --git a/libs/langgraph/langgraph/stream/streaming_handler.py b/libs/langgraph/langgraph/stream/streaming_handler.py index 116ad0da3..4a901bdd3 100644 --- a/libs/langgraph/langgraph/stream/streaming_handler.py +++ b/libs/langgraph/langgraph/stream/streaming_handler.py @@ -142,8 +142,8 @@ class StreamingHandler: """ mux = StreamMux(is_async=is_async) - values_t = ValuesTransformer(is_async=is_async) - messages_t = MessagesTransformer(is_async=is_async) + values_t = ValuesTransformer() + messages_t = MessagesTransformer() all_transformers: list[StreamTransformer] = [values_t, messages_t] if user_transformers: diff --git a/libs/langgraph/langgraph/stream/transformers.py b/libs/langgraph/langgraph/stream/transformers.py index e71648500..9c7bd8444 100644 --- a/libs/langgraph/langgraph/stream/transformers.py +++ b/libs/langgraph/langgraph/stream/transformers.py @@ -2,7 +2,7 @@ from __future__ import annotations from typing import Any -from langgraph.stream._event_log import AsyncEventLog, EventLog, _EventLogBase +from langgraph.stream._event_log import EventLog from langgraph.stream._types import ProtocolEvent, StreamTransformer @@ -15,10 +15,8 @@ class ValuesTransformer(StreamTransformer): _native = True - def __init__(self, *, is_async: bool = False) -> None: - self._log: _EventLogBase[dict[str, Any]] = ( - AsyncEventLog() if is_async else EventLog() - ) + def __init__(self) -> None: + self._log: EventLog[dict[str, Any]] = EventLog() self._latest: dict[str, Any] | None = None self._interrupted = False self._interrupts: list[Any] = [] @@ -61,10 +59,8 @@ class MessagesTransformer(StreamTransformer): _native = True - def __init__(self, *, is_async: bool = False) -> None: - self._log: _EventLogBase[tuple[Any, dict[str, Any]]] = ( - AsyncEventLog() if is_async else EventLog() - ) + def __init__(self) -> None: + self._log: EventLog[tuple[Any, dict[str, Any]]] = EventLog() def init(self) -> dict[str, Any]: return {"messages": self._log} diff --git a/libs/langgraph/tests/test_streaming_handler.py b/libs/langgraph/tests/test_streaming_handler.py index 71e08df04..df6a1410e 100644 --- a/libs/langgraph/tests/test_streaming_handler.py +++ b/libs/langgraph/tests/test_streaming_handler.py @@ -15,7 +15,6 @@ from typing_extensions import TypedDict from langgraph.constants import END, START from langgraph.graph import StateGraph from langgraph.stream import ( - AsyncEventLog, EventLog, StreamChannel, StreamingHandler, @@ -141,6 +140,7 @@ def _build_custom_stream_graph(): class TestEventLog: def test_sync_iteration(self) -> None: log: EventLog[int] = EventLog() + log._bind(is_async=False) log.push(1) log.push(2) log.push(3) @@ -149,6 +149,7 @@ class TestEventLog: def test_multi_cursor(self) -> None: log: EventLog[str] = EventLog() + log._bind(is_async=False) log.push("a") log.push("b") log.close() @@ -158,6 +159,7 @@ class TestEventLog: def test_fail_propagation(self) -> None: log: EventLog[int] = EventLog() + log._bind(is_async=False) log.push(1) log.fail(ValueError("test error")) with pytest.raises(ValueError, match="test error"): @@ -165,7 +167,8 @@ class TestEventLog: @pytest.mark.anyio async def test_async_iteration(self) -> None: - log: AsyncEventLog[int] = AsyncEventLog() + log: EventLog[int] = EventLog() + log._bind(is_async=True) async def producer(): for i in range(3): @@ -178,7 +181,8 @@ class TestEventLog: @pytest.mark.anyio async def test_async_multi_cursor(self) -> None: - log: AsyncEventLog[str] = AsyncEventLog() + log: EventLog[str] = EventLog() + log._bind(is_async=True) log.push("x") log.push("y") log.close() @@ -189,7 +193,8 @@ class TestEventLog: @pytest.mark.anyio async def test_async_fail(self) -> None: - log: AsyncEventLog[int] = AsyncEventLog() + log: EventLog[int] = EventLog() + log._bind(is_async=True) log.push(1) log.fail(RuntimeError("async error")) with pytest.raises(RuntimeError, match="async error"): @@ -199,6 +204,7 @@ class TestEventLog: def test_sync_cursor_yields_items_before_error(self) -> None: """Sync cursor should yield all buffered items before raising.""" log: EventLog[int] = EventLog() + log._bind(is_async=False) log.push(1) log.push(2) log.push(3) @@ -212,7 +218,8 @@ class TestEventLog: @pytest.mark.anyio async def test_async_cursor_yields_items_before_error(self) -> None: """Async cursor should yield all buffered items before raising.""" - log: AsyncEventLog[int] = AsyncEventLog() + log: EventLog[int] = EventLog() + log._bind(is_async=True) log.push(1) log.push(2) log.push(3) @@ -241,19 +248,22 @@ class TestEventLog: def test_empty_log_sync(self) -> None: """Iterating a closed empty log should yield nothing.""" log: EventLog[int] = EventLog() + log._bind(is_async=False) log.close() assert list(log) == [] @pytest.mark.anyio async def test_empty_log_async(self) -> None: """Async-iterating a closed empty log should yield nothing.""" - log: AsyncEventLog[int] = AsyncEventLog() + log: EventLog[int] = EventLog() + log._bind(is_async=True) log.close() assert [item async for item in log] == [] def test_empty_log_fail_sync(self) -> None: """Failing an empty log should raise immediately with no items.""" log: EventLog[int] = EventLog() + log._bind(is_async=False) log.fail(ValueError("empty fail")) with pytest.raises(ValueError, match="empty fail"): list(log) @@ -261,24 +271,43 @@ class TestEventLog: @pytest.mark.anyio async def test_empty_log_fail_async(self) -> None: """Failing an empty log should raise immediately with no items (async).""" - log: AsyncEventLog[int] = AsyncEventLog() + log: EventLog[int] = EventLog() + log._bind(is_async=True) log.fail(ValueError("empty fail")) with pytest.raises(ValueError, match="empty fail"): async for _ in log: pass - def test_sync_has_no_aiter(self) -> None: - """EventLog (sync) should not support async iteration.""" + def test_unbound_iter_raises(self) -> None: + """Iterating an unbound EventLog should raise TypeError.""" log: EventLog[int] = EventLog() log.close() - assert not hasattr(log, "__aiter__") + with pytest.raises(TypeError, match="has not been bound"): + list(log) + + def test_sync_bound_aiter_raises(self) -> None: + """Sync-bound EventLog should reject async iteration.""" + log: EventLog[int] = EventLog() + log._bind(is_async=False) + log.close() + with pytest.raises(TypeError, match="bound to sync mode"): + log.__aiter__() @pytest.mark.anyio - async def test_async_has_no_iter(self) -> None: - """AsyncEventLog should not support sync iteration.""" - log: AsyncEventLog[int] = AsyncEventLog() + async def test_async_bound_iter_raises(self) -> None: + """Async-bound EventLog should reject sync iteration.""" + log: EventLog[int] = EventLog() + log._bind(is_async=True) log.close() - assert not hasattr(log, "__iter__") + with pytest.raises(TypeError, match="bound to async mode"): + iter(log) + + def test_double_bind_raises(self) -> None: + """Binding an already-bound EventLog should raise.""" + log: EventLog[int] = EventLog() + log._bind(is_async=False) + with pytest.raises(RuntimeError, match="already bound"): + log._bind(is_async=True) # --------------------------------------------------------------------------- @@ -289,6 +318,7 @@ class TestEventLog: class TestStreamChannel: def test_push_and_iterate(self) -> None: ch: StreamChannel[str] = StreamChannel("test") + ch._bind(is_async=False) ch.push("a") ch.push("b") ch._close() @@ -297,6 +327,7 @@ class TestStreamChannel: def test_wire_callback(self) -> None: forwarded: list[str] = [] ch: StreamChannel[str] = StreamChannel("test") + ch._bind(is_async=False) ch._wire(lambda item: forwarded.append(item)) ch.push("x") ch.push("y") @@ -307,6 +338,7 @@ class TestStreamChannel: def test_fail_propagation(self) -> None: """_fail() should propagate the error through the underlying log.""" ch: StreamChannel[str] = StreamChannel("test") + ch._bind(is_async=False) ch.push("a") ch._fail(ValueError("channel error")) items: list[str] = [] @@ -317,8 +349,9 @@ class TestStreamChannel: @pytest.mark.anyio async def test_async_iteration(self) -> None: - """Async iteration should delegate to the inner AsyncEventLog.""" - ch: StreamChannel[str] = StreamChannel("test", is_async=True) + """Async iteration should delegate to the inner event log.""" + ch: StreamChannel[str] = StreamChannel("test") + ch._bind(is_async=True) ch.push("x") ch.push("y") ch._close() @@ -328,6 +361,7 @@ class TestStreamChannel: def test_push_without_wire(self) -> None: """Push without a wire callback should still append to the log.""" ch: StreamChannel[int] = StreamChannel("test") + ch._bind(is_async=False) assert ch._wire_fn is None ch.push(42) ch._close() @@ -690,6 +724,7 @@ class TestValuesTransformer: """Values events from subgraphs (non-empty namespace) should be ignored.""" t = ValuesTransformer() t.init() + t._log._bind(is_async=False) t.process(_event("values", {"val": "root"})) t.process(_event("values", {"val": "sub"}, namespace=["sub"])) @@ -703,6 +738,7 @@ class TestValuesTransformer: """Non-values events should be passed through but not captured.""" t = ValuesTransformer() t.init() + t._log._bind(is_async=False) result = t.process(_event("updates", {"x": 1})) assert result is True # passed through @@ -729,6 +765,7 @@ class TestMessagesTransformer: def test_captures_root_messages(self) -> None: t = MessagesTransformer() t.init() + t._log._bind(is_async=False) t.process(_event("messages", ("chunk", {"meta": True}))) t.finalize() @@ -739,6 +776,7 @@ class TestMessagesTransformer: def test_ignores_non_root_namespace(self) -> None: t = MessagesTransformer() t.init() + t._log._bind(is_async=False) t.process(_event("messages", ("chunk", {}), namespace=["sub"])) t.finalize() @@ -747,6 +785,7 @@ class TestMessagesTransformer: def test_ignores_non_messages_methods(self) -> None: t = MessagesTransformer() t.init() + t._log._bind(is_async=False) result = t.process(_event("values", {"v": 1})) assert result is True @@ -756,6 +795,7 @@ class TestMessagesTransformer: def test_fail_propagates(self) -> None: t = MessagesTransformer() t.init() + t._log._bind(is_async=False) t.fail(ValueError("msg error")) with pytest.raises(ValueError, match="msg error"): list(t._log)