add delta overwrite graph coverage

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/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.29"
+__version__ = "0.4.30"

libs/cli/langgraph_cli/cli.py

@@ -159,7 +159,14 @@ OPT_POSTGRES_URI = click.option(
 OPT_API_VERSION = click.option(
     "--api-version",
     type=str,
-    help="API server version to use for the base image. If unspecified, the latest version will be used.",
+    help=(
+        "API server version to use for the base image. If unspecified, the "
+        "latest stable version will be used. Compatible ranges like "
+        "~=0.11.0.dev5 stay on 0.11.0.dev5 while only newer dev builds exist, "
+        "then resolve to the newest matching rc or stable release, for example "
+        "0.11.0rc1 or 0.11.0. Stable-floating ranges like >~=0.11.0.dev5 "
+        "can also pick up future stable releases, for example 0.12.0."
+    ),
 )
 
 OPT_ENGINE_RUNTIME_MODE = click.option(

libs/cli/langgraph_cli/config.py

@@ -9,6 +9,7 @@ from collections import Counter
 from typing import Literal, NamedTuple
 
 import click
+import httpx
 
 from langgraph_cli.schemas import Config, Distros
 from langgraph_cli.uv_lock import python_config_to_docker_uv_lock
@@ -41,6 +42,31 @@ _API_VERSION_PATTERN = re.compile(
     r"(?:\.(?P<patch>\d+))?"
     r"(?:(?:\.|)(?:[A-Za-z][0-9A-Za-z]*))?$"
 )
+_API_VERSION_RANGE_PATTERN = re.compile(r"^(?P<operator>~=|>~=)\s*(?P<version>.+)$")
+_API_VERSION_PART_PATTERN = re.compile(
+    r"^(?P<major>\d+)"
+    r"(?:\.(?P<minor>\d+))?"
+    r"(?:\.(?P<patch>\d+))?"
+    r"(?:(?:\.|)(?P<pre>dev|rc)(?P<pre_n>\d+))?$"
+)
+
+
+class _ParsedApiVersion(NamedTuple):
+    release: tuple[int, ...]
+    prerelease: str | None
+    prerelease_number: int
+
+
+class _ApiVersionRange(NamedTuple):
+    floor: _ParsedApiVersion
+    allow_future_stable: bool
+
+
+_PRERELEASE_ORDER = {
+    "dev": 0,
+    "rc": 1,
+    None: 2,
+}
 
 
 def has_disallowed_build_command_content(command: str) -> bool:
@@ -141,6 +167,133 @@ def _parse_api_version_parts(version_str: str) -> tuple[int, ...]:
     return tuple(int(part) for part in match.groups() if part is not None)
 
 
+def _parse_api_version(version_str: str) -> _ParsedApiVersion:
+    match = _API_VERSION_PART_PATTERN.fullmatch(version_str)
+    if not match:
+        raise ValueError("Version must be major or major.minor or major.minor.patch.")
+    release = tuple(
+        int(part)
+        for part in (
+            match.group("major"),
+            match.group("minor"),
+            match.group("patch"),
+        )
+        if part is not None
+    )
+    prerelease = match.group("pre")
+    prerelease_number = int(match.group("pre_n") or 0)
+    return _ParsedApiVersion(release, prerelease, prerelease_number)
+
+
+def _api_version_sort_key(
+    version: _ParsedApiVersion,
+) -> tuple[tuple[int, ...], int, int]:
+    return (
+        version.release,
+        _PRERELEASE_ORDER[version.prerelease],
+        version.prerelease_number,
+    )
+
+
+def _api_version_upper_bound(version: _ParsedApiVersion) -> tuple[int, ...]:
+    release = version.release
+    if len(release) <= 2:
+        return (release[0] + 1,)
+    return (release[0], release[1] + 1)
+
+
+def _is_compatible_api_version_candidate(
+    candidate: _ParsedApiVersion,
+    version_range: _ApiVersionRange,
+    upper_bound: tuple[int, ...],
+) -> bool:
+    floor = version_range.floor
+    if _api_version_sort_key(candidate) < _api_version_sort_key(floor):
+        return False
+    outside_compatible_range = candidate.release[: len(upper_bound)] >= upper_bound
+    if outside_compatible_range and not version_range.allow_future_stable:
+        return False
+    if outside_compatible_range and candidate.prerelease is not None:
+        return False
+    if floor.prerelease == "dev" and candidate.prerelease == "dev":
+        return candidate == floor
+    return True
+
+
+def _ensure_compatible_api_version_base_image(base_image: str) -> None:
+    if ":" in base_image:
+        raise click.UsageError(
+            "Compatible api_version ranges cannot be used with a tagged base_image."
+        )
+
+
+def _get_pypi_versions(package_name: str) -> list[str]:
+    try:
+        response = httpx.get(
+            f"https://pypi.org/pypi/{package_name}/json",
+            timeout=10,
+        )
+        response.raise_for_status()
+    except httpx.HTTPError as exc:
+        raise click.UsageError(
+            f"Failed to fetch PyPI versions for {package_name}: {exc}"
+        ) from None
+    payload = response.json()
+    releases = payload.get("releases", {})
+    if not isinstance(releases, dict):
+        raise click.UsageError(
+            f"Failed to fetch PyPI versions for {package_name}: invalid response."
+        )
+    return [version for version in releases if isinstance(version, str)]
+
+
+def _resolve_compatible_api_version(
+    api_version: str,
+    base_image: str,
+    version_distro_tag: str,
+) -> str:
+    match = _API_VERSION_RANGE_PATTERN.fullmatch(api_version)
+    if not match:
+        return api_version
+
+    floor_str = match.group("version").strip()
+    try:
+        floor = _parse_api_version(floor_str)
+    except ValueError:
+        raise click.UsageError(
+            f"Invalid compatible api_version range: {api_version}.\n\n"
+            "Use a compatible version range, e.g.:\n"
+            '  "api_version": "~=0.11.0.dev5"\n'
+            "or a stable-floating range, e.g.:\n"
+            '  "api_version": ">~=0.11.0.dev5"'
+        ) from None
+
+    version_range = _ApiVersionRange(
+        floor=floor,
+        allow_future_stable=match.group("operator") == ">~=",
+    )
+    _ensure_compatible_api_version_base_image(base_image)
+    pypi_versions = _get_pypi_versions("langgraph-api")
+
+    candidates: list[tuple[_ParsedApiVersion, str]] = []
+    upper_bound = _api_version_upper_bound(floor)
+    for candidate_str in pypi_versions:
+        try:
+            candidate = _parse_api_version(candidate_str)
+        except ValueError:
+            continue
+        if _is_compatible_api_version_candidate(candidate, version_range, upper_bound):
+            candidates.append((candidate, candidate_str))
+
+    if not candidates:
+        raise click.UsageError(
+            f"No PyPI releases match compatible api_version range {api_version!r} "
+            f"for {base_image} with {version_distro_tag!r}."
+        )
+
+    return max(candidates, key=lambda item: _api_version_sort_key(item[0]))[1]
+
+
 def _is_node_graph(spec: str | dict) -> bool:
     """Check if a graph is a Node.js graph based on the file extension."""
     if isinstance(spec, dict):
@@ -194,7 +347,12 @@ def validate_config(config: Config) -> Config:
             )
     if api_version:
         try:
-            parts = _parse_api_version_parts(api_version)
+            compatible_match = _API_VERSION_RANGE_PATTERN.fullmatch(api_version)
+            parts = _parse_api_version_parts(
+                compatible_match.group("version").strip()
+                if compatible_match
+                else api_version
+            )
             if len(parts) > 3:
                 raise ValueError(
                     "Version must be major or major.minor or major.minor.patch."
@@ -1430,6 +1588,9 @@ def docker_tag(
 
     # Prepend API version if provided
     if api_version:
+        api_version = _resolve_compatible_api_version(
+            api_version, base_image, f"{language}{version_distro_tag}"
+        )
         full_tag = f"{api_version}-{language}{version_distro_tag}"
     elif "/langgraph-server" in base_image and version_distro_tag not in base_image:
         return f"{base_image}-{language}{version_distro_tag}"

libs/cli/pyproject.toml

@@ -23,7 +23,7 @@ dependencies = [
 path = "langgraph_cli/__init__.py"
 [project.optional-dependencies]
 inmem = [
-    "langgraph-api>=0.5.35,<0.12.0a0 ; python_version >= '3.11'",
+    "langgraph-api>=0.5.35,<0.12.0 ; python_version >= '3.11'",
     "langgraph-runtime-inmem>=0.7 ; python_version >= '3.11'",
 ]
 

libs/cli/tests/unit_tests/test_config.py

@@ -2961,6 +2961,163 @@ def test_docker_tag_with_prerelease_api_version(version: str, in_config: bool):
     assert tag == f"langchain/langgraph-api:{version}-py3.11"
 
 
+def test_docker_tag_with_compatible_api_version_promotes_to_latest_patch():
+    config = validate_config(
+        {
+            "python_version": "3.12",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "image_distro": "wolfi",
+            "api_version": "~=0.11.0.dev5",
+        }
+    )
+
+    with patch(
+        "langgraph_cli.config._get_pypi_versions",
+        return_value=[
+            "0.11.0.dev5",
+            "0.11.0.dev6",
+            "0.11.0rc1",
+            "0.11.0",
+            "0.11.1rc1",
+            "0.11.1",
+            "0.12.0rc1",
+        ],
+    ) as get_versions:
+        tag = docker_tag(config)
+
+    get_versions.assert_called_once_with("langgraph-api")
+    assert tag == "langchain/langgraph-api:0.11.1-py3.12-wolfi"
+
+
+def test_docker_tag_with_compatible_api_version_freezes_dev_until_rc():
+    config = validate_config(
+        {
+            "python_version": "3.12",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "api_version": "~=0.11.0.dev5",
+        }
+    )
+
+    with patch(
+        "langgraph_cli.config._get_pypi_versions",
+        return_value=[
+            "0.11.0.dev5",
+            "0.11.0.dev6",
+            "0.11.0.dev7",
+        ],
+    ):
+        tag = docker_tag(config)
+
+    assert tag == "langchain/langgraph-api:0.11.0.dev5-py3.12"
+
+
+def test_docker_tag_with_stable_floating_api_version_promotes_to_future_stable():
+    config = validate_config(
+        {
+            "python_version": "3.12",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "image_distro": "wolfi",
+            "api_version": ">~=0.11.0.dev5",
+        }
+    )
+
+    with patch(
+        "langgraph_cli.config._get_pypi_versions",
+        return_value=[
+            "0.11.0.dev5",
+            "0.11.0.dev6",
+            "0.11.0rc1",
+            "0.11.0",
+            "0.11.1",
+            "0.12.0rc1",
+            "0.12.0",
+            "0.13.0.dev1",
+            "0.13.0",
+        ],
+    ) as get_versions:
+        tag = docker_tag(config)
+
+    get_versions.assert_called_once_with("langgraph-api")
+    assert tag == "langchain/langgraph-api:0.13.0-py3.12-wolfi"
+
+
+def test_docker_tag_with_stable_floating_api_version_ignores_future_prereleases():
+    config = validate_config(
+        {
+            "python_version": "3.12",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "api_version": ">~=0.11.0.dev5",
+        }
+    )
+
+    with patch(
+        "langgraph_cli.config._get_pypi_versions",
+        return_value=[
+            "0.11.0.dev5",
+            "0.11.0",
+            "0.12.0rc1",
+            "0.12.0.dev1",
+        ],
+    ):
+        tag = docker_tag(config)
+
+    assert tag == "langchain/langgraph-api:0.11.0-py3.12"
+
+
+def test_validate_config_rejects_unrecognized_api_version_range_operator():
+    with pytest.raises(click.UsageError, match="Invalid version format"):
+        validate_config(
+            {
+                "python_version": "3.12",
+                "dependencies": ["."],
+                "graphs": {"agent": "./agent.py:graph"},
+                "api_version": "~>=0.11.0.dev5",
+            }
+        )
+
+
+def test_docker_tag_with_compatible_api_version_supports_node_images():
+    config = validate_config(
+        {
+            "node_version": "20",
+            "graphs": {"agent": "./agent.js:graph"},
+            "image_distro": "wolfi",
+            "api_version": "~=1.2.4",
+        }
+    )
+
+    with patch(
+        "langgraph_cli.config._get_pypi_versions",
+        return_value=[
+            "1.2.4",
+            "1.2.5",
+            "1.3.0",
+        ],
+    ) as get_versions:
+        tag = docker_tag(config)
+
+    get_versions.assert_called_once_with("langgraph-api")
+    assert tag == "langchain/langgraphjs-api:1.2.5-node20-wolfi"
+
+
+def test_docker_tag_with_compatible_api_version_rejects_tagged_base_image():
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "api_version": "~=0.11.0.dev5",
+        }
+    )
+
+    with pytest.raises(click.UsageError, match="tagged base_image"):
+        docker_tag(config, base_image="langchain/langgraph-api:0.11.0")
+
+
 def test_config_to_docker_with_api_version():
     """Test config_to_docker function with api_version parameter."""
 

libs/cli/uv-examples/monorepo/uv.lock

@@ -237,14 +237,14 @@ wheels = [
 
 [[package]]
 name = "langchain-protocol"
-version = "0.0.15"
+version = "0.0.17"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/4f/24/9777489d6fbbee64af0c8f96d4f840239c408cf694f3394672807dafc490/langchain_protocol-0.0.15.tar.gz", hash = "sha256:9ab2d11ee73944754f10e037e717098d3a6796f0e58afa9cadda6154e7655ade", size = 5862, upload-time = "2026-05-01T22:30:04.748Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/44/b3/4e2429876c7a35585618caa2b9f9089f7162a6b50562b614ad82ac11c17e/langchain_protocol-0.0.17.tar.gz", hash = "sha256:e7cbe58c205df4b4fd87dc6d5bb23f10e13b236d0e2e1b0b9d05bc2b648f3eea", size = 6026, upload-time = "2026-06-12T18:39:51.923Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/1d/7a/9c97a7b9cbe4c5dc6a44cdb1545450c28f0c8ce89b9c1f0ee7fbad896263/langchain_protocol-0.0.15-py3-none-any.whl", hash = "sha256:461eb794358f83d5e42635a5797799ffec7b4702314e34edf73ac21e75d3ef79", size = 6982, upload-time = "2026-05-01T22:30:03.877Z" },
+    { url = "https://files.pythonhosted.org/packages/13/0a/a1bfe72c6ec856e99773bbd96c8086421e554b3693d0142b9ea009c6ac92/langchain_protocol-0.0.17-py3-none-any.whl", hash = "sha256:982a08fe152586ed10d4ff3d538c2e0b5766e5f307cdea325e10be3f2c17cae6", size = 7096, upload-time = "2026-06-12T18:39:50.973Z" },
 ]
 
 [[package]]

libs/cli/uv-examples/simple/uv.lock

@@ -211,14 +211,14 @@ wheels = [
 
 [[package]]
 name = "langchain-protocol"
-version = "0.0.15"
+version = "0.0.17"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/4f/24/9777489d6fbbee64af0c8f96d4f840239c408cf694f3394672807dafc490/langchain_protocol-0.0.15.tar.gz", hash = "sha256:9ab2d11ee73944754f10e037e717098d3a6796f0e58afa9cadda6154e7655ade", size = 5862, upload-time = "2026-05-01T22:30:04.748Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/44/b3/4e2429876c7a35585618caa2b9f9089f7162a6b50562b614ad82ac11c17e/langchain_protocol-0.0.17.tar.gz", hash = "sha256:e7cbe58c205df4b4fd87dc6d5bb23f10e13b236d0e2e1b0b9d05bc2b648f3eea", size = 6026, upload-time = "2026-06-12T18:39:51.923Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/1d/7a/9c97a7b9cbe4c5dc6a44cdb1545450c28f0c8ce89b9c1f0ee7fbad896263/langchain_protocol-0.0.15-py3-none-any.whl", hash = "sha256:461eb794358f83d5e42635a5797799ffec7b4702314e34edf73ac21e75d3ef79", size = 6982, upload-time = "2026-05-01T22:30:03.877Z" },
+    { url = "https://files.pythonhosted.org/packages/13/0a/a1bfe72c6ec856e99773bbd96c8086421e554b3693d0142b9ea009c6ac92/langchain_protocol-0.0.17-py3-none-any.whl", hash = "sha256:982a08fe152586ed10d4ff3d538c2e0b5766e5f307cdea325e10be3f2c17cae6", size = 7096, upload-time = "2026-06-12T18:39:50.973Z" },
 ]
 
 [[package]]

libs/cli/uv.lock

@@ -1094,7 +1094,7 @@ test = [
 requires-dist = [
     { name = "click", specifier = ">=8.1.7" },
     { name = "httpx", specifier = ">=0.24.0" },
-    { name = "langgraph-api", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.5.35,<0.12.0a0" },
+    { name = "langgraph-api", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.5.35,<0.12.0" },
     { name = "langgraph-runtime-inmem", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.7" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11'", specifier = ">=0.1.0" },
     { name = "pathspec", specifier = ">=0.11.0" },

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/langgraph/channels/delta.py

@@ -177,8 +177,7 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
                 if overwrite_value is not None
                 else self.typ()
             )
-            remaining = [v for i, v in enumerate(values) if i != overwrite_idx]
-            self.value = self.reducer(base, remaining) if remaining else base
+            self.value = base
             return True
         base = self.typ() if self.value is MISSING else self.value
         self.value = self.reducer(base, list(values))

libs/langgraph/langgraph/stream/run_stream.py

@@ -132,13 +132,23 @@ class GraphRunStream:
     def abort(self) -> None:
         """Stop the run early.
 
-        Closes the mux and marks the stream exhausted. The graph
-        iterator is dropped; any in-flight nodes see the closure on
-        their next yield point. Idempotent.
+        Closes the underlying graph iterator (propagating `GeneratorExit`
+        so in-flight nodes and subgraphs are cancelled), closes the mux,
+        and marks the stream exhausted. Idempotent.
         """
         if self._exhausted:
             return
         self._exhausted = True
+        graph_iter = self._graph_iter
+        self._graph_iter = None
+        if (
+            graph_iter is not None
+            and (close := getattr(graph_iter, "close", None)) is not None
+        ):
+            try:
+                close()
+            except Exception:
+                pass
         try:
             self._mux.close()
         except Exception:
@@ -348,6 +358,8 @@ class AsyncGraphRunStream:
         self._scope_list: list[str] = list(mux.scope)
         self._pump_cond = asyncio.Condition()
         self._pumping = False
+        self._anext_task: asyncio.Future[Any] | None = None
+        self._aborting = False
         for key in mux.native_keys:
             setattr(self, key, mux.extensions[key])
         if wire_pump:
@@ -407,7 +419,25 @@ class AsyncGraphRunStream:
 
         try:
             try:
-                part = await self._graph_aiter.__anext__()
+                # Run the pull as a child task so `abort()` can cancel it
+                # mid-flight. Cancelling propagates `CancelledError` into the
+                # graph generator frame -> Pregel loop -> nested subgraph
+                # nodes, which a bare `aclose()` cannot do while the generator
+                # is running ("asynchronous generator is already running").
+                self._anext_task = asyncio.ensure_future(self._graph_aiter.__anext__())
+                try:
+                    part = await self._anext_task
+                except asyncio.CancelledError:
+                    if self._aborting:
+                        # Abort-initiated cancel: stop gracefully.
+                        self._exhausted = True
+                        return False
+                    # Genuine external cancel of this task: also stop the
+                    # in-flight pull, then propagate.
+                    self._anext_task.cancel()
+                    raise
+                finally:
+                    self._anext_task = None
                 event = convert_to_protocol_event(part)
                 self._observe_event(event)
                 await self._mux.apush(event)
@@ -428,15 +458,40 @@ class AsyncGraphRunStream:
     async def abort(self) -> None:
         """Stop the run early.
 
-        Marks the stream exhausted, wakes any pump-waiters, and closes
-        the mux. Any `apush` blocked on backpressure wakes and returns
-        without appending. Idempotent.
+        Marks the stream exhausted and wakes any pump-waiters. Cancels an
+        in-flight pull if one is running, then closes the underlying graph
+        iterator, so running nodes and nested subgraphs are cancelled
+        whether or not a pump is mid-pull. Closes the mux; any `apush`
+        blocked on backpressure wakes and returns without appending.
+        Idempotent.
         """
         async with self._pump_cond:
             if self._exhausted:
                 return
             self._exhausted = True
+            self._aborting = True
+            graph_aiter = self._graph_aiter
+            self._graph_aiter = None
+            anext_task = self._anext_task
             self._pump_cond.notify_all()
+        # If a pump is mid-pull, cancel it so the cancellation propagates
+        # into running nodes and nested subgraphs. Once it settles the
+        # generator is no longer running, so the `aclose()` below is a safe
+        # final cleanup (and handles the no-in-flight-pull case directly).
+        if anext_task is not None and not anext_task.done():
+            anext_task.cancel()
+            try:
+                await anext_task
+            except (asyncio.CancelledError, Exception):
+                pass
+        if (
+            graph_aiter is not None
+            and (aclose := getattr(graph_aiter, "aclose", None)) is not None
+        ):
+            try:
+                await aclose()
+            except Exception:
+                pass
         try:
             await self._mux.aclose()
         except Exception:

libs/langgraph/tests/test_channels.py

@@ -186,6 +186,18 @@ def test_delta_channel_overwrite() -> None:
     assert ch.get()[0].content == "new"
 
 
+def test_delta_channel_overwrite_bypasses_same_step_reducer_writes() -> None:
+    def list_reducer(state: list, writes: list) -> list:
+        out = list(state)
+        for w in writes:
+            out.extend(w)
+        return out
+
+    ch = DeltaChannel(list_reducer, list).from_checkpoint(MISSING)
+    ch.update([[1], Overwrite([50]), [2]])
+    assert ch.get() == [50]
+
+
 def test_delta_channel_remove_message_and_replay() -> None:
     """RemoveMessage must round-trip correctly when writes are replayed."""
     spec = DeltaChannel(_messages_delta_reducer, list)

libs/langgraph/tests/test_pregel.py

@@ -9281,6 +9281,13 @@ def test_send_with_untracked_value_overlapping_keys(
     assert state.values.get("dictionary") == {"session_resource": "legal_value"}
 
 
+def _delta_list_reducer(state: list, writes: Sequence[list]) -> list:
+    out = list(state)
+    for write in writes:
+        out.extend(write)
+    return out
+
+
 @pytest.mark.parametrize("as_json", [False, True])
 def test_overwrite_sequential(
     sync_checkpointer: BaseCheckpointSaver, as_json: bool
@@ -9388,6 +9395,105 @@ def test_overwrite_parallel_error(
         graph.invoke({"messages": ["START"]}, config)
 
 
+@pytest.mark.parametrize("as_json", [False, True])
+def test_delta_channel_overwrite_sequential(
+    sync_checkpointer: BaseCheckpointSaver, as_json: bool
+) -> None:
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(_delta_list_reducer)]
+
+    def node_a(state: State):
+        return {"messages": ["a"]}
+
+    def node_b(state: State):
+        overwrite = {"__overwrite__": ["b"]} if as_json else Overwrite(["b"])
+        return {"messages": overwrite}
+
+    builder = StateGraph(State)
+    builder.add_node("node_a", node_a)
+    builder.add_node("node_b", node_b)
+    builder.add_edge(START, "node_a")
+    builder.add_edge("node_a", "node_b")
+
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "delta-overwrite-sequential"}}
+    result = graph.invoke({"messages": ["START"]}, config)
+    assert result == {"messages": ["b"]}
+
+
+@pytest.mark.parametrize("as_json", [False, True])
+def test_delta_channel_overwrite_parallel(
+    sync_checkpointer: BaseCheckpointSaver, as_json: bool
+) -> None:
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(_delta_list_reducer)]
+
+    def node_a(state: State):
+        return {"messages": ["a"]}
+
+    def node_b(state: State):
+        overwrite = {"__overwrite__": ["b"]} if as_json else Overwrite(["b"])
+        return {"messages": overwrite}
+
+    def node_c(state: State):
+        return {"messages": ["c"]}
+
+    def node_d(state: State):
+        return {"messages": ["d"]}
+
+    builder = StateGraph(State)
+    builder.add_node("node_a", node_a)
+    builder.add_node("node_b", node_b)
+    builder.add_node("node_c", node_c)
+    builder.add_node("node_d", node_d)
+    builder.add_edge(START, "node_a")
+    builder.add_edge("node_a", "node_b")
+    builder.add_edge("node_a", "node_c")
+    builder.add_edge("node_b", "node_d")
+    builder.add_edge("node_c", "node_d")
+
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "delta-overwrite-parallel"}}
+    result = graph.invoke({"messages": ["START"]}, config)
+    assert result == {"messages": ["b", "d"]}
+
+
+@pytest.mark.parametrize("as_json", [False, True])
+def test_delta_channel_overwrite_parallel_error(
+    sync_checkpointer: BaseCheckpointSaver, as_json: bool
+) -> None:
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(_delta_list_reducer)]
+
+    def node_a(state: State):
+        return {"messages": ["a"]}
+
+    def node_b(state: State):
+        overwrite = {"__overwrite__": ["b"]} if as_json else Overwrite(["b"])
+        return {"messages": overwrite}
+
+    def node_c(state: State):
+        overwrite = {"__overwrite__": ["c"]} if as_json else Overwrite(["c"])
+        return {"messages": overwrite}
+
+    builder = StateGraph(State)
+    builder.add_node("node_a", node_a)
+    builder.add_node("node_b", node_b)
+    builder.add_node("node_c", node_c)
+    builder.add_edge(START, "node_a")
+    builder.add_edge("node_a", "node_b")
+    builder.add_edge("node_a", "node_c")
+    builder.add_edge("node_b", END)
+    builder.add_edge("node_c", END)
+
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "delta-overwrite-parallel-error"}}
+    with pytest.raises(
+        InvalidUpdateError, match="Can receive only one Overwrite value per super-step."
+    ):
+        graph.invoke({"messages": ["START"]}, config)
+
+
 def test_fork_does_not_apply_pending_writes(
     sync_checkpointer: BaseCheckpointSaver,
 ) -> None:

libs/langgraph/tests/test_pregel_stream_events_v3.py

@@ -607,6 +607,164 @@ class TestStreamV2Async:
             _ = await anext(aiter(run.values))
         assert run._exhausted is True
 
+    async def test_abort_cancels_running_subgraph(self) -> None:
+        class CountState(TypedDict):
+            count: int
+
+        runs: list[int] = []
+
+        async def sub_node(state: CountState) -> dict:
+            runs.append(state["count"] + 1)
+            await asyncio.sleep(0.05)
+            return {"count": state["count"] + 1}
+
+        sub_graph = (
+            StateGraph(CountState)
+            .add_node("sub_node", sub_node)
+            .set_entry_point("sub_node")
+            .add_conditional_edges(
+                "sub_node",
+                lambda s: END if s["count"] >= 10 else "sub_node",
+            )
+            .compile()
+        )
+
+        async def main_node(state: CountState) -> None:
+            await sub_graph.ainvoke({"count": 0})
+
+        main_graph = (
+            StateGraph(CountState)
+            .add_node("main_node", main_node)
+            .set_entry_point("main_node")
+            .compile()
+        )
+
+        run = await main_graph.astream_events({"count": 0}, version="v3")
+        async for e in run:
+            if (
+                e["method"] == "values"
+                and e["params"]["namespace"]
+                and e["params"]["data"]["count"] >= 2
+            ):
+                break
+        await run.abort()
+        runs_at_abort = len(runs)
+        # Give the (now-cancelled) subgraph a chance to keep looping.
+        await asyncio.sleep(0.3)
+        assert len(runs) == runs_at_abort
+        assert len(runs) < 10
+
+    async def test_abort_cancels_deeply_nested_subgraph(self) -> None:
+        class CountState(TypedDict):
+            count: int
+
+        runs: list[int] = []
+
+        async def deep_node(state: CountState) -> dict:
+            runs.append(state["count"] + 1)
+            await asyncio.sleep(0.05)
+            return {"count": state["count"] + 1}
+
+        # Deepest graph loops until count >= 10.
+        graph = (
+            StateGraph(CountState)
+            .add_node("deep_node", deep_node)
+            .set_entry_point("deep_node")
+            .add_conditional_edges(
+                "deep_node",
+                lambda s: END if s["count"] >= 10 else "deep_node",
+            )
+            .compile()
+        )
+
+        # Wrap it three times: graph -> subgraph -> subgraph -> subgraph.
+        for _ in range(3):
+
+            async def caller(state: CountState, _child: Any = graph) -> dict:
+                return await _child.ainvoke({"count": 0})
+
+            graph = (
+                StateGraph(CountState)
+                .add_node("caller", caller)
+                .set_entry_point("caller")
+                .compile()
+            )
+
+        run = await graph.astream_events({"count": 0}, version="v3")
+        async for e in run:
+            if (
+                e["method"] == "values"
+                and e["params"]["namespace"]
+                and e["params"]["data"]["count"] >= 2
+            ):
+                break
+        await run.abort()
+        runs_at_abort = len(runs)
+        # Give the (now-cancelled) nested subgraph a chance to keep looping.
+        await asyncio.sleep(0.3)
+        assert len(runs) == runs_at_abort
+        assert len(runs) < 10
+
+    async def test_abort_cancels_subgraph_during_inflight_pump(self) -> None:
+        class CountState(TypedDict):
+            count: int
+
+        started = asyncio.Event()
+        cancelled = asyncio.Event()
+
+        async def sub_node(state: CountState) -> dict:
+            started.set()
+            try:
+                # Long-running node: still in flight when abort fires.
+                await asyncio.sleep(5)
+            except asyncio.CancelledError:
+                cancelled.set()
+                raise
+            return {"count": state["count"] + 1}
+
+        sub_graph = (
+            StateGraph(CountState)
+            .add_node("sub_node", sub_node)
+            .set_entry_point("sub_node")
+            .compile()
+        )
+
+        async def main_node(state: CountState) -> None:
+            await sub_graph.ainvoke({"count": 0})
+
+        main_graph = (
+            StateGraph(CountState)
+            .add_node("main_node", main_node)
+            .set_entry_point("main_node")
+            .compile()
+        )
+
+        run = await main_graph.astream_events({"count": 0}, version="v3")
+
+        # A consumer task drives the pump. Once the subgraph node is
+        # running, no further event is produced, so the consumer parks
+        # inside _apump_next awaiting graph_aiter.__anext__() — the
+        # generator is "running" and a plain aclose() would raise.
+        async def consume() -> None:
+            async for _e in run:
+                pass
+
+        consumer = asyncio.create_task(consume())
+        try:
+            await asyncio.wait_for(started.wait(), timeout=2.0)
+            # Let the consumer drain and park in __anext__.
+            await asyncio.sleep(0.05)
+            # Abort from a different task while the consumer is in __anext__.
+            await run.abort()
+            # The in-flight subgraph node must observe cancellation.
+            await asyncio.wait_for(cancelled.wait(), timeout=2.0)
+        finally:
+            consumer.cancel()
+            try:
+                await consumer
+            except asyncio.CancelledError:
+                pass
+
     async def test_extensions_has_native_keys(self) -> None:
         run = await _build_simple_graph().astream_events(
             {"value": "x", "items": []}, version="v3"

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

libs/langgraph/uv.lock

@@ -1721,7 +1721,7 @@ inmem = [
 requires-dist = [
     { name = "click", specifier = ">=8.1.7" },
     { name = "httpx", specifier = ">=0.24.0" },
-    { name = "langgraph-api", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.5.35,<0.12.0a0" },
+    { name = "langgraph-api", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.5.35,<0.12.0" },
     { name = "langgraph-runtime-inmem", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.7" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11'", specifier = ">=0.1.0" },
     { name = "pathspec", specifier = ">=0.11.0" },