release(sdk-py): 0.4.2 (#7955)

## Problem

Fixes #7953.

The v3 SSE and WebSocket stream transports build their default paths by
interpolating `thread_id` directly into the URL:

```python
self._commands_url = commands_path or f"/threads/{thread_id}/commands"
self._stream_url   = stream_path   or f"/threads/{thread_id}/stream/events"
```

This skips the `_quote_path_param` escaping that the rest of the SDK
adopted in #7893. A `thread_id` containing reserved characters or
dot-segments is then normalized by the HTTP/WebSocket stack before
transmission. For example, `thread_id = "../assistants/abc"`:

| | before |
|---|---|
| constructed | `/threads/../assistants/abc/commands` |
| wire path | `/assistants/abc/commands` |

So the value stops being one opaque identifier under
`/threads/{thread_id}/...` and silently hits a different resource.

## Fix

Reuse the existing `_quote_path_param` helper for the **default** paths
in all four transports (`http`, `sync_http`, `ws`, `sync_ws`). Explicit
`commands_path` / `stream_path` overrides are left untouched, so callers
that pass their own paths opt out of encoding as before.

`_quote_path_param("../assistants/abc")` → `..%2Fassistants%2Fabc`,
which the HTTP/WS stack no longer collapses.

## Tests

Adds `tests/streaming/test_transport_path_encoding.py` covering all four
transports:
- default `_commands_url` / `_stream_url` / `_stream_path` are
percent-encoded,
- the actual SSE wire path (async + sync) stays under `/threads/`,
- the built WebSocket URL (async + sync) stays under `/threads/`,
- explicit path overrides are left untouched.

`make format`, `make lint`, and `make test` all pass in `libs/sdk-py`
(491 passed).

.github/workflows/ci.yml

@@ -50,6 +50,8 @@ jobs:
               - '**/uv.lock'
             sdk_py:
               - 'libs/sdk-py/**'
+              - 'libs/langgraph/langgraph/pregel/remote.py'
+              - 'libs/langgraph/langgraph/pregel/_remote_run_stream.py'
 
   lint:
     needs: changes

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.26"
+__version__ = "0.4.27"

libs/cli/langgraph_cli/config.py

@@ -66,34 +66,14 @@ def _get_pip_cleanup_lines(
     to_uninstall: tuple[str] | None,
     pip_installer: Literal["uv", "pip"],
 ) -> str:
-    commands = ["# -- Validate installed dependencies are internally consistent --"]
-    if pip_installer == "uv":
-        commands.append(
-            """RUN uv pip check --system || ( \
-echo "Dependency resolution check failed. One or more installed packages are incompatible."; \
-echo "Pin compatible versions in your dependencies and try again."; \
-exit 1 \
-)"""
-        )
-    elif pip_installer == "pip":
-        commands.append(
-            """RUN python -m pip check || ( \
-echo "Dependency resolution check failed. One or more installed packages are incompatible."; \
-echo "Pin compatible versions in your dependencies and try again."; \
-exit 1 \
-)"""
-        )
-    commands.append(
-        """# -- End dependency validation --
-# -- Ensure user deps didn't inadvertently overwrite langgraph-api
+    commands = [
+        f"""# -- Ensure user deps didn't inadvertently overwrite langgraph-api
 RUN mkdir -p /api/langgraph_api /api/langgraph_runtime /api/langgraph_license && \
 touch /api/langgraph_api/__init__.py /api/langgraph_runtime/__init__.py /api/langgraph_license/__init__.py
-RUN PYTHONDONTWRITEBYTECODE=1 """
-        + install_cmd
-        + """ --no-cache-dir --no-deps -e /api
+RUN PYTHONDONTWRITEBYTECODE=1 {install_cmd} --no-cache-dir --no-deps -e /api
 # -- End of ensuring user deps didn't inadvertently overwrite langgraph-api --
 # -- Removing build deps from the final image ~<:===~~~ --"""
-    )
+    ]
     if to_uninstall:
         for pack in to_uninstall:
             if pack not in _BUILD_TOOLS:

libs/cli/langgraph_cli/dependency_tracking.py

@@ -0,0 +1,141 @@
+"""Detection of tracked Python packages in a local LangGraph project.
+
+Mirrors host-backend's `host.models.dependency_tracking` so that CLI-based
+deploys report the same `tracked_packages` revision metadata that
+GitHub-based deploys do. The host backend strictly validates each entry
+against `<package-name>:<version>` with package-name in `TRACKED_PACKAGES`,
+so the detection rules here must match exactly.
+"""
+
+from __future__ import annotations
+
+import pathlib
+import re
+
+# Single source of truth for which packages the host backend cares about.
+# Keep in sync with host-backend/host/models/tracked_packages.py.
+TRACKED_PACKAGES: tuple[str, ...] = ("google-adk",)
+
+_MAX_READ_BYTES = 5 * 1024 * 1024
+
+_PACKAGES_ALT = "|".join(re.escape(p) for p in TRACKED_PACKAGES)
+
+_DEPS_RE = re.compile(
+    rf"(?<![a-zA-Z0-9_-])({_PACKAGES_ALT})"
+    r"(?:\[[^\]]*\])?"
+    r"\s*((?:(?:==|>=|<=|~=|!=|>|<)\s*[\w.*]+\s*,?\s*)+)"
+)
+
+_UV_LOCK_RE = re.compile(
+    rf'name\s*=\s*"({_PACKAGES_ALT})"\s*\n\s*version\s*=\s*"([^"]+)"'
+)
+
+_BARE_RE = re.compile(rf'(?<![a-zA-Z0-9_-])({_PACKAGES_ALT})(?:\[[^\]]*\])?\s*[,"\'\n]')
+
+_EXTRAS_BRACKET_RE = re.compile(r"\[([a-zA-Z0-9_.\- ,\t]+)\]")
+
+
+def _appears_in_extras(content: str, pkg: str) -> bool:
+    for m in _EXTRAS_BRACKET_RE.finditer(content):
+        for token in m.group(1).split(","):
+            if token.strip() == pkg:
+                return True
+    return False
+
+
+def _read_text(path: pathlib.Path) -> str | None:
+    try:
+        if not path.is_file():
+            return None
+        with open(path, "rb") as f:
+            data = f.read(_MAX_READ_BYTES + 1)
+    except OSError:
+        return None
+    if len(data) > _MAX_READ_BYTES:
+        data = data[:_MAX_READ_BYTES]
+    return data.decode("utf-8", errors="replace")
+
+
+def _find_version_for(
+    pkg: str,
+    lock_content: str | None,
+    pyproject_content: str | None,
+    requirements_content: str | None,
+) -> str | None:
+    if lock_content is not None:
+        for m in _UV_LOCK_RE.finditer(lock_content):
+            if m.group(1) == pkg:
+                return m.group(2)
+    for content in (pyproject_content, requirements_content):
+        if content is None:
+            continue
+        for m in _DEPS_RE.finditer(content):
+            if m.group(1) == pkg:
+                return m.group(2).strip().rstrip(",")
+        for m in _BARE_RE.finditer(content):
+            if m.group(1) == pkg:
+                return "unknown"
+        if _appears_in_extras(content, pkg):
+            return "unknown"
+    return None
+
+
+def _resolved_dep_base(
+    project_root: pathlib.Path, dep_path: str
+) -> pathlib.Path | None:
+    """Return the resolved dep directory if it stays inside the project root."""
+    try:
+        candidate = (project_root / dep_path).resolve()
+    except (OSError, RuntimeError):
+        return None
+    try:
+        candidate.relative_to(project_root)
+    except ValueError:
+        return None
+    return candidate
+
+
+def find_tracked_packages(
+    config: pathlib.Path,
+    config_json: dict,
+) -> list[str]:
+    """Return every tracked package found in deps as `<name>:<version>` entries.
+
+    `config` is the absolute path to `langgraph.json`; dep paths in
+    `config_json["dependencies"]` are resolved relative to its parent.
+    Detection precedence per package: uv.lock resolved > pyproject.toml /
+    requirements.txt specifier > bare reference > extras bracket (last
+    two recorded as "unknown"). Output is ordered by `TRACKED_PACKAGES`.
+    """
+    try:
+        project_root = config.parent.resolve()
+    except (OSError, RuntimeError):
+        return []
+
+    dep_paths = config_json.get("dependencies") or ["."]
+
+    found: dict[str, str] = {}
+
+    for dep_path in dep_paths:
+        if all(pkg in found for pkg in TRACKED_PACKAGES):
+            break
+        if not isinstance(dep_path, str):
+            continue
+        base = _resolved_dep_base(project_root, dep_path)
+        if base is None or not base.is_dir():
+            continue
+
+        lock_content = _read_text(base / "uv.lock")
+        pyproject_content = _read_text(base / "pyproject.toml")
+        requirements_content = _read_text(base / "requirements.txt")
+
+        for pkg in TRACKED_PACKAGES:
+            if pkg in found:
+                continue
+            version = _find_version_for(
+                pkg, lock_content, pyproject_content, requirements_content
+            )
+            if version is not None:
+                found[pkg] = version
+
+    return [f"{pkg}:{found[pkg]}" for pkg in TRACKED_PACKAGES if pkg in found]

libs/cli/langgraph_cli/deploy.py

@@ -20,6 +20,7 @@ from dotenv import dotenv_values, set_key
 import langgraph_cli.config
 from langgraph_cli.analytics import log_command
 from langgraph_cli.constants import DEFAULT_CONFIG
+from langgraph_cli.dependency_tracking import find_tracked_packages
 from langgraph_cli.docker import build_docker_image, can_build_locally
 from langgraph_cli.exec import Runner, subp_exec
 from langgraph_cli.host_backend import HostBackendClient, HostBackendError
@@ -902,6 +903,7 @@ def _run_local_build(
     build_command: str | None,
     docker_build_args: Sequence[str],
     secrets: list[dict[str, str]],
+    tracked_packages: list[str] | None,
 ) -> BuildResult:
     """Build locally with Docker, push to registry, update deployment."""
     # Use buildx to cross-compile for amd64 when running on a non-x86_64 host
@@ -1060,7 +1062,10 @@ def _run_local_build(
         # -- Step: Update deployment --
         _log_deploy_step(step, f"Updating deployment {deployment_id}")
         updated = client.update_deployment(
-            deployment_id, resolved_image, secrets=secrets
+            deployment_id,
+            resolved_image,
+            secrets=secrets,
+            tracked_packages=tracked_packages,
         )
 
     return BuildResult(
@@ -1083,6 +1088,7 @@ def _run_remote_build(
     install_command: str | None,
     build_command: str | None,
     secrets: list[dict[str, str]],
+    tracked_packages: list[str] | None,
 ) -> BuildResult:
     """Upload source tarball and trigger a remote build."""
     from langgraph_cli.archive import create_archive
@@ -1113,6 +1119,7 @@ def _run_remote_build(
         secrets=secrets,
         install_command=install_command,
         build_command=build_command,
+        tracked_packages=tracked_packages,
     )
 
     log_offset: str | None = None
@@ -1597,6 +1604,15 @@ def _deploy_cmd(
     if not deployment_id:
         raise click.ClickException("Failed to determine deployment ID")
 
+    # Scan local sources for tracked packages so the new revision carries
+    # the same metadata GitHub-backed deploys produce. Failures must never
+    # block a deploy.
+    try:
+        tracked_packages = find_tracked_packages(config, config_json) or None
+    except Exception as exc:
+        em.warn(f"Skipped tracked-package scan: {exc}")
+        tracked_packages = None
+
     # -- 3. Build (divergent path) --
     if use_remote_build:
         build_result = _run_remote_build(
@@ -1609,6 +1625,7 @@ def _deploy_cmd(
             install_command=install_command,
             build_command=build_command,
             secrets=secrets,
+            tracked_packages=tracked_packages,
         )
     else:
         build_result = _run_local_build(
@@ -1628,6 +1645,7 @@ def _deploy_cmd(
             build_command=build_command,
             docker_build_args=docker_build_args,
             secrets=secrets,
+            tracked_packages=tracked_packages,
         )
 
     # -- 4. Shared wait + result --

libs/cli/langgraph_cli/host_backend.py

@@ -122,11 +122,14 @@ class HostBackendClient:
         deployment_id: str,
         image_uri: str,
         secrets: list[dict[str, str]] | None = None,
+        tracked_packages: list[str] | None = None,
     ) -> dict[str, Any]:
         payload: dict[str, Any] = {
             "revision_source": "internal_docker",
             "source_revision_config": {"image_uri": image_uri},
         }
+        if tracked_packages:
+            payload["tracked_packages"] = tracked_packages
         if secrets is not None:
             payload["secrets"] = secrets
         return self._request(
@@ -143,6 +146,7 @@ class HostBackendClient:
         secrets: list[dict[str, str]] | None = None,
         install_command: str | None = None,
         build_command: str | None = None,
+        tracked_packages: list[str] | None = None,
     ) -> dict[str, Any]:
         """Trigger a remote build revision with the uploaded tarball."""
         payload: dict[str, Any] = {
@@ -152,6 +156,8 @@ class HostBackendClient:
                 "langgraph_config_path": config_path,
             },
         }
+        if tracked_packages:
+            payload["tracked_packages"] = tracked_packages
 
         source_config: dict[str, Any] = {}
         if install_command is not None:

libs/cli/tests/unit_tests/test_config.py

@@ -1330,8 +1330,6 @@ def test_config_to_docker_pip_installer():
         PATH_TO_CONFIG, config_auto, base_image="langchain/langgraph-api:0.2.47"
     )
     assert "uv pip install --system " in docker_auto
-    assert "RUN uv pip check --system" in docker_auto
-    assert "python -m pip check" not in docker_auto
     assert "rm /usr/bin/uv /usr/bin/uvx" in docker_auto
 
     # Test explicit pip setting
@@ -1340,8 +1338,6 @@ def test_config_to_docker_pip_installer():
         PATH_TO_CONFIG, config_pip, base_image="langchain/langgraph-api:0.2.47"
     )
     assert "uv pip install --system " not in docker_pip
-    assert "RUN uv pip check --system" not in docker_pip
-    assert "RUN python -m pip check" in docker_pip
     assert "pip install" in docker_pip
     assert "rm /usr/bin/uv" not in docker_pip
 
@@ -1351,8 +1347,6 @@ def test_config_to_docker_pip_installer():
         PATH_TO_CONFIG, config_uv, base_image="langchain/langgraph-api:0.2.47"
     )
     assert "uv pip install --system " in docker_uv
-    assert "RUN uv pip check --system" in docker_uv
-    assert "python -m pip check" not in docker_uv
     assert "rm /usr/bin/uv /usr/bin/uvx" in docker_uv
 
     # Test auto behavior with older image (should use pip)
@@ -1363,8 +1357,6 @@ def test_config_to_docker_pip_installer():
         PATH_TO_CONFIG, config_auto_old, base_image="langchain/langgraph-api:0.2.46"
     )
     assert "uv pip install --system " not in docker_auto_old
-    assert "RUN uv pip check --system" not in docker_auto_old
-    assert "RUN python -m pip check" in docker_auto_old
     assert "pip install" in docker_auto_old
     assert "rm /usr/bin/uv" not in docker_auto_old
 
@@ -1374,32 +1366,6 @@ def test_config_to_docker_pip_installer():
         PATH_TO_CONFIG, config_default, base_image="langchain/langgraph-api:0.2.47"
     )
     assert "uv pip install --system " in docker_default
-    assert "RUN uv pip check --system" in docker_default
-    assert "python -m pip check" not in docker_default
-
-
-def test_get_pip_cleanup_lines_selects_check_command_by_installer():
-    cleanup_uv = _get_pip_cleanup_lines(
-        install_cmd="uv pip install --system",
-        to_uninstall=None,
-        pip_installer="uv",
-    )
-    assert "RUN uv pip check --system" in cleanup_uv
-    assert "python -m pip check" not in cleanup_uv
-    assert cleanup_uv.index("RUN uv pip check --system") < cleanup_uv.index(
-        "RUN mkdir -p /api/langgraph_api"
-    )
-
-    cleanup_pip = _get_pip_cleanup_lines(
-        install_cmd="pip install",
-        to_uninstall=None,
-        pip_installer="pip",
-    )
-    assert "RUN python -m pip check" in cleanup_pip
-    assert "uv pip check --system" not in cleanup_pip
-    assert cleanup_pip.index("RUN python -m pip check") < cleanup_pip.index(
-        "RUN mkdir -p /api/langgraph_api"
-    )
 
 
 def test_config_to_docker_uv_lock():
@@ -2524,7 +2490,7 @@ def test_config_to_compose_simple_config():
 def test_config_to_compose_env_vars():
     graphs = {"agent": "./agent.py:graph"}
     expected_compose_stdin = f"""                        OPENAI_API_KEY: "key"
-
+        
         pull_policy: build
         build:
             context: .
@@ -2607,7 +2573,7 @@ def test_config_to_compose_env_file():
 def test_config_to_compose_watch():
     graphs = {"agent": "./agent.py:graph"}
     expected_compose_stdin = f"""\
-
+        
         pull_policy: build
         build:
             context: .
@@ -2633,7 +2599,7 @@ def test_config_to_compose_watch():
                 ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}}'
 {textwrap.indent(textwrap.dedent(FORMATTED_CLEANUP_LINES), "                ")}
                 WORKDIR /deps/outer-unit_tests/unit_tests
-
+        
         develop:
             watch:
                 - path: test_config.json
@@ -2680,7 +2646,7 @@ def test_config_to_compose_end_to_end():
                 ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}}'
 {textwrap.indent(textwrap.dedent(FORMATTED_CLEANUP_LINES), "                ")}
                 WORKDIR /deps/outer-unit_tests/unit_tests
-
+        
         develop:
             watch:
                 - path: test_config.json

libs/cli/tests/unit_tests/test_dependency_tracking.py

@@ -0,0 +1,134 @@
+import pathlib
+
+import pytest
+
+from langgraph_cli.dependency_tracking import (
+    TRACKED_PACKAGES,
+    find_tracked_packages,
+)
+
+
+def _write_project(
+    tmp_path: pathlib.Path,
+    *,
+    dep_subdir: str = ".",
+    uv_lock: str | None = None,
+    pyproject: str | None = None,
+    requirements: str | None = None,
+    dependencies: list[str] | None = None,
+) -> tuple[pathlib.Path, dict]:
+    project_root = tmp_path
+    dep_dir = (project_root / dep_subdir).resolve()
+    dep_dir.mkdir(parents=True, exist_ok=True)
+    if uv_lock is not None:
+        (dep_dir / "uv.lock").write_text(uv_lock)
+    if pyproject is not None:
+        (dep_dir / "pyproject.toml").write_text(pyproject)
+    if requirements is not None:
+        (dep_dir / "requirements.txt").write_text(requirements)
+    config = project_root / "langgraph.json"
+    config.write_text("{}")
+    config_json = {"dependencies": dependencies or [dep_subdir]}
+    return config, config_json
+
+
+def test_uv_lock_resolved_version_preferred(tmp_path: pathlib.Path) -> None:
+    config, config_json = _write_project(
+        tmp_path,
+        uv_lock='name = "google-adk"\nversion = "1.2.3"\n',
+        pyproject='dependencies = ["google-adk>=0.5"]',
+    )
+    assert find_tracked_packages(config, config_json) == ["google-adk:1.2.3"]
+
+
+def test_pyproject_specifier_used_when_no_lock(tmp_path: pathlib.Path) -> None:
+    config, config_json = _write_project(
+        tmp_path,
+        pyproject='dependencies = ["google-adk>=0.5,<2"]',
+    )
+    assert find_tracked_packages(config, config_json) == ["google-adk:>=0.5,<2"]
+
+
+def test_requirements_txt_specifier(tmp_path: pathlib.Path) -> None:
+    config, config_json = _write_project(
+        tmp_path,
+        requirements="google-adk==1.0.0\n",
+    )
+    assert find_tracked_packages(config, config_json) == ["google-adk:==1.0.0"]
+
+
+def test_bare_reference_records_unknown(tmp_path: pathlib.Path) -> None:
+    config, config_json = _write_project(
+        tmp_path,
+        requirements="google-adk\nother-pkg==1.0\n",
+    )
+    assert find_tracked_packages(config, config_json) == ["google-adk:unknown"]
+
+
+def test_extras_bracket_records_unknown(tmp_path: pathlib.Path) -> None:
+    config, config_json = _write_project(
+        tmp_path,
+        pyproject='dependencies = ["deployments-wrap-sdk[google-adk]>=0.0.1"]',
+    )
+    assert find_tracked_packages(config, config_json) == ["google-adk:unknown"]
+
+
+def test_no_match_returns_empty(tmp_path: pathlib.Path) -> None:
+    config, config_json = _write_project(
+        tmp_path,
+        pyproject='dependencies = ["langgraph>=0.2"]',
+    )
+    assert find_tracked_packages(config, config_json) == []
+
+
+def test_traversal_dep_path_is_skipped(tmp_path: pathlib.Path) -> None:
+    outside = tmp_path.parent / "outside-project"
+    outside.mkdir(exist_ok=True)
+    (outside / "uv.lock").write_text('name = "google-adk"\nversion = "9.9.9"\n')
+    project_root = tmp_path / "project"
+    project_root.mkdir()
+    config = project_root / "langgraph.json"
+    config.write_text("{}")
+    config_json = {"dependencies": ["../outside-project"]}
+    assert find_tracked_packages(config, config_json) == []
+
+
+def test_dep_paths_scanned_in_order(tmp_path: pathlib.Path) -> None:
+    project_root = tmp_path
+    (project_root / "first").mkdir()
+    (project_root / "second").mkdir()
+    (project_root / "second" / "uv.lock").write_text(
+        'name = "google-adk"\nversion = "2.0.0"\n'
+    )
+    config = project_root / "langgraph.json"
+    config.write_text("{}")
+    config_json = {"dependencies": ["first", "second"]}
+    assert find_tracked_packages(config, config_json) == ["google-adk:2.0.0"]
+
+
+def test_non_string_dep_entry_ignored(tmp_path: pathlib.Path) -> None:
+    project_root = tmp_path
+    config = project_root / "langgraph.json"
+    config.write_text("{}")
+    config_json = {"dependencies": [123, None]}
+    assert find_tracked_packages(config, config_json) == []
+
+
+def test_oversized_file_is_truncated_not_raised(tmp_path: pathlib.Path) -> None:
+    project_root = tmp_path
+    config = project_root / "langgraph.json"
+    config.write_text("{}")
+    # 6 MB of irrelevant content followed by the tracked-package marker —
+    # the read cap drops the marker, so nothing should be found.
+    padded = ("x" * (6 * 1024 * 1024)) + '\nname = "google-adk"\nversion = "1.0.0"\n'
+    (project_root / "uv.lock").write_text(padded)
+    assert find_tracked_packages(config, {"dependencies": ["."]}) == []
+
+
+@pytest.mark.parametrize("pkg", TRACKED_PACKAGES)
+def test_every_tracked_package_is_detectable(tmp_path: pathlib.Path, pkg: str) -> None:
+    config, config_json = _write_project(
+        tmp_path,
+        uv_lock=f'name = "{pkg}"\nversion = "1.0.0"\n',
+    )
+    assert find_tracked_packages(config, config_json) == [f"{pkg}:1.0.0"]

libs/cli/tests/unit_tests/test_host_backend.py

@@ -1,3 +1,5 @@
+import json
+
 import httpx
 import pytest
 
@@ -176,6 +178,69 @@ def test_update_deployment_no_secrets(client):
     assert result == {"ok": True}
 
 
+def _capturing_client(captured: dict) -> HostBackendClient:
+    def handler(req: httpx.Request) -> httpx.Response:
+        captured["body"] = req.read()
+        return httpx.Response(200, json={"ok": True})
+
+    c = HostBackendClient("https://api.example.com", "key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "key", "Accept": "application/json"},
+        timeout=30,
+    )
+    return c
+
+
+def test_update_deployment_forwards_tracked_packages():
+    captured: dict = {}
+    c = _capturing_client(captured)
+    c.update_deployment(
+        "dep-123",
+        "image:latest",
+        tracked_packages=["google-adk:1.0.0"],
+    )
+    body = json.loads(captured["body"])
+    assert body["tracked_packages"] == ["google-adk:1.0.0"]
+    assert "tracked_packages" not in body["source_revision_config"]
+
+
+def test_update_deployment_omits_tracked_packages_when_absent():
+    captured: dict = {}
+    c = _capturing_client(captured)
+    c.update_deployment("dep-123", "image:latest")
+    body = json.loads(captured["body"])
+    assert "tracked_packages" not in body
+
+
+def test_update_deployment_internal_source_forwards_tracked_packages():
+    captured: dict = {}
+    c = _capturing_client(captured)
+    c.update_deployment_internal_source(
+        "dep-123",
+        source_tarball_path="path/to/tarball",
+        config_path="langgraph.json",
+        tracked_packages=["google-adk:>=0.5"],
+    )
+    body = json.loads(captured["body"])
+    assert body["tracked_packages"] == ["google-adk:>=0.5"]
+    assert body["source_revision_config"]["source_tarball_path"] == "path/to/tarball"
+    assert "tracked_packages" not in body["source_revision_config"]
+
+
+def test_update_deployment_internal_source_omits_tracked_packages_when_absent():
+    captured: dict = {}
+    c = _capturing_client(captured)
+    c.update_deployment_internal_source(
+        "dep-123",
+        source_tarball_path="path/to/tarball",
+        config_path="langgraph.json",
+    )
+    body = json.loads(captured["body"])
+    assert "tracked_packages" not in body
+
+
 def test_list_revisions(client):
     result = client.list_revisions("dep-123", limit=5)
     assert result == {"ok": True}

libs/langgraph/langgraph/_internal/_config.py

@@ -79,6 +79,34 @@ def patch_checkpoint_map(
         return config
 
 
+def _merge_callbacks(base: Callbacks, new: Callbacks) -> Callbacks:
+    """Merge two callbacks values (None / list / BaseCallbackManager).
+
+    Six cases total (3 base types x 2 non-None new types).
+    """
+    if new is None:
+        return base
+    if base is None:
+        return new.copy() if isinstance(new, (list, BaseCallbackManager)) else new
+    if isinstance(new, list):
+        if isinstance(base, list):
+            return base + new
+        if isinstance(base, BaseCallbackManager):
+            mngr = base.copy()
+            for cb in new:
+                mngr.add_handler(cb, inherit=True)
+            return mngr
+    elif isinstance(new, BaseCallbackManager):
+        if isinstance(base, list):
+            mngr = new.copy()
+            for cb in base:
+                mngr.add_handler(cb, inherit=True)
+            return mngr
+        if isinstance(base, BaseCallbackManager):
+            return base.merge(new)
+    raise NotImplementedError(f"Unsupported callback types: {type(base)}, {type(new)}")
+
+
 def merge_configs(*configs: RunnableConfig | None) -> RunnableConfig:
     """Merge multiple configs into one.
 
@@ -113,34 +141,9 @@ def merge_configs(*configs: RunnableConfig | None) -> RunnableConfig:
                 else:
                     base[key] = value
             elif key == "callbacks":
-                base_callbacks = base.get("callbacks")
-                # callbacks can be either None, list[handler] or manager
-                # so merging two callbacks values has 6 cases
-                if isinstance(value, list):
-                    if base_callbacks is None:
-                        base["callbacks"] = value.copy()
-                    elif isinstance(base_callbacks, list):
-                        base["callbacks"] = base_callbacks + value
-                    else:
-                        # base_callbacks is a manager
-                        mngr = base_callbacks.copy()
-                        for callback in value:
-                            mngr.add_handler(callback, inherit=True)
-                        base["callbacks"] = mngr
-                elif isinstance(value, BaseCallbackManager):
-                    # value is a manager
-                    if base_callbacks is None:
-                        base["callbacks"] = value.copy()
-                    elif isinstance(base_callbacks, list):
-                        mngr = value.copy()
-                        for callback in base_callbacks:
-                            mngr.add_handler(callback, inherit=True)
-                        base["callbacks"] = mngr
-                    else:
-                        # base_callbacks is also a manager
-                        base["callbacks"] = base_callbacks.merge(value)
-                else:
-                    raise NotImplementedError
+                base["callbacks"] = _merge_callbacks(
+                    base.get("callbacks"), cast(Callbacks, value)
+                )
             elif key == "recursion_limit":
                 if config["recursion_limit"] != DEFAULT_RECURSION_LIMIT:
                     base["recursion_limit"] = config["recursion_limit"]
@@ -309,7 +312,40 @@ def ensure_config(*configs: RunnableConfig | None) -> RunnableConfig:
         for k, v in config.items():
             if _is_not_empty(v) and k in CONFIG_KEYS:
                 if k == CONF:
-                    empty[k] = cast(dict, v).copy()
+                    # Shallow-merge configurable dicts across configs so values
+                    # bound via with_config(...) (e.g. ls_agent_type) are
+                    # preserved when later configs (e.g. invoke-time) only
+                    # specify a subset of keys like thread_id.
+                    existing = empty.get(k)
+                    empty[k] = (
+                        {**cast(dict, existing), **cast(dict, v)}
+                        if existing
+                        else cast(dict, v).copy()
+                    )
+                elif k == "callbacks":
+                    empty["callbacks"] = _merge_callbacks(
+                        empty.get("callbacks"), cast(Callbacks, v)
+                    )
+                elif k == "metadata":
+                    # Shallow-merge metadata dicts across configs so values
+                    # bound via with_config(...) (e.g. user_id) are preserved
+                    # when later configs supply other metadata keys.
+                    existing = empty.get("metadata")
+                    empty["metadata"] = (
+                        {**cast(dict, existing), **cast(dict, v)}
+                        if existing
+                        else cast(dict, v).copy()
+                    )
+                elif k == "tags":
+                    # Concatenate tags across configs so values bound via
+                    # with_config(...) are preserved when later configs
+                    # supply additional tags. Matches merge_configs.
+                    existing_tags: list[str] | None = empty.get("tags")
+                    empty["tags"] = (
+                        [*existing_tags, *cast(list, v)]
+                        if existing_tags
+                        else list(cast(list, v))
+                    )
                 else:
                     empty[k] = v  # type: ignore[literal-required]
         for k, v in config.items():
@@ -366,3 +402,17 @@ _PROPAGATE_TO_METADATA = frozenset(
         "graph_id",
     )
 )
+
+
+def filter_to_user_tags(tags: Sequence[str] | None) -> list[str] | None:
+    """Drop langgraph's internal `seq:step:*` bookkeeping tags.
+
+    `seq:step:N` tags are added internally to mark sequence steps; everything
+    else (user-supplied tags and any other framework tags) is kept. Returns the
+    surviving tags, or `None` if none remain. Shared by the `messages` and
+    `tasks` stream handlers so both surface the same tag set on their metadata.
+    """
+    if not tags:
+        return None
+    filtered = [t for t in tags if not t.startswith("seq:step")]
+    return filtered or None

libs/langgraph/langgraph/errors.py

@@ -21,6 +21,7 @@ __all__ = (
     "InvalidUpdateError",
     "GraphBubbleUp",
     "GraphInterrupt",
+    "NodeCancelledError",
     "NodeError",
     "NodeInterrupt",
     "NodeTimeoutError",
@@ -164,6 +165,28 @@ class NodeError:
     """Exception raised by the failed node."""
 
 
+class NodeCancelledError(Exception):
+    """Raised when a node body raises ``asyncio.CancelledError`` itself.
+
+    ``asyncio.CancelledError`` is a ``BaseException`` and the pregel runner
+    treats cancelled task futures as silent tear-down (e.g. when it stops
+    sibling tasks after a peer fails). That is the correct behaviour for
+    *framework-initiated* cancellation, but a user node that raises
+    ``asyncio.CancelledError`` from its own body should surface as a node
+    failure, the same way any other exception would.
+
+    The retry layer converts user-raised ``asyncio.CancelledError`` into this
+    type so it flows through the normal error path and the run reports as
+    ``error`` instead of silently succeeding.
+    """
+
+    node: str
+
+    def __init__(self, node: str, message: str | None = None) -> None:
+        super().__init__(message or f"Node {node!r} raised asyncio.CancelledError")
+        self.node = node
+
+
 class NodeTimeoutError(Exception):
     """Raised when a node invocation exceeds one of its configured timeouts.
 

libs/langgraph/langgraph/pregel/_messages.py

@@ -15,6 +15,7 @@ from langchain_core.messages.utils import convert_to_messages
 from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, LLMResult
 from pydantic import BaseModel
 
+from langgraph._internal._config import filter_to_user_tags
 from langgraph._internal._constants import NS_SEP
 from langgraph.constants import TAG_HIDDEN, TAG_NOSTREAM
 from langgraph.pregel.protocol import StreamChunk
@@ -143,9 +144,8 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             ]
             if not self.subgraphs and len(ns) > 0 and ns != self.parent_ns:
                 return
-            if tags:
-                if filtered_tags := [t for t in tags if not t.startswith("seq:step")]:
-                    metadata["tags"] = filtered_tags
+            if (filtered_tags := filter_to_user_tags(tags)) is not None:
+                metadata["tags"] = filtered_tags
             self.metadata[run_id] = (ns, metadata)
 
     def on_llm_new_token(

libs/langgraph/langgraph/pregel/_remote_run_stream.py

@@ -0,0 +1,374 @@
+# libs/langgraph/langgraph/pregel/_remote_run_stream.py
+from __future__ import annotations
+
+import logging
+import sys
+from collections.abc import AsyncIterator, Iterator, Mapping
+from types import TracebackType
+from typing import Any, cast
+
+from langchain_core.runnables import RunnableConfig
+from langgraph_sdk._async.stream import AsyncThreadStream
+from langgraph_sdk._sync.stream import SyncThreadStream
+from langgraph_sdk.client import LangGraphClient, SyncLangGraphClient
+from langgraph_sdk.stream.decoders import DataDecoder
+
+from langgraph.types import Command
+
+logger = logging.getLogger(__name__)
+
+
+def _translate_command_input(input: Any) -> Any:
+    """Translate a local `Command` into the v3 wire `input`, else passthrough.
+
+    The v3 server decides start-vs-resume from thread state (an interrupted
+    run or pending interrupts) and, on resume, wraps the whole `input` as
+    `{"resume": input}` itself. So a resume `Command` must surface its raw
+    `resume` value as the wire `input` (not the serialized dataclass, which
+    the server would double-wrap). The v3 `run.start` path has no `goto` /
+    `update` channel, so those are rejected.
+
+    `langgraph_sdk` is upstream of `langgraph`, so this `Command`-aware
+    marshalling lives here on the adapter (langgraph) side of the boundary.
+    """
+    if isinstance(input, Command):
+        if input.goto or input.update:
+            raise NotImplementedError(
+                "RemoteGraph v3 streaming supports `Command(resume=...)` only; "
+                "`goto` / `update` are not supported by the v3 `run.start` path."
+            )
+        return input.resume
+    return input
+
+
+class _ChannelProjection:
+    """Decoded projection for a wire channel the SDK doesn't type natively.
+
+    Subscribes to `channel` and decodes each event's `params["data"]` through the
+    SDK's `DataDecoder` — the same decoder the SDK's own plain-payload projections
+    (`values` / `updates` / `checkpoints` / `tasks`) use, which yields the item
+    shape that local's `UpdatesTransformer` / `CheckpointsTransformer` /
+    `TasksTransformer` / `CustomTransformer` push, so iterating this matches the
+    corresponding local projection. Iterate with `for` against a sync stream and
+    `async for` against an async stream (matching the underlying SDK). Opening
+    the subscription requires the stream to be entered (`with` / `async with`).
+    """
+
+    def __init__(self, sdk: AsyncThreadStream | SyncThreadStream, channel: str) -> None:
+        self._sdk = sdk
+        self._channel = channel
+
+    def __iter__(self) -> Iterator[Any]:
+        # Sync lane: the sync adapter's SDK returns a sync iterator here.
+        decoder = DataDecoder(self._channel)
+        events = cast(Iterator[Any], self._sdk.subscribe([self._channel]))
+        for event in events:
+            yield from decoder.feed(event)
+
+    def __aiter__(self) -> AsyncIterator[Any]:
+        return self._aiter()
+
+    async def _aiter(self) -> AsyncIterator[Any]:
+        # Async lane: the async adapter's SDK returns an async iterator here.
+        decoder = DataDecoder(self._channel)
+        events = cast(AsyncIterator[Any], self._sdk.subscribe([self._channel]))
+        async for event in events:
+            for item in decoder.feed(event):
+                yield item
+
+
+class _ProjectionRegistry(Mapping[str, Any]):
+    """Read-only name -> projection registry mirroring local `GraphRunStream.extensions`.
+
+    Resolution follows the langchain-protocol wire channels, and every entry
+    yields the same decoded item shape local does (`params.data`):
+
+    - `values` / `messages` / `tool_calls` / `subgraphs` resolve to the SDK's
+      decoded typed projections. `tool_calls` is the `tools` channel — tool
+      *execution* events, distinct from the tool-call *inputs* inside `messages`.
+    - `updates` / `checkpoints` / `tasks` / `custom` have no typed SDK
+      projection, so they resolve to a `_ChannelProjection` that subscribes to
+      the channel and yields `params.data` — matching the local transformer
+      output for those channels.
+    - any other name is a specific custom-extension channel
+      (`thread.extensions[name]`, i.e. `custom:<name>`).
+
+    `lifecycle` is intentionally absent: local derives a status payload from it
+    rather than yielding `params.data`, and the SDK consumes it as control-plane
+    (driving `output` / `interrupted`), so its shape can't be matched — it
+    remains reachable via the raw `events` iterator. `debug` is absent too: it
+    is not a v3 wire channel.
+    """
+
+    # Channels the SDK decodes into typed projections.
+    _TYPED = ("values", "messages", "tool_calls", "subgraphs")
+    # Wire channels with no typed SDK projection — decoded here to match local.
+    _DECODED = ("updates", "checkpoints", "tasks", "custom")
+    _NATIVE = _TYPED + _DECODED
+
+    def __init__(self, sdk: AsyncThreadStream | SyncThreadStream) -> None:
+        self._sdk = sdk
+
+    def __getitem__(self, name: str) -> Any:
+        if name in self._TYPED:
+            return getattr(self._sdk, name)
+        if name in self._DECODED:
+            return _ChannelProjection(self._sdk, name)
+        return self._sdk.extensions[name]
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self._NATIVE)
+
+    def __len__(self) -> int:
+        return len(self._NATIVE)
+
+
+class _RemoteGraphRunStream:
+    """Sync adapter: SyncThreadStream -> GraphRunStream surface."""
+
+    def __init__(
+        self,
+        *,
+        sync_client: SyncLangGraphClient,
+        sdk_thread: SyncThreadStream,
+        input: Any,
+        config: RunnableConfig | None,
+        metadata: dict[str, Any] | None,
+    ) -> None:
+        self._client = sync_client
+        self._sdk = sdk_thread
+        self._start_kwargs: dict[str, Any] = {
+            "input": _translate_command_input(input),
+            "config": config,
+            "metadata": metadata,
+        }
+        self._run_id: str | None = None
+        self._closed = False
+        self._events_iter: Iterator[Any] | None = None
+
+    def __enter__(self) -> _RemoteGraphRunStream:
+        if self._closed:
+            raise RuntimeError("_RemoteGraphRunStream already closed")
+        self._sdk.__enter__()
+        try:
+            result = self._sdk.run.start(**self._start_kwargs)
+        except BaseException:
+            self._sdk.__exit__(*sys.exc_info())
+            raise
+        self._run_id = result["run_id"]
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        self._sdk.__exit__(exc_type, exc, tb)
+
+    @property
+    def output(self) -> Any:
+        return self._sdk.output
+
+    @property
+    def interrupted(self) -> bool:
+        """Whether the remote run is currently paused at an interrupt.
+
+        Reads the SDK's current value without blocking. This differs from
+        local `GraphRunStream.interrupted`, which drives the run to terminal
+        before returning the flag. Sync callers needing a wait-for-interrupt
+        pattern should switch to the async API and drain a projection.
+        """
+        return self._sdk.interrupted
+
+    @property
+    def interrupts(self) -> list[Any]:
+        """Current outstanding interrupt payloads (non-blocking snapshot)."""
+        return list(self._sdk.interrupts)
+
+    @property
+    def values(self) -> Any:
+        """Live state-snapshot projection (mirrors local `run.values`)."""
+        return self._sdk.values
+
+    @property
+    def messages(self) -> Any:
+        """Live message-stream projection (mirrors local `run.messages`)."""
+        return self._sdk.messages
+
+    @property
+    def subgraphs(self) -> Any:
+        """Subgraph-handle projection (mirrors local `run.subgraphs`)."""
+        return self._sdk.subgraphs
+
+    @property
+    def tool_calls(self) -> Any:
+        """Tool-execution projection (the `tools` channel).
+
+        These are tool *execution* events (started / output / finished),
+        distinct from the tool-call *inputs* carried inside `messages`.
+        """
+        return self._sdk.tool_calls
+
+    @property
+    def extensions(self) -> Mapping[str, Any]:
+        """Name -> projection registry (mirrors local `run.extensions`)."""
+        return _ProjectionRegistry(self._sdk)
+
+    def abort(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if self._run_id is not None:
+            try:
+                self._client.runs.cancel(self._sdk.thread_id, self._run_id, wait=False)
+            except Exception:
+                logger.debug("abort: runs.cancel failed", exc_info=True)
+        try:
+            self._sdk.close()
+        except Exception:
+            logger.debug("abort: sdk.close failed", exc_info=True)
+
+    def __iter__(self) -> Iterator[Any]:
+        if self._events_iter is None:
+            self._events_iter = iter(self._sdk.events)
+        return self._events_iter
+
+    def interleave(self, *names: str) -> Iterator[tuple[str, Any]]:
+        yield from self._sdk.interleave_projections(list(names))
+
+
+class _AsyncRemoteGraphRunStream:
+    """Async adapter: AsyncThreadStream -> AsyncGraphRunStream surface."""
+
+    def __init__(
+        self,
+        *,
+        client: LangGraphClient,
+        sdk_thread: AsyncThreadStream,
+        input: Any,
+        config: RunnableConfig | None,
+        metadata: dict[str, Any] | None,
+    ) -> None:
+        self._client = client
+        self._sdk = sdk_thread
+        self._start_kwargs: dict[str, Any] = {
+            "input": _translate_command_input(input),
+            "config": config,
+            "metadata": metadata,
+        }
+        self._run_id: str | None = None
+        self._closed = False
+        self._events_aiter: AsyncIterator[Any] | None = None
+
+    async def __aenter__(self) -> _AsyncRemoteGraphRunStream:
+        if self._closed:
+            raise RuntimeError("_AsyncRemoteGraphRunStream already closed")
+        await self._sdk.__aenter__()
+        try:
+            result = await self._sdk.run.start(**self._start_kwargs)
+        except BaseException:
+            await self._sdk.__aexit__(*sys.exc_info())
+            raise
+        self._run_id = result["run_id"]
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        await self._sdk.__aexit__(exc_type, exc, tb)
+
+    async def output(self) -> Any:
+        """Drive the remote run to completion and return the final state.
+
+        Awaits the SDK's terminal-state awaitable, matching local
+        `AsyncGraphRunStream.output()` (a method, not a property, so
+        `run.output` without `await` fails at type-check time rather than
+        silently yielding a coroutine).
+        """
+        return await self._sdk.output
+
+    async def interrupted(self) -> bool:
+        """Whether the remote run is currently paused at an interrupt.
+
+        Reads the SDK's current value without blocking. This differs from
+        local `AsyncGraphRunStream.interrupted()`, which drives the run to
+        terminal before returning the flag. Callers that need a
+        wait-for-interrupt pattern should drain a projection (e.g.,
+        `async for snap in stream._sdk.values`) until the SDK's paused
+        sentinel fires, then call this method.
+        """
+        return self._sdk.interrupted
+
+    async def interrupts(self) -> list[Any]:
+        """Current outstanding interrupt payloads.
+
+        Non-blocking; reads the SDK's current snapshot. See `interrupted`
+        for the divergence from local v3 semantics.
+        """
+        return list(self._sdk.interrupts)
+
+    @property
+    def values(self) -> Any:
+        """Live state-snapshot projection (mirrors local `run.values`)."""
+        return self._sdk.values
+
+    @property
+    def messages(self) -> Any:
+        """Live message-stream projection (mirrors local `run.messages`)."""
+        return self._sdk.messages
+
+    @property
+    def subgraphs(self) -> Any:
+        """Subgraph-handle projection (mirrors local `run.subgraphs`)."""
+        return self._sdk.subgraphs
+
+    @property
+    def tool_calls(self) -> Any:
+        """Tool-execution projection (the `tools` channel).
+
+        These are tool *execution* events (started / output / finished),
+        distinct from the tool-call *inputs* carried inside `messages`.
+        """
+        return self._sdk.tool_calls
+
+    @property
+    def extensions(self) -> Mapping[str, Any]:
+        """Name -> projection registry (mirrors local `run.extensions`)."""
+        return _ProjectionRegistry(self._sdk)
+
+    async def abort(self) -> None:
+        if self._closed:
+            return
+        self._closed = True
+        if self._run_id is not None:
+            try:
+                await self._client.runs.cancel(
+                    self._sdk.thread_id, self._run_id, wait=False
+                )
+            except Exception:
+                logger.debug("abort: runs.cancel failed", exc_info=True)
+        try:
+            await self._sdk.close()
+        except Exception:
+            logger.debug("abort: sdk.close failed", exc_info=True)
+
+    def __aiter__(self) -> AsyncIterator[Any]:
+        if self._events_aiter is None:
+            self._events_aiter = self._sdk.events.__aiter__()
+        return self._events_aiter
+
+    # Note: deliberately no `interleave()` on the async adapter. Local
+    # `AsyncGraphRunStream` doesn't have one either (async callers compose
+    # with `asyncio.gather` / `asyncio.as_completed`). The sync adapter
+    # provides `interleave()` because sync callers have no comparable
+    # primitive for iterating multiple iterators concurrently.

libs/langgraph/langgraph/pregel/_retry.py

@@ -37,13 +37,23 @@ from langgraph._internal._constants import (
 )
 from langgraph._internal._runnable import create_task_in_config_context
 from langgraph._internal._timeout import sync_timeout_unsupported
-from langgraph.errors import GraphBubbleUp, NodeTimeoutError, ParentCommand
+from langgraph.errors import (
+    GraphBubbleUp,
+    NodeCancelledError,
+    NodeTimeoutError,
+    ParentCommand,
+)
 from langgraph.pregel.protocol import StreamProtocol
 from langgraph.runtime import ExecutionInfo, Runtime
 from langgraph.types import Command, PregelExecutableTask, RetryPolicy, TimeoutPolicy
 
 logger = logging.getLogger(__name__)
 SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11)
+# `asyncio.Task.cancelling()` was added in Python 3.11. It reports the number of
+# pending cancel requests on the task: ``0`` means no external code asked us to
+# cancel — so a ``CancelledError`` observed here was raised by the task body
+# itself (the user's node) rather than by pregel cancelling sibling tasks.
+SUPPORTS_TASK_CANCELLING = sys.version_info >= (3, 11)
 
 
 def _timeout_secs(value: float | timedelta) -> float:
@@ -302,6 +312,28 @@ class _IdleProgressCallbackHandler(BaseCallbackHandler):
     on_custom_event = _touch
 
 
+def _is_user_raised_cancelled() -> bool:
+    """Return True if the in-flight ``CancelledError`` came from the task body.
+
+    Pregel cancels sibling tasks via ``task.cancel()`` when a peer fails, which
+    increments ``asyncio.Task.cancelling()`` on the target before the cancel
+    actually fires. A user node that calls ``raise asyncio.CancelledError()``
+    from inside its own body raises while ``cancelling() == 0``, which is the
+    signal we use to convert the exception into a regular
+    :class:`NodeCancelledError`.
+
+    Returns ``False`` when we can't tell (``cancelling()`` unavailable, or no
+    current task — neither should happen in practice from ``arun_with_retry``)
+    so framework-initiated cancellation continues to propagate unchanged.
+    """
+    if not SUPPORTS_TASK_CANCELLING:
+        return False
+    current = asyncio.current_task()
+    if current is None:
+        return False
+    return current.cancelling() == 0
+
+
 def _drain_cancelled(task: asyncio.Task[Any]) -> None:
     # Mark the abandoned task's exception as retrieved so asyncio doesn't log it.
     with suppress(asyncio.CancelledError):
@@ -600,6 +632,12 @@ def run_with_retry(
         except GraphBubbleUp:
             # if interrupted, end
             raise
+        except asyncio.CancelledError as exc:
+            # A sync node has no asyncio context, so any ``CancelledError`` that
+            # reaches here was raised by the node body itself. Surface it as a
+            # regular exception so the pregel runner panics the run instead of
+            # treating the task as a silent tear-down (LSD-1507).
+            raise NodeCancelledError(task.name) from exc
         except Exception as exc:
             if SUPPORTS_EXC_NOTES:
                 exc.add_note(f"During task with name '{task.name}' and id '{task.id}'")
@@ -736,6 +774,24 @@ async def arun_with_retry(
             # if interrupted, end
             _finish_timed_attempt(config, attempt_ctx)
             raise
+        except asyncio.CancelledError as exc:
+            # ``CancelledError`` reaches us in two very different shapes:
+            #   1. Pregel cancelled this task because a sibling failed
+            #      (``asyncio.Task.cancelling() >= 1``). The framework already
+            #      knows the run is failing and we must let cancellation
+            #      propagate so the watchdog/cleanup code in the runner sees a
+            #      cancelled future.
+            #   2. The node body itself raised ``asyncio.CancelledError`` (
+            #      ``cancelling() == 0``). The runner would otherwise treat
+            #      this as silent tear-down and the run would report
+            #      ``success`` even though the node failed (LSD-1507). Convert
+            #      it into :class:`NodeCancelledError` so it follows the same
+            #      path as any other node failure.
+            if _is_user_raised_cancelled():
+                _finish_timed_attempt(config, attempt_ctx, exc)
+                raise NodeCancelledError(task.name) from exc
+            _finish_timed_attempt(config, attempt_ctx, exc)
+            raise
         except Exception as exc:
             _finish_timed_attempt(config, attempt_ctx, exc)
             if SUPPORTS_EXC_NOTES:

libs/langgraph/langgraph/pregel/debug.py

@@ -6,9 +6,13 @@ from typing import Any
 from uuid import UUID
 
 from langchain_core.runnables import RunnableConfig
-from langgraph.checkpoint.base import CheckpointMetadata, PendingWrite
+from langgraph.checkpoint.base import (
+    EXCLUDED_METADATA_KEYS,
+    CheckpointMetadata,
+    PendingWrite,
+)
 
-from langgraph._internal._config import patch_checkpoint_map
+from langgraph._internal._config import filter_to_user_tags, patch_checkpoint_map
 from langgraph._internal._constants import (
     CONF,
     CONFIG_KEY_CHECKPOINT_NS,
@@ -40,12 +44,31 @@ def map_debug_tasks(tasks: Iterable[PregelExecutableTask]) -> Iterator[TaskPaylo
         if task.config is not None and TAG_HIDDEN in task.config.get("tags", []):
             continue
 
-        yield {
+        payload: TaskPayload = {
             "id": task.id,
             "name": task.name,
             "input": task.input,
             "triggers": task.triggers,
         }
+        # Forward user-meaningful metadata only — drop langgraph's internal
+        # framework keys (langgraph_node/step/triggers/path/checkpoint_ns,
+        # thread_id, ...), which are redundant with the task's own fields and
+        # namespace. Keys like `lc_agent_name`, `ls_integration`, and any
+        # user-supplied metadata ride along. Filtered config tags are folded in
+        # under `tags`, mirroring the messages stream handler. (The comprehension
+        # also yields a fresh dict, so mutating `md` doesn't touch task.config.)
+        if task.config is not None:
+            md = {
+                k: v
+                for k, v in (task.config.get("metadata") or {}).items()
+                if k not in EXCLUDED_METADATA_KEYS
+            }
+            filtered_tags = filter_to_user_tags(task.config.get("tags"))
+            if filtered_tags is not None:
+                md["tags"] = filtered_tags
+            if md:
+                payload["metadata"] = md
+        yield payload
 
 
 def is_multiple_channel_write(value: Any) -> bool:

libs/langgraph/langgraph/pregel/remote.py

@@ -55,6 +55,10 @@ from langgraph._internal._constants import (
     NS_SEP,
 )
 from langgraph.errors import GraphInterrupt, ParentCommand
+from langgraph.pregel._remote_run_stream import (
+    _AsyncRemoteGraphRunStream,
+    _RemoteGraphRunStream,
+)
 from langgraph.pregel.protocol import PregelProtocol, StreamProtocol
 from langgraph.types import (
     All,
@@ -80,6 +84,8 @@ _CONF_DROPLIST = frozenset(
     ),
 )
 
+_V3_SUPPORTED_KWARGS = frozenset({"metadata", "headers"})
+
 
 def _sanitize_config_value(v: Any) -> Any:
     """Recursively sanitize a config value to ensure it contains only primitives."""
@@ -186,6 +192,34 @@ class RemoteGraph(PregelProtocol):
             )
         return self.sync_client
 
+    def _reject_v3_unsupported(
+        self,
+        *,
+        control: Any,
+        transformers: Any,
+        interrupt_before: Any,
+        interrupt_after: Any,
+        extra_kwargs: dict[str, Any],
+    ) -> None:
+        """Raise NotImplementedError for kwargs unsupported by the v3 streaming path."""
+        for name, value in (
+            ("control", control),
+            ("transformers", transformers),
+            ("interrupt_before", interrupt_before),
+            ("interrupt_after", interrupt_after),
+        ):
+            if value:
+                raise NotImplementedError(
+                    f"RemoteGraph.stream_events(version='v3') does not support `{name}=`."
+                )
+        unknown = set(extra_kwargs) - _V3_SUPPORTED_KWARGS
+        if unknown:
+            raise NotImplementedError(
+                f"RemoteGraph.stream_events(version='v3') does not support "
+                f"the following kwargs: {sorted(unknown)!r}. "
+                f"Supported: {sorted(_V3_SUPPORTED_KWARGS)!r}."
+            )
+
     def copy(self, update: dict[str, Any]) -> Self:
         attrs = {**self.__dict__, **update}
         return self.__class__(attrs.pop("assistant_id"), **attrs)
@@ -996,21 +1030,104 @@ class RemoteGraph(PregelProtocol):
             else:
                 yield chunk
 
+    def stream_events(
+        self,
+        input: Any,
+        config: RunnableConfig | None = None,
+        *,
+        version: Literal["v1", "v2", "v3"] = "v2",
+        interrupt_before: All | Sequence[str] | None = None,
+        interrupt_after: All | Sequence[str] | None = None,
+        control: Any = None,
+        transformers: Sequence[Any] | None = None,
+        headers: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        """Stream events from this remote graph.
+
+        For `version="v3"`, returns a `_RemoteGraphRunStream` whose surface
+        matches the local `GraphRunStream`. For other versions, delegates to
+        `Runnable.stream_events`.
+        """
+        if version != "v3":
+            return super().stream_events(input, config, version=version, **kwargs)
+        self._reject_v3_unsupported(
+            control=control,
+            transformers=transformers,
+            interrupt_before=interrupt_before,
+            interrupt_after=interrupt_after,
+            extra_kwargs=kwargs,
+        )
+        sync_client = self._validate_sync_client()
+        sanitized = self._sanitize_config(merge_configs(self.config, config))
+        thread_id = sanitized.get("configurable", {}).pop("thread_id", None)
+        merged_headers = (
+            _merge_tracing_headers(headers) if self.distributed_tracing else headers
+        )
+        sdk_thread = sync_client.threads.stream(
+            thread_id=thread_id,
+            assistant_id=self.assistant_id,
+            headers=merged_headers,
+        )
+        return _RemoteGraphRunStream(
+            sync_client=sync_client,
+            sdk_thread=sdk_thread,
+            input=input,
+            config=sanitized,
+            metadata=kwargs.get("metadata"),
+        )
+
     async def astream_events(
         self,
         input: Any,
         config: RunnableConfig | None = None,
         *,
-        version: Literal["v1", "v2"],
-        include_names: Sequence[All] | None = None,
-        include_types: Sequence[All] | None = None,
-        include_tags: Sequence[All] | None = None,
-        exclude_names: Sequence[All] | None = None,
-        exclude_types: Sequence[All] | None = None,
-        exclude_tags: Sequence[All] | None = None,
+        version: Literal["v1", "v2", "v3"] = "v2",
+        interrupt_before: All | Sequence[str] | None = None,
+        interrupt_after: All | Sequence[str] | None = None,
+        control: Any = None,
+        transformers: Sequence[Any] | None = None,
+        headers: dict[str, str] | None = None,
         **kwargs: Any,
-    ) -> AsyncIterator[dict[str, Any]]:
-        raise NotImplementedError
+    ) -> Any:
+        """Async-stream events from this remote graph.
+
+        For `version="v3"`, awaits to an `_AsyncRemoteGraphRunStream`, matching
+        the local `Pregel.astream_events(version="v3")` awaitable contract:
+        `async with await rg.astream_events(..., version="v3") as run`. For
+        `version="v1"`/`"v2"`, raises NotImplementedError (use `astream`).
+        """
+        if version != "v3":
+            raise NotImplementedError(
+                f"RemoteGraph.astream_events(version={version!r}) is not "
+                "implemented; use astream() for v1/v2 streaming or "
+                "version='v3'."
+            )
+        self._reject_v3_unsupported(
+            control=control,
+            transformers=transformers,
+            interrupt_before=interrupt_before,
+            interrupt_after=interrupt_after,
+            extra_kwargs=kwargs,
+        )
+        client = self._validate_client()
+        sanitized = self._sanitize_config(merge_configs(self.config, config))
+        thread_id = sanitized.get("configurable", {}).pop("thread_id", None)
+        merged_headers = (
+            _merge_tracing_headers(headers) if self.distributed_tracing else headers
+        )
+        sdk_thread = client.threads.stream(
+            thread_id=thread_id,
+            assistant_id=self.assistant_id,
+            headers=merged_headers,
+        )
+        return _AsyncRemoteGraphRunStream(
+            client=client,
+            sdk_thread=sdk_thread,
+            input=input,
+            config=sanitized,
+            metadata=kwargs.get("metadata"),
+        )
 
     @overload
     def invoke(

libs/langgraph/langgraph/stream/_types.py

@@ -35,7 +35,7 @@ class ProtocolEvent(TypedDict):
     """
 
     type: Literal["event"]
-    eventId: NotRequired[str]
+    event_id: NotRequired[str]  # snake_case to match the langchain-protocol wire field
     seq: NotRequired[int]
     method: str  # StreamMode value: "values", "messages", "custom", etc.
     params: _ProtocolEventParams

libs/langgraph/langgraph/stream/transformers.py

@@ -9,7 +9,7 @@ from langchain_core.language_models.chat_model_stream import (
     ChatModelStream,
 )
 from langchain_core.messages import AIMessageChunk, BaseMessage, ToolMessage
-from langchain_protocol.protocol import MessagesData
+from langchain_protocol.protocol import LifecycleCause, MessagesData
 from typing_extensions import NotRequired, TypedDict
 
 from langgraph.errors import GraphDrained, GraphInterrupt
@@ -366,6 +366,7 @@ class LifecyclePayload(TypedDict, total=False):
     namespace: list[str]
     graph_name: NotRequired[str]
     trigger_call_id: NotRequired[str]
+    cause: NotRequired[LifecycleCause]
     error: NotRequired[str]
 
 
@@ -406,6 +407,18 @@ class _TasksLifecycleBase(StreamTransformer):
         # Maps tracked namespace -> task_id of the parent task whose
         # `TaskResultPayload` will close it.
         self._open: dict[tuple[str, ...], str] = {}
+        # lc_agent_name observed at each namespace (first task event wins).
+        # Not read by the base discriminator (which only checks whether the
+        # current task carries an lc_agent_name); maintained as extension state
+        # for subclasses that project named subagents — e.g. a `run.subagents`
+        # transformer reads this to filter to nested runs that have a name.
+        self._lc_by_ns: dict[tuple[str, ...], str | None] = {}
+        # Pregel task_id -> triggering LLM tool_call_id, harvested from a task
+        # whose `input` is a `tool_call_with_context` dict (current shape) or a
+        # list of tool-call dicts (legacy shape). The child subgraph's segment
+        # `node:<task_id>` shares this task_id, so a subagent recovers the tool
+        # call that spawned it (cross-payload).
+        self._pending_tool_calls: dict[str, str] = {}
 
     # --- Template-method hooks (subclass overrides) ---
 
@@ -418,6 +431,8 @@ class _TasksLifecycleBase(StreamTransformer):
         ns: tuple[str, ...],
         graph_name: str | None,
         trigger_call_id: str | None,
+        *,
+        cause: LifecycleCause | None = None,
     ) -> None:
         """Fired once per discovered namespace (first observed task event)."""
         raise NotImplementedError
@@ -443,18 +458,81 @@ class _TasksLifecycleBase(StreamTransformer):
         if "result" in data:
             self._handle_task_result(ns, data)
         else:
-            self._handle_task_start(ns)
+            self._record_identity(ns, data)
+            self._record_pending_tool_calls(data)
+            self._handle_task_start(ns, data)
         # Tasks events are folded into the synthesized projections;
         # suppress from the main event log so iterators don't double-see
         # the same information in two shapes.
         return False
 
-    def _handle_task_start(self, ns: tuple[str, ...]) -> None:
+    def _record_identity(self, ns: tuple[str, ...], data: dict[str, Any]) -> None:
+        """Record this namespace's `lc_agent_name` (first task event wins).
+
+        Runs for every task-start event, including `ns == self.scope` and
+        tracked children. Pregel emits parent-namespace tasks before
+        child-namespace tasks, so under that ordering the parent's identity is
+        recorded by the time a child event is evaluated in `_handle_task_start`.
+        """
+        if ns in self._lc_by_ns:
+            return
+        metadata = data.get("metadata") or {}
+        self._lc_by_ns[ns] = metadata.get("lc_agent_name")
+
+    def _record_pending_tool_calls(self, data: dict[str, Any]) -> None:
+        """Harvest a task's triggering tool_call_id keyed by its task id.
+
+        A tool-dispatch task seeds `task_id -> tool_call_id`; the spawned
+        subgraph's namespace segment `node:<task_id>` shares that id, letting
+        a subagent recover the tool call that caused it across payloads. Two
+        input shapes are handled: the current Pregel push model schedules each
+        tool call as its own task whose `input` is a `tool_call_with_context`
+        dict, while a legacy / batched model passes a list of tool-call dicts.
+        """
+        task_id = data.get("id")
+        if not isinstance(task_id, str):
+            return
+        payload = data.get("input")
+        tool_call_id: str | None = None
+        # Current langgraph schedules each tool call as its own push task
+        # whose input is a `tool_call_with_context` dict.
+        if isinstance(payload, dict) and isinstance(payload.get("tool_call"), dict):
+            candidate = payload["tool_call"].get("id")
+            if isinstance(candidate, str):
+                tool_call_id = candidate
+        # Legacy / batched shape: input is a list of tool-call dicts.
+        elif isinstance(payload, list):
+            for tc in payload:
+                if isinstance(tc, dict) and isinstance(tc.get("id"), str):
+                    tool_call_id = tc["id"]  # first wins
+                    break
+        if tool_call_id is not None:
+            self._pending_tool_calls[task_id] = tool_call_id
+
+    def _handle_task_start(self, ns: tuple[str, ...], data: dict[str, Any]) -> None:
         if not self._should_track(ns) or ns in self._seen:
             return
         self._seen.add(ns)
-        graph_name, trigger_call_id = _parse_ns_segment(ns[-1])
-        self._on_started(ns, graph_name or None, trigger_call_id)
+        parsed_name, trigger_call_id = _parse_ns_segment(ns[-1])
+        metadata = data.get("metadata") or {}
+        child_lc = metadata.get("lc_agent_name")
+        # A subagent boundary is any nested run carrying an lc_agent_name (set
+        # by create_agent). Unnamed runs (lc_agent_name None) are excluded.
+        #
+        # A same-named nested agent — e.g. a subagent that invokes itself — is
+        # surfaced because it re-asserts its own lc_agent_name. The trade-off:
+        # a non-agent subgraph invoked inside a tool inherits the parent's
+        # lc_agent_name and will also surface (named after the parent). A caller
+        # that needs to exclude such a graph can null lc_agent_name in the
+        # config it invokes that graph with.
+        is_subagent = child_lc is not None
+        graph_name = child_lc if is_subagent else (parsed_name or None)
+        cause: LifecycleCause | None = None
+        if is_subagent and trigger_call_id is not None:
+            tool_call_id = self._pending_tool_calls.get(trigger_call_id)
+            if tool_call_id:
+                cause = {"type": "toolCall", "tool_call_id": str(tool_call_id)}
+        self._on_started(ns, graph_name, trigger_call_id, cause=cause)
         if trigger_call_id is not None:
             self._open[ns] = trigger_call_id
 
@@ -553,6 +631,8 @@ class LifecycleTransformer(_TasksLifecycleBase):
         ns: tuple[str, ...],
         graph_name: str | None,
         trigger_call_id: str | None,
+        *,
+        cause: LifecycleCause | None = None,
     ) -> None:
         if trigger_call_id is None:
             # Without a task id we can't correlate a parent-result
@@ -563,6 +643,8 @@ class LifecycleTransformer(_TasksLifecycleBase):
         if graph_name:
             payload["graph_name"] = graph_name
         payload["trigger_call_id"] = trigger_call_id
+        if cause is not None:
+            payload["cause"] = cause
         self._channel.push(payload)
 
     def _on_terminal(
@@ -625,6 +707,8 @@ class SubgraphTransformer(_TasksLifecycleBase):
         ns: tuple[str, ...],
         graph_name: str | None,
         trigger_call_id: str | None,
+        *,
+        cause: LifecycleCause | None = None,
     ) -> None:
         if self._mux is None:
             return
@@ -633,6 +717,10 @@ class SubgraphTransformer(_TasksLifecycleBase):
         except RuntimeError:
             return
         handle_cls = AsyncSubgraphRunStream if child_mux.is_async else SubgraphRunStream
+        # `cause` is intentionally ignored here: it is a wire/lifecycle-channel
+        # concern (carried on `LifecyclePayload`), not something the in-process
+        # subgraph navigation handle exposes. The argument is accepted only to
+        # keep the `_on_started` template signature uniform across transformers.
         handle = handle_cls(
             mux=child_mux,
             path=ns,
@@ -737,7 +825,12 @@ class SubgraphTransformer(_TasksLifecycleBase):
                 for child_ns, status, error in self._pop_terminal_transitions(ns, data):
                     await self._aon_terminal(child_ns, status, error)
             else:
-                self._handle_task_start(ns)
+                # Mirror the sync `process` bookkeeping so the async lane
+                # observes parent identity / tool calls before discriminating
+                # a subagent boundary.
+                self._record_identity(ns, data)
+                self._record_pending_tool_calls(data)
+                self._handle_task_start(ns, data)
             keep = False
         else:
             keep = True

libs/langgraph/langgraph/types.py

@@ -150,6 +150,16 @@ class TaskPayload(TypedDict):
     """Input data passed to the task."""
     triggers: list[str]
     """List of triggers that caused this task to be executed (e.g. channel writes)."""
+    metadata: NotRequired[dict[str, Any]]
+    """Framework-resolved metadata associated with the task.
+
+    Generic dict carrier following the messages-stream pattern. Populated by
+    `map_debug_tasks` from `task.config["metadata"]` when non-empty, so the
+    same keys `stream_mode="messages"` consumers see (e.g. `lc_agent_name`,
+    `langgraph_node`, `langgraph_step`) are available to stream transformers.
+
+    Consumers should ignore unrecognized keys.
+    """
 
 
 class TaskResultPayload(TypedDict):

libs/langgraph/pyproject.toml

@@ -26,7 +26,7 @@ classifiers = [
 dependencies = [
     "langchain-core>=1.4.0,<2",
     "langgraph-checkpoint>=4.1.0,<5.0.0",
-    "langgraph-sdk>=0.3.0,<0.4.0",
+    "langgraph-sdk>=0.4.1,<0.5.0",
     "langgraph-prebuilt>=1.1.0,<1.2.0",
     "xxhash>=3.5.0",
     "pydantic>=2.7.4",

libs/langgraph/tests/test_config_async.py

@@ -1,7 +1,8 @@
 import pytest
-from langchain_core.callbacks import AsyncCallbackManager
+from langchain_core.callbacks import AsyncCallbackManager, BaseCallbackHandler
 
 from langgraph._internal._config import get_async_callback_manager_for_config
+from langgraph.graph import StateGraph
 
 pytestmark = pytest.mark.anyio
 
@@ -17,3 +18,99 @@ def test_new_async_manager_merges_tags_with_config() -> None:
     config = {"callbacks": None, "tags": ["a"]}
     manager = get_async_callback_manager_for_config(config, tags=["b"])
     assert manager.inheritable_tags == ["a", "b"]
+
+
+class _TrackingCallback(BaseCallbackHandler):
+    def __init__(self) -> None:
+        self.called = False
+
+    def on_chain_start(self, *args, **kwargs) -> None:  # noqa: ANN002, ANN003
+        self.called = True
+
+
+async def test_with_config_callbacks_preserved_in_astream_events() -> None:
+    """A callback bound via .with_config(...) must survive when
+    astream_events injects its own internal callback handler.
+
+    Pre-fix: ensure_config overwrites the callbacks key, dropping the
+    bound handler. Post-fix: the handler list is merged.
+    """
+    builder = StateGraph(dict)
+    builder.add_node("node", lambda state: state)
+    builder.add_edge("__start__", "node")
+    cb = _TrackingCallback()
+    graph = builder.compile().with_config({"callbacks": [cb]})
+    async for _ in graph.astream_events({}, version="v2"):
+        pass
+    assert cb.called, "user-bound callback was dropped by ensure_config overwrite"
+
+
+async def test_with_config_configurable_preserved_on_invoke() -> None:
+    """A configurable key bound via .with_config(...) must survive when
+    invoke-time config supplies a different configurable key.
+
+    Pre-fix: ensure_config overwrites the entire configurable dict.
+    Post-fix: the two dicts are shallow-merged per key.
+    """
+    builder = StateGraph(dict)
+    captured: dict = {}
+
+    def node(state, config):  # noqa: ANN001
+        captured.update(config.get("configurable") or {})
+        return state
+
+    builder.add_node("node", node)
+    builder.add_edge("__start__", "node")
+    graph = builder.compile().with_config({"configurable": {"ls_agent_type": "root"}})
+    await graph.ainvoke({}, {"configurable": {"thread_id": "T1"}})
+    assert captured.get("ls_agent_type") == "root", (
+        "bound configurable key was dropped by ensure_config overwrite"
+    )
+    assert captured.get("thread_id") == "T1", "invoke-time key not present"
+
+
+async def test_with_config_metadata_preserved_on_invoke() -> None:
+    """A metadata key bound via .with_config(...) must survive when
+    invoke-time config supplies a different metadata key.
+
+    Pre-fix: ensure_config overwrites the entire metadata dict.
+    Post-fix: the two dicts are shallow-merged per key.
+    """
+    builder = StateGraph(dict)
+    captured: dict = {}
+
+    def node(state, config):  # noqa: ANN001
+        captured.update(config.get("metadata") or {})
+        return state
+
+    builder.add_node("node", node)
+    builder.add_edge("__start__", "node")
+    graph = builder.compile().with_config({"metadata": {"user_id": "U1"}})
+    await graph.ainvoke({}, {"metadata": {"correlation_id": "C1"}})
+    assert captured.get("user_id") == "U1", (
+        "bound metadata key was dropped by ensure_config overwrite"
+    )
+    assert captured.get("correlation_id") == "C1", "invoke-time key not present"
+
+
+async def test_with_config_tags_preserved_on_invoke() -> None:
+    """Tags bound via .with_config(...) must survive when invoke-time
+    config supplies its own tags.
+
+    Pre-fix: ensure_config overwrites the entire tags list.
+    Post-fix: tags are concatenated (matching merge_configs behavior;
+    no deduplication, no sorting).
+    """
+    builder = StateGraph(dict)
+    captured: list = []
+
+    def node(state, config):  # noqa: ANN001
+        captured.extend(config.get("tags") or [])
+        return state
+
+    builder.add_node("node", node)
+    builder.add_edge("__start__", "node")
+    graph = builder.compile().with_config({"tags": ["bound"]})
+    await graph.ainvoke({}, {"tags": ["invoke"]})
+    assert "bound" in captured, "bound tag was dropped by ensure_config overwrite"
+    assert "invoke" in captured, "invoke-time tag not present"

libs/langgraph/tests/test_pregel_debug.py

@@ -0,0 +1,185 @@
+"""Tests for langgraph.pregel.debug helpers."""
+
+from __future__ import annotations
+
+from langgraph.pregel.debug import map_debug_tasks
+
+
+class _FakeTask:
+    """Minimal stand-in for PregelExecutableTask covering only what map_debug_tasks reads."""
+
+    def __init__(
+        self,
+        *,
+        id: str,
+        name: str,
+        input: object,
+        triggers: list[str],
+        config: dict | None,
+    ) -> None:
+        self.id = id
+        self.name = name
+        self.input = input
+        self.triggers = triggers
+        self.config = config
+
+
+def test_map_debug_tasks_forwards_metadata_when_present() -> None:
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={"metadata": {"lc_agent_name": "weather_agent"}},
+    )
+    payloads = list(map_debug_tasks([task]))
+    assert len(payloads) == 1
+    payload = payloads[0]
+    assert payload["id"] == "t1"
+    assert payload["name"] == "tools"
+    assert payload["metadata"] == {"lc_agent_name": "weather_agent"}
+
+
+def test_map_debug_tasks_omits_metadata_when_empty() -> None:
+    # Empty metadata dict in config: don't include metadata in the payload.
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={"metadata": {}},
+    )
+    payloads = list(map_debug_tasks([task]))
+    assert "metadata" not in payloads[0]
+
+
+def test_map_debug_tasks_omits_metadata_when_absent() -> None:
+    # No metadata key in config: don't include metadata in the payload.
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={},
+    )
+    payloads = list(map_debug_tasks([task]))
+    assert "metadata" not in payloads[0]
+
+
+def test_map_debug_tasks_handles_none_config() -> None:
+    # task.config can be None; should not crash.
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config=None,
+    )
+    payloads = list(map_debug_tasks([task]))
+    assert len(payloads) == 1
+    assert "metadata" not in payloads[0]
+
+
+def test_map_debug_tasks_filters_framework_metadata_keys() -> None:
+    """Internal framework keys are dropped from the forwarded metadata; only
+    user-meaningful keys (lc_agent_name, ls_integration, user metadata) ride
+    along. The framework keys (langgraph_*, thread_id, checkpoint_*) are
+    redundant with the task's own fields/namespace.
+    """
+    md = {
+        "lc_agent_name": "weather_agent",
+        "ls_integration": "langchain_create_agent",
+        "my_user_key": "x",
+        "thread_id": "thread-1",
+        "langgraph_step": 1,
+        "langgraph_node": "tools",
+        "langgraph_path": ("__pregel_pull", "tools"),
+        "langgraph_checkpoint_ns": "tools:abc",
+        "checkpoint_ns": "",
+    }
+    task = _FakeTask(
+        id="t1", name="tools", input=[], triggers=["x"], config={"metadata": md}
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert payload["metadata"] == {
+        "lc_agent_name": "weather_agent",
+        "ls_integration": "langchain_create_agent",
+        "my_user_key": "x",
+    }
+
+
+def test_map_debug_tasks_omits_metadata_when_only_framework_keys() -> None:
+    """A task whose metadata is entirely framework keys (e.g. a plain
+    StateGraph node) yields no `metadata` key after filtering.
+    """
+    md = {
+        "thread_id": "thread-1",
+        "langgraph_step": 1,
+        "langgraph_node": "worker",
+        "langgraph_checkpoint_ns": "worker:abc",
+        "checkpoint_ns": "",
+    }
+    task = _FakeTask(
+        id="t1", name="worker", input=[], triggers=["x"], config={"metadata": md}
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert "metadata" not in payload
+
+
+def test_map_debug_tasks_metadata_is_copied_not_referenced() -> None:
+    """Mutating the source config after emission must not affect the
+    payload — TaskPayload.metadata is a defensive copy.
+    """
+    md = {"lc_agent_name": "a"}
+    task = _FakeTask(
+        id="t1", name="tools", input=[], triggers=["x"], config={"metadata": md}
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    md["lc_agent_name"] = "MUTATED"
+    assert payload["metadata"]["lc_agent_name"] == "a"
+
+
+def test_map_debug_tasks_folds_filtered_tags_into_metadata() -> None:
+    """Config tags are folded into TaskPayload.metadata under `tags`, with
+    langchain's internal `seq:step:*` tags filtered out — mirroring the
+    messages stream handler so both channels surface the same tag set."""
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={
+            "metadata": {"lc_agent_name": "weather_agent"},
+            "tags": ["seq:step:1", "user-tag", "session-123"],
+        },
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert payload["metadata"]["lc_agent_name"] == "weather_agent"
+    assert payload["metadata"]["tags"] == ["user-tag", "session-123"]
+
+
+def test_map_debug_tasks_omits_tags_when_only_seq_step() -> None:
+    """If the only tags are internal `seq:step:*` markers, no `tags` key is
+    added (matches the messages handler's `if filtered_tags:` guard)."""
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={"metadata": {"lc_agent_name": "a"}, "tags": ["seq:step:1"]},
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert "tags" not in payload["metadata"]
+
+
+def test_map_debug_tasks_adds_tags_even_without_other_metadata() -> None:
+    """Filtered tags surface even when config has no metadata dict."""
+    task = _FakeTask(
+        id="t1",
+        name="tools",
+        input=[],
+        triggers=["x"],
+        config={"tags": ["user-tag"]},
+    )
+    payload = next(iter(map_debug_tasks([task])))
+    assert payload["metadata"] == {"tags": ["user-tag"]}

libs/langgraph/tests/test_remote_graph_v3.py

@@ -0,0 +1,658 @@
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from langgraph.pregel._remote_run_stream import (
+    _AsyncRemoteGraphRunStream,
+    _ChannelProjection,
+    _ProjectionRegistry,
+    _RemoteGraphRunStream,
+    _translate_command_input,
+)
+from langgraph.pregel.remote import (
+    _V3_SUPPORTED_KWARGS,
+    RemoteGraph,
+)
+from langgraph.types import Command
+
+
+def _make_sync_adapter(*, run_start_returns=None, run_start_raises=None):
+    sync_client = MagicMock()
+    sdk_thread = MagicMock()
+    sdk_thread.thread_id = "thread-abc"
+    sdk_thread.__enter__ = MagicMock(return_value=sdk_thread)
+    sdk_thread.__exit__ = MagicMock(return_value=None)
+    sdk_thread.run = MagicMock()
+    if run_start_raises is not None:
+        sdk_thread.run.start = MagicMock(side_effect=run_start_raises)
+    else:
+        sdk_thread.run.start = MagicMock(
+            return_value=run_start_returns or {"run_id": "run-xyz"}
+        )
+    adapter = _RemoteGraphRunStream(
+        sync_client=sync_client,
+        sdk_thread=sdk_thread,
+        input={"x": 1},
+        config={"configurable": {}},
+        metadata=None,
+    )
+    return adapter, sync_client, sdk_thread
+
+
+def test_enter_calls_sdk_enter_then_run_start_and_captures_run_id():
+    adapter, _, sdk_thread = _make_sync_adapter()
+    with adapter as stream:
+        assert stream is adapter
+        sdk_thread.__enter__.assert_called_once()
+        sdk_thread.run.start.assert_called_once_with(
+            input={"x": 1}, config={"configurable": {}}, metadata=None
+        )
+        assert adapter._run_id == "run-xyz"
+
+
+def test_exit_delegates_to_sdk_exit_with_exc_info():
+    adapter, _, sdk_thread = _make_sync_adapter()
+    with adapter:
+        pass
+    sdk_thread.__exit__.assert_called_once_with(None, None, None)
+
+
+def test_enter_unwinds_sdk_cm_when_run_start_raises():
+    adapter, _, sdk_thread = _make_sync_adapter(
+        run_start_raises=RuntimeError("start boom")
+    )
+    with pytest.raises(RuntimeError, match="start boom"):
+        with adapter:
+            pytest.fail("body should not run")
+    sdk_thread.__enter__.assert_called_once()
+    sdk_thread.__exit__.assert_called_once()
+    exc_info = sdk_thread.__exit__.call_args.args
+    assert exc_info[0] is RuntimeError
+    assert isinstance(exc_info[1], RuntimeError)
+    assert adapter._run_id is None
+
+
+def test_output_interrupted_interrupts_passthrough():
+    adapter, _, sdk_thread = _make_sync_adapter()
+    sdk_thread.output = {"foo": 1}
+    sdk_thread.interrupted = True
+    sdk_thread.interrupts = [{"interrupt_id": "i1", "namespace": [], "value": "v"}]
+    with adapter as stream:
+        assert stream.output == {"foo": 1}
+        assert stream.interrupted is True
+        assert stream.interrupts == [
+            {"interrupt_id": "i1", "namespace": [], "value": "v"}
+        ]
+
+
+def test_sync_projection_attrs_forward_to_sdk():
+    adapter, _, sdk_thread = _make_sync_adapter()
+    sdk_thread.values = object()
+    sdk_thread.messages = object()
+    sdk_thread.tool_calls = object()
+    sdk_thread.subgraphs = object()
+    with adapter as stream:
+        assert stream.values is sdk_thread.values
+        assert stream.messages is sdk_thread.messages
+        assert stream.tool_calls is sdk_thread.tool_calls
+        assert stream.subgraphs is sdk_thread.subgraphs
+        assert set(stream.extensions) == set(_ProjectionRegistry._NATIVE)
+        assert stream.extensions["values"] is sdk_thread.values
+
+
+def test_projection_registry_typed_decoded_and_custom():
+    sdk = MagicMock()
+    sdk.values = object()
+    sdk.messages = object()
+    sdk.tool_calls = object()
+    sdk.subgraphs = object()
+    custom_named = object()
+    sdk.extensions = {"my_custom": custom_named}
+    registry = _ProjectionRegistry(sdk)
+
+    # Typed channels resolve to the SDK's decoded projections.
+    assert registry["values"] is sdk.values
+    assert registry["tool_calls"] is sdk.tool_calls
+    assert registry["subgraphs"] is sdk.subgraphs
+    # Channels without a typed projection resolve to a decoding _ChannelProjection.
+    ckpt = registry["checkpoints"]
+    assert isinstance(ckpt, _ChannelProjection)
+    assert ckpt._channel == "checkpoints"
+    assert isinstance(registry["updates"], _ChannelProjection)
+    # A non-protocol name is a specific custom-extension channel.
+    assert registry["my_custom"] is custom_named
+    # Enumerable set is the typed + decoded channels (no `lifecycle`, no `debug`).
+    assert list(registry) == [
+        "values",
+        "messages",
+        "tool_calls",
+        "subgraphs",
+        "updates",
+        "checkpoints",
+        "tasks",
+        "custom",
+    ]
+    assert len(registry) == 8
+
+
+def test_channel_projection_decodes_params_data():
+    sdk = MagicMock()
+    # Real wire events carry `method`; the SDK `DataDecoder` yields matching
+    # events' `params.data` and skips dataless and off-channel ones.
+    sdk.subscribe = MagicMock(
+        return_value=iter(
+            [
+                {"method": "checkpoints", "params": {"data": {"n": 1}}},
+                {"method": "checkpoints", "params": {}},  # no data -> skipped
+                {"method": "checkpoints", "params": {"data": {"n": 2}}},
+                {"method": "lifecycle", "params": {"data": {"n": 3}}},  # other channel
+            ]
+        )
+    )
+    proj = _ChannelProjection(sdk, "checkpoints")
+    assert list(proj) == [{"n": 1}, {"n": 2}]
+    sdk.subscribe.assert_called_once_with(["checkpoints"])
+
+
+@pytest.mark.anyio
+async def test_channel_projection_decodes_params_data_async():
+    """Async lane mirrors the sync lane: `async for` over the SDK's async
+    subscription, decoded through the same `DataDecoder`."""
+
+    class _FakeAsyncEvents:
+        def __init__(self, items):
+            self._items = list(items)
+
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self):
+            if not self._items:
+                raise StopAsyncIteration
+            return self._items.pop(0)
+
+    sdk = MagicMock()
+    sdk.subscribe = MagicMock(
+        return_value=_FakeAsyncEvents(
+            [
+                {"method": "checkpoints", "params": {"data": {"n": 1}}},
+                {"method": "checkpoints", "params": {}},  # no data -> skipped
+                {"method": "checkpoints", "params": {"data": {"n": 2}}},
+                {"method": "lifecycle", "params": {"data": {"n": 3}}},  # other channel
+            ]
+        )
+    )
+    proj = _ChannelProjection(sdk, "checkpoints")
+    assert [item async for item in proj] == [{"n": 1}, {"n": 2}]
+    sdk.subscribe.assert_called_once_with(["checkpoints"])
+
+
+def test_sync_adapter_translates_command_input():
+    sync_client = MagicMock()
+    sdk_thread = MagicMock()
+    adapter = _RemoteGraphRunStream(
+        sync_client=sync_client,
+        sdk_thread=sdk_thread,
+        input=Command(resume="go"),
+        config=None,
+        metadata=None,
+    )
+    assert adapter._start_kwargs["input"] == "go"
+
+
+def test_iter_caches_first_subscription():
+    adapter, _, sdk_thread = _make_sync_adapter()
+    fake_events = [object(), object(), object()]
+    sdk_thread.events = iter(fake_events)
+    with adapter as stream:
+        first = iter(stream)
+        second = iter(stream)
+        assert first is second
+        assert list(first) == fake_events
+
+
+def test_abort_cancels_run_and_closes_sdk():
+    adapter, sync_client, sdk_thread = _make_sync_adapter()
+    with adapter as stream:
+        stream.abort()
+        sync_client.runs.cancel.assert_called_once_with(
+            "thread-abc", "run-xyz", wait=False
+        )
+        sdk_thread.close.assert_called_once()
+
+
+def test_abort_before_enter_skips_cancel_but_closes_sdk():
+    adapter, sync_client, sdk_thread = _make_sync_adapter()
+    adapter.abort()
+    sync_client.runs.cancel.assert_not_called()
+    sdk_thread.close.assert_called_once()
+
+
+def test_abort_is_idempotent():
+    adapter, sync_client, sdk_thread = _make_sync_adapter()
+    with adapter as stream:
+        stream.abort()
+        stream.abort()
+        assert sync_client.runs.cancel.call_count == 1
+        assert sdk_thread.close.call_count == 1
+
+
+def test_abort_swallows_cancel_failure_and_still_closes():
+    adapter, sync_client, sdk_thread = _make_sync_adapter()
+    sync_client.runs.cancel.side_effect = RuntimeError("cancel boom")
+    with adapter as stream:
+        stream.abort()
+        sdk_thread.close.assert_called_once()
+
+
+def test_sync_interleave_delegates_to_interleave_projections():
+    adapter, _, sdk_thread = _make_sync_adapter()
+    pairs = [("values", {"x": 1}), ("messages", object())]
+    sdk_thread.interleave_projections.return_value = pairs
+    with adapter as stream:
+        result = list(stream.interleave("values", "messages"))
+    assert result == pairs
+    sdk_thread.interleave_projections.assert_called_once_with(["values", "messages"])
+
+
+def test_async_adapter_has_no_interleave():
+    """Async adapter intentionally lacks `interleave` (mirrors local
+    `AsyncGraphRunStream`, which doesn't have one either). Async callers
+    compose with `asyncio.gather` / `asyncio.as_completed`.
+    """
+    assert not hasattr(_AsyncRemoteGraphRunStream, "interleave")
+
+
+def _make_async_adapter(*, run_start_returns=None, run_start_raises=None):
+    client = MagicMock()
+    client.runs.cancel = AsyncMock()
+    sdk_thread = MagicMock()
+    sdk_thread.thread_id = "thread-abc"
+    sdk_thread.__aenter__ = AsyncMock(return_value=sdk_thread)
+    sdk_thread.__aexit__ = AsyncMock(return_value=None)
+    sdk_thread.close = AsyncMock()
+    sdk_thread.run = MagicMock()
+    if run_start_raises is not None:
+        sdk_thread.run.start = AsyncMock(side_effect=run_start_raises)
+    else:
+        sdk_thread.run.start = AsyncMock(
+            return_value=run_start_returns or {"run_id": "run-xyz"}
+        )
+    adapter = _AsyncRemoteGraphRunStream(
+        client=client,
+        sdk_thread=sdk_thread,
+        input={"x": 1},
+        config={"configurable": {}},
+        metadata=None,
+    )
+    return adapter, client, sdk_thread
+
+
+@pytest.mark.anyio
+async def test_aenter_calls_sdk_aenter_then_run_start_and_captures_run_id():
+    adapter, _, sdk_thread = _make_async_adapter()
+    async with adapter as stream:
+        assert stream is adapter
+        sdk_thread.__aenter__.assert_awaited_once()
+        sdk_thread.run.start.assert_awaited_once_with(
+            input={"x": 1}, config={"configurable": {}}, metadata=None
+        )
+        assert adapter._run_id == "run-xyz"
+
+
+@pytest.mark.anyio
+async def test_aexit_delegates_to_sdk_aexit():
+    adapter, _, sdk_thread = _make_async_adapter()
+    async with adapter:
+        pass
+    sdk_thread.__aexit__.assert_awaited_once_with(None, None, None)
+
+
+@pytest.mark.anyio
+async def test_aenter_unwinds_sdk_cm_when_run_start_raises():
+    adapter, _, sdk_thread = _make_async_adapter(
+        run_start_raises=RuntimeError("start boom")
+    )
+    with pytest.raises(RuntimeError, match="start boom"):
+        async with adapter:
+            pytest.fail("body should not run")
+    sdk_thread.__aenter__.assert_awaited_once()
+    sdk_thread.__aexit__.assert_awaited_once()
+    assert adapter._run_id is None
+
+
+@pytest.mark.anyio
+async def test_async_output_interrupted_interrupts_passthrough():
+    adapter, _, sdk_thread = _make_async_adapter()
+
+    async def _fake_output_awaitable():
+        return {"foo": 1}
+
+    sdk_thread.output = _fake_output_awaitable()
+    sdk_thread.interrupted = True
+    sdk_thread.interrupts = [{"interrupt_id": "i1", "namespace": [], "value": "v"}]
+    async with adapter as stream:
+        assert await stream.output() == {"foo": 1}
+        assert await stream.interrupted() is True
+        assert await stream.interrupts() == [
+            {"interrupt_id": "i1", "namespace": [], "value": "v"}
+        ]
+
+
+@pytest.mark.anyio
+async def test_async_projection_attrs_forward_to_sdk():
+    adapter, _, sdk_thread = _make_async_adapter()
+    sdk_thread.values = object()
+    sdk_thread.messages = object()
+    sdk_thread.tool_calls = object()
+    sdk_thread.subgraphs = object()
+    async with adapter as stream:
+        assert stream.values is sdk_thread.values
+        assert stream.messages is sdk_thread.messages
+        assert stream.tool_calls is sdk_thread.tool_calls
+        assert stream.subgraphs is sdk_thread.subgraphs
+        assert set(stream.extensions) == set(_ProjectionRegistry._NATIVE)
+        assert stream.extensions["messages"] is sdk_thread.messages
+
+
+@pytest.mark.anyio
+async def test_aiter_caches_first_subscription():
+    adapter, _, sdk_thread = _make_async_adapter()
+
+    class _FakeAsyncEvents:
+        def __init__(self, items):
+            self._items = list(items)
+
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self):
+            if not self._items:
+                raise StopAsyncIteration
+            return self._items.pop(0)
+
+    sdk_thread.events = _FakeAsyncEvents([object(), object()])
+    async with adapter as stream:
+        first = stream.__aiter__()
+        second = stream.__aiter__()
+        assert first is second
+
+
+@pytest.mark.anyio
+async def test_async_abort_cancels_run_and_closes_sdk():
+    adapter, client, sdk_thread = _make_async_adapter()
+    async with adapter as stream:
+        await stream.abort()
+        client.runs.cancel.assert_awaited_once_with("thread-abc", "run-xyz", wait=False)
+        sdk_thread.close.assert_awaited_once()
+
+
+@pytest.mark.anyio
+async def test_async_abort_before_aenter_skips_cancel():
+    adapter, client, sdk_thread = _make_async_adapter()
+    await adapter.abort()
+    client.runs.cancel.assert_not_awaited()
+    sdk_thread.close.assert_awaited_once()
+
+
+@pytest.mark.anyio
+async def test_async_abort_swallows_cancel_failure():
+    adapter, client, sdk_thread = _make_async_adapter()
+    client.runs.cancel.side_effect = RuntimeError("cancel boom")
+    async with adapter as stream:
+        await stream.abort()
+        sdk_thread.close.assert_awaited_once()
+
+
+def _make_remote_graph() -> RemoteGraph:
+    sync_client = MagicMock()
+    async_client = MagicMock()
+    rg = RemoteGraph(
+        "agent",
+        client=async_client,
+        sync_client=sync_client,
+    )
+    return rg
+
+
+def test_reject_v3_unsupported_passes_when_all_clear():
+    rg = _make_remote_graph()
+    rg._reject_v3_unsupported(
+        control=None,
+        transformers=None,
+        interrupt_before=None,
+        interrupt_after=None,
+        extra_kwargs={},
+    )
+
+
+@pytest.mark.parametrize(
+    "kwarg_name,kwarg_value",
+    [
+        ("control", object()),
+        ("transformers", [object()]),
+        ("interrupt_before", ["node_a"]),
+        ("interrupt_after", ["node_b"]),
+    ],
+)
+def test_reject_v3_unsupported_raises_per_kwarg(kwarg_name, kwarg_value):
+    rg = _make_remote_graph()
+    kwargs = dict(
+        control=None,
+        transformers=None,
+        interrupt_before=None,
+        interrupt_after=None,
+        extra_kwargs={},
+    )
+    kwargs[kwarg_name] = kwarg_value
+    with pytest.raises(NotImplementedError, match=f"`{kwarg_name}=`"):
+        rg._reject_v3_unsupported(**kwargs)
+
+
+def test_reject_v3_unsupported_raises_on_unknown_extra_kwarg():
+    rg = _make_remote_graph()
+    with pytest.raises(NotImplementedError, match="context"):
+        rg._reject_v3_unsupported(
+            control=None,
+            transformers=None,
+            interrupt_before=None,
+            interrupt_after=None,
+            extra_kwargs={"context": {}},
+        )
+
+
+def test_reject_v3_unsupported_allows_metadata_and_headers():
+    rg = _make_remote_graph()
+    rg._reject_v3_unsupported(
+        control=None,
+        transformers=None,
+        interrupt_before=None,
+        interrupt_after=None,
+        extra_kwargs={"metadata": {"a": 1}, "headers": {"X": "y"}},
+    )
+
+
+def test_translate_command_input_surfaces_raw_resume_value():
+    # The v3 server wraps the resume `input` as {"resume": input} itself, so the
+    # wire `input` must be the raw resume value, not the serialized dataclass.
+    assert _translate_command_input(Command(resume="go")) == "go"
+    assert _translate_command_input(Command(resume={"id": "v"})) == {"id": "v"}
+
+
+def test_translate_command_input_rejects_goto_and_update():
+    with pytest.raises(NotImplementedError, match="goto"):
+        _translate_command_input(Command(goto="node_b"))
+    with pytest.raises(NotImplementedError, match="update"):
+        _translate_command_input(Command(update={"a": 1}))
+
+
+def test_translate_command_input_passes_through_non_command():
+    assert _translate_command_input({"a": 1}) == {"a": 1}
+    assert _translate_command_input(None) is None
+
+
+def test_v3_supported_kwargs_known_set():
+    assert _V3_SUPPORTED_KWARGS == frozenset({"metadata", "headers"})
+
+
+def test_stream_events_v3_constructs_sdk_thread_with_sanitized_args():
+    sync_client = MagicMock()
+    sdk_thread = MagicMock()
+    sync_client.threads.stream.return_value = sdk_thread
+    rg = RemoteGraph(
+        "agent",
+        client=MagicMock(),
+        sync_client=sync_client,
+    )
+    result = rg.stream_events(
+        {"input_key": 1},
+        config={"configurable": {"thread_id": "t1", "user": "u"}},
+        version="v3",
+    )
+    assert isinstance(result, _RemoteGraphRunStream)
+    sync_client.threads.stream.assert_called_once()
+    call = sync_client.threads.stream.call_args
+    assert call.kwargs["thread_id"] == "t1"
+    assert call.kwargs["assistant_id"] == "agent"
+    assert call.kwargs["headers"] is None
+
+
+def test_stream_events_v3_passes_none_thread_id_when_absent():
+    sync_client = MagicMock()
+    sync_client.threads.stream.return_value = MagicMock()
+    rg = RemoteGraph("agent", client=MagicMock(), sync_client=sync_client)
+    rg.stream_events({"x": 1}, version="v3")
+    call = sync_client.threads.stream.call_args
+    assert call.kwargs["thread_id"] is None
+
+
+def test_stream_events_v3_rejects_unsupported_kwargs_before_sdk_call():
+    sync_client = MagicMock()
+    rg = RemoteGraph("agent", client=MagicMock(), sync_client=sync_client)
+    with pytest.raises(NotImplementedError, match="control"):
+        rg.stream_events({"x": 1}, version="v3", control=object())
+    sync_client.threads.stream.assert_not_called()
+
+
+def test_stream_events_v3_translates_command_input():
+    sync_client = MagicMock()
+    sync_client.threads.stream.return_value = MagicMock()
+    rg = RemoteGraph("agent", client=MagicMock(), sync_client=sync_client)
+    # Resume Command surfaces its raw resume value as the wire `input`; the v3
+    # server wraps it as {"resume": input} once it detects the interrupt.
+    adapter = rg.stream_events(Command(resume="go"), version="v3")
+    assert adapter._start_kwargs["input"] == "go"
+
+
+def test_stream_events_v3_rejects_goto_update_command():
+    rg = RemoteGraph("agent", client=MagicMock(), sync_client=MagicMock())
+    with pytest.raises(NotImplementedError, match="goto"):
+        rg.stream_events(Command(goto="node_b"), version="v3")
+
+
+def test_stream_events_v3_strips_checkpoint_keys_from_configurable():
+    sync_client = MagicMock()
+    sync_client.threads.stream.return_value = MagicMock()
+    rg = RemoteGraph("agent", client=MagicMock(), sync_client=sync_client)
+    adapter = rg.stream_events(
+        {"x": 1},
+        config={
+            "configurable": {
+                "thread_id": "t1",
+                "checkpoint_id": "c1",
+                "checkpoint_ns": "ns",
+                "user": "u",
+            }
+        },
+        version="v3",
+    )
+    sent_config = adapter._start_kwargs["config"]
+    assert "checkpoint_id" not in sent_config["configurable"]
+    assert "checkpoint_ns" not in sent_config["configurable"]
+    assert sent_config["configurable"]["user"] == "u"
+
+
+def test_stream_events_v3_merges_tracing_headers_when_distributed_tracing(
+    monkeypatch,
+):
+    from langgraph.pregel import remote as remote_mod
+
+    sync_client = MagicMock()
+    sync_client.threads.stream.return_value = MagicMock()
+    rg = RemoteGraph(
+        "agent",
+        client=MagicMock(),
+        sync_client=sync_client,
+        distributed_tracing=True,
+    )
+    captured = {}
+
+    def fake_merge(headers):
+        captured["arg"] = headers
+        return {"x-ls-trace": "1", **(headers or {})}
+
+    monkeypatch.setattr(remote_mod, "_merge_tracing_headers", fake_merge)
+    rg.stream_events({"x": 1}, version="v3", headers={"X-Custom": "y"})
+    assert captured["arg"] == {"X-Custom": "y"}
+    sent_headers = sync_client.threads.stream.call_args.kwargs["headers"]
+    assert sent_headers["x-ls-trace"] == "1"
+    assert sent_headers["X-Custom"] == "y"
+
+
+def test_stream_events_v3_passes_headers_unchanged_without_tracing():
+    sync_client = MagicMock()
+    sync_client.threads.stream.return_value = MagicMock()
+    rg = RemoteGraph(
+        "agent",
+        client=MagicMock(),
+        sync_client=sync_client,
+        distributed_tracing=False,
+    )
+    rg.stream_events({"x": 1}, version="v3", headers={"X-Custom": "y"})
+    sent_headers = sync_client.threads.stream.call_args.kwargs["headers"]
+    assert sent_headers == {"X-Custom": "y"}
+
+
+def test_stream_events_non_v3_delegates_to_super():
+    rg = RemoteGraph("agent", client=MagicMock(), sync_client=MagicMock())
+    sync_client_attr = rg.sync_client
+    try:
+        rg.stream_events({"x": 1}, version="v2")
+    except Exception:
+        pass
+    sync_client_attr.threads.stream.assert_not_called()
+
+
+@pytest.mark.anyio
+async def test_astream_events_v3_constructs_sdk_thread():
+    client = MagicMock()
+    sdk_thread = MagicMock()
+    client.threads.stream.return_value = sdk_thread
+    rg = RemoteGraph("agent", client=client, sync_client=MagicMock())
+    result = await rg.astream_events(
+        {"x": 1},
+        config={"configurable": {"thread_id": "t1"}},
+        version="v3",
+    )
+    assert isinstance(result, _AsyncRemoteGraphRunStream)
+    call = client.threads.stream.call_args
+    assert call.kwargs["thread_id"] == "t1"
+    assert call.kwargs["assistant_id"] == "agent"
+
+
+@pytest.mark.anyio
+async def test_astream_events_v3_rejects_unsupported_kwargs():
+    client = MagicMock()
+    rg = RemoteGraph("agent", client=client, sync_client=MagicMock())
+    with pytest.raises(NotImplementedError, match="transformers"):
+        await rg.astream_events({"x": 1}, version="v3", transformers=[object()])
+    client.threads.stream.assert_not_called()
+
+
+@pytest.mark.anyio
+async def test_astream_events_non_v3_raises_not_implemented():
+    rg = RemoteGraph("agent", client=MagicMock(), sync_client=MagicMock())
+    with pytest.raises(NotImplementedError, match="not implemented"):
+        await rg.astream_events({"x": 1}, version="v2")

libs/langgraph/tests/test_retry.py

@@ -1,4 +1,5 @@
 import asyncio
+import contextlib
 import operator
 import sys
 import threading
@@ -35,7 +36,13 @@ from langgraph._internal._runnable import RunnableCallable
 from langgraph._internal._timeout import coerce_timeout_policy
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
-from langgraph.errors import GraphInterrupt, NodeError, NodeTimeoutError, ParentCommand
+from langgraph.errors import (
+    GraphInterrupt,
+    NodeCancelledError,
+    NodeError,
+    NodeTimeoutError,
+    ParentCommand,
+)
 from langgraph.func import entrypoint, task
 from langgraph.graph import END, START, StateGraph, add_messages
 from langgraph.pregel import NodeBuilder, Pregel
@@ -63,6 +70,18 @@ NEEDS_CONTEXTVARS = pytest.mark.skipif(
     reason="Python 3.11+ is required for async contextvars support",
 )
 
+# `asyncio.Task.cancelling()` is Python 3.11+. The LSD-1507 fix in
+# `langgraph/pregel/_retry.py` falls back to a no-op on 3.10 (preserves the
+# existing CancelledError-as-silent-tear-down behaviour) because there is no
+# reliable way to distinguish user-raised from framework-initiated
+# cancellation without that API. Tests for the converted behaviour gate on
+# the same Python version boundary.
+NEEDS_TASK_CANCELLING = pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="LSD-1507 user-cancellation conversion requires Python 3.11+ "
+    "(asyncio.Task.cancelling)",
+)
+
 
 def test_should_retry_on_single_exception():
     """Test retry with a single exception type."""
@@ -2802,3 +2821,123 @@ def test_error_handler_resumes_after_crash_multiple_nodes():
     assert call_count["handler_b"] == 2  # ran again on resume
     assert "recovered_a:a" in result["results"]
     assert "recovered_b:b" in result["results"]
+
+
+@NEEDS_TASK_CANCELLING
+@pytest.mark.anyio
+async def test_arun_with_retry_user_raised_cancelled_becomes_node_cancelled():
+    class UserCancelsProc:
+        async def ainvoke(self, input, config):
+            raise asyncio.CancelledError("nope")
+
+    task = _make_task(UserCancelsProc(), name="user-cancel")
+
+    with pytest.raises(NodeCancelledError) as excinfo:
+        await arun_with_retry(task, retry_policy=None)
+    assert excinfo.value.node == "user-cancel"
+    # original CancelledError chained for debugging
+    assert isinstance(excinfo.value.__cause__, asyncio.CancelledError)
+
+
+@NEEDS_TASK_CANCELLING
+@pytest.mark.anyio
+async def test_arun_with_retry_user_raised_cancelled_with_timeout_policy():
+    """The timeout path runs the node in a child task; the conversion must
+    still trigger for user-raised ``CancelledError``."""
+
+    class UserCancelsProc:
+        async def ainvoke(self, input, config):
+            raise asyncio.CancelledError
+
+    task = _make_task(
+        UserCancelsProc(), timeout=_idle_timeout(1.0), name="user-cancel-timed"
+    )
+
+    with pytest.raises(NodeCancelledError) as excinfo:
+        await arun_with_retry(task, retry_policy=None)
+    assert excinfo.value.node == "user-cancel-timed"
+
+
+def test_run_with_retry_sync_node_raising_cancelled_becomes_node_cancelled():
+    class SyncUserCancelsProc:
+        def invoke(self, input, config):
+            raise asyncio.CancelledError("sync nope")
+
+    task = _make_task(SyncUserCancelsProc(), timeout=None, name="sync-user-cancel")
+
+    with pytest.raises(NodeCancelledError) as excinfo:
+        run_with_retry(task, retry_policy=None)
+    assert excinfo.value.node == "sync-user-cancel"
+    assert isinstance(excinfo.value.__cause__, asyncio.CancelledError)
+
+
+@NEEDS_TASK_CANCELLING
+@pytest.mark.anyio
+async def test_arun_with_retry_external_cancel_propagates_as_cancelled():
+    """When the asyncio task running ``arun_with_retry`` is cancelled from the
+    outside, the cancellation must still propagate as
+    ``asyncio.CancelledError``. Converting it to ``NodeCancelledError`` would
+    break the runner's ability to cancel sibling tasks during cleanup."""
+
+    started = asyncio.Event()
+    observed: list[BaseException] = []
+
+    class SlowProc:
+        async def ainvoke(self, input, config):
+            started.set()
+            await asyncio.sleep(10.0)
+            return "never"
+
+    task = _make_task(SlowProc(), timeout=None, name="external-cancel")
+
+    async def runner():
+        try:
+            await arun_with_retry(task, retry_policy=None)
+        except BaseException as exc:
+            observed.append(exc)
+            raise
+
+    bg = asyncio.create_task(runner())
+    await started.wait()
+    bg.cancel()
+    # We expect the cancellation to surface to us as well; swallow it here so
+    # the test runner's own task isn't poisoned by the cancel.
+    with contextlib.suppress(asyncio.CancelledError):
+        await bg
+    assert observed, "runner did not observe any exception"
+    # Framework cancellation must remain a CancelledError, not be rewritten as
+    # NodeCancelledError.
+    assert isinstance(observed[0], asyncio.CancelledError)
+    assert not isinstance(observed[0], NodeCancelledError)
+
+
+@NEEDS_TASK_CANCELLING
+@pytest.mark.anyio
+async def test_pregel_user_raised_cancellederror_fails_run():
+    """End-to-end: a two-branch graph where one branch raises
+    ``asyncio.CancelledError`` must fail the run instead of returning
+    a partial-success state. This is the LSD-1507 customer scenario."""
+
+    class _S(TypedDict, total=False):
+        vals: Annotated[list[str], operator.add]
+
+    async def ok(state: _S) -> _S:
+        return {"vals": ["ok"]}
+
+    async def boom(state: _S) -> _S:
+        raise asyncio.CancelledError("user-raised in node body")
+
+    graph = (
+        StateGraph(_S)
+        .add_node("ok", ok)
+        .add_node("boom", boom)
+        .add_edge(START, "ok")
+        .add_edge(START, "boom")
+        .add_edge("ok", END)
+        .add_edge("boom", END)
+        .compile()
+    )
+
+    with pytest.raises(NodeCancelledError) as excinfo:
+        await graph.ainvoke({"vals": []})
+    assert excinfo.value.node == "boom"

libs/langgraph/tests/test_stream_lifecycle_transformer.py

@@ -35,20 +35,25 @@ def _tasks_start(
     *,
     task_id: str,
     name: str,
+    metadata: dict[str, Any] | None = None,
+    input: Any = None,
 ) -> dict[str, Any]:
     """Build a `tasks` ProtocolEvent carrying a TaskPayload (start)."""
+    data: dict[str, Any] = {
+        "id": task_id,
+        "name": name,
+        "input": input,
+        "triggers": [],
+    }
+    if metadata is not None:
+        data["metadata"] = metadata
     return {
         "type": "event",
         "method": "tasks",
         "params": {
             "namespace": namespace,
             "timestamp": TS,
-            "data": {
-                "id": task_id,
-                "name": name,
-                "input": None,
-                "triggers": [],
-            },
+            "data": data,
         },
     }
 
@@ -404,3 +409,272 @@ def test_stream_events_v3_with_nested_parent_ns_scopes_lifecycle() -> None:
         assert ns[:1] == ("outer:abc",), (
             f"namespace {ns} not within scoped prefix ('outer:abc',)"
         )
+
+
+# ---------------------------------------------------------------------------
+# Parsed-segment fallback (no subagent boundary)
+# ---------------------------------------------------------------------------
+
+
+def test_no_metadata_falls_through_to_existing_behavior() -> None:
+    """Tasks events without metadata produce the same output as before T4."""
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool"))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["graph_name"] == "agent"
+    assert payload["trigger_call_id"] == "abc"
+    assert "cause" not in payload
+
+
+def test_empty_metadata_dict_falls_through() -> None:
+    """An explicit empty metadata dict is treated the same as no metadata."""
+    mux = _build_lifecycle_mux()
+    mux.push(_tasks_start(["agent:abc"], task_id="t1", name="tool", metadata={}))
+
+    [payload] = _drain_lifecycle(mux)
+    assert payload["graph_name"] == "agent"
+    assert "cause" not in payload
+
+
+# ---------------------------------------------------------------------------
+# Subagent discrimination via lc_agent_name transition
+# ---------------------------------------------------------------------------
+#
+# A nested task is a subagent iff its metadata["lc_agent_name"] is present and
+# differs from its PARENT namespace's lc_agent_name. These tests replicate the
+# empirically-verified `create_agent` stream shape synthetically:
+#
+#   - A supervisor created via `create_agent(name="supervisor")` emits its own
+#     node tasks (model, tools) at ns=(), each with
+#     metadata["lc_agent_name"] == "supervisor".
+#   - The parent `tools` task at ns=() carries a `tool_call_with_context`
+#     dict as its `input` (with the LLM tool_call_id at
+#     input["tool_call"]["id"]) and a task `id`. (A legacy / batched shape
+#     passes a list of tool-call dicts instead; both are exercised below.)
+#   - When a tool body invokes an inner `create_agent(name="weather_agent")`,
+#     the inner agent's node tasks stream at ns=("tools:<taskid>",) with
+#     metadata["lc_agent_name"] == "weather_agent", sharing the SAME <taskid>
+#     as the parent `tools` task.
+#   - A plain StateGraph (no name) inherits the parent's lc_agent_name, so its
+#     child lc == parent lc -> NOT a subagent.
+
+
+def test_lifecycle_uses_lc_agent_name_for_subagent() -> None:
+    """A nested run whose lc_agent_name differs from its parent's is a subagent.
+
+    graph_name becomes the child's lc_agent_name; cause is recovered by joining
+    the child segment's task-id to the parent push task's tool call. This uses
+    the production `tool_call_with_context` dict input shape current langgraph
+    emits (tool_call_id at input["tool_call"]["id"]).
+    """
+    mux = _build_lifecycle_mux()
+    # Supervisor's `tools` push task at scope ns: carries its own lc_agent_name
+    # and a `tool_call_with_context` dict as `input`. Each tool call is its own
+    # push task, and the task id seeds the child segment.
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "supervisor"},
+            input={
+                "__type": "tool_call_with_context",
+                "tool_call": {
+                    "name": "call_weather",
+                    "args": {"city": "Boston"},
+                    "id": "call_w",
+                    "type": "tool_call",
+                },
+                "state": {},
+            },
+        )
+    )
+    # Inner weather_agent's first node task streams under the parent `tools`
+    # task's namespace segment (shared task id) with its own lc_agent_name.
+    mux.push(
+        _tasks_start(
+            ["tools:tools_task_1"],
+            task_id="inner_model_1",
+            name="model",
+            metadata={"lc_agent_name": "weather_agent"},
+        )
+    )
+
+    payloads = _drain_lifecycle(mux)
+    started = [p for p in payloads if p["event"] == "started"]
+    [subagent] = [p for p in started if p["namespace"] == ["tools:tools_task_1"]]
+    assert subagent["graph_name"] == "weather_agent", (
+        "graph_name should be the child's lc_agent_name, not the parsed segment"
+    )
+    assert subagent["cause"] == {"type": "toolCall", "tool_call_id": "call_w"}, (
+        "cause should recover the triggering tool_call_id from the parent push "
+        "task's tool_call_with_context input via the shared task id"
+    )
+
+
+def test_lifecycle_subagent_cause_from_legacy_list_input() -> None:
+    """cause recovery also handles the legacy / batched list input shape.
+
+    A parent task whose `input` is a list of tool-call dicts (rather than a
+    `tool_call_with_context` dict) still seeds the tool_call_id join.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "supervisor"},
+            input=[{"name": "call_weather", "args": {"city": "SF"}, "id": "call_w"}],
+        )
+    )
+    mux.push(
+        _tasks_start(
+            ["tools:tools_task_1"],
+            task_id="inner_model_1",
+            name="model",
+            metadata={"lc_agent_name": "weather_agent"},
+        )
+    )
+
+    payloads = _drain_lifecycle(mux)
+    started = [p for p in payloads if p["event"] == "started"]
+    [subagent] = [p for p in started if p["namespace"] == ["tools:tools_task_1"]]
+    assert subagent["graph_name"] == "weather_agent"
+    assert subagent["cause"] == {"type": "toolCall", "tool_call_id": "call_w"}
+
+
+def test_lifecycle_same_name_nested_run_is_surfaced() -> None:
+    """A nested run whose lc_agent_name matches the parent's is still surfaced.
+
+    A subagent that invokes itself re-asserts its own lc_agent_name, so child
+    lc == parent lc. The discriminator surfaces any nested run carrying an
+    lc_agent_name, so the recursive call is reported (named after the agent,
+    with the triggering tool call as cause).
+
+    Trade-off: a non-agent subgraph that merely inherited the parent's
+    lc_agent_name would also surface here. That is accepted; a caller can null
+    lc_agent_name in the config it invokes such a graph with to exclude it.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "weather_agent"},
+            input={
+                "__type": "tool_call_with_context",
+                "tool_call": {
+                    "name": "recurse",
+                    "args": {},
+                    "id": "call_x",
+                    "type": "tool_call",
+                },
+                "state": {},
+            },
+        )
+    )
+    # The agent invokes itself: the nested run re-asserts the SAME lc_agent_name.
+    mux.push(
+        _tasks_start(
+            ["tools:tools_task_1"],
+            task_id="inner_node_1",
+            name="model",
+            metadata={"lc_agent_name": "weather_agent"},
+        )
+    )
+
+    payloads = _drain_lifecycle(mux)
+    started = [p for p in payloads if p["event"] == "started"]
+    [nested] = [p for p in started if p["namespace"] == ["tools:tools_task_1"]]
+    assert nested["graph_name"] == "weather_agent", (
+        "a same-named nested run (e.g. self-recursion) must still be surfaced"
+    )
+    assert nested["cause"] == {"type": "toolCall", "tool_call_id": "call_x"}
+
+
+def test_lifecycle_unnamed_nested_agent_is_not_subagent() -> None:
+    """A nested run with lc_agent_name None is excluded (not a subagent)."""
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "supervisor"},
+            input={
+                "__type": "tool_call_with_context",
+                "tool_call": {
+                    "name": "lookup",
+                    "args": {},
+                    "id": "call_x",
+                    "type": "tool_call",
+                },
+                "state": {},
+            },
+        )
+    )
+    mux.push(
+        _tasks_start(
+            ["plain:tools_task_1"],
+            task_id="inner_node_1",
+            name="inner_node",
+            metadata={"lc_agent_name": None},
+        )
+    )
+
+    payloads = _drain_lifecycle(mux)
+    started = [p for p in payloads if p["event"] == "started"]
+    [nested] = [p for p in started if p["namespace"] == ["plain:tools_task_1"]]
+    assert nested["graph_name"] == "plain"
+    assert "cause" not in nested
+
+
+def test_lifecycle_subagent_terminal_roundtrip() -> None:
+    """A detected subagent closes with `completed` when its parent task results.
+
+    Pushes the subagent's `started` (via the `tool_call_with_context` parent
+    plus the child task event) and then the parent push task's terminal
+    result, asserting the namespace is closed and the `started` payload's
+    projected graph_name / cause survive the roundtrip.
+    """
+    mux = _build_lifecycle_mux()
+    mux.push(
+        _tasks_start(
+            [],
+            task_id="tools_task_1",
+            name="tools",
+            metadata={"lc_agent_name": "supervisor"},
+            input={
+                "__type": "tool_call_with_context",
+                "tool_call": {
+                    "name": "call_weather",
+                    "args": {"city": "Boston"},
+                    "id": "call_w",
+                    "type": "tool_call",
+                },
+                "state": {},
+            },
+        )
+    )
+    mux.push(
+        _tasks_start(
+            ["tools:tools_task_1"],
+            task_id="inner_model_1",
+            name="model",
+            metadata={"lc_agent_name": "weather_agent"},
+        )
+    )
+    # The parent push task (id=tools_task_1, at scope ns) finishes, closing
+    # the subagent subgraph that streamed under `tools:tools_task_1`.
+    mux.push(_tasks_result([], task_id="tools_task_1", name="tools"))
+
+    payloads = _drain_lifecycle(mux)
+    ns = ["tools:tools_task_1"]
+    subagent = [p for p in payloads if p["namespace"] == ns]
+    assert [p["event"] for p in subagent] == ["started", "completed"]
+    started, _completed = subagent
+    assert started["graph_name"] == "weather_agent"
+    assert started["cause"] == {"type": "toolCall", "tool_call_id": "call_w"}

libs/langgraph/tests/test_utils.py

@@ -15,12 +15,14 @@ from unittest.mock import MagicMock, patch
 
 import langsmith
 import pytest
+from langchain_core.callbacks import BaseCallbackHandler, CallbackManager
 from langchain_core.runnables import RunnableConfig
 from langchain_core.tracers import LangChainTracer
 from typing_extensions import NotRequired, Required, TypedDict
 
 from langgraph._internal._config import (
     _is_not_empty,
+    _merge_callbacks,
     ensure_config,
     get_callback_manager_for_config,
 )
@@ -427,3 +429,128 @@ def test_callback_manager_copies_configurable_ids_to_tracing_metadata() -> None:
         "thread_id": "th-123",
         "user_id": "uid-1",
     }
+
+
+class _TrackingCB(BaseCallbackHandler):
+    """Minimal callback handler used only as a sentinel for merge tests."""
+
+    def __init__(self, tag: str) -> None:
+        self.tag = tag
+
+    def __eq__(self, other: object) -> bool:
+        return isinstance(other, _TrackingCB) and self.tag == other.tag
+
+    def __hash__(self) -> int:
+        return hash(self.tag)
+
+
+def test_merge_callbacks_none_base_list_new() -> None:
+    cb = _TrackingCB("a")
+    merged = _merge_callbacks(None, [cb])
+    assert merged == [cb]
+
+
+def test_merge_callbacks_list_base_list_new() -> None:
+    a, b = _TrackingCB("a"), _TrackingCB("b")
+    merged = _merge_callbacks([a], [b])
+    assert merged == [a, b]
+
+
+def test_merge_callbacks_list_base_manager_new() -> None:
+    a = _TrackingCB("a")
+    mgr = CallbackManager(handlers=[_TrackingCB("b")])
+    merged = _merge_callbacks([a], mgr)
+    assert isinstance(merged, CallbackManager)
+    assert _TrackingCB("a") in merged.handlers
+    assert _TrackingCB("b") in merged.handlers
+
+
+def test_merge_callbacks_manager_base_list_new() -> None:
+    mgr = CallbackManager(handlers=[_TrackingCB("a")])
+    b = _TrackingCB("b")
+    merged = _merge_callbacks(mgr, [b])
+    assert isinstance(merged, CallbackManager)
+    assert _TrackingCB("a") in merged.handlers
+    assert _TrackingCB("b") in merged.handlers
+
+
+def test_merge_callbacks_manager_base_manager_new() -> None:
+    mgr_a = CallbackManager(handlers=[_TrackingCB("a")])
+    mgr_b = CallbackManager(handlers=[_TrackingCB("b")])
+    merged = _merge_callbacks(mgr_a, mgr_b)
+    assert isinstance(merged, CallbackManager)
+    assert _TrackingCB("a") in merged.handlers
+    assert _TrackingCB("b") in merged.handlers
+
+
+def test_merge_callbacks_none_base_none_new() -> None:
+    merged = _merge_callbacks(None, None)
+    assert merged is None
+
+
+def test_ensure_config_merges_configurable_across_configs() -> None:
+    a = {"configurable": {"ls_agent_type": "root"}}
+    b = {"configurable": {"thread_id": "T1"}}
+    merged = ensure_config(a, b)
+    assert merged["configurable"]["ls_agent_type"] == "root"
+    assert merged["configurable"]["thread_id"] == "T1"
+
+
+def test_ensure_config_configurable_later_wins_per_key() -> None:
+    a = {"configurable": {"shared": "from_a", "only_a": "A"}}
+    b = {"configurable": {"shared": "from_b", "only_b": "B"}}
+    merged = ensure_config(a, b)
+    assert merged["configurable"]["shared"] == "from_b"  # later wins per key
+    assert merged["configurable"]["only_a"] == "A"
+    assert merged["configurable"]["only_b"] == "B"
+
+
+def test_ensure_config_merges_metadata_across_configs() -> None:
+    a = {"metadata": {"user_id": "U1"}}
+    b = {"metadata": {"correlation_id": "C1"}}
+    merged = ensure_config(a, b)
+    assert merged["metadata"]["user_id"] == "U1"
+    assert merged["metadata"]["correlation_id"] == "C1"
+
+
+def test_ensure_config_metadata_later_wins_per_key() -> None:
+    a = {"metadata": {"shared": "from_a"}}
+    b = {"metadata": {"shared": "from_b"}}
+    merged = ensure_config(a, b)
+    assert merged["metadata"]["shared"] == "from_b"
+
+
+def test_ensure_config_merges_tags_across_configs() -> None:
+    a = {"tags": ["alpha"]}
+    b = {"tags": ["beta"]}
+    merged = ensure_config(a, b)
+    assert merged["tags"] == ["alpha", "beta"]
+
+
+def test_ensure_config_tags_concat_preserves_order_and_duplicates() -> None:
+    # Plain concat (matches merge_configs in this file — no dedup, no sort).
+    a = {"tags": ["shared", "alpha"]}
+    b = {"tags": ["shared", "beta"]}
+    merged = ensure_config(a, b)
+    assert merged["tags"] == ["shared", "alpha", "shared", "beta"]
+
+
+def test_ensure_config_merges_callbacks_across_configs() -> None:
+    a_cb = _TrackingCB("a")
+    b_cb = _TrackingCB("b")
+    merged = ensure_config({"callbacks": [a_cb]}, {"callbacks": [b_cb]})
+    assert merged["callbacks"] == [a_cb, b_cb]
+
+
+def test_ensure_config_none_inputs_ignored() -> None:
+    # mixed with None should not raise
+    merged = ensure_config(None, {"tags": ["t"]}, None)
+    assert merged["tags"] == ["t"]
+
+
+def test_ensure_config_empty_inputs() -> None:
+    # everything empty -> defaults
+    merged = ensure_config()
+    assert merged["tags"] == []
+    assert merged["configurable"] == {}
+    assert merged["callbacks"] is None

libs/sdk-py/integration/graph/tools_agent.py

@@ -1,13 +1,13 @@
-"""create_agent-based example exercising the v3 ``tools`` channel.
+"""create_agent-based example exercising the v3 `tools` channel.
 
-`thread.tool_calls` and the underlying ``tools`` channel only emit
+`thread.tool_calls` and the underlying `tools` channel only emit
 events when an actual model issues a tool call through langchain's
-agent stack. The synthetic ``streaming_graph.py`` hand-builds
+agent stack. The synthetic `streaming_graph.py` hand-builds
 `AIMessage(tool_calls=[...])` and a `ToolMessage` via the messages
 reducer — that gets persisted in state but never produces tool-call
 telemetry on the wire. This graph fixes that by going through
-`create_agent` with a real tool, driven by a `GenericFakeChatModel` so
-the test stays hermetic (no `ANTHROPIC_API_KEY` required).
+`create_agent` with a real tool, driven by a hermetic fake chat model
+(no `ANTHROPIC_API_KEY` required).
 
 Flow on `run.start`:
 
@@ -25,7 +25,8 @@ from typing import Any
 
 from langchain.agents import create_agent
 from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
-from langchain_core.messages import AIMessage
+from langchain_core.messages import AIMessage, BaseMessage, ToolMessage
+from langchain_core.outputs import ChatGeneration, ChatResult
 from langchain_core.tools import tool
 
 
@@ -36,51 +37,56 @@ def search(query: str) -> str:
 
 
 class _ToolBindingFakeChatModel(FakeMessagesListChatModel):
-    """Fake chat model that satisfies `create_agent`'s ``bind_tools`` call.
+    """Stateless fake chat model driving a single `search` tool call.
 
-    ``create_agent`` calls ``model.bind_tools(tools)`` to attach the tool
-    schema (``langchain/agents/factory.py:1284``). The base
-    ``FakeMessagesListChatModel`` inherits ``BaseChatModel.bind_tools``,
-    which raises ``NotImplementedError``. We don't actually need the
-    bound schema — the fake replays scripted ``AIMessage``s with their
-    own ``tool_calls`` field — so override ``bind_tools`` as a no-op.
+    `create_agent` calls `model.bind_tools(tools)` to attach the tool
+    schema (`langchain/agents/factory.py:1284`). The base
+    `FakeMessagesListChatModel` inherits `BaseChatModel.bind_tools`,
+    which raises `NotImplementedError`, so `bind_tools` is overridden as
+    a no-op (the reply is hand-built and already carries `tool_calls`).
 
-    ``FakeMessagesListChatModel`` is preferred over
-    ``GenericFakeChatModel`` here because the latter's ``_stream``
-    breaks the message into content chunks and **drops ``tool_calls``**
-    when content is empty, causing the v2 streaming path inside
-    ``create_agent`` to raise ``RuntimeError("v2 stream finished
-    without producing a message")``. ``FakeMessagesListChatModel``
-    falls back to the default ``_stream`` that yields the whole
-    message in one chunk, preserving ``tool_calls``.
+    The reply is derived from conversation state rather than a cycling
+    response list: the `search` tool call is issued until a `ToolMessage`
+    appears, then a terminating `AIMessage`. This avoids the response-index
+    parity flake where `FakeMessagesListChatModel.responses` is shared
+    process-wide and cycles `0 -> 1 -> 0`; a run that started mid-cycle
+    (e.g. on a reused server worker) would reply `"done."` first and emit
+    no tool call. Being order-independent, every run emits exactly one
+    tool call regardless of how many times the model was previously called.
+
+    `FakeMessagesListChatModel` is subclassed (rather than
+    `GenericFakeChatModel`) because the latter's `_stream` breaks the
+    message into content chunks and drops `tool_calls` when content is
+    empty, causing the v2 streaming path inside `create_agent` to raise
+    `RuntimeError("v2 stream finished without producing a message")`.
+    The inherited `_stream` yields the whole message in one chunk,
+    preserving `tool_calls`.
     """
 
     def bind_tools(self, tools: Any, **kwargs: Any) -> _ToolBindingFakeChatModel:
         return self
 
-
-# Two scripted turns. ``FakeMessagesListChatModel`` cycles through
-# ``responses`` (resetting to index 0 after the last) so the graph
-# can be run many times without restart; per run, ``create_agent``
-# invokes the model exactly twice (once to issue the tool call,
-# once after the tool result to produce the terminating answer).
-_supervisor_responses: list[AIMessage] = [
-    AIMessage(
-        content="",
-        id="ai-tools-1",
-        tool_calls=[
-            {
-                "id": "tc-1",
-                "name": "search",
-                "args": {"query": "v3"},
-            }
-        ],
-    ),
-    AIMessage(content="done.", id="ai-tools-2"),
-]
+    def _generate(
+        self,
+        messages: list[BaseMessage],
+        stop: list[str] | None = None,
+        run_manager: Any = None,
+        **kwargs: Any,
+    ) -> ChatResult:
+        if any(isinstance(m, ToolMessage) for m in messages):
+            response = AIMessage(content="done.", id="ai-tools-done")
+        else:
+            response = AIMessage(
+                content="",
+                id="ai-tools-call",
+                tool_calls=[{"id": "tc-1", "name": "search", "args": {"query": "v3"}}],
+            )
+        return ChatResult(generations=[ChatGeneration(message=response)])
 
 
-_supervisor_model = _ToolBindingFakeChatModel(responses=_supervisor_responses)
+# `responses` is a required field on `FakeMessagesListChatModel`, but the
+# overridden `_generate` derives its reply from state and never reads it.
+_supervisor_model = _ToolBindingFakeChatModel(responses=[])
 
 
 graph = create_agent(

libs/sdk-py/langgraph_sdk/__init__.py

@@ -3,6 +3,6 @@ from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
 from langgraph_sdk.encryption.types import EncryptionContext
 
-__version__ = "0.3.15"
+__version__ = "0.4.2"
 
 __all__ = ["Auth", "Encryption", "EncryptionContext", "get_client", "get_sync_client"]

libs/sdk-py/langgraph_sdk/_async/stream.py

@@ -18,7 +18,7 @@ import contextlib
 import random
 from collections.abc import AsyncGenerator, AsyncIterator, Generator, Mapping
 from dataclasses import dataclass, field
-from typing import Any, Literal, TypedDict
+from typing import Any, Literal, TypedDict, cast
 
 from langchain_core.language_models.chat_model_stream import AsyncChatModelStream
 from langchain_protocol import Event, SubscribeParams
@@ -26,6 +26,16 @@ from langchain_protocol import Event, SubscribeParams
 from langgraph_sdk._async.http import HttpClient
 from langgraph_sdk.schema import QueryParamTypes
 from langgraph_sdk.stream.controller import _SeenEventIds
+from langgraph_sdk.stream.decoders import (
+    DataDecoder,
+    Decoder,
+    ExtensionsDecoder,
+    MessagesDecoder,
+    SubgraphsDecoder,
+    ToolCallsDecoder,
+    validate_interleave_channels,
+)
+from langgraph_sdk.stream.subscription import compute_union_filter, infer_channel
 from langgraph_sdk.stream.transport import (
     AsyncProtocolTransport,
     EventStreamHandle,
@@ -348,6 +358,7 @@ class _ValuesProjection:
             raise RuntimeError("AsyncThreadStream not entered — use `async with`.")
         params: SubscribeParams = {"channels": ["values"]}
         sub = self._thread._register_subscription(params)
+        decoder = DataDecoder("values")
         try:
             await self._thread._reconcile_stream(params)
             self._thread._ensure_fanout_running()
@@ -357,12 +368,8 @@ class _ValuesProjection:
                 item = await sub.queue.get()
                 if item is None:
                     return
-                params_field = item.get("params") or {}
-                data = (
-                    params_field.get("data") if isinstance(params_field, dict) else None
-                )
-                if data is not None:
-                    yield data
+                for out in decoder.feed(item):
+                    yield out
         finally:
             self._thread._unregister_subscription(sub.id)
 
@@ -397,7 +404,13 @@ class _MessagesProjection:
             return
         params = _exact_namespace_params(["messages"], self._namespace)
         sub = self._thread._register_subscription(params)
-        active: dict[str, AsyncChatModelStream] = {}
+        decoder = MessagesDecoder(
+            namespace=self._namespace,
+            stream_factory=lambda *, namespace, node, message_id: AsyncChatModelStream(
+                namespace=namespace, node=node, message_id=message_id
+            ),
+        )
+        registered: list[AsyncChatModelStream] = []
         try:
             await self._thread._reconcile_stream(params)
             self._thread._ensure_fanout_running()
@@ -405,59 +418,12 @@ class _MessagesProjection:
                 item = await sub.queue.get()
                 if item is None:
                     return
-                params_field = item.get("params") or {}
-                if _event_namespace(params_field) != self._namespace:
-                    continue
-                data = (
-                    params_field.get("data") if isinstance(params_field, dict) else None
-                )
-                if not isinstance(data, dict):
-                    continue
-                event_type = data.get("event")
-                if event_type == "message-start":
-                    message_id = _message_event_id(data)
-                    key = _message_route_key(data, fallback=message_id)
-                    metadata = (
-                        data.get("metadata")
-                        if isinstance(data.get("metadata"), dict)
-                        else {}
-                    )
-                    stream = AsyncChatModelStream(
-                        namespace=list(self._namespace),
-                        node=metadata.get("langgraph_node") if metadata else None,
-                        message_id=message_id,
-                    )
-                    active[key] = stream
+                for stream in decoder.feed(item):
                     self._thread._register_active_message_stream(stream)
-                    stream.dispatch(data)
+                    registered.append(stream)
                     yield stream
-                else:
-                    key = _message_route_key(data)
-                    stream = active.get(key)
-                    if stream is None and key == "__single__" and len(active) == 1:
-                        # Content-block events (content-block-start /
-                        # content-block-delta / content-block-finish /
-                        # message-finish) don't carry the message ``id``
-                        # on the wire, so ``_message_route_key`` returns
-                        # ``__single__`` while the active stream was
-                        # registered under ``message:<id>``. When exactly
-                        # one stream is active, that mismatch is
-                        # unambiguous -- the events belong to it.
-                        # Events that DO carry an explicit id which
-                        # doesn't match any active stream are still
-                        # dropped (orphan-delta safety, see
-                        # ``test_messages_orphan_delta_without_matching_key_is_dropped``).
-                        stream = next(iter(active.values()))
-                    if stream is None:
-                        continue
-                    stream.dispatch(data)
-                    if event_type in ("message-finish", "error"):
-                        self._thread._unregister_active_message_stream(stream)
-                        for route_key, candidate in list(active.items()):
-                            if candidate is stream:
-                                del active[route_key]
         finally:
-            for stream in active.values():
+            for stream in registered:
                 self._thread._unregister_active_message_stream(stream)
             self._thread._unregister_subscription(sub.id)
 
@@ -950,10 +916,17 @@ class _SubgraphsProjection:
             raise RuntimeError("AsyncThreadStream not entered - use `async with`.")
         params = _subgraph_subscription_params(self._scope)
         sub = self._thread._register_subscription(params)
-        seen: set[tuple[str, ...]] = set()
-        active: dict[tuple[str, ...], ScopedStreamHandle] = {}
-        # Activate root inbox so scope-level messages events consumed here are
-        # forwarded to `thread.messages` even after the shared SSE ends.
+        decoder = SubgraphsDecoder(
+            scope=self._scope,
+            handle_factory=lambda *, path, graph_name, trigger_call_id: (
+                ScopedStreamHandle(
+                    thread=self._thread,
+                    path=path,
+                    graph_name=graph_name,
+                    trigger_call_id=trigger_call_id,
+                )
+            ),
+        )
         root_inbox: asyncio.Queue[Event | None] | None = (
             self._thread._activate_root_messages_inbox() if not self._scope else None
         )
@@ -965,80 +938,14 @@ class _SubgraphsProjection:
                 if item is None:
                     return
                 params_field = item.get("params") or {}
-                namespace = _event_namespace(params_field)
-                data = (
-                    params_field.get("data") if isinstance(params_field, dict) else None
-                )
-                if not isinstance(data, dict):
-                    continue
-                method = item.get("method")
-
-                # Route events at a child namespace (or deeper) to that child
-                # handle's channel inbox so sequential child-projection
-                # consumption works without opening a second SSE.
-                ns_tuple = tuple(namespace)
-                routed_to_child = False
-                for child_path, child_handle in active.items():
-                    child_len = len(child_path)
-                    if (
-                        len(ns_tuple) >= child_len
-                        and ns_tuple[:child_len] == child_path
-                    ):
-                        child_handle._push_event(item)
-                        routed_to_child = True
-                        break
-
-                # Scope-level messages events are not routed to any child; forward
-                # them to the root inbox so `thread.messages` can drain them after
-                # this projection finishes (dedup prevents the SSE from replaying).
                 if (
-                    not routed_to_child
-                    and root_inbox is not None
-                    and method == "messages"
-                    and tuple(namespace) == self._scope
+                    root_inbox is not None
+                    and item.get("method") == "messages"
+                    and tuple(_event_namespace(params_field)) == self._scope
                 ):
                     root_inbox.put_nowait(item)
-
-                if method == "tasks":
-                    if "result" in data:
-                        self._apply_tasks_result(namespace, data, active)
-                    elif _is_direct_child(namespace, self._scope):
-                        path = tuple(namespace)
-                        if path not in seen:
-                            seen.add(path)
-                            graph_name, trigger_call_id = _parse_namespace_segment(
-                                path[-1]
-                            )
-                            handle = ScopedStreamHandle(
-                                thread=self._thread,
-                                path=path,
-                                graph_name=graph_name or None,
-                                trigger_call_id=trigger_call_id,
-                            )
-                            active[path] = handle
-                            yield handle
-                elif (
-                    method == "lifecycle"
-                    and data.get("event") == "started"
-                    and _is_direct_child(namespace, self._scope)
-                ):
-                    # ``create_deep_agent`` and similar surfaces signal
-                    # subagent invocation via a child-namespace
-                    # ``lifecycle: started`` event rather than a ``tasks``
-                    # event. JS does the same (see ``langgraphjs``
-                    # ``stream/handles/subgraphs.ts``).
-                    path = tuple(namespace)
-                    if path not in seen:
-                        seen.add(path)
-                        graph_name, trigger_call_id = _parse_namespace_segment(path[-1])
-                        handle = ScopedStreamHandle(
-                            thread=self._thread,
-                            path=path,
-                            graph_name=graph_name or None,
-                            trigger_call_id=trigger_call_id,
-                        )
-                        active[path] = handle
-                        yield handle
+                for handle in decoder.feed(item):
+                    yield handle
         finally:
             # Determine terminal status from the parent run's lifecycle result.
             # If _run_done resolved as errored, force-complete remaining children
@@ -1049,32 +956,13 @@ class _SubgraphsProjection:
                 result = run_done.result()
                 if isinstance(result, _RunTerminal) and result.status == "errored":
                     terminal_status = "failed"
-            for handle in active.values():
+            for handle in decoder._active.values():
                 if handle.status == "started":
                     handle._finish(terminal_status)
             self._thread._unregister_subscription(sub.id)
             if root_inbox is not None:
                 root_inbox.put_nowait(None)
 
-    def _apply_tasks_result(
-        self,
-        namespace: list[str],
-        data: dict[str, Any],
-        active: dict[tuple[str, ...], ScopedStreamHandle],
-    ) -> None:
-        result_id = data.get("id")
-        if not result_id:
-            return
-        parent_path = tuple(namespace)
-        for child_path, handle in list(active.items()):
-            if child_path[:-1] != parent_path:
-                continue
-            if handle.trigger_call_id != result_id:
-                continue
-            status, error = _terminal_from_tasks_result(data)
-            handle._finish(status, error)
-            del active[child_path]
-
 
 class ToolCallHandle:
     """Async handle for one root-scope tool call."""
@@ -1161,7 +1049,18 @@ class _ToolCallsProjection:
             raise RuntimeError("AsyncThreadStream not entered - use `async with`.")
         params = _exact_namespace_params(["tools"], self._namespace)
         sub = self._thread._register_subscription(params)
-        active: dict[str, ToolCallHandle] = {}
+        decoder = ToolCallsDecoder(
+            namespace=self._namespace,
+            handle_factory=lambda *, tool_call_id, name, input, namespace: (
+                ToolCallHandle(
+                    tool_call_id=tool_call_id,
+                    name=name,
+                    input=input,
+                    namespace=namespace,
+                )
+            ),
+        )
+        registered: list[ToolCallHandle] = []
         try:
             await self._thread._reconcile_stream(params)
             self._thread._ensure_fanout_running()
@@ -1169,52 +1068,10 @@ class _ToolCallsProjection:
                 item = await sub.queue.get()
                 if item is None:
                     return
-                params_field = item.get("params") or {}
-                if _event_namespace(params_field) != self._namespace:
-                    continue
-                data = (
-                    params_field.get("data") if isinstance(params_field, dict) else None
-                )
-                if not isinstance(data, dict):
-                    continue
-                event_type = data.get("event")
-                tool_call_id = data.get("tool_call_id")
-                if not isinstance(tool_call_id, str):
-                    continue
-
-                if event_type == "tool-started":
-                    tool_name = data.get("tool_name")
-                    if not isinstance(tool_name, str):
-                        tool_name = ""
-                    handle = ToolCallHandle(
-                        tool_call_id=tool_call_id,
-                        name=tool_name,
-                        input=data.get("input"),
-                        namespace=list(self._namespace),
-                    )
-                    active[tool_call_id] = handle
+                for handle in decoder.feed(item):
                     self._thread._register_active_tool_call(handle)
+                    registered.append(handle)
                     yield handle
-                elif event_type == "tool-output-delta":
-                    handle = active.get(tool_call_id)
-                    delta = data.get("delta")
-                    if handle is not None and isinstance(delta, str):
-                        handle._push_delta(delta)
-                elif event_type == "tool-finished":
-                    handle = active.pop(tool_call_id, None)
-                    if handle is not None:
-                        self._thread._unregister_active_tool_call(handle)
-                        handle._finish(data.get("output"))
-                elif event_type == "tool-error":
-                    handle = active.pop(tool_call_id, None)
-                    if handle is not None:
-                        self._thread._unregister_active_tool_call(handle)
-                        message = data.get("message")
-                        handle._fail(
-                            RuntimeError(
-                                str(message) if message else "Tool call errored"
-                            )
-                        )
         finally:
             # Read terminal error from _run_done if it is already resolved.
             # We do NOT block here: callers who need a terminal observation
@@ -1226,14 +1083,15 @@ class _ToolCallsProjection:
             if run_done is not None and run_done.done() and not run_done.cancelled():
                 terminal = run_done.result()
                 terminal_err = terminal.error
-            err = (
+            err: BaseException = (
                 terminal_err
                 if terminal_err is not None
                 else RuntimeError("Tool call stream closed before terminal tool event.")
             )
-            for handle in active.values():
-                self._thread._unregister_active_tool_call(handle)
+            for handle in list(decoder._active.values()):
                 handle._fail(err)
+            for handle in registered:
+                self._thread._unregister_active_tool_call(handle)
             self._thread._unregister_subscription(sub.id)
 
 
@@ -1280,6 +1138,7 @@ class _ExtensionProjection:
         if self._namespace:
             params["namespaces"] = [self._namespace]
         sub = self._thread._register_subscription(params)
+        decoder = ExtensionsDecoder(name=self._name)
         try:
             if self._thread._closed:
                 return
@@ -1289,12 +1148,8 @@ class _ExtensionProjection:
                 item = await sub.queue.get()
                 if item is None:
                     return
-                event_params = item.get("params") or {}
-                data = (
-                    event_params.get("data") if isinstance(event_params, dict) else None
-                )
-                if isinstance(data, dict):
-                    yield data
+                for out in decoder.feed(item):
+                    yield out
         finally:
             self._thread._unregister_subscription(sub.id)
 
@@ -1558,6 +1413,174 @@ class AsyncThreadStream:
             params["depth"] = depth
         return self._subscription_iter(params)
 
+    async def interleave_projections(
+        self, channels: list[str]
+    ) -> AsyncIterator[tuple[str, Any]]:
+        """Yield `(channel_name, item)` tuples across multiple projections.
+
+        One shared subscription drives all per-channel decoders; items arrive
+        in server-emit order (the SDK analog of `GraphRunStream.interleave`).
+
+        Args:
+            channels: Flat list of `"values"`, `"messages"`, `"tool_calls"`,
+                `"subgraphs"`, and/or extension names. Built-ins yield their
+                typed item (snapshot dict / `AsyncChatModelStream` /
+                `ToolCallHandle` / `ScopedStreamHandle`); an extension yields
+                its payload dict, keyed by the bare extension name.
+
+        Note:
+            Handles and streams are yielded eagerly (before their sub-stream
+            completes), so items arrive interleaved in real time. To receive a
+            fully-resolved handle (output already populated), use the dedicated
+            `thread.tool_calls` / `thread.messages` projections instead.
+        """
+        validate_interleave_channels(channels)
+        if self._transport is None:
+            raise RuntimeError("AsyncThreadStream not entered — use `async with`.")
+        decoders: dict[str, Decoder] = {}
+        sub_params: list[SubscribeParams] = []
+        for ch in channels:
+            if ch == "values":
+                decoders[ch] = DataDecoder("values")
+                sub_params.append({"channels": ["values"]})
+            elif ch in ("updates", "checkpoints", "tasks"):
+                # Plain payload channels (local Updates/Checkpoints/Tasks
+                # analog). Root-scope filter is load-bearing: a co-requested
+                # unscoped `values` widens the merged subscription to all
+                # namespaces, so the decoder itself keeps subgraph payloads out.
+                decoders[ch] = DataDecoder(ch, namespace=[])
+                sub_params.append(_exact_namespace_params([ch], []))
+            elif ch == "messages":
+                decoders[ch] = MessagesDecoder(
+                    namespace=[],
+                    stream_factory=lambda *, namespace, node, message_id: (
+                        AsyncChatModelStream(
+                            namespace=namespace, node=node, message_id=message_id
+                        )
+                    ),
+                )
+                sub_params.append(_exact_namespace_params(["messages"], []))
+            elif ch == "tool_calls":
+                decoders[ch] = ToolCallsDecoder(
+                    namespace=[],
+                    handle_factory=lambda *, tool_call_id, name, input, namespace: (
+                        ToolCallHandle(
+                            tool_call_id=tool_call_id,
+                            name=name,
+                            input=input,
+                            namespace=namespace,
+                        )
+                    ),
+                )
+                sub_params.append(_exact_namespace_params(["tools"], []))
+            elif ch == "subgraphs":
+                decoders[ch] = SubgraphsDecoder(
+                    scope=(),
+                    handle_factory=lambda *, path, graph_name, trigger_call_id: (
+                        ScopedStreamHandle(
+                            thread=self,
+                            path=path,
+                            graph_name=graph_name,
+                            trigger_call_id=trigger_call_id,
+                        )
+                    ),
+                )
+                sub_params.append(_subgraph_subscription_params(()))
+            else:
+                decoders[ch] = ExtensionsDecoder(name=ch)
+                sub_params.append({"channels": [f"custom:{ch}"]})
+        if not sub_params:
+            return
+        merged = cast(
+            SubscribeParams,
+            compute_union_filter(cast(list[dict[str, Any]], sub_params)),
+        )
+        subgraphs = decoders.get("subgraphs")
+        # Track decoder-created handles so teardown can finalize anything still
+        # in flight; otherwise an awaiting `handle.output` / `handle.messages`
+        # would hang after an early break or run termination.
+        registered_tool_calls: list[ToolCallHandle] = []
+        registered_message_streams: list[AsyncChatModelStream] = []
+        try:
+            async for event in self._subscription_iter(merged):
+                if subgraphs is not None:
+                    for item in subgraphs.feed(event):
+                        yield ("subgraphs", item)
+                wire = infer_channel(event)
+                public = self._interleave_public_name(wire)
+                # subgraphs is driven separately above (it consumes all events); never dispatch it here.
+                if public is not None and public != "subgraphs":
+                    decoder = decoders.get(public)
+                    if decoder is not None:
+                        for item in decoder.feed(event):
+                            if public == "tool_calls":
+                                self._register_active_tool_call(item)
+                                registered_tool_calls.append(item)
+                            elif public == "messages":
+                                self._register_active_message_stream(item)
+                                registered_message_streams.append(item)
+                            yield (public, item)
+        finally:
+            self._finalize_interleave_decoders(
+                decoders.get("tool_calls"),
+                subgraphs,
+                registered_tool_calls,
+                registered_message_streams,
+            )
+
+    @staticmethod
+    def _interleave_public_name(wire: str | None) -> str | None:
+        """Map a wire channel name to the public channel name used in interleave tuples."""
+        if wire is None:
+            return None
+        if wire == "tools":
+            return "tool_calls"
+        if wire.startswith("custom:"):
+            return wire[len("custom:") :]
+        return wire  # values, messages (tasks/lifecycle pass through with no decoder match)
+
+    def _finalize_interleave_decoders(
+        self,
+        tool_calls: Decoder | None,
+        subgraphs: Decoder | None,
+        registered_tool_calls: list[ToolCallHandle],
+        registered_message_streams: list[AsyncChatModelStream],
+    ) -> None:
+        """Finalize in-flight handles when `interleave_projections` tears down.
+
+        Mirrors the terminal handling of the dedicated `_ToolCallsProjection` /
+        `_SubgraphsProjection`: in-flight tool calls are failed (so awaiting
+        `handle.output` can't hang) and discovered subgraph children are
+        force-completed with the run's terminal status.
+        """
+        run_done = self._run_done
+        resolved = (
+            run_done.result()
+            if run_done is not None and run_done.done() and not run_done.cancelled()
+            else None
+        )
+        if isinstance(tool_calls, ToolCallsDecoder):
+            err: BaseException = (
+                resolved.error
+                if resolved is not None and resolved.error is not None
+                else RuntimeError("Tool call stream closed before terminal tool event.")
+            )
+            for handle in list(tool_calls._active.values()):
+                handle._fail(err)
+        for handle in registered_tool_calls:
+            self._unregister_active_tool_call(handle)
+        for stream in registered_message_streams:
+            self._unregister_active_message_stream(stream)
+        if isinstance(subgraphs, SubgraphsDecoder):
+            terminal_status: SubgraphStatus = (
+                "failed"
+                if isinstance(resolved, _RunTerminal) and resolved.status == "errored"
+                else "completed"
+            )
+            for child in subgraphs._active.values():
+                if child.status == "started":
+                    child._finish(terminal_status)
+
     async def _subscription_iter(
         self, params: SubscribeParams
     ) -> AsyncGenerator[Event, None]:

libs/sdk-py/langgraph_sdk/_sync/stream.py

@@ -17,13 +17,23 @@ import queue
 import threading
 from collections.abc import Iterator, Mapping
 from dataclasses import dataclass
-from typing import Any, Literal, TypedDict
+from typing import Any, Literal, TypedDict, cast
 
 from langchain_core.language_models.chat_model_stream import ChatModelStream
 from langchain_protocol import Event, SubscribeParams
 
 from langgraph_sdk._sync.http import SyncHttpClient
 from langgraph_sdk.schema import QueryParamTypes
+from langgraph_sdk.stream.decoders import (
+    DataDecoder,
+    Decoder,
+    ExtensionsDecoder,
+    MessagesDecoder,
+    SubgraphsDecoder,
+    ToolCallsDecoder,
+    validate_interleave_channels,
+)
+from langgraph_sdk.stream.subscription import compute_union_filter, infer_channel
 from langgraph_sdk.stream.sync_controller import SyncStreamController, _SyncSubscription
 from langgraph_sdk.stream.transport import (
     SyncEventStreamHandle,
@@ -295,6 +305,7 @@ class _SyncValuesProjection:
             raise RuntimeError("SyncThreadStream not entered — use `with`.")
         params: SubscribeParams = {"channels": ["values"]}
         sub = self._thread._register_subscription(params)
+        decoder = DataDecoder("values")
         try:
             self._thread._reconcile_stream(params)
             self._thread._ensure_fanout_running()
@@ -304,12 +315,7 @@ class _SyncValuesProjection:
                 item = sub.queue.get()
                 if item is None:
                     return
-                params_field = item.get("params") or {}
-                data = (
-                    params_field.get("data") if isinstance(params_field, dict) else None
-                )
-                if data is not None:
-                    yield data
+                yield from decoder.feed(cast(dict[str, Any], item))
         finally:
             self._thread._unregister_subscription(sub.id)
 
@@ -340,87 +346,35 @@ class _SyncMessagesProjection:
             return
         params = _exact_namespace_params(["messages"], self._namespace)
         sub = self._thread._register_subscription(params)
-        active: dict[str, ChatModelStream] = {}
+        decoder = MessagesDecoder(
+            namespace=self._namespace,
+            stream_factory=lambda *, namespace, node, message_id: ChatModelStream(
+                namespace=namespace, node=node, message_id=message_id
+            ),
+        )
+        registered: list[ChatModelStream] = []
+        pending: list[ChatModelStream] = []
         try:
             self._thread._reconcile_stream(params)
             self._thread._ensure_fanout_running()
             while True:
                 item = sub.queue.get()
                 if item is None:
+                    # EOF: surface remaining streams (possibly incomplete) in start order.
+                    while pending:
+                        yield pending.pop(0)
                     return
-                params_field = item.get("params") or {}
-                if _event_namespace(params_field) != self._namespace:
-                    continue
-                data = (
-                    params_field.get("data") if isinstance(params_field, dict) else None
-                )
-                if not isinstance(data, dict):
-                    continue
-                event_type = data.get("event")
-                if event_type == "message-start":
-                    message_id = _message_event_id(data)
-                    key = _message_route_key(data, fallback=message_id)
-                    metadata = (
-                        data.get("metadata")
-                        if isinstance(data.get("metadata"), dict)
-                        else {}
-                    )
-                    stream = ChatModelStream(
-                        namespace=list(self._namespace),
-                        node=metadata.get("langgraph_node") if metadata else None,
-                        message_id=message_id,
-                    )
-                    active[key] = stream
+                for stream in decoder.feed(cast(dict[str, Any], item)):
                     self._thread._register_active_message_stream(stream)
-                    stream.dispatch(data)
-                    # Pre-dispatch all remaining events for this message so the
-                    # caller can access str(message.text) inside a for loop.
-                    while not stream._done:
-                        next_item = sub.queue.get()
-                        if next_item is None:
-                            sub.queue.put(None)
-                            break
-                        next_params = next_item.get("params") or {}
-                        next_data = (
-                            next_params.get("data")
-                            if isinstance(next_params, dict)
-                            else None
-                        )
-                        if not isinstance(next_data, dict):
-                            continue
-                        next_event_type = next_data.get("event")
-                        next_key = _message_route_key(next_data)
-                        target = active.get(next_key)
-                        if (
-                            target is None
-                            and next_key == "__single__"
-                            and len(active) == 1
-                        ):
-                            target = next(iter(active.values()))
-                        if target is not None:
-                            target.dispatch(next_data)
-                            if next_event_type in ("message-finish", "error"):
-                                self._thread._unregister_active_message_stream(target)
-                                for rk, cand in list(active.items()):
-                                    if cand is target:
-                                        del active[rk]
-                    yield stream
-                else:
-                    key = _message_route_key(data)
-                    stream = active.get(key)
-                    if stream is None and key == "__single__" and len(active) == 1:
-                        stream = next(iter(active.values()))
-                    if stream is None:
-                        continue
-                    stream.dispatch(data)
-                    if event_type in ("message-finish", "error"):
-                        self._thread._unregister_active_message_stream(stream)
-                        for route_key, candidate in list(active.items()):
-                            if candidate is stream:
-                                del active[route_key]
+                    registered.append(stream)
+                    pending.append(stream)
+                # Surface in start order once each (and all earlier) streams are done,
+                # so `str(stream.text)` is ready on yield (sync pre-dispatch contract).
+                while pending and pending[0]._done:
+                    yield pending.pop(0)
         finally:
-            for s in active.values():
-                self._thread._unregister_active_message_stream(s)
+            for stream in registered:
+                self._thread._unregister_active_message_stream(stream)
             self._thread._unregister_subscription(sub.id)
 
 
@@ -595,100 +549,37 @@ class _SyncToolCallsProjection:
             raise RuntimeError("SyncThreadStream not entered — use `with`.")
         params = _exact_namespace_params(["tools"], self._namespace)
         sub = self._thread._register_subscription(params)
-        active: dict[str, SyncToolCallHandle] = {}
+        decoder = ToolCallsDecoder(
+            namespace=self._namespace,
+            handle_factory=lambda *, tool_call_id, name, input, namespace: (
+                SyncToolCallHandle(
+                    tool_call_id=tool_call_id,
+                    name=name,
+                    input=input,
+                    namespace=namespace,
+                )
+            ),
+        )
+        registered: list[SyncToolCallHandle] = []
+        pending: list[SyncToolCallHandle] = []
         try:
             self._thread._reconcile_stream(params)
             self._thread._ensure_fanout_running()
             while True:
                 item = sub.queue.get()
                 if item is None:
+                    # EOF: surface remaining handles (possibly incomplete) in start order.
+                    while pending:
+                        yield pending.pop(0)
                     return
-                params_field = item.get("params") or {}
-                if _event_namespace(params_field) != self._namespace:
-                    continue
-                data = (
-                    params_field.get("data") if isinstance(params_field, dict) else None
-                )
-                if not isinstance(data, dict):
-                    continue
-                event_type = data.get("event")
-                tool_call_id = data.get("tool_call_id")
-                if not isinstance(tool_call_id, str):
-                    continue
-                if event_type == "tool-started":
-                    tool_name = data.get("tool_name")
-                    if not isinstance(tool_name, str):
-                        tool_name = ""
-                    handle = SyncToolCallHandle(
-                        tool_call_id=tool_call_id,
-                        name=tool_name,
-                        input=data.get("input"),
-                        namespace=list(self._namespace),
-                    )
-                    active[tool_call_id] = handle
+                for handle in decoder.feed(cast(dict[str, Any], item)):
                     self._thread._register_active_tool_call(handle)
-                    # Pre-dispatch events until this tool call completes so that
-                    # `call.output` is resolved when the caller receives the handle.
-                    while not handle.done:
-                        next_item = sub.queue.get()
-                        if next_item is None:
-                            sub.queue.put(None)
-                            break
-                        next_params = next_item.get("params") or {}
-                        if _event_namespace(next_params) != self._namespace:
-                            continue
-                        next_data = (
-                            next_params.get("data")
-                            if isinstance(next_params, dict)
-                            else None
-                        )
-                        if not isinstance(next_data, dict):
-                            continue
-                        next_event_type = next_data.get("event")
-                        next_tcid = next_data.get("tool_call_id")
-                        if not isinstance(next_tcid, str):
-                            continue
-                        if next_event_type == "tool-output-delta":
-                            h = active.get(next_tcid)
-                            delta = next_data.get("delta")
-                            if h is not None and isinstance(delta, str):
-                                h._push_delta(delta)
-                        elif next_event_type == "tool-finished":
-                            h = active.pop(next_tcid, None)
-                            if h is not None:
-                                self._thread._unregister_active_tool_call(h)
-                                h._finish(next_data.get("output"))
-                        elif next_event_type == "tool-error":
-                            h = active.pop(next_tcid, None)
-                            if h is not None:
-                                self._thread._unregister_active_tool_call(h)
-                                message = next_data.get("message")
-                                h._fail(
-                                    RuntimeError(
-                                        str(message) if message else "Tool call errored"
-                                    )
-                                )
-                    yield handle
-                elif event_type == "tool-output-delta":
-                    h = active.get(tool_call_id)
-                    delta = data.get("delta")
-                    if h is not None and isinstance(delta, str):
-                        h._push_delta(delta)
-                elif event_type == "tool-finished":
-                    h = active.pop(tool_call_id, None)
-                    if h is not None:
-                        self._thread._unregister_active_tool_call(h)
-                        h._finish(data.get("output"))
-                elif event_type == "tool-error":
-                    h = active.pop(tool_call_id, None)
-                    if h is not None:
-                        self._thread._unregister_active_tool_call(h)
-                        message = data.get("message")
-                        h._fail(
-                            RuntimeError(
-                                str(message) if message else "Tool call errored"
-                            )
-                        )
+                    registered.append(handle)
+                    pending.append(handle)
+                # Surface in start order once each (and all earlier) handles are done,
+                # so `call.output` is resolved when the caller receives the handle.
+                while pending and pending[0].done:
+                    yield pending.pop(0)
         finally:
             # Read terminal error from _run_done if it is already resolved.
             # We do NOT block here: callers who need a terminal observation
@@ -708,9 +599,10 @@ class _SyncToolCallsProjection:
                 if terminal_err is not None
                 else RuntimeError("Tool call stream closed before terminal tool event.")
             )
-            for h in active.values():
-                self._thread._unregister_active_tool_call(h)
-                h._fail(err)
+            for handle in list(decoder._active.values()):
+                handle._fail(err)
+            for handle in registered:
+                self._thread._unregister_active_tool_call(handle)
             self._thread._unregister_subscription(sub.id)
 
 
@@ -1067,8 +959,17 @@ class _SyncSubgraphsProjection:
             raise RuntimeError("SyncThreadStream not entered — use `with`.")
         params = _subgraph_subscription_params(self._scope)
         sub = self._thread._register_subscription(params)
-        seen: set[tuple[str, ...]] = set()
-        active: dict[tuple[str, ...], SyncScopedStreamHandle] = {}
+        decoder = SubgraphsDecoder(
+            scope=self._scope,
+            handle_factory=lambda *, path, graph_name, trigger_call_id: (
+                SyncScopedStreamHandle(
+                    thread=self._thread,
+                    path=path,
+                    graph_name=graph_name,
+                    trigger_call_id=trigger_call_id,
+                )
+            ),
+        )
         root_inbox: queue.Queue[Event | None] | None = (
             self._thread._activate_root_messages_inbox() if not self._scope else None
         )
@@ -1080,71 +981,14 @@ class _SyncSubgraphsProjection:
                 if item is None:
                     return
                 params_field = item.get("params") or {}
-                namespace = _event_namespace(params_field)
-                data = (
-                    params_field.get("data") if isinstance(params_field, dict) else None
-                )
-                if not isinstance(data, dict):
-                    continue
-                method = item.get("method")
-
-                ns_tuple = tuple(namespace)
-                routed_to_child = False
-                for child_path, child_handle in active.items():
-                    child_len = len(child_path)
-                    if (
-                        len(ns_tuple) >= child_len
-                        and ns_tuple[:child_len] == child_path
-                    ):
-                        child_handle._push_event(item)
-                        routed_to_child = True
-                        break
-
                 if (
-                    not routed_to_child
-                    and root_inbox is not None
-                    and method == "messages"
-                    and tuple(namespace) == self._scope
+                    root_inbox is not None
+                    and item.get("method") == "messages"
+                    and tuple(_event_namespace(params_field)) == self._scope
                 ):
                     root_inbox.put_nowait(item)
-
-                if method == "tasks":
-                    if "result" in data:
-                        self._apply_tasks_result(namespace, data, active)
-                    elif _is_direct_child(namespace, self._scope):
-                        path = tuple(namespace)
-                        if path not in seen:
-                            seen.add(path)
-                            graph_name, trigger_call_id = _parse_namespace_segment(
-                                path[-1]
-                            )
-                            handle = SyncScopedStreamHandle(
-                                thread=self._thread,
-                                path=path,
-                                graph_name=graph_name or None,
-                                trigger_call_id=trigger_call_id,
-                            )
-                            active[path] = handle
-                            yield handle
-                elif (
-                    method == "lifecycle"
-                    and data.get("event") == "started"
-                    and _is_direct_child(namespace, self._scope)
-                ):
-                    # ``create_deep_agent`` subagent discovery: child-
-                    # namespace ``lifecycle: started`` rather than ``tasks``.
-                    path = tuple(namespace)
-                    if path not in seen:
-                        seen.add(path)
-                        graph_name, trigger_call_id = _parse_namespace_segment(path[-1])
-                        handle = SyncScopedStreamHandle(
-                            thread=self._thread,
-                            path=path,
-                            graph_name=graph_name or None,
-                            trigger_call_id=trigger_call_id,
-                        )
-                        active[path] = handle
-                        yield handle
+                for handle in decoder.feed(cast(dict[str, Any], item)):
+                    yield handle
         finally:
             # Determine terminal status from the run's lifecycle result.
             # If _run_done resolved as errored, force-complete remaining children
@@ -1158,32 +1002,13 @@ class _SyncSubgraphsProjection:
                         terminal_status = "failed"
                 except Exception:
                     pass
-            for handle in active.values():
+            for handle in decoder._active.values():
                 if handle.status == "started":
                     handle._finish(terminal_status)
             self._thread._unregister_subscription(sub.id)
             if root_inbox is not None:
                 root_inbox.put_nowait(None)
 
-    def _apply_tasks_result(
-        self,
-        namespace: list[str],
-        data: dict[str, Any],
-        active: dict[tuple[str, ...], SyncScopedStreamHandle],
-    ) -> None:
-        result_id = data.get("id")
-        if not result_id:
-            return
-        parent_path = tuple(namespace)
-        for child_path, handle in list(active.items()):
-            if child_path[:-1] != parent_path:
-                continue
-            if handle.trigger_call_id != result_id:
-                continue
-            status, error = _terminal_from_tasks_result(data)
-            handle._finish(status, error)
-            del active[child_path]
-
 
 class _SyncExtensionsProjection:
     """Mapping from extension name to custom event payload stream.
@@ -1230,6 +1055,7 @@ class _SyncExtensionProjection:
         if self._namespace:
             params["namespaces"] = [self._namespace]
         sub = self._thread._register_subscription(params)
+        decoder = ExtensionsDecoder(name=self._name)
         try:
             if self._thread._closed:
                 return
@@ -1239,12 +1065,7 @@ class _SyncExtensionProjection:
                 item = sub.queue.get()
                 if item is None:
                     return
-                event_params = item.get("params") or {}
-                data = (
-                    event_params.get("data") if isinstance(event_params, dict) else None
-                )
-                if isinstance(data, dict):
-                    yield data
+                yield from decoder.feed(cast(dict[str, Any], item))
         finally:
             self._thread._unregister_subscription(sub.id)
 
@@ -1461,6 +1282,173 @@ class SyncThreadStream:
             params["depth"] = depth
         return self._subscription_iter(params)
 
+    def interleave_projections(self, channels: list[str]) -> Iterator[tuple[str, Any]]:
+        """Yield `(channel_name, item)` tuples across multiple projections.
+
+        One shared subscription drives all per-channel decoders; items arrive
+        in server-emit order (the SDK analog of `GraphRunStream.interleave`).
+
+        Args:
+            channels: Flat list of `"values"`, `"messages"`, `"tool_calls"`,
+                `"subgraphs"`, and/or extension names. Built-ins yield their
+                typed item (snapshot dict / `ChatModelStream` /
+                `SyncToolCallHandle` / `SyncScopedStreamHandle`); an extension
+                yields its payload dict, keyed by the bare extension name.
+
+        Note:
+            Handles and streams are yielded eagerly (before their sub-stream
+            completes), so items arrive interleaved in real time. To receive a
+            fully-resolved handle (output already populated), use the dedicated
+            `thread.tool_calls` / `thread.messages` projections instead.
+        """
+        validate_interleave_channels(channels)
+        if self._transport is None:
+            raise RuntimeError("SyncThreadStream not entered — use `with`.")
+        decoders: dict[str, Decoder] = {}
+        sub_params: list[dict[str, Any]] = []
+        for ch in channels:
+            if ch == "values":
+                decoders[ch] = DataDecoder("values")
+                sub_params.append({"channels": ["values"]})
+            elif ch in ("updates", "checkpoints", "tasks"):
+                # Plain payload channels (local Updates/Checkpoints/Tasks
+                # analog). Root-scope filter is load-bearing: a co-requested
+                # unscoped `values` widens the merged subscription to all
+                # namespaces, so the decoder itself keeps subgraph payloads out.
+                decoders[ch] = DataDecoder(ch, namespace=[])
+                sub_params.append(dict(_exact_namespace_params([ch], [])))
+            elif ch == "messages":
+                decoders[ch] = MessagesDecoder(
+                    namespace=[],
+                    stream_factory=lambda *, namespace, node, message_id: (
+                        ChatModelStream(
+                            namespace=namespace, node=node, message_id=message_id
+                        )
+                    ),
+                )
+                sub_params.append(dict(_exact_namespace_params(["messages"], [])))
+            elif ch == "tool_calls":
+                decoders[ch] = ToolCallsDecoder(
+                    namespace=[],
+                    handle_factory=lambda *, tool_call_id, name, input, namespace: (
+                        SyncToolCallHandle(
+                            tool_call_id=tool_call_id,
+                            name=name,
+                            input=input,
+                            namespace=namespace,
+                        )
+                    ),
+                )
+                sub_params.append(dict(_exact_namespace_params(["tools"], [])))
+            elif ch == "subgraphs":
+                decoders[ch] = SubgraphsDecoder(
+                    scope=(),
+                    handle_factory=lambda *, path, graph_name, trigger_call_id: (
+                        SyncScopedStreamHandle(
+                            thread=self,
+                            path=path,
+                            graph_name=graph_name,
+                            trigger_call_id=trigger_call_id,
+                        )
+                    ),
+                )
+                sub_params.append(dict(_subgraph_subscription_params(())))
+            else:
+                decoders[ch] = ExtensionsDecoder(name=ch)
+                sub_params.append({"channels": [f"custom:{ch}"]})
+        if not sub_params:
+            return
+        merged = cast(
+            SubscribeParams,
+            compute_union_filter(cast("list[dict[str, Any]]", sub_params)),
+        )
+        subgraphs = decoders.get("subgraphs")
+        # Track decoder-created handles so teardown can finalize anything still
+        # in flight; otherwise an awaiting `handle.output` / `handle.messages`
+        # would block after an early break or run termination.
+        registered_tool_calls: list[SyncToolCallHandle] = []
+        registered_message_streams: list[ChatModelStream] = []
+        try:
+            for event in self._subscription_iter(merged):
+                if subgraphs is not None:
+                    for item in subgraphs.feed(event):
+                        yield ("subgraphs", item)
+                wire = infer_channel(event)
+                public = self._interleave_public_name(wire)
+                # subgraphs is driven separately above (it consumes all events); never dispatch it here.
+                if public is not None and public != "subgraphs":
+                    decoder = decoders.get(public)
+                    if decoder is not None:
+                        for item in decoder.feed(event):
+                            if public == "tool_calls":
+                                self._register_active_tool_call(item)
+                                registered_tool_calls.append(item)
+                            elif public == "messages":
+                                self._register_active_message_stream(item)
+                                registered_message_streams.append(item)
+                            yield (public, item)
+        finally:
+            self._finalize_interleave_decoders(
+                decoders.get("tool_calls"),
+                subgraphs,
+                registered_tool_calls,
+                registered_message_streams,
+            )
+
+    @staticmethod
+    def _interleave_public_name(wire: str | None) -> str | None:
+        """Map a wire channel name to the public channel name used in interleave tuples."""
+        if wire is None:
+            return None
+        if wire == "tools":
+            return "tool_calls"
+        if wire.startswith("custom:"):
+            return wire[len("custom:") :]
+        return wire  # values, messages (tasks/lifecycle pass through with no decoder match)
+
+    def _finalize_interleave_decoders(
+        self,
+        tool_calls: Decoder | None,
+        subgraphs: Decoder | None,
+        registered_tool_calls: list[SyncToolCallHandle],
+        registered_message_streams: list[ChatModelStream],
+    ) -> None:
+        """Finalize in-flight handles when `interleave_projections` tears down.
+
+        Mirrors the terminal handling of the dedicated `_SyncToolCallsProjection`
+        / `_SyncSubgraphsProjection`: in-flight tool calls are failed (so a
+        blocking `handle.output` can't hang) and discovered subgraph children
+        are force-completed with the run's terminal status.
+        """
+        run_done = self._run_done
+        resolved: _RunTerminal | None = None
+        if run_done is not None and run_done.done():
+            try:
+                resolved = run_done.result(timeout=0)
+            except Exception:
+                resolved = None
+        if isinstance(tool_calls, ToolCallsDecoder):
+            err: BaseException = (
+                resolved.error
+                if resolved is not None and resolved.error is not None
+                else RuntimeError("Tool call stream closed before terminal tool event.")
+            )
+            for handle in list(tool_calls._active.values()):
+                handle._fail(err)
+        for handle in registered_tool_calls:
+            self._unregister_active_tool_call(handle)
+        for stream in registered_message_streams:
+            self._unregister_active_message_stream(stream)
+        if isinstance(subgraphs, SubgraphsDecoder):
+            terminal_status: SubgraphStatus = (
+                "failed"
+                if resolved is not None and resolved.status == "errored"
+                else "completed"
+            )
+            for child in subgraphs._active.values():
+                if child.status == "started":
+                    child._finish(terminal_status)
+
     def _subscription_iter(self, params: SubscribeParams) -> Iterator[Event]:
         sub = self._register_subscription(params)
         try:

libs/sdk-py/langgraph_sdk/stream/decoders.py

@@ -0,0 +1,359 @@
+"""Per-channel event → items state machines.
+
+Used both by the projection iterators (`_ValuesProjection`,
+`_MessagesProjection`, `_ToolCallsProjection`, `_SubgraphsProjection`) on
+`AsyncThreadStream` / `SyncThreadStream`, and by `interleave_projections`,
+which drives multiple decoders from one shared subscription.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable, Mapping
+from typing import Any, Literal, Protocol
+
+#: Channel names the public ``interleave_projections`` API accepts as built-ins.
+SUPPORTED_INTERLEAVE_CHANNELS = (
+    "values",
+    "messages",
+    "tool_calls",
+    "subgraphs",
+    "updates",
+    "checkpoints",
+    "tasks",
+)
+
+#: Channel names that ``infer_channel`` recognizes as first-class protocol
+#: methods but that ``interleave_projections`` has no decoder for. Routing them
+#: to the extension/``custom:`` fallback would subscribe to a channel that never
+#: matches and silently yield nothing, so they are rejected up front (fail
+#: closed). ``lifecycle`` is control-plane (drives run output/interrupt); ``tools``
+#: is the wire alias for the public ``tool_calls`` channel.
+RESERVED_INTERLEAVE_CHANNELS = frozenset({"lifecycle", "tools", "input"})
+
+
+def validate_interleave_channels(channels: list[str]) -> None:
+    """Reject reserved protocol channel names before they hit the fallback.
+
+    Genuine extension names pass through untouched; only names that
+    ``infer_channel`` treats as built-in methods without an interleave decoder
+    are rejected, so a typo'd or unsupported protocol channel surfaces an error
+    instead of an empty stream.
+    """
+    for ch in channels:
+        if ch in RESERVED_INTERLEAVE_CHANNELS:
+            hint = ' (use "tool_calls")' if ch == "tools" else ""
+            raise ValueError(
+                f"{ch!r} is not a valid interleave_projections channel{hint}. "
+                f"Supported channels: {', '.join(SUPPORTED_INTERLEAVE_CHANNELS)}, "
+                "or an extension name."
+            )
+
+
+def _event_namespace(params_field: Any) -> list[str]:
+    if not isinstance(params_field, dict):
+        return []
+    namespace = params_field.get("namespace") or []
+    return list(namespace) if isinstance(namespace, list) else []
+
+
+def _message_event_id(data: dict[str, Any]) -> str | None:
+    message_id = data.get("id") or data.get("message_id")
+    return str(message_id) if message_id is not None else None
+
+
+def _message_route_key(data: dict[str, Any], fallback: str | None = None) -> str:
+    """Return the routing key for a message-channel event in `active`.
+
+    Keys on `message_id` when available so concurrent messages that share the
+    same `run_id` (two AI turns in one agent step) route to independent streams
+    rather than colliding on a shared `run:<run_id>` slot.
+    """
+    message_id = _message_event_id(data)
+    if message_id is not None:
+        return f"message:{message_id}"
+    if fallback is not None:
+        return f"message:{fallback}"
+    return "__single__"
+
+
+SubgraphStatus = Literal["started", "completed", "failed", "interrupted"]
+
+
+def _parse_namespace_segment(segment: str) -> tuple[str, str | None]:
+    name, sep, task_id = segment.partition(":")
+    return name, task_id if sep else None
+
+
+def _terminal_from_tasks_result(
+    data: dict[str, Any],
+) -> tuple[SubgraphStatus, str | None]:
+    if data.get("interrupts"):
+        return "interrupted", None
+    error = data.get("error")
+    if error:
+        return "failed", str(error)
+    return "completed", None
+
+
+def _is_direct_child(namespace: list[str], scope: tuple[str, ...]) -> bool:
+    return len(namespace) == len(scope) + 1 and tuple(namespace[: len(scope)]) == scope
+
+
+class Decoder(Protocol):
+    def feed(self, event: Mapping[str, Any]) -> Iterable[Any]: ...
+
+
+class DataDecoder:
+    """Yields `params.data` from events of a single `method`.
+
+    Covers the channels whose projection is just "emit the payload": `values`,
+    `updates`, `checkpoints`, `tasks` — the SDK analog of local's
+    `Values`/`Updates`/`Checkpoints`/`TasksTransformer`, all of which push
+    `params["data"]` unchanged. The REST-state seeding for `values` stays at
+    the projection layer; it is a one-shot pre-stream fetch, not part of the
+    event state machine.
+
+    Args:
+        method: The protocol `method` this decoder consumes.
+        namespace: When not `None`, events whose namespace differs are ignored
+            (scope filter, mirroring the local transformers' `namespace != scope`
+            check). `None` consumes every namespace — the historical `values`
+            projection behavior, where subscription scoping is handled upstream.
+    """
+
+    def __init__(self, method: str, namespace: list[str] | None = None):
+        self._method = method
+        self._namespace = list(namespace) if namespace is not None else None
+
+    def feed(self, event: Mapping[str, Any]) -> Iterable[Any]:
+        if event.get("method") != self._method:
+            return
+        params = event.get("params") or {}
+        if self._namespace is not None and _event_namespace(params) != self._namespace:
+            return
+        data = params.get("data")
+        if data is not None:
+            yield data
+
+
+class MessagesDecoder:
+    """Yields one chat-model stream per `message-start` event.
+
+    Subsequent events route to the matching stream via `stream.dispatch(data)`.
+    Mirrors the per-event body of `_MessagesProjection._messages_iter`
+    (`_async/stream.py:404-458`). The subscription open/close and the
+    `_root_messages_inbox` drain branch stay at the projection layer.
+
+    Args:
+        namespace: Events whose namespace differs are ignored (scope filter).
+        stream_factory: Keyword-only `(namespace, node, message_id) -> stream`.
+            Sync binds `ChatModelStream`; async binds `AsyncChatModelStream`.
+    """
+
+    def __init__(
+        self,
+        namespace: list[str],
+        stream_factory: Callable[..., Any],
+    ):
+        self._namespace = list(namespace)
+        self._stream_factory = stream_factory
+        self._active: dict[str, Any] = {}  # route_key -> stream
+
+    def feed(self, event: Mapping[str, Any]) -> Iterable[Any]:
+        if event.get("method") != "messages":
+            return
+        params = event.get("params") or {}
+        if _event_namespace(params) != self._namespace:
+            return
+        data = params.get("data")
+        if not isinstance(data, dict):
+            return
+        if data.get("event") == "message-start":
+            message_id = _message_event_id(data)
+            key = _message_route_key(data, fallback=message_id)
+            metadata = (
+                data.get("metadata") if isinstance(data.get("metadata"), dict) else {}
+            )
+            stream = self._stream_factory(
+                namespace=list(self._namespace),
+                node=metadata.get("langgraph_node") if metadata else None,
+                message_id=message_id,
+            )
+            self._active[key] = stream
+            stream.dispatch(data)
+            yield stream
+        else:
+            key = _message_route_key(data)
+            stream = self._active.get(key)
+            if stream is None and key == "__single__" and len(self._active) == 1:
+                stream = next(iter(self._active.values()))
+            if stream is None:
+                return
+            stream.dispatch(data)
+            if data.get("event") in ("message-finish", "error"):
+                for route_key, candidate in list(self._active.items()):
+                    if candidate is stream:
+                        del self._active[route_key]
+
+
+class ToolCallsDecoder:
+    """Yields one tool-call handle per `tool-started` event.
+
+    Mirrors the per-event body of `_ToolCallsProjection._tool_calls_iter`
+    (`_async/stream.py:1168-1217`). The thread register/unregister and the
+    terminal-error-on-close finally stay at the projection / wrapper layer.
+
+    Args:
+        namespace: Events whose namespace differs are ignored.
+        handle_factory: Keyword-only `(tool_call_id, name, input, namespace) -> handle`.
+    """
+
+    def __init__(self, namespace: list[str], handle_factory: Callable[..., Any]):
+        self._namespace = list(namespace)
+        self._handle_factory = handle_factory
+        self._active: dict[str, Any] = {}
+
+    def feed(self, event: Mapping[str, Any]) -> Iterable[Any]:
+        if event.get("method") != "tools":
+            return
+        params = event.get("params") or {}
+        if _event_namespace(params) != self._namespace:
+            return
+        data = params.get("data")
+        if not isinstance(data, dict):
+            return
+        tool_call_id = data.get("tool_call_id")
+        if not isinstance(tool_call_id, str):
+            return
+        event_type = data.get("event")
+        if event_type == "tool-started":
+            name = data.get("tool_name")
+            handle = self._handle_factory(
+                tool_call_id=tool_call_id,
+                name=name if isinstance(name, str) else "",
+                input=data.get("input"),
+                namespace=list(self._namespace),
+            )
+            self._active[tool_call_id] = handle
+            yield handle
+        elif event_type == "tool-output-delta":
+            handle = self._active.get(tool_call_id)
+            delta = data.get("delta")
+            if handle is not None and isinstance(delta, str):
+                handle._push_delta(delta)
+        elif event_type == "tool-finished":
+            handle = self._active.pop(tool_call_id, None)
+            if handle is not None:
+                handle._finish(data.get("output"))
+        elif event_type == "tool-error":
+            handle = self._active.pop(tool_call_id, None)
+            if handle is not None:
+                message = data.get("message")
+                handle._fail(
+                    RuntimeError(str(message) if message else "Tool call errored")
+                )
+
+
+class SubgraphsDecoder:
+    """Discovers child subgraph handles and fans out events to active ones.
+
+    Mirrors the per-event body of `_SubgraphsProjection._subgraphs_iter`
+    (`_async/stream.py:963-1041`) plus `_apply_tasks_result`. Root-inbox
+    forwarding and terminal-status-on-close stay at the projection / wrapper
+    layer.
+
+    Args:
+        scope: Tuple-form namespace of this decoder's parent. `()` for root.
+        handle_factory: Keyword-only `(path, graph_name, trigger_call_id) -> handle`.
+    """
+
+    def __init__(self, scope: tuple[str, ...], handle_factory: Callable[..., Any]):
+        self._scope = scope
+        self._handle_factory = handle_factory
+        self._active: dict[tuple[str, ...], Any] = {}
+        self._seen: set[tuple[str, ...]] = set()
+
+    def feed(self, event: Mapping[str, Any]) -> Iterable[Any]:
+        params = event.get("params") or {}
+        namespace = _event_namespace(params)
+        data = params.get("data")
+        if not isinstance(data, dict):
+            return
+        method = event.get("method")
+
+        # 1. Fanout: first active child whose path prefixes this namespace.
+        ns_tuple = tuple(namespace)
+        for child_path, child_handle in self._active.items():
+            child_len = len(child_path)
+            if len(ns_tuple) >= child_len and ns_tuple[:child_len] == child_path:
+                child_handle._push_event(event)
+                break
+
+        # 2 + 3. Discovery / status from tasks; discovery from lifecycle.
+        if method == "tasks":
+            if "result" in data:
+                self._apply_tasks_result(namespace, data)
+            elif _is_direct_child(namespace, self._scope):
+                yield from self._discover(namespace)
+        elif (
+            method == "lifecycle"
+            and data.get("event") == "started"
+            and _is_direct_child(namespace, self._scope)
+        ):
+            yield from self._discover(namespace)
+
+    def _discover(self, namespace: list[str]) -> Iterable[Any]:
+        path = tuple(namespace)
+        if path in self._seen:
+            return
+        self._seen.add(path)
+        graph_name, trigger_call_id = _parse_namespace_segment(path[-1])
+        handle = self._handle_factory(
+            path=path,
+            graph_name=graph_name or None,
+            trigger_call_id=trigger_call_id,
+        )
+        self._active[path] = handle
+        yield handle
+
+    def _apply_tasks_result(self, namespace: list[str], data: dict[str, Any]) -> None:
+        result_id = data.get("id")
+        if not result_id:
+            return
+        parent_path = tuple(namespace)
+        for child_path, handle in list(self._active.items()):
+            if child_path[:-1] != parent_path:
+                continue
+            if handle.trigger_call_id != result_id:
+                continue
+            status, error = _terminal_from_tasks_result(data)
+            handle._finish(status, error)
+            del self._active[child_path]
+
+
+class ExtensionsDecoder:
+    """Yields `params.data` from one named custom channel.
+
+    Mirrors `_ExtensionProjection._iter` (`_async/stream.py:1278-1299`), with
+    an added name filter so it can share one subscription in interleave.
+
+    Args:
+        name: The extension name. Only `custom` events whose `data["name"]`
+            matches are consumed.
+    """
+
+    def __init__(self, name: str):
+        if not name:
+            raise ValueError("extension name must be non-empty.")
+        self._name = name
+
+    def feed(self, event: Mapping[str, Any]) -> Iterable[Any]:
+        if event.get("method") != "custom":
+            return
+        params = event.get("params") or {}
+        data = params.get("data")
+        if not isinstance(data, dict):
+            return
+        if data.get("name") != self._name:
+            return
+        yield data

libs/sdk-py/langgraph_sdk/stream/transport/http.py

@@ -20,6 +20,7 @@ import httpx
 import orjson
 from langchain_protocol import Event
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.sse import BytesLineDecoder, SSEDecoder
 from langgraph_sdk.stream.transport.base import (
     EventStreamHandle,
@@ -49,8 +50,12 @@ class ProtocolSseTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_url = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_url = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._max_queue_size = max_queue_size
         self._closed = False

libs/sdk-py/langgraph_sdk/stream/transport/sync_http.py

@@ -10,6 +10,7 @@ import httpx
 import orjson
 from langchain_protocol import Event
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.sse import BytesLineDecoder, SSEDecoder
 from langgraph_sdk.stream.transport.base import (
     SyncEventStreamHandle,
@@ -31,8 +32,12 @@ class SyncProtocolSseTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_url = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_url = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._closed = False
         self._open_responses: list[httpx.Response] = []

libs/sdk-py/langgraph_sdk/stream/transport/sync_ws.py

@@ -11,6 +11,7 @@ import orjson
 from langchain_protocol import Event
 from websockets.sync.client import connect as websocket_connect
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.stream.transport.base import (
     SyncEventStreamHandle,
     build_event_stream_body,
@@ -36,8 +37,12 @@ class SyncProtocolWebSocketTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_path = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_path = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._connect = connect
         self._ping_interval = ping_interval

libs/sdk-py/langgraph_sdk/stream/transport/ws.py

@@ -13,6 +13,7 @@ from langchain_protocol import Event
 from websockets.asyncio.client import connect as websocket_connect
 from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK
 
+from langgraph_sdk._shared.utilities import _quote_path_param
 from langgraph_sdk.stream.transport.base import (
     EventStreamHandle,
     build_event_stream_body,
@@ -39,8 +40,12 @@ class ProtocolWebSocketTransport:
     ) -> None:
         self._client = client
         self.thread_id = thread_id
-        self._commands_url = commands_path or f"/threads/{thread_id}/commands"
-        self._stream_path = stream_path or f"/threads/{thread_id}/stream/events"
+        self._commands_url = (
+            commands_path or f"/threads/{_quote_path_param(thread_id)}/commands"
+        )
+        self._stream_path = (
+            stream_path or f"/threads/{_quote_path_param(thread_id)}/stream/events"
+        )
         self._default_headers: dict[str, str] = dict(headers or {})
         self._connect = connect
         self._max_queue_size = max_queue_size

libs/sdk-py/tests/integration/test_remote_graph_v3.py

@@ -0,0 +1,133 @@
+"""Integration tests for RemoteGraph v3 streaming.
+
+Tests the end-to-end wiring: RemoteGraph -> langgraph_sdk client.threads.stream(...) ->
+docker-running langgraph-api -> SSE projections -> adapter classes.
+
+Run with: pytest tests/integration/test_remote_graph_v3.py -m integration
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import pytest
+from langchain_core.runnables import RunnableConfig
+from langgraph.pregel.remote import RemoteGraph
+from langgraph.types import Command
+
+pytestmark = pytest.mark.integration
+
+URL = "http://localhost:2024"
+
+# Input shapes matching what the integration graphs expect.
+# `agent` graph (streaming_graph.py): AgentState has messages, value, items.
+# `tools_agent` graph (tools_agent.py): create_agent graph expects messages list.
+_AGENT_INPUT = {"messages": [], "value": "init", "items": []}
+_TOOLS_AGENT_INPUT = {"messages": [{"role": "user", "content": "search for v3"}]}
+
+
+@pytest.fixture
+def remote_agent() -> RemoteGraph:
+    return RemoteGraph("agent", url=URL)
+
+
+@pytest.fixture
+def remote_tools_agent() -> RemoteGraph:
+    return RemoteGraph("tools_agent", url=URL)
+
+
+async def test_async_happy_path_yields_output(remote_tools_agent: RemoteGraph) -> None:
+    """tools_agent completes without interrupt; ``await stream.output`` drives
+    the run to terminal via the lifecycle watcher (no explicit event iteration
+    needed — the SSE subscription stays open by design after run completion)."""
+    async with await remote_tools_agent.astream_events(
+        _TOOLS_AGENT_INPUT,
+        version="v3",
+    ) as stream:
+        output = await stream.output()
+        assert output is not None
+        assert (await stream.interrupted()) is False
+
+
+async def test_async_interrupt_path_surfaces_interrupts(
+    remote_agent: RemoteGraph,
+) -> None:
+    """agent graph hits ask_human; interrupted must be True with >= 1 interrupt.
+
+    Note: interrupts pause the run but DON'T resolve `_run_done` (only
+    `completed` / `failed` lifecycle phases do), so `await stream.output()`
+    would hang. The adapter doesn't expose `interleave()` on the async
+    side (mirrors local `AsyncGraphRunStream`), so drain the `values`
+    projection directly until the run reports it is interrupted.
+    """
+    async with await remote_agent.astream_events(
+        _AGENT_INPUT,
+        version="v3",
+    ) as stream:
+        async for _ in stream.values:
+            if await stream.interrupted():
+                break
+        assert (await stream.interrupted()) is True
+        interrupts = await stream.interrupts()
+        assert len(interrupts) >= 1
+
+
+async def test_async_resume_after_interrupt(remote_agent: RemoteGraph) -> None:
+    """Interrupt the agent at ask_human, then resume the SAME thread with
+    `Command(resume=...)`.
+
+    Validates the v3 resume path end-to-end. The client sends the raw resume
+    value as `input` (not a serialized Command); the server detects the
+    thread's pending interrupt from persisted state — which survives the first
+    session's close — and wraps it as `Command(resume=...)`, driving the run
+    past `ask_human` to completion (the graph interrupts only once).
+    """
+    thread_id = str(uuid.uuid4())
+    config: RunnableConfig = {"configurable": {"thread_id": thread_id}}
+
+    # First session: drive until the agent pauses at the ask_human interrupt.
+    async with await remote_agent.astream_events(
+        _AGENT_INPUT,
+        config=config,
+        version="v3",
+    ) as stream:
+        async for _ in stream.values:
+            if await stream.interrupted():
+                break
+        assert (await stream.interrupted()) is True
+
+    # Second session on the same thread: resume with the human's answer. The
+    # run continues past ask_human to completion with no further interrupt.
+    async with await remote_agent.astream_events(
+        Command(resume="yes"),
+        config=config,
+        version="v3",
+    ) as stream:
+        output = await stream.output()
+        assert output is not None
+        assert (await stream.interrupted()) is False
+
+
+def test_sync_happy_path_yields_output(remote_tools_agent: RemoteGraph) -> None:
+    """Sync stream: tools_agent completes; ``stream.output`` (sync property)
+    blocks until terminal."""
+    with remote_tools_agent.stream_events(
+        _TOOLS_AGENT_INPUT,
+        version="v3",
+    ) as stream:
+        output = stream.output
+        assert output is not None
+        assert stream.interrupted is False
+
+
+async def test_abort_mid_run_cancels_server_side(
+    remote_tools_agent: RemoteGraph,
+) -> None:
+    """Abort immediately after run.start; reaching the end without exception
+    confirms abort + __aexit__ cleanup worked."""
+    async with await remote_tools_agent.astream_events(
+        _TOOLS_AGENT_INPUT,
+        version="v3",
+    ) as stream:
+        await stream.abort()
+    # Reaching here without unhandled exceptions confirms abort + __aexit__ succeeded.

libs/sdk-py/tests/streaming/_events.py

@@ -75,6 +75,18 @@ def values_event(
     return _base(seq, "values", namespace or [], data or {"values": {}})
 
 
+def updates_event(
+    seq: int = 0, namespace: list[str] | None = None, **data: Any
+) -> dict[str, Any]:
+    return _base(seq, "updates", namespace or [], data or {})
+
+
+def checkpoints_event(
+    seq: int = 0, namespace: list[str] | None = None, **data: Any
+) -> dict[str, Any]:
+    return _base(seq, "checkpoints", namespace or [], data or {})
+
+
 def custom_event(
     seq: int = 0, name: str = "ext", namespace: list[str] | None = None, **data: Any
 ) -> dict[str, Any]:

libs/sdk-py/tests/streaming/test_decoders.py

@@ -0,0 +1,460 @@
+"""Unit tests for the per-channel Decoders.
+
+Each test drives a single decoder with synthetic events from `_events` and
+asserts the items the decoder yields.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from langgraph_sdk.stream.decoders import (
+    DataDecoder,
+    ExtensionsDecoder,
+    MessagesDecoder,
+    SubgraphsDecoder,
+    ToolCallsDecoder,
+)
+from streaming._events import (
+    checkpoints_event,
+    custom_event,
+    lifecycle_completed_event,
+    lifecycle_started_event,
+    message_error_event,
+    message_finish_event,
+    message_start_event,
+    message_text_delta_event,
+    tasks_result_event,
+    tasks_start_event,
+    tool_error_event,
+    tool_finished_event,
+    tool_output_delta_event,
+    tool_started_event,
+    updates_event,
+    values_event,
+)
+
+
+def test_data_decoder_yields_params_data():
+    decoder = DataDecoder("values")
+    assert list(decoder.feed(values_event(seq=1, x=1))) == [{"x": 1}]
+    assert list(decoder.feed(values_event(seq=2, x=2, y=3))) == [{"x": 2, "y": 3}]
+
+
+def test_data_decoder_ignores_other_methods():
+    decoder = DataDecoder("values")
+    assert list(decoder.feed(lifecycle_completed_event(seq=1))) == []
+    assert list(decoder.feed(updates_event(seq=2, foo=1))) == []
+
+
+def test_data_decoder_handles_updates_checkpoints_tasks_methods():
+    assert list(DataDecoder("updates").feed(updates_event(seq=1, node={"x": 1}))) == [
+        {"node": {"x": 1}}
+    ]
+    assert list(
+        DataDecoder("checkpoints").feed(checkpoints_event(seq=2, ts="t", v=4))
+    ) == [{"ts": "t", "v": 4}]
+    # tasks payloads pass through verbatim as data dicts
+    [item] = list(DataDecoder("tasks").feed(tasks_start_event(seq=3, task_id="a")))
+    assert item["id"] == "a"
+
+
+def test_data_decoder_namespace_none_yields_regardless_of_namespace():
+    decoder = DataDecoder("checkpoints", namespace=None)
+    assert list(decoder.feed(checkpoints_event(seq=1, namespace=["child"], v=1))) == [
+        {"v": 1}
+    ]
+
+
+def test_data_decoder_namespace_filter_drops_non_matching_namespace():
+    decoder = DataDecoder("checkpoints", namespace=[])
+    # root-namespace event is yielded; child-namespace event is filtered out
+    assert list(decoder.feed(checkpoints_event(seq=1, v=1))) == [{"v": 1}]
+    assert list(decoder.feed(checkpoints_event(seq=2, namespace=["child"], v=2))) == []
+
+
+class _FakeStream:
+    """Stand-in for AsyncChatModelStream/ChatModelStream in decoder tests."""
+
+    def __init__(self, *, namespace, node, message_id):
+        self.namespace = namespace
+        self.node = node
+        self.message_id = message_id
+        self.dispatched: list[dict] = []
+
+    def dispatch(self, data):
+        self.dispatched.append(data)
+
+
+def _factory(*, namespace, node, message_id):
+    return _FakeStream(namespace=namespace, node=node, message_id=message_id)
+
+
+def test_messages_decoder_yields_stream_on_message_start():
+    decoder = MessagesDecoder(namespace=[], stream_factory=_factory)
+    streams = list(
+        decoder.feed(message_start_event(seq=1, message_id="m-1", node="agent"))
+    )
+    assert len(streams) == 1
+    assert streams[0].message_id == "m-1"
+    assert streams[0].node == "agent"
+    # The start event is dispatched into the stream too (matches stream.py:432).
+    assert (
+        streams[0].dispatched and streams[0].dispatched[0]["event"] == "message-start"
+    )
+
+
+def test_messages_decoder_dispatches_delta_to_active_stream():
+    decoder = MessagesDecoder(namespace=[], stream_factory=_factory)
+    [stream] = list(decoder.feed(message_start_event(seq=1, message_id="m-1")))
+    delta = message_text_delta_event(seq=2, message_id="m-1", text="hi")
+    assert list(decoder.feed(delta)) == []
+    assert stream.dispatched[-1]["event"] == "content-block-delta"
+
+
+def test_messages_decoder_finish_retires_stream():
+    decoder = MessagesDecoder(namespace=[], stream_factory=_factory)
+    [stream] = list(decoder.feed(message_start_event(seq=1, message_id="m-1")))
+    list(decoder.feed(message_finish_event(seq=2, message_id="m-1")))
+    assert stream.dispatched[-1]["event"] == "message-finish"
+    assert all(s is not stream for s in decoder._active.values())
+    [again] = list(decoder.feed(message_start_event(seq=3, message_id="m-1")))
+    assert again is not stream
+
+
+def test_messages_decoder_error_retires_stream():
+    decoder = MessagesDecoder(namespace=[], stream_factory=_factory)
+    [stream] = list(decoder.feed(message_start_event(seq=1, message_id="m-1")))
+    list(decoder.feed(message_error_event(seq=2, message_id="m-1", message="boom")))
+    assert stream.dispatched[-1]["event"] == "error"
+    assert all(s is not stream for s in decoder._active.values())
+
+
+def test_messages_decoder_single_fallback_routes_idless_events():
+    decoder = MessagesDecoder(namespace=[], stream_factory=_factory)
+    [stream] = list(decoder.feed(message_start_event(seq=1, message_id="m-1")))
+    list(decoder.feed(message_text_delta_event(seq=2, text="x")))  # no message_id
+    assert stream.dispatched[-1]["event"] == "content-block-delta"
+
+
+def test_messages_decoder_drops_idful_events_for_unknown_stream():
+    decoder = MessagesDecoder(namespace=[], stream_factory=_factory)
+    [stream] = list(decoder.feed(message_start_event(seq=1, message_id="m-1")))
+    list(decoder.feed(message_text_delta_event(seq=2, message_id="ghost", text="x")))
+    assert all(d["event"] != "content-block-delta" for d in stream.dispatched)
+
+
+def test_messages_decoder_ignores_other_namespaces():
+    decoder = MessagesDecoder(namespace=[], stream_factory=_factory)
+    assert (
+        list(
+            decoder.feed(
+                message_start_event(seq=1, namespace=["child"], message_id="m-1")
+            )
+        )
+        == []
+    )
+
+
+def test_messages_decoder_drops_idless_delta_when_multiple_active():
+    decoder = MessagesDecoder(namespace=[], stream_factory=_factory)
+    [a] = list(decoder.feed(message_start_event(seq=1, message_id="m-1")))
+    [b] = list(decoder.feed(message_start_event(seq=2, message_id="m-2")))
+    # id-less delta is ambiguous with two active streams -> dropped, routed to neither
+    list(decoder.feed(message_text_delta_event(seq=3, text="x")))
+    assert all(d["event"] != "content-block-delta" for d in a.dispatched)
+    assert all(d["event"] != "content-block-delta" for d in b.dispatched)
+
+
+class _FakeToolHandle:
+    def __init__(self, *, tool_call_id, name, input, namespace):
+        self.tool_call_id = tool_call_id
+        self.name = name
+        self.input = input
+        self.namespace = namespace
+        self.deltas: list[str] = []
+        self.finished_output: Any = None
+        self.finished = False
+        self.error: BaseException | None = None
+
+    def _push_delta(self, delta):
+        self.deltas.append(delta)
+
+    def _finish(self, output):
+        self.finished = True
+        self.finished_output = output
+
+    def _fail(self, exc):
+        self.error = exc
+
+
+def _tool_factory(*, tool_call_id, name, input, namespace):
+    return _FakeToolHandle(
+        tool_call_id=tool_call_id, name=name, input=input, namespace=namespace
+    )
+
+
+def test_tool_calls_decoder_yields_handle_on_start():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    [handle] = list(
+        decoder.feed(tool_started_event(seq=1, tool_call_id="tc-1", tool_name="search"))
+    )
+    assert handle.tool_call_id == "tc-1"
+    assert handle.name == "search"
+
+
+def test_tool_calls_decoder_routes_delta_finish_and_error():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    [h] = list(
+        decoder.feed(tool_started_event(seq=1, tool_call_id="tc-1", tool_name="search"))
+    )
+    list(decoder.feed(tool_output_delta_event(seq=2, tool_call_id="tc-1", delta="x")))
+    assert h.deltas == ["x"]
+    list(
+        decoder.feed(
+            tool_finished_event(seq=3, tool_call_id="tc-1", output={"ok": True})
+        )
+    )
+    assert h.finished and h.finished_output == {"ok": True}
+
+    decoder2 = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    [h2] = list(
+        decoder2.feed(
+            tool_started_event(seq=1, tool_call_id="tc-2", tool_name="search")
+        )
+    )
+    list(decoder2.feed(tool_error_event(seq=2, tool_call_id="tc-2", message="boom")))
+    assert isinstance(h2.error, RuntimeError) and "boom" in str(h2.error)
+
+
+def test_tool_calls_decoder_drops_events_for_unknown_id():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    assert (
+        list(
+            decoder.feed(
+                tool_output_delta_event(seq=1, tool_call_id="ghost", delta="x")
+            )
+        )
+        == []
+    )
+
+
+def test_tool_calls_decoder_finish_and_error_retire_handle():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    [h] = list(
+        decoder.feed(tool_started_event(seq=1, tool_call_id="tc-1", tool_name="search"))
+    )
+    list(decoder.feed(tool_finished_event(seq=2, tool_call_id="tc-1")))
+    # retired: a late delta for the same id is now dropped
+    list(
+        decoder.feed(tool_output_delta_event(seq=3, tool_call_id="tc-1", delta="late"))
+    )
+    assert h.deltas == []
+
+
+def test_tool_calls_decoder_ignores_other_namespaces():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    assert (
+        list(
+            decoder.feed(
+                tool_started_event(seq=1, namespace=["child"], tool_call_id="tc-1")
+            )
+        )
+        == []
+    )
+
+
+def test_tool_calls_decoder_error_retires_handle():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    [h] = list(
+        decoder.feed(tool_started_event(seq=1, tool_call_id="tc-1", tool_name="search"))
+    )
+    list(decoder.feed(tool_error_event(seq=2, tool_call_id="tc-1", message="boom")))
+    # retired: a late delta for the same id is now dropped
+    list(
+        decoder.feed(tool_output_delta_event(seq=3, tool_call_id="tc-1", delta="late"))
+    )
+    assert h.deltas == []
+
+
+def test_tool_calls_decoder_skips_non_str_tool_call_id():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    # Build a tools event whose data.tool_call_id is not a string.
+    bad = tool_started_event(seq=1, tool_call_id="tc-1", tool_name="search")
+    bad["params"]["data"]["tool_call_id"] = 123
+    assert list(decoder.feed(bad)) == []
+
+
+def test_tool_calls_decoder_skips_non_str_delta():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    [h] = list(
+        decoder.feed(tool_started_event(seq=1, tool_call_id="tc-1", tool_name="search"))
+    )
+    evt = tool_output_delta_event(seq=2, tool_call_id="tc-1", delta="x")
+    evt["params"]["data"]["delta"] = 123  # non-str delta must be ignored
+    list(decoder.feed(evt))
+    assert h.deltas == []
+
+
+def test_tool_calls_decoder_defaults_missing_tool_name_to_empty():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    evt = tool_started_event(seq=1, tool_call_id="tc-1", tool_name="search")
+    del evt["params"]["data"]["tool_name"]  # absent tool_name -> handle.name == ""
+    [h] = list(decoder.feed(evt))
+    assert h.name == ""
+
+
+def test_tool_calls_decoder_error_message_defaults_when_blank():
+    decoder = ToolCallsDecoder(namespace=[], handle_factory=_tool_factory)
+    [h] = list(
+        decoder.feed(tool_started_event(seq=1, tool_call_id="tc-1", tool_name="search"))
+    )
+    evt = tool_error_event(seq=2, tool_call_id="tc-1")
+    evt["params"]["data"]["message"] = ""  # blank -> default message
+    list(decoder.feed(evt))
+    assert str(h.error) == "Tool call errored"
+
+
+class _FakeScopedHandle:
+    def __init__(self, *, path, graph_name, trigger_call_id):
+        self.path = path
+        self.graph_name = graph_name
+        self.trigger_call_id = trigger_call_id
+        self.status = "started"
+        self.error = None
+        self.events: list[dict] = []
+
+    def _push_event(self, event):
+        self.events.append(event)
+
+    def _finish(self, status, error=None):
+        if self.status != "started":
+            return
+        self.status = status
+        self.error = error
+
+
+def _scoped_factory(*, path, graph_name, trigger_call_id):
+    return _FakeScopedHandle(
+        path=path, graph_name=graph_name, trigger_call_id=trigger_call_id
+    )
+
+
+def test_subgraphs_decoder_discovers_on_lifecycle_started_once():
+    decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory)
+    [h] = list(decoder.feed(lifecycle_started_event(seq=1, namespace=["child"])))
+    assert h.path == ("child",)
+    assert list(decoder.feed(lifecycle_started_event(seq=2, namespace=["child"]))) == []
+
+
+def test_subgraphs_decoder_discovers_on_tasks_start_without_result():
+    decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory)
+    [h] = list(decoder.feed(tasks_start_event(seq=1, namespace=["child"])))
+    assert h.path == ("child",)
+
+
+def test_subgraphs_decoder_parses_graph_name_and_trigger_from_segment():
+    decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory)
+    [h] = list(decoder.feed(tasks_start_event(seq=1, namespace=["agent:call-1"])))
+    assert h.graph_name == "agent"
+    assert h.trigger_call_id == "call-1"
+
+
+def test_subgraphs_decoder_fans_out_events_to_active_child():
+    decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory)
+    [h] = list(decoder.feed(lifecycle_started_event(seq=1, namespace=["child"])))
+    inner = message_start_event(seq=2, namespace=["child"], message_id="m")
+    assert list(decoder.feed(inner)) == []
+    assert inner in h.events  # whole event pushed, not just data
+
+
+def test_subgraphs_decoder_fans_out_grandchild_to_direct_child():
+    decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory)
+    [h] = list(decoder.feed(lifecycle_started_event(seq=1, namespace=["child"])))
+    grand = message_start_event(seq=2, namespace=["child", "grand"], message_id="m")
+    list(decoder.feed(grand))
+    assert grand in h.events
+
+
+def test_subgraphs_decoder_tasks_result_at_parent_finalizes_child():
+    decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory)
+    # child discovered with a colon segment -> trigger_call_id == "call-1"
+    [h] = list(decoder.feed(tasks_start_event(seq=1, namespace=["agent:call-1"])))
+    assert h.trigger_call_id == "call-1"
+    # the finalizing tasks-result is emitted at the PARENT (root) namespace,
+    # with id (task_id) matching the child's trigger_call_id
+    list(
+        decoder.feed(
+            tasks_result_event(seq=2, namespace=[], task_id="call-1", result={"ok": 1})
+        )
+    )
+    assert h.status == "completed"
+    # finalized + removed from active: later child-namespace events no longer fan out
+    later = message_start_event(seq=3, namespace=["agent:call-1"], message_id="m")
+    list(decoder.feed(later))
+    assert later not in h.events
+
+
+def test_subgraphs_decoder_tasks_result_failed_status():
+    decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory)
+    [h] = list(decoder.feed(tasks_start_event(seq=1, namespace=["agent:call-1"])))
+    list(
+        decoder.feed(
+            tasks_result_event(seq=2, namespace=[], task_id="call-1", error="boom")
+        )
+    )
+    assert h.status == "failed"
+    assert h.error == "boom"
+
+
+def test_subgraphs_decoder_tasks_result_interrupted_status():
+    decoder = SubgraphsDecoder(scope=(), handle_factory=_scoped_factory)
+    [h] = list(decoder.feed(tasks_start_event(seq=1, namespace=["agent:call-1"])))
+    list(
+        decoder.feed(
+            tasks_result_event(
+                seq=2, namespace=[], task_id="call-1", interrupts=[{"id": "i-1"}]
+            )
+        )
+    )
+    assert h.status == "interrupted"
+
+
+def test_subgraphs_decoder_ignores_unrelated_and_scope_itself():
+    decoder = SubgraphsDecoder(scope=("root",), handle_factory=_scoped_factory)
+    # not a direct child of ("root",): wrong depth / wrong prefix
+    assert list(decoder.feed(lifecycle_started_event(seq=1, namespace=["other"]))) == []
+    # the scope's own namespace is not a discovery
+    assert list(decoder.feed(lifecycle_started_event(seq=2, namespace=["root"]))) == []
+
+
+def test_extensions_decoder_yields_full_data_for_matching_name():
+    decoder = ExtensionsDecoder(name="foo")
+    # custom_event(name="foo", x=1) -> params.data = {"name": "foo", "x": 1}
+    assert list(decoder.feed(custom_event(seq=1, name="foo", x=1))) == [
+        {"name": "foo", "x": 1}
+    ]
+
+
+def test_extensions_decoder_ignores_other_extension_names():
+    decoder = ExtensionsDecoder(name="foo")
+    assert list(decoder.feed(custom_event(seq=1, name="bar", x=1))) == []
+
+
+def test_extensions_decoder_ignores_non_custom_methods():
+    decoder = ExtensionsDecoder(name="foo")
+    assert list(decoder.feed(lifecycle_completed_event(seq=1))) == []
+
+
+def test_extensions_decoder_ignores_non_dict_data():
+    decoder = ExtensionsDecoder(name="foo")
+    evt = custom_event(seq=1, name="foo", x=1)
+    evt["params"]["data"] = "not-a-dict"
+    assert list(decoder.feed(evt)) == []
+
+
+def test_extensions_decoder_rejects_empty_name():
+    import pytest
+
+    with pytest.raises(ValueError):
+        ExtensionsDecoder(name="")

libs/sdk-py/tests/streaming/test_sync_projections.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import cast
+from typing import Any, cast
 
 import httpx
 import pytest
@@ -562,3 +562,79 @@ def test_sync_tool_call_handle_deltas_single_consumer_guard():
     # Second access: must raise immediately (before any iteration).
     with pytest.raises(RuntimeError, match="single consumer"):
         _ = handle.deltas
+
+
+def test_sync_messages_subscription_pre_dispatches_before_yield():
+    """Over a live subscription, str(stream.text) must work immediately on yield."""
+    fake = SyncFakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            message_start_event(seq=1, message_id="msg-1"),
+            message_text_delta_event(seq=2, text="hello", message_id="msg-1"),
+            message_text_finish_event(seq=3, text="hello", message_id="msg-1"),
+            message_finish_event(seq=4, message_id="msg-1"),
+            lifecycle_completed_event(seq=5),
+        ]
+    )
+    fake.set_state({})
+    collected: list[str] = []
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            for stream in thread.messages:
+                collected.append(str(stream.text))
+    assert collected == ["hello"]
+
+
+def test_sync_tool_calls_subscription_resolves_output_before_yield():
+    """Over a live subscription, call.output is resolved when the handle is yielded."""
+    fake = SyncFakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            tool_started_event(seq=1, tool_call_id="call-1", tool_name="search"),
+            tool_finished_event(seq=2, tool_call_id="call-1", output={"ok": True}),
+            lifecycle_completed_event(seq=3),
+        ]
+    )
+    fake.set_state({})
+    outputs: list[Any] = []
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            for call in thread.tool_calls:
+                outputs.append(call.output)  # resolved (blocking) on yield
+    assert outputs == [{"ok": True}]
+
+
+# ---------------------------------------------------------------------------
+# Regression: interleaved-concurrent tool calls must BOTH surface
+# ---------------------------------------------------------------------------
+
+
+def test_sync_tool_calls_interleaved_concurrent_calls_both_surface():
+    """Two tool calls whose events interleave must BOTH be yielded (regression:
+    the pre-decoder read-ahead silently dropped the second concurrent call)."""
+    fake = SyncFakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            tool_started_event(seq=1, tool_call_id="call-a", tool_name="search"),
+            tool_started_event(seq=2, tool_call_id="call-b", tool_name="lookup"),
+            tool_finished_event(seq=3, tool_call_id="call-a", output={"a": 1}),
+            tool_finished_event(seq=4, tool_call_id="call-b", output={"b": 2}),
+            lifecycle_completed_event(seq=5),
+        ]
+    )
+    fake.set_state({})
+    seen = []
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            for call in thread.tool_calls:
+                seen.append(call.tool_call_id)
+    assert sorted(seen) == ["call-a", "call-b"]

libs/sdk-py/tests/streaming/test_sync_thread_stream.py

@@ -602,3 +602,310 @@ def test_v3_streaming_sync_surface_smoke():
     assert tools_result[0].name == "search"  # ty: ignore[unresolved-attribute]
     assert results["progress"] == [{"name": "progress", "step": 1}]
     assert final == {"final": True}
+
+
+# ---------------------------------------------------------------------------
+# interleave_projections tests
+# ---------------------------------------------------------------------------
+
+
+def test_interleave_projections_single_channel_values():
+    from streaming._events import (
+        lifecycle_completed_event,
+        lifecycle_started_event,
+        values_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({"counter": 0})
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            values_event(seq=1, counter=1),
+            values_event(seq=2, counter=2),
+            lifecycle_completed_event(seq=3),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            items = []
+            for ch, item in thread.interleave_projections(["values"]):
+                items.append((ch, item))
+    assert ("values", {"counter": 1}) in items
+    assert ("values", {"counter": 2}) in items
+    assert all(ch == "values" for ch, _ in items)
+
+
+def test_interleave_projections_values_and_messages_arrival_order():
+    from streaming._events import (
+        lifecycle_completed_event,
+        lifecycle_started_event,
+        message_finish_event,
+        message_start_event,
+        values_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({"counter": 0})
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            values_event(seq=1, counter=1),
+            message_start_event(seq=2, message_id="m-1"),
+            values_event(seq=3, counter=2),
+            message_finish_event(seq=4, message_id="m-1"),
+            lifecycle_completed_event(seq=5),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            order = []
+            for ch, _ in thread.interleave_projections(["values", "messages"]):
+                order.append(ch)
+                if len(order) >= 3:
+                    break
+    assert order[:3] == ["values", "messages", "values"]
+
+
+def test_interleave_projections_mixes_builtin_and_extension():
+    from streaming._events import (
+        custom_event,
+        lifecycle_completed_event,
+        lifecycle_started_event,
+        values_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({"counter": 0})
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            values_event(seq=1, counter=1),
+            custom_event(seq=2, name="foo", hello="world"),
+            lifecycle_completed_event(seq=3),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            items = []
+            for ch, item in thread.interleave_projections(["values", "foo"]):
+                items.append((ch, item))
+    assert ("values", {"counter": 1}) in items
+    assert ("foo", {"name": "foo", "hello": "world"}) in items
+
+
+def test_interleave_projections_tool_calls_uses_public_name():
+    from streaming._events import (
+        lifecycle_completed_event,
+        lifecycle_started_event,
+        tool_finished_event,
+        tool_started_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({})
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            tool_started_event(seq=1, tool_call_id="call-1", tool_name="search"),
+            tool_finished_event(seq=2, tool_call_id="call-1", output={"ok": True}),
+            lifecycle_completed_event(seq=3),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            names = []
+            handle = None
+            for ch, item in thread.interleave_projections(["tool_calls"]):
+                names.append(ch)
+                if handle is None:
+                    handle = item
+                break
+    assert names == ["tool_calls"]
+    assert handle is not None
+    assert handle.tool_call_id == "call-1"
+
+
+def test_interleave_projections_subgraphs_discovers_child():
+    from streaming._events import (
+        lifecycle_completed_event,
+        lifecycle_started_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({})
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            lifecycle_started_event(seq=1, namespace=["child"]),
+            lifecycle_completed_event(seq=2),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            discovered = []
+            for ch, handle in thread.interleave_projections(["subgraphs"]):
+                discovered.append((ch, handle.path))
+    assert ("subgraphs", ("child",)) in discovered
+
+
+def test_interleave_projections_inflight_tool_call_failed_on_break():
+    """A tool handle held past an early break is failed in teardown, never left hanging."""
+    from streaming._events import (
+        lifecycle_completed_event,
+        lifecycle_started_event,
+        tool_started_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({})
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            tool_started_event(seq=1, tool_call_id="call-1", tool_name="search"),
+            # no tool-finished: the call is still in flight when the consumer breaks
+            lifecycle_completed_event(seq=2),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            handle = None
+            for _, item in thread.interleave_projections(["tool_calls"]):
+                handle = item
+                break
+            assert handle is not None
+            # Without teardown finalization this blocks forever; the bounded
+            # timeout turns a regression into a TimeoutError, not a RuntimeError.
+            with pytest.raises(RuntimeError):
+                handle._result.result(timeout=2)
+
+
+def test_interleave_projections_inflight_subgraph_finished_on_terminal():
+    """A discovered subgraph child with no terminal tasks-result is force-completed."""
+    from streaming._events import (
+        lifecycle_completed_event,
+        lifecycle_started_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({})
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            lifecycle_started_event(seq=1, namespace=["child"]),
+            # no tasks-result for the child: it is still "started" at run end
+            lifecycle_completed_event(seq=2),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            child = None
+            for _, handle in thread.interleave_projections(["subgraphs"]):
+                child = handle
+            assert child is not None
+            assert child.status == "completed"
+
+
+@pytest.mark.parametrize("channel", ["lifecycle", "tools", "input"])
+def test_interleave_projections_rejects_reserved_channel(channel):
+    """Reserved protocol channel names raise instead of silently no-op'ing.
+
+    `infer_channel` treats these as first-class methods, but they have no
+    interleave decoder, so routing them to the extension/`custom:` fallback
+    would subscribe to a channel that never matches and yield nothing. Fail
+    closed. (`updates`/`checkpoints`/`tasks` are supported and tested below.)
+    """
+    from streaming._events import (
+        lifecycle_completed_event,
+        lifecycle_started_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({})
+    fake.script([lifecycle_started_event(seq=0), lifecycle_completed_event(seq=1)])
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with (
+            threads.stream(thread_id="t-1", assistant_id="agent") as thread,
+            pytest.raises(ValueError, match=channel),
+        ):
+            for _ in thread.interleave_projections([channel]):
+                pass
+
+
+def test_interleave_projections_data_channels_yield_payloads():
+    """`updates`/`checkpoints`/`tasks` yield their raw `params.data` payloads."""
+    from streaming._events import (
+        checkpoints_event,
+        lifecycle_completed_event,
+        lifecycle_started_event,
+        tasks_start_event,
+        updates_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({})
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            updates_event(seq=1, node={"v": 1}),
+            checkpoints_event(seq=2, ts="t-0", v=4),
+            tasks_start_event(seq=3, task_id="task-9"),
+            lifecycle_completed_event(seq=4),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            items = list(
+                thread.interleave_projections(["updates", "checkpoints", "tasks"])
+            )
+    assert ("updates", {"node": {"v": 1}}) in items
+    assert ("checkpoints", {"ts": "t-0", "v": 4}) in items
+    assert any(ch == "tasks" and item.get("id") == "task-9" for ch, item in items)
+
+
+def test_interleave_projections_data_channel_scoped_to_root_namespace():
+    """A child-namespace checkpoint must not leak into a root interleave."""
+    from streaming._events import (
+        checkpoints_event,
+        lifecycle_completed_event,
+        lifecycle_started_event,
+    )
+
+    fake = SyncFakeServer()
+    fake.set_state({"counter": 0})
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            checkpoints_event(seq=1, namespace=["child"], scope="child"),
+            checkpoints_event(seq=2, scope="root"),
+            lifecycle_completed_event(seq=3),
+        ]
+    )
+    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
+        threads = SyncThreadsClient(SyncHttpClient(raw))
+        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            thread.run.start(input={})
+            checkpoints = [
+                item
+                for ch, item in thread.interleave_projections(["values", "checkpoints"])
+                if ch == "checkpoints"
+            ]
+    assert {"scope": "root"} in checkpoints
+    assert {"scope": "child"} not in checkpoints

libs/sdk-py/tests/streaming/test_thread_stream.py

@@ -17,9 +17,17 @@ from langgraph_sdk.stream.transport import (
     ProtocolWebSocketTransport,
 )
 from streaming._events import (
+    checkpoints_event,
+    custom_event,
     lifecycle_completed_event,
     lifecycle_event,
     lifecycle_started_event,
+    message_finish_event,
+    message_start_event,
+    tasks_start_event,
+    tool_finished_event,
+    tool_started_event,
+    updates_event,
     values_event,
 )
 from streaming._fake_server import FakeServer
@@ -971,3 +979,255 @@ async def test_v3_streaming_async_surface_smoke():
     assert tool_calls[0].name == "search"
     assert progress == [{"name": "progress", "step": 1}]
     assert final == {"final": True}
+
+
+async def test_interleave_projections_single_channel_values():
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            values_event(seq=1, counter=1),
+            values_event(seq=2, counter=2),
+            lifecycle_completed_event(seq=3),
+        ]
+    )
+    fake.set_state({"counter": 0})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            await thread.run.start(input={})
+            items = []
+            async for ch, item in thread.interleave_projections(["values"]):
+                items.append((ch, item))
+    assert ("values", {"counter": 1}) in items
+    assert ("values", {"counter": 2}) in items
+    assert all(ch == "values" for ch, _ in items)
+
+
+async def test_interleave_projections_values_and_messages_arrival_order():
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            values_event(seq=1, counter=1),
+            message_start_event(seq=2, message_id="m-1"),
+            values_event(seq=3, counter=2),
+            message_finish_event(seq=4, message_id="m-1"),
+            lifecycle_completed_event(seq=5),
+        ]
+    )
+    fake.set_state({"counter": 0})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            await thread.run.start(input={})
+            order = []
+            async for ch, _ in thread.interleave_projections(["values", "messages"]):
+                order.append(ch)
+                if len(order) >= 3:
+                    break
+    assert order[:3] == ["values", "messages", "values"]
+
+
+async def test_interleave_projections_mixes_builtin_and_extension():
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            values_event(seq=1, counter=1),
+            custom_event(seq=2, name="foo", hello="world"),
+            lifecycle_completed_event(seq=3),
+        ]
+    )
+    fake.set_state({"counter": 0})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            await thread.run.start(input={})
+            items = []
+            async for ch, item in thread.interleave_projections(["values", "foo"]):
+                items.append((ch, item))
+    assert ("values", {"counter": 1}) in items
+    # extension payload is the whole params.data including "name"; tuple uses bare name "foo"
+    assert ("foo", {"name": "foo", "hello": "world"}) in items
+
+
+async def test_interleave_projections_tool_calls_uses_public_name():
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            tool_started_event(seq=1, tool_call_id="call-1", tool_name="search"),
+            tool_finished_event(seq=2, tool_call_id="call-1", output={"ok": True}),
+            lifecycle_completed_event(seq=3),
+        ]
+    )
+    fake.set_state({})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            await thread.run.start(input={})
+            names = []
+            async for ch, item in thread.interleave_projections(["tool_calls"]):
+                names.append(ch)
+                assert item.tool_call_id == "call-1"  # real ToolCallHandle
+    # tuple uses the PUBLIC name "tool_calls", never the wire name "tools"
+    assert names == ["tool_calls"]
+
+
+async def test_interleave_projections_subgraphs_discovers_child():
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            lifecycle_started_event(seq=1, namespace=["child"]),
+            lifecycle_completed_event(seq=2),
+        ]
+    )
+    fake.set_state({})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            await thread.run.start(input={})
+            discovered = []
+            async for ch, handle in thread.interleave_projections(["subgraphs"]):
+                discovered.append((ch, handle.path))
+    assert ("subgraphs", ("child",)) in discovered
+
+
+async def test_interleave_projections_inflight_tool_call_failed_on_break():
+    """A tool handle held past an early break is failed in teardown, never left hanging."""
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            tool_started_event(seq=1, tool_call_id="call-1", tool_name="search"),
+            # no tool-finished: the call is still in flight when the consumer breaks
+            lifecycle_completed_event(seq=2),
+        ]
+    )
+    fake.set_state({})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            await thread.run.start(input={})
+            handle = None
+            async for _, item in thread.interleave_projections(["tool_calls"]):
+                handle = item
+                break
+            assert handle is not None
+            # Without teardown finalization this would hang forever; wait_for
+            # turns a regression into a TimeoutError rather than a RuntimeError.
+            with pytest.raises(RuntimeError):
+                await asyncio.wait_for(handle.output, timeout=2)
+
+
+async def test_interleave_projections_inflight_subgraph_finished_on_terminal():
+    """A discovered subgraph child with no terminal tasks-result is force-completed."""
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            lifecycle_started_event(seq=1, namespace=["child"]),
+            # no tasks-result for the child: it is still "started" at run end
+            lifecycle_completed_event(seq=2),
+        ]
+    )
+    fake.set_state({})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            await thread.run.start(input={})
+            child = None
+            async for _, handle in thread.interleave_projections(["subgraphs"]):
+                child = handle
+            assert child is not None
+            assert child.status == "completed"
+
+
+@pytest.mark.parametrize("channel", ["lifecycle", "tools", "input"])
+async def test_interleave_projections_rejects_reserved_channel(channel):
+    """Reserved protocol channel names raise instead of silently no-op'ing.
+
+    `infer_channel` treats these as first-class methods, but they have no
+    interleave decoder, so routing them to the extension/`custom:` fallback
+    would subscribe to a channel that never matches and yield nothing. Fail
+    closed. (`updates`/`checkpoints`/`tasks` are supported and tested below.)
+    """
+    fake = FakeServer()
+    fake.script([lifecycle_started_event(seq=0), lifecycle_completed_event(seq=1)])
+    fake.set_state({})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            with pytest.raises(ValueError, match=channel):
+                async for _ in thread.interleave_projections([channel]):
+                    pass
+
+
+async def test_interleave_projections_data_channels_yield_payloads():
+    """`updates`/`checkpoints`/`tasks` yield their raw `params.data` payloads."""
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            updates_event(seq=1, node={"v": 1}),
+            checkpoints_event(seq=2, ts="t-0", v=4),
+            tasks_start_event(seq=3, task_id="task-9"),
+            lifecycle_completed_event(seq=4),
+        ]
+    )
+    fake.set_state({})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            await thread.run.start(input={})
+            items = []
+            async for ch, item in thread.interleave_projections(
+                ["updates", "checkpoints", "tasks"]
+            ):
+                items.append((ch, item))
+    assert ("updates", {"node": {"v": 1}}) in items
+    assert ("checkpoints", {"ts": "t-0", "v": 4}) in items
+    assert any(ch == "tasks" and item.get("id") == "task-9" for ch, item in items)
+
+
+async def test_interleave_projections_data_channel_scoped_to_root_namespace():
+    """A child-namespace checkpoint must not leak into a root interleave.
+
+    `values` subscribes unscoped, so `compute_union_filter` widens the merged
+    subscription to all namespaces; the `DataDecoder` root filter is what keeps
+    a subgraph checkpoint out of the root projection (mirrors local scope).
+    """
+    fake = FakeServer()
+    fake.script(
+        [
+            lifecycle_started_event(seq=0),
+            checkpoints_event(seq=1, namespace=["child"], scope="child"),
+            checkpoints_event(seq=2, scope="root"),
+            lifecycle_completed_event(seq=3),
+        ]
+    )
+    fake.set_state({"counter": 0})
+    asgi = httpx.ASGITransport(app=fake.app)
+    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
+        threads = ThreadsClient(HttpClient(raw))
+        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
+            await thread.run.start(input={})
+            checkpoints = []
+            async for ch, item in thread.interleave_projections(
+                ["values", "checkpoints"]
+            ):
+                if ch == "checkpoints":
+                    checkpoints.append(item)
+    assert {"scope": "root"} in checkpoints
+    assert {"scope": "child"} not in checkpoints

libs/sdk-py/tests/streaming/test_transport_path_encoding.py

@@ -0,0 +1,136 @@
+"""Regression tests for #7953: v3 stream transports must percent-encode
+`thread_id` in their default paths so a value containing reserved characters
+or dot-segments stays an opaque identifier under `/threads/{thread_id}/...`
+instead of being normalized into a different resource path by the HTTP stack.
+"""
+
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from langgraph_sdk.stream.transport.base import build_websocket_url
+from langgraph_sdk.stream.transport.http import ProtocolSseTransport
+from langgraph_sdk.stream.transport.sync_http import SyncProtocolSseTransport
+from langgraph_sdk.stream.transport.sync_ws import SyncProtocolWebSocketTransport
+from langgraph_sdk.stream.transport.ws import ProtocolWebSocketTransport
+
+# A thread_id that escapes the /threads/ namespace if interpolated raw: an HTTP
+# client collapses `/threads/../assistants/abc/...` to `/assistants/abc/...`.
+TRAVERSAL_THREAD_ID = "../assistants/abc"
+ENCODED_COMMANDS_PATH = "/threads/..%2Fassistants%2Fabc/commands"
+ENCODED_STREAM_PATH = "/threads/..%2Fassistants%2Fabc/stream/events"
+
+
+@pytest.mark.anyio
+async def test_async_sse_default_paths_encode_thread_id():
+    transport = ProtocolSseTransport(
+        client=httpx.AsyncClient(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_url == ENCODED_STREAM_PATH
+
+
+def test_sync_sse_default_paths_encode_thread_id():
+    transport = SyncProtocolSseTransport(
+        client=httpx.Client(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_url == ENCODED_STREAM_PATH
+
+
+@pytest.mark.anyio
+async def test_async_ws_default_paths_encode_thread_id():
+    transport = ProtocolWebSocketTransport(
+        client=httpx.AsyncClient(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_path == ENCODED_STREAM_PATH
+
+
+def test_sync_ws_default_paths_encode_thread_id():
+    transport = SyncProtocolWebSocketTransport(
+        client=httpx.Client(), thread_id=TRAVERSAL_THREAD_ID
+    )
+    assert transport._commands_url == ENCODED_COMMANDS_PATH
+    assert transport._stream_path == ENCODED_STREAM_PATH
+
+
+async def test_async_sse_wire_path_stays_under_threads_namespace():
+    """The path that actually goes on the wire must not be normalized away."""
+    seen: list[str] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request.url.raw_path.decode("ascii"))
+        return httpx.Response(202)
+
+    async with httpx.AsyncClient(
+        transport=httpx.MockTransport(handler),
+        base_url="https://example.com",
+        trust_env=False,
+    ) as client:
+        transport = ProtocolSseTransport(client=client, thread_id=TRAVERSAL_THREAD_ID)
+        await transport.send_command({"id": 1, "method": "noop", "params": {}})
+
+    assert seen[0] == ENCODED_COMMANDS_PATH
+
+
+def test_sync_sse_wire_path_stays_under_threads_namespace():
+    seen: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request.url.raw_path.decode("ascii"))
+        return httpx.Response(202)
+
+    with httpx.Client(
+        transport=httpx.MockTransport(handler),
+        base_url="https://example.com",
+        trust_env=False,
+    ) as client:
+        transport = SyncProtocolSseTransport(
+            client=client, thread_id=TRAVERSAL_THREAD_ID
+        )
+        transport.send_command({"id": 1, "method": "noop", "params": {}})
+
+    assert seen[0] == ENCODED_COMMANDS_PATH
+
+
+@pytest.mark.anyio
+async def test_async_ws_url_stays_under_threads_namespace():
+    transport = ProtocolWebSocketTransport(
+        client=httpx.AsyncClient(base_url="https://example.com/api"),
+        thread_id=TRAVERSAL_THREAD_ID,
+    )
+    url = build_websocket_url(transport._client.base_url, transport._stream_path)
+    assert url == "wss://example.com/api/threads/..%2Fassistants%2Fabc/stream/events"
+
+
+def test_sync_ws_url_stays_under_threads_namespace():
+    transport = SyncProtocolWebSocketTransport(
+        client=httpx.Client(base_url="https://example.com/api"),
+        thread_id=TRAVERSAL_THREAD_ID,
+    )
+    url = build_websocket_url(transport._client.base_url, transport._stream_path)
+    assert url == "wss://example.com/api/threads/..%2Fassistants%2Fabc/stream/events"
+
+
+@pytest.mark.anyio
+async def test_explicit_path_overrides_are_left_untouched():
+    """Callers passing explicit paths opt out of default encoding entirely."""
+    sse = ProtocolSseTransport(
+        client=httpx.AsyncClient(),
+        thread_id=TRAVERSAL_THREAD_ID,
+        commands_path="/custom/commands",
+        stream_path="/custom/events",
+    )
+    assert sse._commands_url == "/custom/commands"
+    assert sse._stream_url == "/custom/events"
+
+    ws = ProtocolWebSocketTransport(
+        client=httpx.AsyncClient(),
+        thread_id=TRAVERSAL_THREAD_ID,
+        commands_path="/custom/commands",
+        stream_path="/custom/events",
+    )
+    assert ws._commands_url == "/custom/commands"
+    assert ws._stream_path == "/custom/events"