use cause in lifecycle

This commit is contained in:
Christian Bromann
2026-04-23 16:13:49 -07:00
parent 1bf0f1b7f8
commit f30a11055b
3 changed files with 31 additions and 39 deletions
+4 -21
View File
@@ -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)
@@ -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)
@@ -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: