mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-26 17:42:24 +02:00
Eugene YurtsevandGitHub
f178eb821e
Fixes incorrect `Command.PARENT` bubbling when checkpoint namespaces include numeric task-disambiguation segments like `|1`. In some nested-invoke/fanout scenarios, the runtime inserts a purely-numeric namespace segment between `name:task_id` segments (e.g. `parent_first:<id>|1|node:<id>`). The previous ParentCommand rewrite logic only handled numeric segments at the end of the namespace, which could produce a malformed parent graph identifier (e.g. `parent_first:<id>|1`) and prevent the command from routing to the intended parent node. This change normalizes checkpoint namespaces by dropping numeric segments before computing the parent namespace in both sync and async retry paths. Added a minimal regression test that exercises the nested-invoke case and asserts that `Command(graph=Command.PARENT, goto=...)` reliably routes to the parent graph, regardless of whether the jump comes from the first or second nested invocation.
54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from typing_extensions import TypedDict
|
|
|
|
from langgraph.graph import END, START, StateGraph
|
|
from langgraph.types import Command
|
|
|
|
|
|
def test_parent_command_from_nested_subgraph() -> None:
|
|
class ParentState(TypedDict):
|
|
jump_from_idx: int
|
|
|
|
class ChildState(TypedDict):
|
|
jump: bool
|
|
|
|
child_builder: StateGraph[ChildState] = StateGraph(ChildState)
|
|
|
|
def child_node(state: ChildState) -> Command | ChildState:
|
|
if state["jump"]:
|
|
return Command(graph=Command.PARENT, goto="parent_second")
|
|
return state
|
|
|
|
child_builder.add_node("node", child_node)
|
|
child_builder.add_edge(START, "node")
|
|
|
|
child_0 = child_builder.compile()
|
|
child_1 = child_builder.compile()
|
|
|
|
parent_builder: StateGraph[ParentState] = StateGraph(ParentState)
|
|
|
|
def parent_first(state: ParentState) -> ParentState:
|
|
child_0.invoke({"jump": state["jump_from_idx"] == 1})
|
|
if state["jump_from_idx"] == 1:
|
|
raise AssertionError("Shouldn't be here")
|
|
|
|
child_1.invoke({"jump": state["jump_from_idx"] == 2})
|
|
if state["jump_from_idx"] == 2:
|
|
raise AssertionError("Shouldn't be here")
|
|
|
|
return state
|
|
|
|
def parent_second(state: ParentState) -> ParentState:
|
|
return state
|
|
|
|
parent_builder.add_node("parent_first", parent_first)
|
|
parent_builder.add_node("parent_second", parent_second)
|
|
parent_builder.add_edge(START, "parent_first")
|
|
parent_builder.add_edge("parent_second", END)
|
|
|
|
graph = parent_builder.compile()
|
|
|
|
assert graph.invoke({"jump_from_idx": 1}) == {"jump_from_idx": 1}
|
|
assert graph.invoke({"jump_from_idx": 2}) == {"jump_from_idx": 2}
|