diff --git a/libs/sdk-py/integration/scripts/test_subgraphs.py b/libs/sdk-py/integration/scripts/test_subgraphs.py index 109a57c63..823bd9061 100644 --- a/libs/sdk-py/integration/scripts/test_subgraphs.py +++ b/libs/sdk-py/integration/scripts/test_subgraphs.py @@ -37,7 +37,7 @@ async def _drain_subgraphs(thread) -> list: # Just count handle paths; deep message iteration on scoped # handles has its own draining pattern and isn't the goal of # this test (which exercises subgraph discovery via - # child-namespace ``lifecycle: started``). + # child-namespace `lifecycle: started/running`). children.append({"path": child.path}) return children diff --git a/libs/sdk-py/langgraph_sdk/_async/stream.py b/libs/sdk-py/langgraph_sdk/_async/stream.py index 2ebc2d111..be8e4e9fe 100644 --- a/libs/sdk-py/langgraph_sdk/_async/stream.py +++ b/libs/sdk-py/langgraph_sdk/_async/stream.py @@ -537,12 +537,11 @@ def _is_direct_child(namespace: list[str], scope: tuple[str, ...]) -> bool: def _subgraph_subscription_params(scope: tuple[str, ...]) -> SubscribeParams: # Subscribe to tasks + messages + tools + lifecycle without a depth limit # so all descendant-namespace events are captured in one SSE and buffered - # into each child handle's inbox. ``lifecycle`` is included so child- - # namespace ``started`` events (the canonical signal for - # ``create_deep_agent``-style subagent discovery, matching JS behavior) - # reach ``_subgraphs_iter``; servers that surface child invocations via - # ``tasks`` events instead are also handled via the existing ``method == - # "tasks"`` branch. + # into each child handle's inbox. `lifecycle` is included so child + # started/running events (the canonical signal for `create_deep_agent`- + # style subagent discovery, matching JS behavior) reach `_subgraphs_iter`; + # servers that surface child invocations via `tasks` events instead are + # also handled via the existing `method == "tasks"` branch. return { "channels": ["messages", "tasks", "tools", "lifecycle"], "namespaces": [list(scope)], @@ -1991,7 +1990,9 @@ class AsyncThreadStream: params = event.get("params") or {} data = params.get("data") if isinstance(params, dict) else None phase = data.get("event") if isinstance(data, dict) else None - payload_namespace = data.get("namespace") if isinstance(data, dict) else None + payload_namespace = ( + data.get("namespace") if isinstance(data, dict) else None + ) if isinstance(payload_namespace, list) and payload_namespace: return if phase in ("started", "running"): diff --git a/libs/sdk-py/langgraph_sdk/_sync/stream.py b/libs/sdk-py/langgraph_sdk/_sync/stream.py index ff72fda5a..6a6e45284 100644 --- a/libs/sdk-py/langgraph_sdk/_sync/stream.py +++ b/libs/sdk-py/langgraph_sdk/_sync/stream.py @@ -114,9 +114,9 @@ def _is_direct_child(namespace: list[str], scope: tuple[str, ...]) -> bool: def _subgraph_subscription_params(scope: tuple[str, ...]) -> SubscribeParams: - # Includes ``lifecycle`` so child-namespace ``started`` events (the - # ``create_deep_agent`` subagent discovery signal, matching JS) - # reach ``_subgraphs_iter`` alongside ``tasks``-based discovery. + # Includes `lifecycle` so child started/running events (the + # `create_deep_agent` subagent discovery signal, matching JS) reach + # `_subgraphs_iter` alongside `tasks`-based discovery. return { "channels": ["messages", "tasks", "tools", "lifecycle"], "namespaces": [list(scope)], @@ -1624,7 +1624,9 @@ class SyncThreadStream: params = event.get("params") or {} data = params.get("data") if isinstance(params, dict) else None phase = data.get("event") if isinstance(data, dict) else None - payload_namespace = data.get("namespace") if isinstance(data, dict) else None + payload_namespace = ( + data.get("namespace") if isinstance(data, dict) else None + ) if isinstance(payload_namespace, list) and payload_namespace: return if phase in ("started", "running"): diff --git a/libs/sdk-py/langgraph_sdk/stream/decoders.py b/libs/sdk-py/langgraph_sdk/stream/decoders.py index af07884d3..cce9845bf 100644 --- a/libs/sdk-py/langgraph_sdk/stream/decoders.py +++ b/libs/sdk-py/langgraph_sdk/stream/decoders.py @@ -9,7 +9,7 @@ which drives multiple decoders from one shared subscription. from __future__ import annotations from collections.abc import Callable, Iterable, Mapping -from typing import Any, Literal, Protocol +from typing import Any, Literal, Protocol, cast #: Channel names the public ``interleave_projections`` API accepts as built-ins. SUPPORTED_INTERLEAVE_CHANNELS = ( @@ -61,7 +61,7 @@ def _lifecycle_payload_namespace( ) -> list[str]: namespace = data.get("namespace") if isinstance(namespace, list): - return list(namespace) + return cast(list[str], list(namespace)) return _event_namespace(params_field) @@ -310,7 +310,9 @@ class SubgraphsDecoder: yield from self._discover(namespace) elif method == "lifecycle": event_type = data.get("event") - if event_type == "started" and _is_direct_child(namespace, self._scope): + if event_type in ("started", "running") and _is_direct_child( + namespace, self._scope + ): yield from self._discover(namespace) elif event_type in ("completed", "failed", "interrupted"): self._apply_lifecycle_terminal(namespace, data) diff --git a/libs/sdk-py/tests/streaming/test_decoders.py b/libs/sdk-py/tests/streaming/test_decoders.py index 94f9c10c7..6dcb5d9ac 100644 --- a/libs/sdk-py/tests/streaming/test_decoders.py +++ b/libs/sdk-py/tests/streaming/test_decoders.py @@ -375,6 +375,24 @@ def test_subgraphs_decoder_discovers_on_forwarded_lifecycle_payload_namespace(): assert h.trigger_call_id == "call-1" +def test_subgraphs_decoder_discovers_on_forwarded_lifecycle_running_status(): + decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory) + [h] = list( + decoder.feed( + _forwarded_lifecycle_event( + seq=1, + event="running", + namespace=["child:call-1"], + graph_name="child", + trigger_call_id="call-1", + ) + ) + ) + assert h.path == ("child:call-1",) + assert h.graph_name == "child" + assert h.trigger_call_id == "call-1" + + def test_subgraphs_decoder_completes_on_forwarded_lifecycle_payload_namespace(): decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory) [h] = list( diff --git a/libs/sdk-py/tests/streaming/test_scoped_handles.py b/libs/sdk-py/tests/streaming/test_scoped_handles.py index a3ad52905..024b18b69 100644 --- a/libs/sdk-py/tests/streaming/test_scoped_handles.py +++ b/libs/sdk-py/tests/streaming/test_scoped_handles.py @@ -81,7 +81,7 @@ async def test_subgraphs_yields_handle_from_forwarded_lifecycle_namespace(): lifecycle_started_event(seq=0), _forwarded_lifecycle_event( seq=1, - event="started", + event="running", namespace=["worker:abc"], graph_name="worker", trigger_call_id="abc", diff --git a/libs/sdk-py/tests/streaming/test_sync_scoped_handles.py b/libs/sdk-py/tests/streaming/test_sync_scoped_handles.py index fdd6f3dbb..205fc8148 100644 --- a/libs/sdk-py/tests/streaming/test_sync_scoped_handles.py +++ b/libs/sdk-py/tests/streaming/test_sync_scoped_handles.py @@ -140,7 +140,7 @@ def test_sync_subgraphs_yields_handle_from_forwarded_lifecycle_namespace(): lifecycle_started_event(seq=0), _forwarded_lifecycle_event( seq=1, - event="started", + event="running", namespace=["worker:abc"], graph_name="worker", trigger_call_id="abc",