From 8e2f23e0519c8db5e3980d2dc8b657d577f97abc Mon Sep 17 00:00:00 2001 From: Christian Bromann Date: Mon, 27 Apr 2026 08:02:59 -0700 Subject: [PATCH] move more over --- libs/langgraph/langgraph/pregel/main.py | 22 +- libs/langgraph/langgraph/stream/_convert.py | 41 ++- libs/langgraph/langgraph/stream/_mux.py | 16 +- libs/langgraph/langgraph/stream/_types.py | 4 +- .../langgraph/stream/transformers.py | 267 +++++++++++++++++- libs/langgraph/pyproject.toml | 2 +- libs/langgraph/tests/test_pregel_stream_v2.py | 24 ++ .../tests/test_stream_messages_transformer.py | 76 +++++ .../tests/test_stream_subgraph_transformer.py | 169 ++++++++++- libs/langgraph/uv.lock | 21 +- 10 files changed, 618 insertions(+), 24 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/main.py b/libs/langgraph/langgraph/pregel/main.py index 8891ddd76..1e621302c 100644 --- a/libs/langgraph/langgraph/pregel/main.py +++ b/libs/langgraph/langgraph/pregel/main.py @@ -375,11 +375,13 @@ def _build_stream_factories( from langgraph.stream.transformers import ( MessagesTransformer, SubgraphTransformer, + ToolLifecycleTransformer, ValuesTransformer, ) builtins: list[Callable[..., Any]] = [ ValuesTransformer, + ToolLifecycleTransformer, MessagesTransformer, SubgraphTransformer, ] @@ -3361,6 +3363,9 @@ class Pregel( interrupt_before: All | Sequence[str] | None = None, interrupt_after: All | Sequence[str] | None = None, transformers: Sequence[Any] | None = None, + stream_modes: Sequence[StreamMode] | None = None, + output_keys: str | Sequence[str] | None = None, + **kwargs: Any, ) -> Any: """Start a sync v2 streaming run driven by transformer projections. @@ -3387,16 +3392,19 @@ class Pregel( factories = _build_stream_factories(self._stream_transformers, transformers) mux = StreamMux(factories=factories, is_async=False) - stream_modes = _collect_stream_modes(mux) + requested_stream_modes = set(_collect_stream_modes(mux)) + requested_stream_modes.update(stream_modes or ()) graph_iter = iter( self.stream( input, _merge_v2_messages_flag(config), - stream_mode=stream_modes, + stream_mode=list(requested_stream_modes), subgraphs=True, version="v2", + output_keys=output_keys, interrupt_before=interrupt_before, interrupt_after=interrupt_after, + **kwargs, ) ) return GraphRunStream(graph_iter, mux) @@ -3409,6 +3417,9 @@ class Pregel( interrupt_before: All | Sequence[str] | None = None, interrupt_after: All | Sequence[str] | None = None, transformers: Sequence[Any] | None = None, + stream_modes: Sequence[StreamMode] | None = None, + output_keys: str | Sequence[str] | None = None, + **kwargs: Any, ) -> Any: """Async counterpart to `stream_v2`. @@ -3429,15 +3440,18 @@ class Pregel( factories = _build_stream_factories(self._stream_transformers, transformers) mux = StreamMux(factories=factories, is_async=True) - stream_modes = _collect_stream_modes(mux) + requested_stream_modes = set(_collect_stream_modes(mux)) + requested_stream_modes.update(stream_modes or ()) graph_aiter = self.astream( input, _merge_v2_messages_flag(config), - stream_mode=stream_modes, + stream_mode=list(requested_stream_modes), subgraphs=True, version="v2", + output_keys=output_keys, interrupt_before=interrupt_before, interrupt_after=interrupt_after, + **kwargs, ).__aiter__() return AsyncGraphRunStream(graph_aiter, mux) diff --git a/libs/langgraph/langgraph/stream/_convert.py b/libs/langgraph/langgraph/stream/_convert.py index c1a14b5b0..07f33c87c 100644 --- a/libs/langgraph/langgraph/stream/_convert.py +++ b/libs/langgraph/langgraph/stream/_convert.py @@ -7,6 +7,31 @@ from langgraph.stream._types import ProtocolEvent, _ProtocolEventParams from langgraph.types import StreamPart +def _is_v2_messages_payload(data: Any) -> bool: + return isinstance(data, dict) and isinstance(data.get("event"), str) + + +def _normalize_messages_data(data: dict[str, Any]) -> dict[str, Any]: + """Normalize Python Core message fields to the protocol wire shape.""" + normalized = {**data} + if ( + normalized["event"] == "message-start" + and "id" not in normalized + and isinstance(normalized.get("message_id"), str) + ): + normalized["id"] = normalized["message_id"] + if ( + normalized["event"] + in ("content-block-start", "content-block-delta", "content-block-finish") + and "content" not in normalized + and isinstance(normalized.get("content_block"), dict) + ): + normalized["content"] = normalized["content_block"] + normalized.pop("message_id", None) + normalized.pop("content_block", None) + return normalized + + def convert_to_protocol_event(part: StreamPart) -> ProtocolEvent: """Convert a v2 StreamPart to a ProtocolEvent. @@ -18,11 +43,25 @@ def convert_to_protocol_event(part: StreamPart) -> ProtocolEvent: The equivalent ProtocolEvent. """ part_dict = cast(dict[str, Any], part) + data = part_dict["data"] params: _ProtocolEventParams = { "namespace": list(part_dict["ns"]), "timestamp": int(time.time() * 1000), - "data": part_dict["data"], + "data": data, } + if ( + part_dict["type"] == "messages" + and isinstance(data, tuple) + and len(data) == 2 + and _is_v2_messages_payload(data[0]) + and isinstance(data[1], dict) + ): + payload, metadata = data + params["data"] = _normalize_messages_data(payload) + if isinstance(metadata.get("langgraph_node"), str): + params["node"] = metadata["langgraph_node"] + if isinstance(metadata.get("run_id"), str): + params["run_id"] = metadata["run_id"] if "interrupts" in part_dict: params["interrupts"] = part_dict["interrupts"] return { diff --git a/libs/langgraph/langgraph/stream/_mux.py b/libs/langgraph/langgraph/stream/_mux.py index ea9196942..4639ef6ac 100644 --- a/libs/langgraph/langgraph/stream/_mux.py +++ b/libs/langgraph/langgraph/stream/_mux.py @@ -227,6 +227,18 @@ class StreamMux: """Return the transformer that owns the projection at `key`, if any.""" return self._transformer_by_key.get(key) + def emit(self, event: ProtocolEvent) -> None: + """Append a protocol event directly to the main log. + + Built-in transformers use this for protocol repair events that + must appear before the source event they are processing. Direct + emission intentionally bypasses the transformer pipeline, but + still lets this mux remain the only local sequencing authority. + """ + self._seq += 1 + event["seq"] = self._seq + self._events.push(event) + def push(self, event: ProtocolEvent) -> None: """Route an event through all transformers, then append to the main log. @@ -473,10 +485,8 @@ class StreamMux: visible in the main event log but are not passed through transformers' `process()` methods. """ - self._seq += 1 event: ProtocolEvent = { "type": "event", - "seq": self._seq, "method": f"custom:{channel_name}", "params": { "namespace": [], @@ -484,4 +494,4 @@ class StreamMux: "data": item, }, } - self._events.push(event) + self.emit(event) diff --git a/libs/langgraph/langgraph/stream/_types.py b/libs/langgraph/langgraph/stream/_types.py index 1a6c5ce22..56612cad4 100644 --- a/libs/langgraph/langgraph/stream/_types.py +++ b/libs/langgraph/langgraph/stream/_types.py @@ -22,6 +22,8 @@ class _ProtocolEventParams(TypedDict): namespace: list[str] timestamp: int data: Any + node: NotRequired[str] + run_id: NotRequired[str] interrupts: NotRequired[tuple[Any, ...]] @@ -35,7 +37,7 @@ class ProtocolEvent(TypedDict): """ type: Literal["event"] - eventId: NotRequired[str] + event_id: NotRequired[str] seq: NotRequired[int] method: str # StreamMode value: "values", "messages", "custom", etc. params: _ProtocolEventParams diff --git a/libs/langgraph/langgraph/stream/transformers.py b/libs/langgraph/langgraph/stream/transformers.py index 112fd1776..db93a4714 100644 --- a/libs/langgraph/langgraph/stream/transformers.py +++ b/libs/langgraph/langgraph/stream/transformers.py @@ -36,6 +36,106 @@ _TERMINAL_STATUSES: frozenset[SubgraphStatus] = frozenset( ) +def _is_record(value: Any) -> bool: + return isinstance(value, dict) + + +def _to_chat_model_stream_event(event: MessagesData) -> MessagesData: + """Convert wire-shaped message fields to ChatModelStream's internal shape.""" + event_type = event.get("event") + converted: dict[str, Any] = dict(event) + if ( + event_type == "message-start" + and "message_id" not in converted + and isinstance(converted.get("id"), str) + ): + converted["message_id"] = converted["id"] + if ( + event_type in ("content-block-start", "content-block-delta", "content-block-finish") + and "content_block" not in converted + and isinstance(converted.get("content"), dict) + ): + converted["content_block"] = converted["content"] + return cast("MessagesData", converted) + + +def _message_event_id(event: MessagesData) -> str | None: + raw_id = event.get("id") or event.get("message_id") + return str(raw_id) if raw_id is not None else None + + +def _content_block_start_skeleton(content: Any) -> dict[str, Any] | None: + """Return a minimal content-block-start payload for a delta/finish block.""" + if not _is_record(content) or not isinstance(content.get("type"), str): + return None + + block_type = content["type"] + skeleton: dict[str, Any] = {"type": block_type} + if block_type == "text": + skeleton["text"] = "" + elif block_type == "reasoning": + skeleton["reasoning"] = "" + elif block_type in ("tool_call", "tool_call_chunk"): + skeleton["type"] = "tool_call_chunk" + if isinstance(content.get("id"), str): + skeleton["id"] = content["id"] + if isinstance(content.get("name"), str): + skeleton["name"] = content["name"] + skeleton["args"] = "" + elif block_type in ("server_tool_call", "server_tool_call_chunk"): + skeleton["type"] = "server_tool_call_chunk" + if isinstance(content.get("id"), str): + skeleton["id"] = content["id"] + if isinstance(content.get("name"), str): + skeleton["name"] = content["name"] + skeleton["args"] = "" + return skeleton + + +def _copy_event( + source: ProtocolEvent, + *, + method: str, + namespace: list[str], + data: Any, +) -> ProtocolEvent: + params = {**source["params"], "namespace": namespace, "data": data} + return {"type": "event", "method": method, "params": params} + + +def _message_repair_key(event: ProtocolEvent, run_id: str) -> str: + namespace_key = "\x1f".join(event["params"]["namespace"]) + return f"{namespace_key}\x1e{run_id}" + + +def _extract_tool_calls_from_values(data: Any) -> dict[str, dict[str, Any]]: + if not _is_record(data): + return {} + messages = data.get("messages") + if not isinstance(messages, list): + return {} + known: dict[str, dict[str, Any]] = {} + for message in messages: + if not _is_record(message): + continue + tool_calls = message.get("tool_calls") + if not isinstance(tool_calls, list): + continue + for tool_call in tool_calls: + if not _is_record(tool_call): + continue + tool_call_id = tool_call.get("id") + if not isinstance(tool_call_id, str): + continue + name = tool_call.get("name") + args = tool_call.get("args") + known[tool_call_id] = { + "tool_name": name if isinstance(name, str) else "", + "input": args if _is_record(args) else {}, + } + return known + + class ValuesTransformer(StreamTransformer): """Capture values events as a drainable stream of state snapshots. @@ -88,6 +188,86 @@ class ValuesTransformer(StreamTransformer): return True +class ToolLifecycleTransformer(StreamTransformer): + """Repair tool-start events needed for deterministic subagent discovery. + + Some subagent frameworks expose a tool-caused subgraph lifecycle before + a LangChain tool callback has emitted the matching `tool-started` + frame. Core can infer the missing start from the latest values snapshot + (`messages[*].tool_calls`) and emit it before the lifecycle event leaves + the mux, keeping remote clients from guessing from values snapshots. + """ + + scope_exact = False + required_stream_modes = ("values", "tools", "lifecycle") + + def __init__(self, scope: tuple[str, ...] = ()) -> None: + super().__init__(scope) + self._known_tool_calls: dict[str, dict[str, Any]] = {} + self._emitted_tool_starts: set[str] = set() + self._mux: StreamMux | None = None + + def init(self) -> dict[str, Any]: + return {} + + def _on_register(self, mux: StreamMux) -> None: + self._mux = mux + + def process(self, event: ProtocolEvent) -> bool: + method = event["method"] + data = event["params"]["data"] + if method == "values": + self._known_tool_calls.update(_extract_tool_calls_from_values(data)) + return True + if method == "tools" and _is_record(data): + if ( + data.get("event") == "tool-started" + and isinstance(data.get("tool_call_id"), str) + ): + tool_call_id = cast("str", data["tool_call_id"]) + if tool_call_id in self._emitted_tool_starts: + return False + self._emitted_tool_starts.add(tool_call_id) + return True + if method == "lifecycle": + self._emit_missing_tool_started(event) + return True + + def _emit_missing_tool_started(self, event: ProtocolEvent) -> None: + if self._mux is None: + return + data = event["params"]["data"] + if not _is_record(data) or data.get("event") != "started": + return + cause = data.get("cause") + if not _is_record(cause) or cause.get("type") != "toolCall": + return + tool_call_id = cause.get("tool_call_id") + if not isinstance(tool_call_id, str): + return + if tool_call_id in self._emitted_tool_starts: + return + known = self._known_tool_calls.get(tool_call_id) + if known is None: + return + + self._emitted_tool_starts.add(tool_call_id) + namespace = event["params"]["namespace"] + self._mux.emit( + _copy_event( + event, + method="tools", + namespace=namespace[:-1], + data={ + "event": "tool-started", + "tool_call_id": tool_call_id, + "tool_name": known["tool_name"], + "input": known["input"], + }, + ) + ) + + class MessagesTransformer(StreamTransformer): """Capture messages events as ChatModelStream objects. @@ -145,12 +325,17 @@ class MessagesTransformer(StreamTransformer): # Correlate protocol events back to a ChatModelStream by run_id # (attached to the event's metadata by StreamMessagesHandler). self._by_run: dict[str, ChatModelStream] = {} + self._started_blocks: dict[str, set[int]] = {} + self._mux: StreamMux | None = None self._pump_fn: Callable[[], bool] | None = None self._apump_fn: Callable[[], Awaitable[bool]] | None = None def init(self) -> dict[str, Any]: return {"messages": self._log} + def _on_register(self, mux: StreamMux) -> None: + self._mux = mux + def _bind_pump(self, fn: Callable[[], bool]) -> None: """Wire the sync pull callback. Called by GraphRunStream._wire_request_more.""" self._pump_fn = fn @@ -209,14 +394,30 @@ class MessagesTransformer(StreamTransformer): # to a subgraph and are routed by `SubgraphTransformer`. ns = tuple(params["namespace"]) depth = len(self.scope) - if len(ns) > depth + 1 or ns[:depth] != self.scope: + if ns[:depth] != self.scope: return True - payload, metadata = params["data"] - node: str | None = metadata.get("langgraph_node") - run_id = str(metadata.get("run_id", "")) if metadata else "" + raw_data = params["data"] + metadata: dict[str, Any] = {} + if isinstance(raw_data, tuple) and len(raw_data) == 2: + payload, raw_metadata = raw_data + metadata = raw_metadata if isinstance(raw_metadata, dict) else {} + else: + payload = raw_data + node = params.get("node") + if not isinstance(node, str): + node = metadata.get("langgraph_node") + if not isinstance(node, str): + node = None + raw_run_id = params.get("run_id", metadata.get("run_id")) + run_id = str(raw_run_id) if raw_run_id is not None else "" if isinstance(payload, dict) and "event" in payload: + self._repair_content_block_lifecycle( + event, cast("MessagesData", payload), run_id=run_id + ) + if len(ns) > depth + 1: + return True self._route_protocol_event( cast("MessagesData", payload), run_id=run_id, node=node ) @@ -237,23 +438,69 @@ class MessagesTransformer(StreamTransformer): run_id: str, node: str | None, ) -> None: + stream_event = _to_chat_model_stream_event(event) event_type = event.get("event") if event_type == "message-start": - message_id = event.get("message_id") + message_id = _message_event_id(event) stream = self._make_stream( namespace=list(self.scope), node=node, - message_id=str(message_id) if message_id is not None else None, + message_id=message_id, ) - self._by_run[run_id] = stream + self._by_run[run_id or message_id or ""] = stream self._log.push(stream) - stream.dispatch(event) + stream.dispatch(stream_event) elif run_id in self._by_run: stream = self._by_run[run_id] - stream.dispatch(event) + stream.dispatch(stream_event) if event_type == "message-finish": del self._by_run[run_id] + def _repair_content_block_lifecycle( + self, + source: ProtocolEvent, + event: MessagesData, + *, + run_id: str, + ) -> None: + if self._mux is None: + return + event_type = event.get("event") + key = _message_repair_key(source, run_id) + if event_type == "message-start": + self._started_blocks[key] = set() + return + if event_type == "content-block-start": + index = event.get("index") + if isinstance(index, int): + self._started_blocks.setdefault(key, set()).add(index) + return + if event_type in ("content-block-delta", "content-block-finish"): + index = event.get("index") + if not isinstance(index, int): + return + started = self._started_blocks.setdefault(key, set()) + if index in started: + return + skeleton = _content_block_start_skeleton(event.get("content")) + if skeleton is None: + return + started.add(index) + self._mux.emit( + _copy_event( + source, + method="messages", + namespace=list(source["params"]["namespace"]), + data={ + "event": "content-block-start", + "index": index, + "content": skeleton, + }, + ) + ) + elif event_type == "message-finish": + self._started_blocks.pop(key, None) + def _route_whole_message(self, message: BaseMessage, *, node: str | None) -> None: stream = self._make_stream( namespace=list(self.scope), @@ -267,12 +514,14 @@ class MessagesTransformer(StreamTransformer): def finalize(self) -> None: """Clear any routing state — streams close themselves via `message-finish`.""" self._by_run.clear() + self._started_blocks.clear() def fail(self, err: BaseException) -> None: """Propagate run error to any streams still open when the graph fails.""" for stream in list(self._by_run.values()): stream.fail(err) self._by_run.clear() + self._started_blocks.clear() class SubgraphRunStream(BaseRunStream): diff --git a/libs/langgraph/pyproject.toml b/libs/langgraph/pyproject.toml index 2573395fb..8d0298e83 100644 --- a/libs/langgraph/pyproject.toml +++ b/libs/langgraph/pyproject.toml @@ -24,7 +24,7 @@ classifiers = [ 'Programming Language :: Python :: 3.13', ] dependencies = [ - "langchain-core==1.3.0a2", + "langchain-core>=1.3.2", "langgraph-checkpoint>=2.1.0,<5.0.0", "langgraph-sdk>=0.3.0,<0.4.0", "langgraph-prebuilt>=1.0.9,<1.1.0", diff --git a/libs/langgraph/tests/test_pregel_stream_v2.py b/libs/langgraph/tests/test_pregel_stream_v2.py index 233c40498..61da4e3e4 100644 --- a/libs/langgraph/tests/test_pregel_stream_v2.py +++ b/libs/langgraph/tests/test_pregel_stream_v2.py @@ -804,6 +804,30 @@ class TestConvertToProtocolEvent: assert isinstance(event["params"]["namespace"], list) assert event["params"]["namespace"] == ["a", "b", "c"] + def test_messages_conversion_uses_wire_shape(self) -> None: + part = { + "type": "messages", + "ns": ("call_model:task-1",), + "data": ( + { + "event": "content-block-delta", + "index": 0, + "content_block": {"type": "text", "text": "hi"}, + }, + {"langgraph_node": "call_model", "run_id": "run-1"}, + ), + } + event = convert_to_protocol_event(part) + assert event["method"] == "messages" + assert event["params"]["namespace"] == ["call_model:task-1"] + assert event["params"]["node"] == "call_model" + assert event["params"]["run_id"] == "run-1" + assert event["params"]["data"] == { + "event": "content-block-delta", + "index": 0, + "content": {"type": "text", "text": "hi"}, + } + # --------------------------------------------------------------------------- # StreamMux unit tests diff --git a/libs/langgraph/tests/test_stream_messages_transformer.py b/libs/langgraph/tests/test_stream_messages_transformer.py index f7e47da96..78d0c9487 100644 --- a/libs/langgraph/tests/test_stream_messages_transformer.py +++ b/libs/langgraph/tests/test_stream_messages_transformer.py @@ -257,6 +257,82 @@ class TestProtocolEventRouting: (stream,) = [*log._items] assert stream.node == "my_llm" + def test_wire_shape_routes_to_chat_model_stream(self) -> None: + t, log = _make_sync_transformer() + for evt in ( + {"event": "message-start", "role": "ai", "id": "msg-1"}, + { + "event": "content-block-start", + "index": 0, + "content": {"type": "text", "text": ""}, + }, + { + "event": "content-block-delta", + "index": 0, + "content": {"type": "text", "text": "hello"}, + }, + { + "event": "content-block-finish", + "index": 0, + "content": {"type": "text", "text": "hello"}, + }, + {"event": "message-finish", "reason": "stop"}, + ): + t.process( + { + "type": "event", + "method": "messages", + "params": { + "namespace": [], + "timestamp": TS, + "data": evt, + "node": "llm", + "run_id": "run-1", + }, + } + ) + (stream,) = [*log._items] + assert stream.message_id == "msg-1" + assert str(stream.text) == "hello" + + def test_missing_content_block_start_is_synthesized_in_main_log(self) -> None: + mux = StreamMux([MessagesTransformer()], is_async=False) + events = iter(mux._events) + + for evt in ( + {"event": "message-start", "role": "ai", "id": "msg-1"}, + { + "event": "content-block-delta", + "index": 0, + "content": {"type": "text", "text": "hello"}, + }, + ): + mux.push( + { + "type": "event", + "method": "messages", + "params": { + "namespace": [], + "timestamp": TS, + "data": evt, + "run_id": "run-1", + }, + } + ) + mux.close() + + emitted = list(events) + assert [event["params"]["data"]["event"] for event in emitted] == [ + "message-start", + "content-block-start", + "content-block-delta", + ] + assert emitted[1]["params"]["data"]["content"] == { + "type": "text", + "text": "", + } + assert emitted[0]["seq"] < emitted[1]["seq"] < emitted[2]["seq"] + # --------------------------------------------------------------------------- # Non-streaming (whole AIMessage) fallback diff --git a/libs/langgraph/tests/test_stream_subgraph_transformer.py b/libs/langgraph/tests/test_stream_subgraph_transformer.py index ee6591e55..4baf5db0f 100644 --- a/libs/langgraph/tests/test_stream_subgraph_transformer.py +++ b/libs/langgraph/tests/test_stream_subgraph_transformer.py @@ -20,6 +20,7 @@ from langgraph.stream.transformers import ( MessagesTransformer, SubgraphRunStream, SubgraphTransformer, + ToolLifecycleTransformer, ValuesTransformer, ) from langgraph.types import interrupt @@ -75,7 +76,12 @@ def _subscribe(log: EventLog) -> None: # --------------------------------------------------------------------------- -_FACTORIES = [ValuesTransformer, MessagesTransformer, SubgraphTransformer] +_FACTORIES = [ + ValuesTransformer, + ToolLifecycleTransformer, + MessagesTransformer, + SubgraphTransformer, +] def _handle_values_items(handle: SubgraphRunStream) -> list: @@ -135,6 +141,167 @@ class TestSubgraphTransformerUnit: assert handle.cause == {"type": "toolCall", "tool_call_id": "call_abc"} assert handle.status == "started" + def test_tool_started_is_synthesized_before_tool_caused_lifecycle(self) -> None: + mux, transformer = self._mux() + events = iter(mux._events) + mux.push( + _values( + { + "messages": [ + { + "tool_calls": [ + { + "id": "call_abc", + "name": "task", + "args": {"subagent_type": "researcher"}, + } + ] + } + ] + }, + namespace=[], + ) + ) + mux.push( + _lifecycle( + "started", + namespace=["task:child"], + graph_name="child", + cause={"type": "toolCall", "tool_call_id": "call_abc"}, + ) + ) + + mux.close() + tool_started, lifecycle_started = list(events)[1:3] + assert tool_started["method"] == "tools" + assert tool_started["params"]["namespace"] == [] + assert tool_started["params"]["data"] == { + "event": "tool-started", + "tool_call_id": "call_abc", + "tool_name": "task", + "input": {"subagent_type": "researcher"}, + } + assert lifecycle_started["method"] == "lifecycle" + assert tool_started["seq"] < lifecycle_started["seq"] + assert self._handle(transformer).path == ("task:child",) + + def test_core_golden_trace_uses_js_wire_shape_and_ordering(self) -> None: + mux, _transformer = self._mux() + events = iter(mux._events) + mux.push( + _values( + { + "messages": [ + { + "tool_calls": [ + { + "id": "call_abc", + "name": "task", + "args": {"subagent_type": "researcher"}, + } + ] + } + ] + }, + namespace=[], + ) + ) + for data in ( + {"event": "message-start", "id": "msg-1", "role": "ai"}, + { + "event": "content-block-delta", + "index": 0, + "content": {"type": "text", "text": "hi"}, + }, + ): + mux.push( + { + "type": "event", + "method": "messages", + "params": { + "namespace": ["call_model:task-1"], + "timestamp": TS, + "data": data, + "run_id": "run-1", + }, + } + ) + mux.push( + _lifecycle( + "started", + namespace=["task:child"], + graph_name="child", + cause={"type": "toolCall", "tool_call_id": "call_abc"}, + ) + ) + mux.close() + + trace = [ + (event["method"], event["params"]["namespace"], event["params"]["data"]) + for event in events + ] + assert trace == [ + ( + "values", + [], + { + "messages": [ + { + "tool_calls": [ + { + "id": "call_abc", + "name": "task", + "args": {"subagent_type": "researcher"}, + } + ] + } + ] + }, + ), + ( + "messages", + ["call_model:task-1"], + {"event": "message-start", "id": "msg-1", "role": "ai"}, + ), + ( + "messages", + ["call_model:task-1"], + { + "event": "content-block-start", + "index": 0, + "content": {"type": "text", "text": ""}, + }, + ), + ( + "messages", + ["call_model:task-1"], + { + "event": "content-block-delta", + "index": 0, + "content": {"type": "text", "text": "hi"}, + }, + ), + ( + "tools", + [], + { + "event": "tool-started", + "tool_call_id": "call_abc", + "tool_name": "task", + "input": {"subagent_type": "researcher"}, + }, + ), + ( + "lifecycle", + ["task:child"], + { + "event": "started", + "graph_name": "child", + "cause": {"type": "toolCall", "tool_call_id": "call_abc"}, + }, + ), + ] + def test_status_transitions(self) -> None: mux, transformer = self._mux() mux.push(_lifecycle("started", namespace=["t:c"], graph_name="c")) diff --git a/libs/langgraph/uv.lock b/libs/langgraph/uv.lock index a18fc116b..a4c59bb51 100644 --- a/libs/langgraph/uv.lock +++ b/libs/langgraph/uv.lock @@ -1348,10 +1348,11 @@ wheels = [ [[package]] name = "langchain-core" -version = "1.3.0a2" +version = "1.3.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "jsonpatch" }, + { name = "langchain-protocol" }, { name = "langsmith" }, { name = "packaging" }, { name = "pydantic" }, @@ -1360,9 +1361,21 @@ dependencies = [ { name = "typing-extensions" }, { name = "uuid-utils" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/af/bc/0bff31fcaff174d86031cc713471a3e85ed4ec8e5cd95ad0217f2aced20e/langchain_core-1.3.0a2.tar.gz", hash = "sha256:52d978c84552b74b9a3f16c1fced84f9e27cc96d7a67c601925ce6cbc4ea3cf9", size = 854580, upload-time = "2026-04-13T14:37:55.745Z" } +sdist = { url = "https://files.pythonhosted.org/packages/a8/03/7219502e8ca728d65eb44d7a3eb60239230742a70dbfc9241b9bfd61c4ab/langchain_core-1.3.2.tar.gz", hash = "sha256:fd7a50b2f28ba561fd9d7f5d2760bc9e06cf00cdf820a3ccafe88a94ffa8d5b7", size = 911813, upload-time = "2026-04-24T15:49:23.699Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/0e/14/03c09686602567059f26af29de0c44546a83af2f2aa29925e61040e43ea2/langchain_core-1.3.0a2-py3-none-any.whl", hash = "sha256:9e929a34f0b0c6c1255e395a1de34f8626893ceb4cdae550a22a0bd18c87be54", size = 510233, upload-time = "2026-04-13T14:37:54.277Z" }, + { url = "https://files.pythonhosted.org/packages/7d/d5/8fa4431007cbb7cfed7590f4d6a5dea3ad724f4174d248f6642ef5ce7d05/langchain_core-1.3.2-py3-none-any.whl", hash = "sha256:d44a66127f9f8db735bdfd0ab9661bccb47a97113cfd3f2d89c74864422b7274", size = 542390, upload-time = "2026-04-24T15:49:21.991Z" }, +] + +[[package]] +name = "langchain-protocol" +version = "0.0.12" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/5c/51/1157009b6f94e6e58be58fa8b620187d657909a8b36a6bf5b0c52a2711f6/langchain_protocol-0.0.12.tar.gz", hash = "sha256:5e14c434290a705c9510fdb1a83ecf7561a5e6e0dfd053930ade80dba069269f", size = 6408, upload-time = "2026-04-25T01:05:01.489Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/95/82/3431e3061c917439589fa88a6b23c9bc0e154cba0f05d2e895a68c76ff74/langchain_protocol-0.0.12-py3-none-any.whl", hash = "sha256:402b61f42d4139692528cf37226c367bb6efc8ff8165b29380accb0abfece7b2", size = 6639, upload-time = "2026-04-25T01:05:00.487Z" }, ] [[package]] @@ -1439,7 +1452,7 @@ test = [ [package.metadata] requires-dist = [ - { name = "langchain-core", specifier = "==1.3.0a2" }, + { name = "langchain-core", specifier = ">=1.3.2" }, { name = "langgraph-checkpoint", editable = "../checkpoint" }, { name = "langgraph-prebuilt", editable = "../prebuilt" }, { name = "langgraph-sdk", editable = "../sdk-py" },