diff --git a/libs/langgraph/langgraph/_internal/_config.py b/libs/langgraph/langgraph/_internal/_config.py index 8bbbacbf5..b79b24c26 100644 --- a/libs/langgraph/langgraph/_internal/_config.py +++ b/libs/langgraph/langgraph/_internal/_config.py @@ -20,6 +20,7 @@ from langchain_core.runnables.config import ( from langgraph.checkpoint.base import CheckpointMetadata from langgraph._internal._constants import ( + _CHECKPOINT_COORDINATE_KEYS, CONF, CONFIG_KEY_CHECKPOINT_ID, CONFIG_KEY_CHECKPOINT_MAP, @@ -342,6 +343,28 @@ def ensure_config(*configs: RunnableConfig | None) -> RunnableConfig: if _is_not_empty(v) }, ) + # An explicit config that supplies its own checkpoint coordinate (a + # thread_id, or any checkpoint_ns/checkpoint_id/checkpoint_map) is addressing + # its own checkpoint lineage, so drop the inherited ambient configurable + # rather than merging over it: a child graph invoked inside a parent node + # would otherwise write its checkpoints under the parent's namespace and + # never find them again. An explicit thread_id resets even when it equals the + # ambient one, since a child reusing the parent's thread id still addresses + # its own root namespace, not the parent task's. Configs that only refine + # other keys keep the ambient and shallow-merge over it below. + if empty.get(CONF): + for config in configs: + if config is None: + continue + explicit_configurable = config.get(CONF) + if not explicit_configurable: + continue + if any( + _is_not_empty(explicit_configurable.get(k)) + for k in _CHECKPOINT_COORDINATE_KEYS + ): + empty[CONF] = {} + break for config in configs: if config is None: continue diff --git a/libs/langgraph/langgraph/_internal/_constants.py b/libs/langgraph/langgraph/_internal/_constants.py index 360f7f275..58a4e7917 100644 --- a/libs/langgraph/langgraph/_internal/_constants.py +++ b/libs/langgraph/langgraph/_internal/_constants.py @@ -95,6 +95,15 @@ NULL_TASK_ID = sys.intern("00000000-0000-0000-0000-000000000000") OVERWRITE = sys.intern("__overwrite__") # dict key for the overwrite value, used as `{'__overwrite__': value}` +# Checkpoint coordinate keys: when any of these appear in an explicit +# configurable, the caller is addressing its own checkpoint lineage. +_CHECKPOINT_COORDINATE_KEYS = ( + CONFIG_KEY_THREAD_ID, + CONFIG_KEY_CHECKPOINT_NS, + CONFIG_KEY_CHECKPOINT_ID, + CONFIG_KEY_CHECKPOINT_MAP, +) + # redefined to avoid circular import with langgraph.constants _TAG_HIDDEN = sys.intern("langsmith:hidden") diff --git a/libs/langgraph/tests/test_subgraph_persistence.py b/libs/langgraph/tests/test_subgraph_persistence.py index 58dfcf596..c320a094e 100644 --- a/libs/langgraph/tests/test_subgraph_persistence.py +++ b/libs/langgraph/tests/test_subgraph_persistence.py @@ -639,3 +639,49 @@ def test_stateful_namespace_isolation( "broccoli round 2", "Veggie: broccoli round 2", ] + + +def test_child_with_own_thread_id_keeps_namespace( + sync_checkpointer: BaseCheckpointSaver, +) -> None: + """A child graph invoked from inside a parent node with its own thread_id + must store and read its checkpoint under its own namespace, not inherit the + parent task's checkpoint_ns. + """ + + class ChildState(TypedDict): + count: int + + def child_node(state: ChildState) -> dict: + return {"count": (state.get("count") or 0) + 1} + + child = ( + StateGraph(ChildState) + .add_node("n", child_node) + .add_edge(START, "n") + .compile(checkpointer=sync_checkpointer) + ) + + child_thread = str(uuid4()) + child_config = {"configurable": {"thread_id": child_thread}} + + def parent_node(state: ParentState) -> dict: + child.invoke({}, config=child_config) + return {"result": "ok"} + + parent = ( + StateGraph(ParentState) + .add_node("p", parent_node) + .add_edge(START, "p") + .compile(checkpointer=sync_checkpointer) + ) + parent_config = {"configurable": {"thread_id": str(uuid4())}} + + parent.invoke({"result": ""}, config=parent_config) + state1 = child.get_state(child_config) + assert state1.values.get("count") == 1 + assert state1.config["configurable"]["checkpoint_ns"] == "" + + parent.invoke({"result": ""}, config=parent_config) + state2 = child.get_state(child_config) + assert state2.values.get("count") == 2 diff --git a/libs/langgraph/tests/test_subgraph_persistence_async.py b/libs/langgraph/tests/test_subgraph_persistence_async.py index 759df5549..ba4ca1a91 100644 --- a/libs/langgraph/tests/test_subgraph_persistence_async.py +++ b/libs/langgraph/tests/test_subgraph_persistence_async.py @@ -660,3 +660,50 @@ async def test_stateful_namespace_isolation_async( "broccoli round 2", "Veggie: broccoli round 2", ] + + +@NEEDS_CONTEXTVARS +async def test_child_with_own_thread_id_keeps_namespace_async( + async_checkpointer: BaseCheckpointSaver, +) -> None: + """A child graph invoked from inside a parent node with its own thread_id + must store and read its checkpoint under its own namespace, not inherit the + parent task's checkpoint_ns. + """ + + class ChildState(TypedDict): + count: int + + def child_node(state: ChildState) -> dict: + return {"count": (state.get("count") or 0) + 1} + + child = ( + StateGraph(ChildState) + .add_node("n", child_node) + .add_edge(START, "n") + .compile(checkpointer=async_checkpointer) + ) + + child_thread = str(uuid4()) + child_config = {"configurable": {"thread_id": child_thread}} + + async def parent_node(state: ParentState) -> dict: + await child.ainvoke({}, config=child_config) + return {"result": "ok"} + + parent = ( + StateGraph(ParentState) + .add_node("p", parent_node) + .add_edge(START, "p") + .compile(checkpointer=async_checkpointer) + ) + parent_config = {"configurable": {"thread_id": str(uuid4())}} + + await parent.ainvoke({"result": ""}, config=parent_config) + state1 = await child.aget_state(child_config) + assert state1.values.get("count") == 1 + assert state1.config["configurable"]["checkpoint_ns"] == "" + + await parent.ainvoke({"result": ""}, config=parent_config) + state2 = await child.aget_state(child_config) + assert state2.values.get("count") == 2 diff --git a/libs/langgraph/tests/test_utils.py b/libs/langgraph/tests/test_utils.py index d962727b8..523528561 100644 --- a/libs/langgraph/tests/test_utils.py +++ b/libs/langgraph/tests/test_utils.py @@ -506,6 +506,95 @@ def test_ensure_config_configurable_later_wins_per_key() -> None: assert merged["configurable"]["only_b"] == "B" +def test_ensure_config_explicit_configurable_replaces_ambient() -> None: + # An explicit checkpoint coordinate (here a new thread_id) starts a fresh + # lineage and drops the ambient run context (e.g. a parent task's + # checkpoint_ns), so a child graph does not inherit it. + from langchain_core.runnables.config import var_child_runnable_config + + token = var_child_runnable_config.set( + {"configurable": {"checkpoint_ns": "p:parent-task", "checkpoint_id": "cid"}} + ) + try: + merged = ensure_config({"configurable": {"thread_id": "child"}}) + finally: + var_child_runnable_config.reset(token) + assert merged["configurable"]["thread_id"] == "child" + assert "checkpoint_ns" not in merged["configurable"] + assert "checkpoint_id" not in merged["configurable"] + + +def test_ensure_config_ambient_inherited_when_no_explicit_configurable() -> None: + # With no explicit configurable, the ambient run context is inherited + # unchanged (stateless subgraph / interrupt-resume pattern). + from langchain_core.runnables.config import var_child_runnable_config + + token = var_child_runnable_config.set( + {"configurable": {"checkpoint_ns": "p:parent-task"}} + ) + try: + merged = ensure_config({"tags": ["t"]}) + finally: + var_child_runnable_config.reset(token) + assert merged["configurable"]["checkpoint_ns"] == "p:parent-task" + + +def test_ensure_config_explicit_configurables_still_merge_over_ambient() -> None: + # A new thread_id drops the ambient, but explicit configs still shallow-merge + # among themselves, so a with_config(...) value (ls_agent_type) survives + # alongside an invoke-time thread_id. + from langchain_core.runnables.config import var_child_runnable_config + + token = var_child_runnable_config.set( + {"configurable": {"checkpoint_ns": "p:parent-task"}} + ) + try: + merged = ensure_config( + {"configurable": {"ls_agent_type": "root"}}, + {"configurable": {"thread_id": "child"}}, + ) + finally: + var_child_runnable_config.reset(token) + assert merged["configurable"]["ls_agent_type"] == "root" + assert merged["configurable"]["thread_id"] == "child" + assert "checkpoint_ns" not in merged["configurable"] + + +def test_ensure_config_non_coordinate_config_keeps_ambient_checkpoint_ns() -> None: + # A nested subagent is invoked with a non-coordinate configurable key + # (ls_agent_type) and no thread_id; it must keep the inherited checkpoint_ns + # so it stays a discoverable child of the parent run (deepagents `task` tool). + from langchain_core.runnables.config import var_child_runnable_config + + token = var_child_runnable_config.set( + {"configurable": {"thread_id": "parent", "checkpoint_ns": "p:parent-task"}} + ) + try: + merged = ensure_config({"configurable": {"ls_agent_type": "subagent"}}) + finally: + var_child_runnable_config.reset(token) + assert merged["configurable"]["ls_agent_type"] == "subagent" + assert merged["configurable"]["checkpoint_ns"] == "p:parent-task" + assert merged["configurable"]["thread_id"] == "parent" + + +def test_ensure_config_same_thread_id_still_clears_ambient() -> None: + # A child that reuses the parent's thread_id is still addressing its own root + # namespace on that thread, so the parent task's checkpoint_ns must not leak + # in; otherwise the child writes state that get_state cannot read back. + from langchain_core.runnables.config import var_child_runnable_config + + token = var_child_runnable_config.set( + {"configurable": {"thread_id": "shared", "checkpoint_ns": "p:parent-task"}} + ) + try: + merged = ensure_config({"configurable": {"thread_id": "shared"}}) + finally: + var_child_runnable_config.reset(token) + assert merged["configurable"]["thread_id"] == "shared" + assert "checkpoint_ns" not in merged["configurable"] + + def test_ensure_config_merges_metadata_across_configs() -> None: a = {"metadata": {"user_id": "U1"}} b = {"metadata": {"correlation_id": "C1"}}