From febf241b3001b1e4bf2259ad9fb527232327db3f Mon Sep 17 00:00:00 2001 From: Eugene Yurtsev Date: Fri, 7 Nov 2025 11:50:47 -0500 Subject: [PATCH] x --- libs/langgraph/langgraph/pregel/_retry.py | 18 +++--- libs/langgraph/tests/test_pregel.py | 72 +++++++++++++++++++++++ 2 files changed, 81 insertions(+), 9 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/_retry.py b/libs/langgraph/langgraph/pregel/_retry.py index 554a90a59..b2bf08b08 100644 --- a/libs/langgraph/langgraph/pregel/_retry.py +++ b/libs/langgraph/langgraph/pregel/_retry.py @@ -9,7 +9,7 @@ from collections.abc import Awaitable, Callable, Sequence from dataclasses import replace from typing import Any -from langgraph._internal._config import patch_configurable +from langgraph._internal._config import patch_configurable, recast_checkpoint_ns from langgraph._internal._constants import ( CONF, CONFIG_KEY_CHECKPOINT_NS, @@ -43,16 +43,16 @@ def run_with_retry( except ParentCommand as exc: ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS] cmd = exc.args[0] - if cmd.graph in (ns, task.name): + if cmd.graph in (recast_checkpoint_ns(ns), task.name): # this command is for the current graph, handle it for w in task.writers: w.invoke(cmd, config) break elif cmd.graph == Command.PARENT: # this command is for the parent graph, assign it to the parent - parts = ns.split(NS_SEP) - if parts[-1].isdigit(): - parts.pop() + # normalize namespace by removing task IDs + recast_ns = recast_checkpoint_ns(ns) + parts = recast_ns.split(NS_SEP) parent_ns = NS_SEP.join(parts[:-1]) exc.args = (replace(cmd, graph=parent_ns),) # bubble up @@ -138,16 +138,16 @@ async def arun_with_retry( except ParentCommand as exc: ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS] cmd = exc.args[0] - if cmd.graph in (ns, task.name): + if cmd.graph in (recast_checkpoint_ns(ns), task.name): # this command is for the current graph, handle it for w in task.writers: w.invoke(cmd, config) break elif cmd.graph == Command.PARENT: # this command is for the parent graph, assign it to the parent - parts = ns.split(NS_SEP) - if parts[-1].isdigit(): - parts.pop() + # normalize namespace by removing task IDs + recast_ns = recast_checkpoint_ns(ns) + parts = recast_ns.split(NS_SEP) parent_ns = NS_SEP.join(parts[:-1]) exc.args = (replace(cmd, graph=parent_ns),) # bubble up diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 73e403f81..d5ae7b861 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -7911,6 +7911,78 @@ def test_parent_command_goto( } +@pytest.mark.parametrize("subgraph_persist", [True, False]) +def test_parent_command_goto_deeply_nested( + sync_checkpointer: BaseCheckpointSaver, subgraph_persist: bool +) -> None: + """Test Command.PARENT with goto in deeply nested graphs (3+ levels). + + This tests the fix for issue #6409 where Command.PARENT with goto + would fail in graphs with 3 or more levels of nesting due to + namespace comparison issues. + """ + class State(TypedDict): + messages: Annotated[list[str], operator.add] + + # Level 3 (deepest): sub_sub_graph + def sub_sub_node(state): + """Returns Command.PARENT to jump to grandparent's node.""" + return Command( + graph=Command.PARENT, + goto="sub_node_3", + update={"messages": ["sub_sub_node"]}, + ) + + sub_sub_builder = StateGraph(State) + sub_sub_builder.add_node("sub_sub_node", sub_sub_node) + sub_sub_builder.add_edge(START, "sub_sub_node") + sub_sub_graph = sub_sub_builder.compile(checkpointer=subgraph_persist) + + # Level 2 (middle): sub_graph + def sub_node_1(state): + return {"messages": ["sub_node_1"]} + + def sub_node_3(state): + """Target node for Command.PARENT goto.""" + return {"messages": ["sub_node_3"]} + + sub_builder = StateGraph(State) + sub_builder.add_node("sub_node_1", sub_node_1) + sub_builder.add_node("sub_node_2", sub_sub_graph) + sub_builder.add_node("sub_node_3", sub_node_3) + sub_builder.add_edge(START, "sub_node_1") + sub_builder.add_edge("sub_node_1", "sub_node_2") + sub_graph = sub_builder.compile(checkpointer=subgraph_persist) + + # Level 1 (top): main_graph + def main_node_1(state): + return {"messages": ["main_node_1"]} + + main_builder = StateGraph(State) + main_builder.add_node("main_node_1", main_node_1) + main_builder.add_node("main_node_2", sub_graph) + main_builder.add_edge(START, "main_node_1") + main_builder.add_edge("main_node_1", "main_node_2") + main_graph = main_builder.compile(sync_checkpointer, name="main") + + config = {"configurable": {"thread_id": 1}} + + result = main_graph.invoke(input={"messages": ["start"]}, config=config) + + # Verify the execution order includes all expected nodes. + # Note: When subgraphs have persistent checkpointers, parent state + # is passed down, which may cause message duplication at subgraph boundaries. + # The key assertion is that: + # 1. All expected messages appear in order + # 2. sub_node_3 executed (proving Command.PARENT goto worked) + expected_messages = ["main_node_1", "sub_node_1", "sub_sub_node", "sub_node_3"] + assert all(msg in result["messages"] for msg in expected_messages), ( + f"Expected all messages {expected_messages} to be in result {result['messages']}" + ) + # Verify sub_node_3 executed last (the Command.PARENT goto target) + assert result["messages"][-1] == "sub_node_3" + + @pytest.mark.parametrize("with_timeout", [True, False]) def test_timeout_with_parent_command( sync_checkpointer: BaseCheckpointSaver, with_timeout: bool