diff --git a/libs/langgraph/langgraph/stream/transformers.py b/libs/langgraph/langgraph/stream/transformers.py index d08a47c1e..29e247b5e 100644 --- a/libs/langgraph/langgraph/stream/transformers.py +++ b/libs/langgraph/langgraph/stream/transformers.py @@ -419,6 +419,11 @@ class _TasksLifecycleBase(StreamTransformer): # `node:` shares this task_id, so a subagent recovers the tool # call that spawned it (cross-payload). self._pending_tool_calls: dict[str, str] = {} + # `cause` for the current `_on_started` dispatch, set immediately before + # the call and read by overrides that surface it. Keeping `cause` off the + # `_on_started` signature means overrides predating it (e.g. deepagents' + # `SubagentTransformer`) don't break. + self._pending_cause: LifecycleCause | None = None # --- Template-method hooks (subclass overrides) --- @@ -431,10 +436,11 @@ class _TasksLifecycleBase(StreamTransformer): ns: tuple[str, ...], graph_name: str | None, trigger_call_id: str | None, - *, - cause: LifecycleCause | None = None, ) -> None: - """Fired once per discovered namespace (first observed task event).""" + """Fired once per discovered namespace (first observed task event). + + The triggering `cause` (if any) is available as `self._pending_cause`. + """ raise NotImplementedError def _on_terminal( @@ -532,7 +538,10 @@ class _TasksLifecycleBase(StreamTransformer): tool_call_id = self._pending_tool_calls.get(trigger_call_id) if tool_call_id: cause = {"type": "toolCall", "tool_call_id": str(tool_call_id)} - self._on_started(ns, graph_name, trigger_call_id, cause=cause) + # Deliver `cause` via instance state, not the call signature, so + # `_on_started` stays backward-compatible with overrides predating it. + self._pending_cause = cause + self._on_started(ns, graph_name, trigger_call_id) if trigger_call_id is not None: self._open[ns] = trigger_call_id @@ -631,8 +640,6 @@ class LifecycleTransformer(_TasksLifecycleBase): ns: tuple[str, ...], graph_name: str | None, trigger_call_id: str | None, - *, - cause: LifecycleCause | None = None, ) -> None: if trigger_call_id is None: # Without a task id we can't correlate a parent-result @@ -643,6 +650,7 @@ class LifecycleTransformer(_TasksLifecycleBase): if graph_name: payload["graph_name"] = graph_name payload["trigger_call_id"] = trigger_call_id + cause = self._pending_cause if cause is not None: payload["cause"] = cause self._channel.push(payload) @@ -707,8 +715,6 @@ class SubgraphTransformer(_TasksLifecycleBase): ns: tuple[str, ...], graph_name: str | None, trigger_call_id: str | None, - *, - cause: LifecycleCause | None = None, ) -> None: if self._mux is None: return @@ -717,10 +723,9 @@ class SubgraphTransformer(_TasksLifecycleBase): except RuntimeError: return handle_cls = AsyncSubgraphRunStream if child_mux.is_async else SubgraphRunStream - # `cause` is intentionally ignored here: it is a wire/lifecycle-channel - # concern (carried on `LifecyclePayload`), not something the in-process - # subgraph navigation handle exposes. The argument is accepted only to - # keep the `_on_started` template signature uniform across transformers. + # The triggering `cause` (on `self._pending_cause`) is a wire/lifecycle + # concern carried on `LifecyclePayload`; the in-process subgraph + # navigation handle does not expose it. handle = handle_cls( mux=child_mux, path=ns, diff --git a/libs/langgraph/tests/test_stream_lifecycle_transformer.py b/libs/langgraph/tests/test_stream_lifecycle_transformer.py index 74bbc5560..893b0b3b7 100644 --- a/libs/langgraph/tests/test_stream_lifecycle_transformer.py +++ b/libs/langgraph/tests/test_stream_lifecycle_transformer.py @@ -25,6 +25,7 @@ from langgraph.stream._mux import StreamMux from langgraph.stream.transformers import ( LifecyclePayload, LifecycleTransformer, + _TasksLifecycleBase, ) TS = int(time.time() * 1000) @@ -678,3 +679,34 @@ def test_lifecycle_subagent_terminal_roundtrip() -> None: started, _completed = subagent assert started["graph_name"] == "weather_agent" assert started["cause"] == {"type": "toolCall", "tool_call_id": "call_w"} + + +def test_on_started_override_without_cause_is_backward_compatible() -> None: + """An `_on_started` override with the original 3-arg signature must work. + + `cause` is delivered via `self._pending_cause`, not the call signature, so + older/third-party subclasses (e.g. deepagents' `SubagentTransformer`) that + override `_on_started(self, ns, graph_name, trigger_call_id)` keep working — + no `TypeError: _on_started() got an unexpected keyword argument 'cause'`. + """ + seen: list[tuple] = [] + + class _LegacyTransformer(_TasksLifecycleBase): + def init(self) -> dict[str, Any]: + return {} + + def _should_track(self, ns: tuple[str, ...]) -> bool: + return len(ns) == 1 # direct child of the root scope + + # Deliberately omits `cause` — mirrors a pre-`cause` override. + def _on_started(self, ns, graph_name, trigger_call_id) -> None: # type: ignore[override] + seen.append((ns, graph_name, trigger_call_id)) + + def _on_terminal(self, ns, status, error) -> None: + pass + + transformer = _LegacyTransformer(scope=()) + # Must not raise: the base calls `_on_started` without a `cause` kwarg. + transformer.process(_tasks_start(["agent:abc123"], task_id="abc123", name="agent")) + + assert seen == [(("agent:abc123",), "agent", "abc123")]