fix: nested subgraph inherits parent checkpoint_ns (regression in 1.2.3) (#8053)

Closes #8038

## Description

The `ensure_config` merge introduced in #7926 caused a child graph
invoked inside a parent node to inherit the parent task's
`checkpoint_ns` from the ambient run context
(`var_child_runnable_config`), so the child's checkpoints were written
under an unreadable namespace and re-ran from scratch each turn (#8038).
The first explicitly passed `configurable` that carries a checkpoint
coordinate (a `thread_id`, or any
`checkpoint_ns`/`checkpoint_id`/`checkpoint_map`) now replaces the
ambient one, while subsequent explicit configs still shallow-merge —
preserving #7926's `with_config(...)` semantics.

## Contract

`ensure_config` merges an explicit `configurable` over the ambient run
context (`var_child_runnable_config`) with one rule: **an explicit
`configurable` that supplies its own checkpoint coordinate addresses its
own checkpoint lineage, so the ambient `configurable` is dropped rather
than merged over.** Coordinate keys are `thread_id`, `checkpoint_ns`,
`checkpoint_id`, and `checkpoint_map` (grouped as
`_CHECKPOINT_COORDINATE_KEYS`). A non-coordinate `configurable` keeps
the ambient and shallow-merges over it.

Below, each case shows the parent/child graph wiring that triggers it
and the resulting namespacing. The child is always a compiled subgraph
invoked from inside a parent node.

### 1. Subgraph invoked with no new config → ambient inherited

```python
child = StateGraph(ChildState).add_node("n", child_node).add_edge(START, "n").compile(checkpointer=checkpointer)

def parent_node(state):
    child.invoke({}, config=None)          # no explicit configurable
    return {"result": "ok"}

parent = StateGraph(ParentState).add_node("p", parent_node).add_edge(START, "p").compile(checkpointer=checkpointer)
parent.invoke({"result": ""}, config={"configurable": {"thread_id": "parent"}})
```
Child inherits the parent task's `checkpoint_ns` (`p:<parent-task>`);
its checkpoints are written as a discoverable child of the parent run.
Pre-#7926 behavior, unchanged.

### 2. Subgraph invoked with a new thread_id → ambient dropped

```python
child = StateGraph(ChildState).add_node("n", child_node).add_edge(START, "n").compile(checkpointer=checkpointer)
child_config = {"configurable": {"thread_id": str(uuid4())}}

def parent_node(state):
    child.invoke({}, config=child_config)  # explicit new thread_id
    return {"result": "ok"}

parent = StateGraph(ParentState).add_node("p", parent_node).add_edge(START, "p").compile(checkpointer=checkpointer)
parent.invoke({"result": ""}, config={"configurable": {"thread_id": "parent"}})
```
Child starts its own lineage on `child_config`'s thread; `checkpoint_ns
== ""`; `child.get_state(child_config)` reads back state across repeated
parent turns. Fixes #8038.

### 3. Subgraph invoked with the same thread_id as parent → ambient
still dropped

```python
child = StateGraph(ChildState).add_node("n", child_node).add_edge(START, "n").compile(checkpointer=checkpointer)

def parent_node(state):
    # reuses the parent's thread_id as the child's own
    child.invoke({}, config={"configurable": {"thread_id": state["parent_thread"]}})
    return {"result": "ok"}

parent = StateGraph(ParentState).add_node("p", parent_node).add_edge(START, "p").compile(checkpointer=checkpointer)
parent_thread = str(uuid4())
parent.invoke({"parent_thread": parent_thread, "result": ""}, config={"configurable": {"thread_id": parent_thread}})
```
Child addresses its own root namespace on the shared thread;
`checkpoint_ns == ""`. The parent task's `checkpoint_ns` must not leak
in, or `child.get_state({"configurable": {"thread_id": parent_thread}})`
returns empty state and the child re-runs from scratch each turn.

### 4. Subagent invoked with a non-coordinate key only → ambient
inherited

```python
child = StateGraph(ChildState).add_node("n", child_node).add_edge(START, "n").compile(checkpointer=checkpointer)

def parent_node(state):
    # ls_agent_type is not a checkpoint coordinate, so ambient is kept
    child.invoke({}, config={"configurable": {"ls_agent_type": "subagent"}})
    return {"result": "ok"}

parent = StateGraph(ParentState).add_node("p", parent_node).add_edge(START, "p").compile(checkpointer=checkpointer)
parent.invoke({"result": ""}, config={"configurable": {"thread_id": "parent"}})
```
Child remains a discoverable child of the parent run; ambient
`thread_id` and `checkpoint_ns` preserved (deepagents `task` tool
pattern).

### 5. with_config(...) + invoke-time thread_id → ambient dropped, then
merged

```python
# child compiled with a non-coordinate configurable via with_config
child = (
    StateGraph(ChildState)
    .add_node("n", child_node)
    .add_edge(START, "n")
    .compile(checkpointer=checkpointer)
    .with_config({"configurable": {"ls_agent_type": "root"}})
)

def parent_node(state):
    # invoke-time config supplies the thread_id; with_config's ls_agent_type survives
    child.invoke({}, config={"configurable": {"thread_id": "child"}})
    return {"result": "ok"}

parent = StateGraph(ParentState).add_node("p", parent_node).add_edge(START, "p").compile(checkpointer=checkpointer)
parent.invoke({"result": ""}, config={"configurable": {"thread_id": "parent"}})
```
First coordinate-bearing config (`thread_id`) drops the ambient;
subsequent explicit configs still shallow-merge, so `ls_agent_type`
survives alongside `thread_id`. Preserves #7926 `with_config(...)`
semantics.

### Regression note

Cases 1, 4, and 5 are the pre-#7926 behavior and are preserved
unchanged. Cases 2 and 3 fix the regression introduced by #7926: an
explicit `thread_id` resets the ambient even when it equals the ambient
thread id, because a child reusing the parent's thread id still
addresses its own root namespace on that thread, not the parent task's.

## Self-Hosted Release Note
Fix regression where a nested subgraph with its own `thread_id` invoked
inside a parent node lost its persisted state across turns.

## Test Plan
- [x] `pytest tests/test_subgraph_persistence.py -k
test_child_with_own_thread_id_keeps_namespace` (case 2)
- [x] `pytest tests/test_utils.py -k
ensure_config_explicit_configurable_replaces_ambient` (case 2)
- [x] `pytest tests/test_utils.py -k
ensure_config_ambient_inherited_when_no_explicit_configurable` (case 1)
- [x] `pytest tests/test_utils.py -k
ensure_config_non_coordinate_config_keeps_ambient_checkpoint_ns` (case
4)
- [x] `pytest tests/test_utils.py -k
ensure_config_explicit_configurables_still_merge_over_ambient` (case 5)
- [x] `pytest tests/test_utils.py -k
ensure_config_same_thread_id_still_clears_ambient` (case 3)

Made by [Open SWE](https://openswe.vercel.app)

---------

Co-authored-by: Nick Hollon <274035459+nick-hollon-lc@users.noreply.github.com>
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

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

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")
 

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

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

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"}}