From f30a11055bc5ae9169449d0fd5d659f6fc39f0e0 Mon Sep 17 00:00:00 2001 From: Christian Bromann Date: Thu, 23 Apr 2026 16:13:22 -0700 Subject: [PATCH] use cause in lifecycle --- libs/langgraph/langgraph/pregel/_lifecycle.py | 25 +++---------------- .../langgraph/stream/transformers.py | 20 ++++++++++----- .../tests/test_stream_subgraph_transformer.py | 25 ++++++++++--------- 3 files changed, 31 insertions(+), 39 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/_lifecycle.py b/libs/langgraph/langgraph/pregel/_lifecycle.py index 7875a748b..293bdffef 100644 --- a/libs/langgraph/langgraph/pregel/_lifecycle.py +++ b/libs/langgraph/langgraph/pregel/_lifecycle.py @@ -166,24 +166,6 @@ class StreamLifecycleHandler(BaseCallbackHandler, _StreamingCallbackHandler): return () return tuple(cast(str, nskey).split(NS_SEP))[:-1] - @staticmethod - def _trigger_call_id(metadata: dict[str, Any] | None) -> str | None: - """Extract `trigger_call_id` from task metadata if present. - - The task that spawned a nested `Pregel` has its task id encoded - in `langgraph_checkpoint_ns`'s last segment as - `node_name:task_id`. Returns the `task_id` portion, which - parents can correlate with their `tools` / `tasks` events. - """ - if not metadata: - return None - nskey = cast(str | None, metadata.get("langgraph_checkpoint_ns")) - if not nskey: - return None - last = nskey.split(NS_SEP)[-1] - _, sep, task_id = last.rpartition(":") - return task_id if sep else None - def _emit(self, ns: tuple[str, ...], payload: dict[str, Any]) -> None: self.stream((ns, "lifecycle", payload)) @@ -248,9 +230,10 @@ class StreamLifecycleHandler(BaseCallbackHandler, _StreamingCallbackHandler): payload: dict[str, Any] = {"event": "started"} if name: payload["graph_name"] = name - trigger_call_id = self._trigger_call_id(metadata) - if trigger_call_id: - payload["trigger_call_id"] = trigger_call_id + # `cause` is intentionally not populated here: pregel does not know + # what on the parent namespace triggered this subgraph. Product- + # specific stream transformers populate `cause` before events + # reach the wire. See LifecycleCause in the protocol definition. self._emit(ns, payload) self._pending_running.add(ns) diff --git a/libs/langgraph/langgraph/stream/transformers.py b/libs/langgraph/langgraph/stream/transformers.py index f5ac39bf5..112fd1776 100644 --- a/libs/langgraph/langgraph/stream/transformers.py +++ b/libs/langgraph/langgraph/stream/transformers.py @@ -9,7 +9,12 @@ from langchain_core.language_models.chat_model_stream import ( ChatModelStream, ) from langchain_core.messages import AIMessageChunk, BaseMessage -from langchain_protocol.protocol import CheckpointRef, LifecycleData, MessagesData +from langchain_protocol.protocol import ( + CheckpointRef, + LifecycleCause, + LifecycleData, + MessagesData, +) from langgraph.errors import GraphInterrupt from langgraph.stream._event_log import EventLog @@ -285,8 +290,11 @@ class SubgraphRunStream(BaseRunStream): Lifecycle fields update in place as events arrive: - `path`: the namespace tuple — stable for the life of the handle. - - `graph_name` / `trigger_call_id`: set once from the `started` - payload. + - `graph_name` / `cause`: set once from the `started` payload. + `cause` is populated by product-specific stream transformers + (see `LifecycleCause` in the protocol definition); pregel itself + emits no `cause`, so it may be `None` for subgraphs not covered + by a product transformer. - `status`: advances `started` → `running` → `completed` / `failed` / `interrupted`. - `error` / `checkpoint`: set on the terminal event when present. @@ -303,12 +311,12 @@ class SubgraphRunStream(BaseRunStream): mux: StreamMux, *, graph_name: str | None = None, - trigger_call_id: str | None = None, + cause: LifecycleCause | None = None, ) -> None: super().__init__(mux) self.path: tuple[str, ...] = path self.graph_name: str | None = graph_name - self.trigger_call_id: str | None = trigger_call_id + self.cause: LifecycleCause | None = cause self.status: SubgraphStatus = "started" self.error: str | None = None self.checkpoint: CheckpointRef | None = None @@ -427,7 +435,7 @@ class SubgraphTransformer(StreamTransformer): path=ns, mux=child_mux, graph_name=data.get("graph_name"), - trigger_call_id=data.get("trigger_call_id"), + cause=data.get("cause"), ) self._by_ns[ns] = handle self._root_log.push(handle) diff --git a/libs/langgraph/tests/test_stream_subgraph_transformer.py b/libs/langgraph/tests/test_stream_subgraph_transformer.py index 6a137a6ef..ee6591e55 100644 --- a/libs/langgraph/tests/test_stream_subgraph_transformer.py +++ b/libs/langgraph/tests/test_stream_subgraph_transformer.py @@ -32,14 +32,14 @@ def _lifecycle( *, namespace: list[str] | None = None, graph_name: str | None = None, - trigger_call_id: str | None = None, + cause: dict[str, Any] | None = None, error: str | None = None, ) -> ProtocolEvent: data: dict[str, Any] = {"event": event} if graph_name is not None: data["graph_name"] = graph_name - if trigger_call_id is not None: - data["trigger_call_id"] = trigger_call_id + if cause is not None: + data["cause"] = cause if error is not None: data["error"] = error return { @@ -125,14 +125,14 @@ class TestSubgraphTransformerUnit: "started", namespace=["task_a:child"], graph_name="child", - trigger_call_id="task_a", + cause={"type": "toolCall", "tool_call_id": "call_abc"}, ) ) handle = self._handle(transformer) assert handle.path == ("task_a:child",) assert handle.graph_name == "child" - assert handle.trigger_call_id == "task_a" + assert handle.cause == {"type": "toolCall", "tool_call_id": "call_abc"} assert handle.status == "started" def test_status_transitions(self) -> None: @@ -371,10 +371,10 @@ class TestSubgraphTransformerAsyncEndToEnd: assert child.status == "completed" -class TestSubgraphTriggerCallId: - """Confirm `trigger_call_id` flows from real pregel metadata.""" +class TestSubgraphCause: + """Pregel core emits no `cause`; product transformers populate it.""" - def test_trigger_call_id_populated_end_to_end(self) -> None: + def test_cause_not_populated_by_pregel(self) -> None: graph = _build_nested_graph() run = graph.stream_v2({"value": "", "items": []}) @@ -382,14 +382,15 @@ class TestSubgraphTriggerCallId: assert len(collected) == 1 child = collected[0] - # The child's single-segment path encodes `node_name:task_id`. - # Both the parsed task_id (`trigger_call_id`) and the segment - # should match the same task_id suffix. + # The child's single-segment path still encodes `node_name:task_id` + # (that's pregel's internal namespace format), but `cause` is now + # product-agnostic and must be populated by a stream transformer, + # not by pregel itself. assert ":" in child.path[0] node_name, _, task_id = child.path[0].partition(":") assert node_name == "sub" assert task_id # non-empty - assert child.trigger_call_id == task_id + assert child.cause is None class TestSubgraphInterrupt: