From 3eb73e8ad2de66c2a088b30ffbf098bb186fbfb3 Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Wed, 29 Apr 2026 09:43:49 -0400 Subject: [PATCH] feat(langgraph): native v2 projections for custom, updates, checkpoints, debug, tasks (#7640) --- libs/langgraph/langgraph/stream/__init__.py | 10 + .../langgraph/stream/transformers.py | 185 +++++ .../tests/test_stream_data_transformers.py | 694 ++++++++++++++++++ 3 files changed, 889 insertions(+) create mode 100644 libs/langgraph/tests/test_stream_data_transformers.py diff --git a/libs/langgraph/langgraph/stream/__init__.py b/libs/langgraph/langgraph/stream/__init__.py index 1830597d0..294b760b2 100644 --- a/libs/langgraph/langgraph/stream/__init__.py +++ b/libs/langgraph/langgraph/stream/__init__.py @@ -14,15 +14,23 @@ from langgraph.stream.run_stream import ( ) from langgraph.stream.stream_channel import StreamChannel from langgraph.stream.transformers import ( + CheckpointsTransformer, + CustomTransformer, + DebugTransformer, LifecyclePayload, LifecycleTransformer, SubgraphStatus, SubgraphTransformer, + TasksTransformer, + UpdatesTransformer, ) __all__ = [ "AsyncGraphRunStream", "AsyncSubgraphRunStream", + "CheckpointsTransformer", + "CustomTransformer", + "DebugTransformer", "GraphRunStream", "LifecyclePayload", "LifecycleTransformer", @@ -32,4 +40,6 @@ __all__ = [ "SubgraphRunStream", "SubgraphStatus", "SubgraphTransformer", + "TasksTransformer", + "UpdatesTransformer", ] diff --git a/libs/langgraph/langgraph/stream/transformers.py b/libs/langgraph/langgraph/stream/transformers.py index fc2009b13..27ad12376 100644 --- a/libs/langgraph/langgraph/stream/transformers.py +++ b/libs/langgraph/langgraph/stream/transformers.py @@ -82,6 +82,76 @@ class ValuesTransformer(StreamTransformer): return True +class CustomTransformer(StreamTransformer): + """Capture custom events as a drainable stream of arbitrary payloads. + + Nodes emit custom data via `get_stream_writer()`. This transformer + surfaces those events on `run.custom` as a `StreamChannel[Any]`, + preserving payloads in arrival order. + + Only events at the run's own scope are captured; custom data from + deeper subgraphs is available on the respective subgraph handle's + `.custom` projection. + + Native transformer — `run.custom` is a direct attribute. + """ + + _native = True + required_stream_modes = ("custom",) + + def __init__(self, scope: tuple[str, ...] = ()) -> None: + super().__init__(scope) + self._log: StreamChannel[Any] = StreamChannel() + self._scope_list: list[str] = list(scope) + + def init(self) -> dict[str, Any]: + return {"custom": self._log} + + def process(self, event: ProtocolEvent) -> bool: + if event["method"] != "custom": + return True + params = event["params"] + if params["namespace"] != self._scope_list: + return True + self._log.push(params["data"]) + return True + + +class UpdatesTransformer(StreamTransformer): + """Capture updates events as a drainable stream of node outputs. + + Surfaces `stream_mode="updates"` data on `run.updates` as a + `StreamChannel[dict[str, Any]]`. Each item is a dict mapping a node + (or task) name to the update it returned after a step. + + Only events at the run's own scope are captured; updates from deeper + subgraphs are available on the respective subgraph handle's + `.updates` projection. + + Native transformer — `run.updates` is a direct attribute. + """ + + _native = True + required_stream_modes = ("updates",) + + def __init__(self, scope: tuple[str, ...] = ()) -> None: + super().__init__(scope) + self._log: StreamChannel[dict[str, Any]] = StreamChannel() + self._scope_list: list[str] = list(scope) + + def init(self) -> dict[str, Any]: + return {"updates": self._log} + + def process(self, event: ProtocolEvent) -> bool: + if event["method"] != "updates": + return True + params = event["params"] + if params["namespace"] != self._scope_list: + return True + self._log.push(params["data"]) + return True + + class MessagesTransformer(StreamTransformer): """Capture messages events as ChatModelStream objects. @@ -741,3 +811,118 @@ class SubgraphTransformer(_TasksLifecycleBase): handle.path, exc_info=True, ) + + +class CheckpointsTransformer(StreamTransformer): + """Capture checkpoint events as a drainable stream. + + Surfaces `stream_mode="checkpoints"` data on `run.checkpoints` as + a `StreamChannel[dict[str, Any]]`. Each item is in the same format + as returned by `get_state()`. + + Checkpoint events are only emitted when a checkpointer is configured + on the graph. When no checkpointer is present, the projection exists + but receives no events. + + Only events at the run's own scope are captured; checkpoint data from + deeper subgraphs is available on the respective subgraph handle's + `.checkpoints` projection. + + Native transformer — `run.checkpoints` is a direct attribute. + """ + + _native = True + required_stream_modes = ("checkpoints",) + + def __init__(self, scope: tuple[str, ...] = ()) -> None: + super().__init__(scope) + self._log: StreamChannel[dict[str, Any]] = StreamChannel() + self._scope_list: list[str] = list(scope) + + def init(self) -> dict[str, Any]: + return {"checkpoints": self._log} + + def process(self, event: ProtocolEvent) -> bool: + if event["method"] != "checkpoints": + return True + params = event["params"] + if params["namespace"] != self._scope_list: + return True + self._log.push(params["data"]) + return True + + +class DebugTransformer(StreamTransformer): + """Capture debug events as a drainable stream. + + Surfaces `stream_mode="debug"` data on `run.debug` as a + `StreamChannel[dict[str, Any]]`. Each item is a debug event with + step-level detail (checkpoint snapshots, task payloads, and + task results wrapped with step number and timestamp). + + Only events at the run's own scope are captured; debug data from + deeper subgraphs is available on the respective subgraph handle's + `.debug` projection. + + Native transformer — `run.debug` is a direct attribute. + """ + + _native = True + required_stream_modes = ("debug",) + + def __init__(self, scope: tuple[str, ...] = ()) -> None: + super().__init__(scope) + self._log: StreamChannel[dict[str, Any]] = StreamChannel() + self._scope_list: list[str] = list(scope) + + def init(self) -> dict[str, Any]: + return {"debug": self._log} + + def process(self, event: ProtocolEvent) -> bool: + if event["method"] != "debug": + return True + params = event["params"] + if params["namespace"] != self._scope_list: + return True + self._log.push(params["data"]) + return True + + +class TasksTransformer(StreamTransformer): + """Capture raw task events as a drainable stream. + + Surfaces `stream_mode="tasks"` data on `run.tasks` as a + `StreamChannel[dict[str, Any]]`. Each item is a task payload + (start or result). + + `LifecycleTransformer` and `SubgraphTransformer` also consume + `tasks` events for subgraph discovery and lifecycle tracking. + This transformer captures the raw payloads independently for + consumers who need task-level detail. + + Only events at the run's own scope are captured; task data from + deeper subgraphs is available on the respective subgraph handle's + `.tasks` projection. + + Native transformer — `run.tasks` is a direct attribute. + """ + + _native = True + required_stream_modes = ("tasks",) + + def __init__(self, scope: tuple[str, ...] = ()) -> None: + super().__init__(scope) + self._log: StreamChannel[dict[str, Any]] = StreamChannel() + self._scope_list: list[str] = list(scope) + + def init(self) -> dict[str, Any]: + return {"tasks": self._log} + + def process(self, event: ProtocolEvent) -> bool: + if event["method"] != "tasks": + return True + params = event["params"] + if params["namespace"] != self._scope_list: + return True + self._log.push(params["data"]) + return True diff --git a/libs/langgraph/tests/test_stream_data_transformers.py b/libs/langgraph/tests/test_stream_data_transformers.py new file mode 100644 index 000000000..61b105ccc --- /dev/null +++ b/libs/langgraph/tests/test_stream_data_transformers.py @@ -0,0 +1,694 @@ +"""Tests for CustomTransformer, UpdatesTransformer, CheckpointsTransformer, DebugTransformer, TasksTransformer. + +These transformers capture raw protocol events for their respective stream +modes and expose them as native projections on the run stream (run.custom, +run.updates, run.checkpoints, run.debug, run.tasks). Tests dispatch synthetic +protocol events through a StreamMux to isolate transformer logic; the final +group exercises real graphs through stream_v2. +""" + +from __future__ import annotations + +import operator +import time +from typing import Annotated, Any + +from typing_extensions import TypedDict + +from langgraph.constants import END, START +from langgraph.graph import StateGraph +from langgraph.stream._mux import StreamMux +from langgraph.stream.stream_channel import StreamChannel +from langgraph.stream.transformers import ( + CheckpointsTransformer, + CustomTransformer, + DebugTransformer, + LifecycleTransformer, + TasksTransformer, + UpdatesTransformer, +) + +TS = int(time.time() * 1000) + + +def _custom_event(namespace: list[str], data: Any) -> dict[str, Any]: + return { + "type": "event", + "method": "custom", + "params": {"namespace": namespace, "timestamp": TS, "data": data}, + } + + +def _checkpoints_event(namespace: list[str], data: Any) -> dict[str, Any]: + return { + "type": "event", + "method": "checkpoints", + "params": {"namespace": namespace, "timestamp": TS, "data": data}, + } + + +def _debug_event(namespace: list[str], data: Any) -> dict[str, Any]: + return { + "type": "event", + "method": "debug", + "params": {"namespace": namespace, "timestamp": TS, "data": data}, + } + + +def _tasks_event(namespace: list[str], data: Any) -> dict[str, Any]: + return { + "type": "event", + "method": "tasks", + "params": {"namespace": namespace, "timestamp": TS, "data": data}, + } + + +def _updates_event(namespace: list[str], data: Any) -> dict[str, Any]: + return { + "type": "event", + "method": "updates", + "params": {"namespace": namespace, "timestamp": TS, "data": data}, + } + + +def _arm(mux: StreamMux, transformer: Any) -> None: + """Force projection logs to accept pushes (skip lazy-subscribe gate).""" + mux._events._subscribed = True + transformer._log._subscribed = True + + +def _drain(transformer: Any) -> list[Any]: + return list(transformer._log._items) + + +# --------------------------------------------------------------------------- +# CustomTransformer +# --------------------------------------------------------------------------- + + +def test_custom_captures_root_scope_events() -> None: + t = CustomTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_custom_event([], {"status": "processing"})) + mux.push(_custom_event([], {"status": "done"})) + + items = _drain(t) + assert items == [{"status": "processing"}, {"status": "done"}] + + +def test_custom_ignores_subgraph_scope_events() -> None: + t = CustomTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_custom_event(["subgraph:abc"], {"from": "child"})) + + assert _drain(t) == [] + + +def test_custom_scoped_transformer_captures_own_scope() -> None: + t = CustomTransformer(scope=("agent:abc",)) + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_custom_event([], {"from": "root"})) + mux.push(_custom_event(["agent:abc"], {"from": "self"})) + mux.push(_custom_event(["agent:abc", "deep:def"], {"from": "child"})) + + items = _drain(t) + assert items == [{"from": "self"}] + + +def test_custom_preserves_any_payload_type() -> None: + t = CustomTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_custom_event([], "string_payload")) + mux.push(_custom_event([], 42)) + mux.push(_custom_event([], [1, 2, 3])) + + assert _drain(t) == ["string_payload", 42, [1, 2, 3]] + + +def test_custom_does_not_suppress_from_main_log() -> None: + t = CustomTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_custom_event([], "data")) + + methods = [evt["method"] for evt in mux._events._items] + assert "custom" in methods + + +def test_custom_ignores_other_methods() -> None: + t = CustomTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push( + { + "type": "event", + "method": "values", + "params": {"namespace": [], "timestamp": TS, "data": {}}, + } + ) + assert _drain(t) == [] + + +def test_custom_required_stream_modes() -> None: + assert CustomTransformer.required_stream_modes == ("custom",) + + +def test_custom_is_native() -> None: + assert getattr(CustomTransformer, "_native", False) is True + + +def test_custom_init_returns_correct_key() -> None: + t = CustomTransformer() + projection = t.init() + assert "custom" in projection + assert isinstance(projection["custom"], StreamChannel) + + +# --------------------------------------------------------------------------- +# CheckpointsTransformer +# --------------------------------------------------------------------------- + + +def test_checkpoints_captures_root_scope_events() -> None: + t = CheckpointsTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + checkpoint_data = {"values": {"x": 1}, "next": ["node_b"]} + mux.push(_checkpoints_event([], checkpoint_data)) + + items = _drain(t) + assert items == [checkpoint_data] + + +def test_checkpoints_ignores_subgraph_events() -> None: + t = CheckpointsTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_checkpoints_event(["child:abc"], {"values": {"x": 1}})) + + assert _drain(t) == [] + + +def test_checkpoints_scoped_transformer() -> None: + t = CheckpointsTransformer(scope=("sub:abc",)) + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_checkpoints_event([], {"from": "root"})) + mux.push(_checkpoints_event(["sub:abc"], {"from": "self"})) + + assert _drain(t) == [{"from": "self"}] + + +def test_checkpoints_does_not_suppress_from_main_log() -> None: + t = CheckpointsTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_checkpoints_event([], {"values": {}})) + + methods = [evt["method"] for evt in mux._events._items] + assert "checkpoints" in methods + + +def test_checkpoints_required_stream_modes() -> None: + assert CheckpointsTransformer.required_stream_modes == ("checkpoints",) + + +def test_checkpoints_is_native() -> None: + assert getattr(CheckpointsTransformer, "_native", False) is True + + +# --------------------------------------------------------------------------- +# DebugTransformer +# --------------------------------------------------------------------------- + + +def test_debug_captures_root_scope_events() -> None: + t = DebugTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + debug_data = { + "step": 0, + "type": "checkpoint", + "timestamp": "2026-01-01T00:00:00Z", + "payload": {"values": {"x": 1}}, + } + mux.push(_debug_event([], debug_data)) + + items = _drain(t) + assert items == [debug_data] + + +def test_debug_ignores_subgraph_events() -> None: + t = DebugTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_debug_event(["child:abc"], {"step": 0, "type": "task"})) + + assert _drain(t) == [] + + +def test_debug_captures_multiple_event_types() -> None: + t = DebugTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_debug_event([], {"step": 0, "type": "checkpoint", "payload": {}})) + mux.push(_debug_event([], {"step": 1, "type": "task", "payload": {}})) + mux.push(_debug_event([], {"step": 1, "type": "task_result", "payload": {}})) + + items = _drain(t) + assert len(items) == 3 + assert [d["type"] for d in items] == ["checkpoint", "task", "task_result"] + + +def test_debug_does_not_suppress_from_main_log() -> None: + t = DebugTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_debug_event([], {"step": 0})) + + methods = [evt["method"] for evt in mux._events._items] + assert "debug" in methods + + +def test_debug_required_stream_modes() -> None: + assert DebugTransformer.required_stream_modes == ("debug",) + + +def test_debug_is_native() -> None: + assert getattr(DebugTransformer, "_native", False) is True + + +# --------------------------------------------------------------------------- +# TasksTransformer +# --------------------------------------------------------------------------- + + +def test_tasks_captures_root_scope_events() -> None: + t = TasksTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + task_start = {"id": "t1", "name": "my_node", "input": None, "triggers": []} + mux.push(_tasks_event([], task_start)) + + items = _drain(t) + assert items == [task_start] + + +def test_tasks_captures_start_and_result() -> None: + t = TasksTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + start = {"id": "t1", "name": "a", "input": None, "triggers": []} + result = {"id": "t1", "name": "a", "result": {"output": 42}, "error": None} + mux.push(_tasks_event([], start)) + mux.push(_tasks_event([], result)) + + items = _drain(t) + assert items == [start, result] + + +def test_tasks_ignores_subgraph_events() -> None: + t = TasksTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_tasks_event(["child:abc"], {"id": "t1", "name": "x"})) + + assert _drain(t) == [] + + +def test_tasks_scoped_transformer() -> None: + t = TasksTransformer(scope=("agent:abc",)) + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_tasks_event([], {"id": "t1"})) + mux.push(_tasks_event(["agent:abc"], {"id": "t2"})) + mux.push(_tasks_event(["agent:abc", "deep:def"], {"id": "t3"})) + + assert _drain(t) == [{"id": "t2"}] + + +def test_tasks_does_not_suppress_from_main_log() -> None: + """TasksTransformer returns True — it doesn't suppress tasks events. + + (LifecycleTransformer suppresses them, but that's independent.) + """ + t = TasksTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_tasks_event([], {"id": "t1"})) + + methods = [evt["method"] for evt in mux._events._items] + assert "tasks" in methods + + +def test_tasks_required_stream_modes() -> None: + assert TasksTransformer.required_stream_modes == ("tasks",) + + +def test_tasks_is_native() -> None: + assert getattr(TasksTransformer, "_native", False) is True + + +# --------------------------------------------------------------------------- +# UpdatesTransformer +# --------------------------------------------------------------------------- + + +def test_updates_captures_root_scope_events() -> None: + t = UpdatesTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + update = {"my_node": {"value": "hello!"}} + mux.push(_updates_event([], update)) + + items = _drain(t) + assert items == [update] + + +def test_updates_captures_multiple_steps() -> None: + t = UpdatesTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_updates_event([], {"node_a": {"x": 1}})) + mux.push(_updates_event([], {"node_b": {"x": 2}})) + + items = _drain(t) + assert items == [{"node_a": {"x": 1}}, {"node_b": {"x": 2}}] + + +def test_updates_ignores_subgraph_events() -> None: + t = UpdatesTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_updates_event(["child:abc"], {"inner_node": {"v": 1}})) + + assert _drain(t) == [] + + +def test_updates_scoped_transformer() -> None: + t = UpdatesTransformer(scope=("agent:abc",)) + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_updates_event([], {"from": "root"})) + mux.push(_updates_event(["agent:abc"], {"from": "self"})) + + assert _drain(t) == [{"from": "self"}] + + +def test_updates_does_not_suppress_from_main_log() -> None: + t = UpdatesTransformer() + mux = StreamMux([t], is_async=False) + _arm(mux, t) + + mux.push(_updates_event([], {"n": {}})) + + methods = [evt["method"] for evt in mux._events._items] + assert "updates" in methods + + +def test_updates_required_stream_modes() -> None: + assert UpdatesTransformer.required_stream_modes == ("updates",) + + +def test_updates_is_native() -> None: + assert getattr(UpdatesTransformer, "_native", False) is True + + +# --------------------------------------------------------------------------- +# Cross-transformer: unrelated events pass through +# --------------------------------------------------------------------------- + + +def test_unrelated_events_ignored_by_all() -> None: + """Non-matching method events don't land in any transformer's log.""" + transformers = [ + CustomTransformer(), + UpdatesTransformer(), + CheckpointsTransformer(), + DebugTransformer(), + TasksTransformer(), + ] + mux = StreamMux(transformers, is_async=False) + mux._events._subscribed = True + for t in transformers: + t._log._subscribed = True + + mux.push( + { + "type": "event", + "method": "values", + "params": {"namespace": [], "timestamp": TS, "data": {"x": 1}}, + } + ) + + for t in transformers: + assert list(t._log._items) == [] + + +# --------------------------------------------------------------------------- +# End-to-end: real graphs through stream_v2 +# --------------------------------------------------------------------------- + + +class _State(TypedDict): + value: str + items: Annotated[list[str], operator.add] + + +def _my_node(state: _State) -> dict[str, Any]: + from langgraph.config import get_stream_writer + + writer = get_stream_writer() + writer({"status": "working", "node": "my_node"}) + return {"value": state["value"] + "!", "items": ["done"]} + + +def _make_simple_graph() -> Any: + builder = StateGraph(_State, input_schema=_State) + builder.add_node("my_node", _my_node) + builder.add_edge(START, "my_node") + builder.add_edge("my_node", END) + return builder.compile() + + +def test_stream_v2_custom_projection_opt_in() -> None: + """run.custom surfaces get_stream_writer() payloads when opted in.""" + graph = _make_simple_graph() + run = graph.stream_v2( + {"value": "hello", "items": []}, transformers=[CustomTransformer] + ) + + custom_events = list(run.custom) + assert len(custom_events) >= 1 + assert any(e.get("status") == "working" for e in custom_events) + + +def test_stream_v2_custom_and_values_coexist() -> None: + """Both run.custom and run.values work in the same run.""" + graph = _make_simple_graph() + run = graph.stream_v2( + {"value": "hello", "items": []}, transformers=[CustomTransformer] + ) + + custom_events = list(run.custom) + assert run.output is not None + assert run.output["value"] == "hello!" + assert len(custom_events) >= 1 + + +def test_stream_v2_tasks_projection_opt_in() -> None: + """run.tasks surfaces raw task events when opted in via transformers=.""" + graph = _make_simple_graph() + run = graph.stream_v2({"value": "x", "items": []}, transformers=[TasksTransformer]) + + tasks_events = list(run.tasks) + assert len(tasks_events) >= 1 + names = [t.get("name") for t in tasks_events if "name" in t] + assert "my_node" in names + + +def test_stream_v2_debug_projection_opt_in() -> None: + """run.debug surfaces debug events when opted in via transformers=.""" + graph = _make_simple_graph() + run = graph.stream_v2({"value": "x", "items": []}, transformers=[DebugTransformer]) + + debug_events = list(run.debug) + assert len(debug_events) >= 1 + types = {d.get("type") for d in debug_events} + assert types & {"checkpoint", "task", "task_result"} + + +def test_stream_v2_updates_projection_opt_in() -> None: + """run.updates surfaces node output dicts when opted in via transformers=.""" + graph = _make_simple_graph() + run = graph.stream_v2( + {"value": "x", "items": []}, transformers=[UpdatesTransformer] + ) + + updates = list(run.updates) + assert len(updates) >= 1 + node_names = {k for u in updates for k in u if k != "__interrupt__"} + assert "my_node" in node_names + + +def test_stream_v2_all_transformers_interleaved() -> None: + """All five transformers registered together, consumed via interleave.""" + graph = _make_simple_graph() + run = graph.stream_v2( + {"value": "x", "items": []}, + transformers=[ + CustomTransformer, + UpdatesTransformer, + CheckpointsTransformer, + DebugTransformer, + TasksTransformer, + ], + ) + + collected: dict[str, list[Any]] = { + "custom": [], + "updates": [], + "debug": [], + "tasks": [], + } + for name, item in run.interleave("custom", "updates", "debug", "tasks"): + collected[name].append(item) + + assert len(collected["custom"]) >= 1 + assert len(collected["updates"]) >= 1 + assert len(collected["tasks"]) >= 1 + assert len(collected["debug"]) >= 1 + types = {d.get("type") for d in collected["debug"]} + assert types & {"checkpoint", "task", "task_result"} + node_names = {k for u in collected["updates"] for k in u if k != "__interrupt__"} + assert "my_node" in node_names + + assert run.output is not None + assert run.output["value"] == "x!" + + +def test_stream_v2_all_transformers_with_checkpointer() -> None: + """All transformers with a checkpointer — run.checkpoints populated.""" + from langgraph.checkpoint.memory import InMemorySaver + + builder = StateGraph(_State, input_schema=_State) + builder.add_node("my_node", _my_node) + builder.add_edge(START, "my_node") + builder.add_edge("my_node", END) + graph = builder.compile(checkpointer=InMemorySaver()) + + run = graph.stream_v2( + {"value": "x", "items": []}, + config={"configurable": {"thread_id": "test-all"}}, + transformers=[ + CustomTransformer, + UpdatesTransformer, + CheckpointsTransformer, + DebugTransformer, + TasksTransformer, + ], + ) + + collected: dict[str, list[Any]] = { + "custom": [], + "updates": [], + "checkpoints": [], + "debug": [], + "tasks": [], + } + for name, item in run.interleave( + "custom", "updates", "checkpoints", "debug", "tasks" + ): + collected[name].append(item) + + assert len(collected["checkpoints"]) >= 1 + assert len(collected["custom"]) >= 1 + + +def test_stream_v2_checkpoints_projection_opt_in() -> None: + """run.checkpoints surfaces checkpoint data when opted in with a checkpointer.""" + from langgraph.checkpoint.memory import InMemorySaver + + builder = StateGraph(_State, input_schema=_State) + builder.add_node("my_node", _my_node) + builder.add_edge(START, "my_node") + builder.add_edge("my_node", END) + graph = builder.compile(checkpointer=InMemorySaver()) + + run = graph.stream_v2( + {"value": "x", "items": []}, + config={"configurable": {"thread_id": "test-ckpt-standalone"}}, + transformers=[CheckpointsTransformer], + ) + + checkpoints = list(run.checkpoints) + assert len(checkpoints) >= 1 + + +# --------------------------------------------------------------------------- +# TasksTransformer + LifecycleTransformer co-registration +# --------------------------------------------------------------------------- + + +def test_tasks_and_lifecycle_coregistration() -> None: + """When both are in the same StreamMux, LifecycleTransformer suppresses + tasks events from the main log (returns False) while TasksTransformer + still captures them into its own log. + """ + lifecycle = LifecycleTransformer() + tasks = TasksTransformer() + mux = StreamMux([lifecycle, tasks], is_async=False) + mux._events._subscribed = True + tasks._log._subscribed = True + lifecycle._channel._subscribed = True + + task_data = {"id": "t1", "name": "my_node", "input": None, "triggers": []} + mux.push(_tasks_event([], task_data)) + + assert _drain(tasks) == [task_data] + + methods = [evt["method"] for evt in mux._events._items] + assert "tasks" not in methods + + +def test_tasks_and_lifecycle_coregistration_e2e() -> None: + """E2e: TasksTransformer captures task events even when LifecycleTransformer + is present and suppressing them from the main log. + """ + graph = _make_simple_graph() + run = graph.stream_v2( + {"value": "x", "items": []}, + transformers=[TasksTransformer], + ) + + tasks_events = list(run.tasks) + assert len(tasks_events) >= 1 + names = [t.get("name") for t in tasks_events if "name" in t] + assert "my_node" in names