release(langgraph): 1.2.17 (#9418)

Bumps langgraph from 1.2.16 to 1.2.17.

Includes:

- Apply an explicit `recursion_limit` equal to the default (#9417)
- Keep the caller's namespace in `RemoteGraph.stream(subgraphs=True)`
(#9401)

libs/cli/generate_schema.py

@@ -23,6 +23,7 @@ from langgraph_cli.schemas import (
     GraphDef,
     HttpConfig,
     IndexConfig,
+    PriorityConfig,
     SecurityConfig,
     SerdeConfig,
     StoreConfig,
@@ -113,6 +114,7 @@ def add_descriptions_to_schema(schema, cls):
                 StoreConfig,
                 IndexConfig,
                 AuthConfig,
+                PriorityConfig,
                 SecurityConfig,
                 HttpConfig,
                 CorsConfig,

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.37"
+__version__ = "0.4.38"

libs/cli/langgraph_cli/cli.py

@@ -847,6 +847,7 @@ def dev(
         store=config_json.get("store"),
         wait_for_client=wait_for_client,
         auth=config_json.get("auth"),
+        priority=config_json.get("priority"),
         http=config_json.get("http"),
         ui=config_json.get("ui"),
         ui_config=config_json.get("ui_config"),

libs/cli/langgraph_cli/config.py

@@ -442,6 +442,7 @@ def validate_config(
         "store": config.get("store"),
         "auth": config.get("auth"),
         "encryption": config.get("encryption"),
+        "priority": config.get("priority"),
         "http": config.get("http"),
         # Pass through webhooks config so it can be injected into the image
         "webhooks": config.get("webhooks"),
@@ -606,6 +607,19 @@ def validate_config(
                     f"Invalid encryption.path format: '{encryption_conf['path']}'. "
                     "Must be in format './path/to/file.py:attribute_name'"
                 )
+    # Validate priority config
+    if priority_conf := config.get("priority"):
+        if "path" in priority_conf:
+            if ":" not in priority_conf["path"]:
+                raise ValueError(
+                    f"Invalid priority.path format: '{priority_conf['path']}'. "
+                    "Must be in format './path/to/file.py:attribute_name'"
+                )
+            if _is_node_graph(priority_conf["path"]):
+                raise ValueError(
+                    "JavaScript priority callbacks are not supported: "
+                    f"'{priority_conf['path']}'. Use a Python file or module instead."
+                )
     if http_conf := config.get("http"):
         if "app" in http_conf:
             if ":" not in http_conf["app"]:
@@ -649,6 +663,7 @@ _KNOWN_CONFIG_KEYS = {
     "store",
     "auth",
     "encryption",
+    "priority",
     "http",
     "webhooks",
     "checkpointer",
@@ -1080,6 +1095,49 @@ def _update_encryption_path(
     )
 
 
+def _update_priority_path(
+    config_path: pathlib.Path, config: Config, local_deps: LocalDeps
+) -> None:
+    """Update priority.path to use Docker container paths."""
+    priority_conf = config.get("priority")
+    if not priority_conf or not (path_str := priority_conf.get("path")):
+        return
+
+    module_str, sep, attr_str = path_str.partition(":")
+    if not sep or not module_str.startswith("."):
+        return  # Already validated or absolute path
+
+    resolved = config_path.parent / module_str
+    if not resolved.exists():
+        raise FileNotFoundError(
+            f"Priority file not found: {resolved} (from {path_str})"
+        )
+    if not resolved.is_file():
+        raise IsADirectoryError(f"Priority path must be a file: {resolved}")
+
+    # Check faux packages first (they take precedence)
+    for faux_path, (_, destpath) in local_deps.faux_pkgs.items():
+        if resolved.is_relative_to(faux_path):
+            new_path = f"{destpath}/{resolved.relative_to(faux_path)}:{attr_str}"
+            priority_conf["path"] = new_path
+            return
+
+    # Check real packages
+    for real_path in local_deps.real_pkgs:
+        if resolved.is_relative_to(real_path):
+            new_path = (
+                f"/deps/{real_path.name}/{resolved.relative_to(real_path)}:{attr_str}"
+            )
+            priority_conf["path"] = new_path
+            return
+
+    raise ValueError(
+        f"Priority file '{resolved}' not covered by dependencies.\n"
+        "Add its parent directory to the 'dependencies' array in your config.\n"
+        f"Current dependencies: {config['dependencies']}"
+    )
+
+
 def _update_checkpointer_path(
     config_path: pathlib.Path, config: Config, local_deps: LocalDeps
 ) -> None:
@@ -1213,6 +1271,9 @@ def _build_runtime_env_vars(config: Config) -> list[str]:
     if (encryption_config := config.get("encryption")) is not None:
         env_vars.append(f"ENV LANGGRAPH_ENCRYPTION='{json.dumps(encryption_config)}'")
 
+    if (priority_config := config.get("priority")) is not None:
+        env_vars.append(f"ENV LANGGRAPH_PRIORITY='{json.dumps(priority_config)}'")
+
     if (http_config := config.get("http")) is not None:
         env_vars.append(f"ENV LANGGRAPH_HTTP='{json.dumps(http_config)}'")
 
@@ -1380,6 +1441,8 @@ def python_config_to_docker(
     _update_auth_path(config_path, config, local_deps)
     # Rewrite encryption path, so it points to the correct location in the Docker container
     _update_encryption_path(config_path, config, local_deps)
+    # Rewrite priority path, so it points to the correct location in the Docker container
+    _update_priority_path(config_path, config, local_deps)
     # Rewrite checkpointer path, so it points to the correct location in the Docker container
     _update_checkpointer_path(config_path, config, local_deps)
     # Rewrite HTTP app path, so it points to the correct location in the Docker container

libs/cli/langgraph_cli/schemas.py

@@ -373,6 +373,27 @@ class EncryptionConfig(TypedDict, total=False):
     """
 
 
+class PriorityConfig(TypedDict, total=False):
+    """Configuration for custom run priority logic.
+
+    Allows you to decide each thread's priority settings with a Python function
+    that the server calls at the first run on a thread with no settings, which
+    can come long after the thread was created."""
+
+    path: str
+    """Required. Path to an instance of the Priority() class that registers an assign function.
+
+    Format: "path/to/file.py:my_priority"
+
+    Example:
+        {
+            "priority": {
+                "path": "./priority.py:my_priority"
+            }
+        }
+    """
+
+
 class CorsConfig(TypedDict, total=False):
     """Specifies Cross-Origin Resource Sharing (CORS) rules for your server.
 
@@ -753,6 +774,13 @@ class Config(TypedDict, total=False):
     Allows you to implement custom encryption for sensitive data stored in the database.
     """
 
+    priority: PriorityConfig | None
+    """Optional. Custom run priority config, including the path to your Python priority logic.
+
+    Allows you to decide each thread's priority settings at the first run on a thread with no settings,
+    which can come long after the thread was created.
+    """
+
     http: HttpConfig | None
     """Optional. Configuration for the built-in HTTP server, controlling which custom routes are exposed
     and how cross-origin requests are handled.
@@ -784,6 +812,7 @@ __all__ = [
     "CheckpointerConfig",
     "AuthConfig",
     "EncryptionConfig",
+    "PriorityConfig",
     "HttpConfig",
     "MiddlewareOrders",
     "Distros",

libs/cli/langgraph_cli/uv_lock.py

@@ -910,6 +910,7 @@ def python_config_to_docker_uv_lock(
     for section, key in [
         ("auth", "path"),
         ("encryption", "path"),
+        ("priority", "path"),
         ("checkpointer", "path"),
         ("http", "app"),
     ]:

libs/cli/schemas/schema.json

@@ -196,6 +196,17 @@
                 }
               ]
             },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
             "source": {
               "anyOf": [
                 {
@@ -438,6 +449,17 @@
                 }
               ]
             },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
             "source": {
               "$ref": "#/$defs/UvSource"
             },
@@ -666,6 +688,17 @@
                 }
               ]
             },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
             "source": {
               "anyOf": [
                 {
@@ -1170,6 +1203,18 @@
       },
       "required": []
     },
+    "PriorityConfig": {
+      "title": "PriorityConfig",
+      "description": "Configuration for custom run priority logic.\n\nAllows you to decide each thread's priority settings with a Python function\nthat the server calls at the first run on a thread with no settings, which\ncan come long after the thread was created.",
+      "type": "object",
+      "properties": {
+        "path": {
+          "type": "string",
+          "description": "Required. Path to an instance of the Priority() class that registers an assign function.\n"
+        }
+      },
+      "required": []
+    },
     "UvSource": {
       "title": "UvSource",
       "description": "Deployment source rooted at a uv project or workspace.",

libs/cli/schemas/schema.v0.json

@@ -196,6 +196,17 @@
                 }
               ]
             },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
             "source": {
               "anyOf": [
                 {
@@ -438,6 +449,17 @@
                 }
               ]
             },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
             "source": {
               "$ref": "#/$defs/UvSource"
             },
@@ -666,6 +688,17 @@
                 }
               ]
             },
+            "priority": {
+              "anyOf": [
+                {
+                  "$ref": "#/$defs/PriorityConfig"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
+            },
             "source": {
               "anyOf": [
                 {
@@ -1170,6 +1203,18 @@
       },
       "required": []
     },
+    "PriorityConfig": {
+      "title": "PriorityConfig",
+      "description": "Configuration for custom run priority logic.\n\nAllows you to decide each thread's priority settings with a Python function\nthat the server calls at the first run on a thread with no settings, which\ncan come long after the thread was created.",
+      "type": "object",
+      "properties": {
+        "path": {
+          "type": "string",
+          "description": "Required. Path to an instance of the Priority() class that registers an assign function.\n"
+        }
+      },
+      "required": []
+    },
     "UvSource": {
       "title": "UvSource",
       "description": "Deployment source rooted at a uv project or workspace.",

libs/cli/tests/unit_tests/cli/test_cli.py

@@ -2,10 +2,12 @@ import json
 import pathlib
 import re
 import shutil
+import sys
 import tempfile
 import textwrap
 from contextlib import contextmanager
 from pathlib import Path
+from unittest.mock import MagicMock
 
 import click
 from click.testing import CliRunner
@@ -371,6 +373,32 @@ def test_dev_command_requires_ssl_certfile_and_keyfile_together(tmp_path) -> Non
     assert "Both --ssl-certfile and --ssl-keyfile must be provided" in result.output
 
 
+def test_dev_command_passes_priority_to_the_server(tmp_path, monkeypatch) -> None:
+    config_path = tmp_path / "langgraph.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "dependencies": ["."],
+                "graphs": {"agent": "./agent.py:graph"},
+                "priority": {"path": "./priority.py:priority"},
+            }
+        ),
+        encoding="utf-8",
+    )
+    # The test environment doesn't install langgraph-api, so stand in for its server.
+    server = MagicMock()
+    monkeypatch.setitem(sys.modules, "langgraph_api.cli", server)
+    # dev appends the working directory and its dependencies to sys.path.
+    monkeypatch.setattr(sys, "path", list(sys.path))
+
+    result = CliRunner().invoke(cli, ["dev", "--config", str(config_path)])
+
+    assert result.exit_code == 0, result.output
+    assert server.run_server.call_args.kwargs["priority"] == {
+        "path": "./priority.py:priority"
+    }
+
+
 def test_deploy_list_command(monkeypatch) -> None:
     runner = CliRunner()
     captured: dict[str, str] = {}

libs/cli/tests/unit_tests/test_config.py

@@ -16,6 +16,7 @@ from langgraph_cli.config import (
     config_to_docker,
     default_base_image,
     docker_tag,
+    get_unknown_keys,
     has_disallowed_build_command_content,
     validate_config,
     validate_config_file,
@@ -155,6 +156,7 @@ def test_validate_config():
         "store": None,
         "auth": None,
         "encryption": None,
+        "priority": None,
         "webhooks": None,
         "checkpointer": None,
         "http": None,
@@ -184,6 +186,7 @@ def test_validate_config():
         "store": None,
         "auth": None,
         "encryption": None,
+        "priority": None,
         "webhooks": None,
         "checkpointer": None,
         "http": None,
@@ -1289,6 +1292,109 @@ def test_config_to_docker_python_encryption_formatted():
     )
 
 
+@pytest.mark.parametrize(
+    "path", ["./priority.py:priority", "my_package.priority:priority"]
+)
+def test_config_to_docker_python_priority(path: str):
+    # Test that priority config is kept by validation and isn't an unknown key
+    raw_config = {
+        "python_version": "3.11",
+        "graphs": {"agent": "./agent.py:graph"},
+        "dependencies": ["."],
+        "priority": {"path": path},
+    }
+    validated = validate_config(raw_config)
+
+    assert validated["priority"] == {"path": path}
+    assert get_unknown_keys(raw_config) == []
+
+
+def test_config_to_docker_python_priority_bad_path():
+    # Test that invalid priority path format raises ValueError
+    with pytest.raises(ValueError, match="Invalid priority.path format"):
+        validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "./agent.py:graph"},
+                "dependencies": ["."],
+                "priority": {"path": "./priority.py"},  # Missing :attribute
+            }
+        )
+
+
+@pytest.mark.parametrize("ext", [".ts", ".mts", ".cts", ".js", ".mjs", ".cjs"])
+def test_config_to_docker_python_priority_js_path(ext: str):
+    # Test that a JavaScript priority path is rejected
+    with pytest.raises(ValueError, match="JavaScript priority callbacks"):
+        validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {"agent": "./agent.py:graph"},
+                "dependencies": ["."],
+                "priority": {"path": f"./priority{ext}:priority"},
+            }
+        )
+
+
+def test_config_to_docker_nodejs_priority():
+    # Test that a Node.js-only project can name a Python priority file
+    actual_docker_stdin, _ = config_to_docker(
+        PATH_TO_CONFIG,
+        validate_config(
+            {
+                "node_version": "20",
+                "graphs": {"agent": "./graphs/agent.js:graph"},
+                "priority": {"path": "./priority.py:priority"},
+            }
+        ),
+        base_image="langchain/langgraphjs-api",
+    )
+    assert (
+        """ENV LANGGRAPH_PRIORITY='{"path": "./priority.py:priority"}'"""
+        in actual_docker_stdin
+    )
+
+
+def test_config_to_docker_python_priority_formatted():
+    # Test that priority config is written to the Docker output with its container path
+    actual_docker_stdin, _ = config_to_docker(
+        PATH_TO_CONFIG,
+        validate_config(
+            {
+                "python_version": "3.11",
+                "dependencies": ["."],
+                "graphs": {"agent": "./graphs/agent.py:graph"},
+                "priority": {"path": "./agent.py:my_priority"},
+            }
+        ),
+        base_image="langchain/langgraph-api",
+    )
+    assert (
+        "ENV LANGGRAPH_PRIORITY="
+        """'{"path": "/deps/outer-unit_tests/unit_tests/agent.py:my_priority"}'"""
+        in actual_docker_stdin
+    )
+
+
+def test_config_to_docker_python_priority_not_in_dependencies():
+    # Test that a priority file outside the dependencies raises ValueError
+    with pytest.raises(
+        ValueError, match="Priority file .* not covered by dependencies"
+    ):
+        config_to_docker(
+            PATH_TO_CONFIG,
+            validate_config(
+                {
+                    "python_version": "3.11",
+                    "dependencies": ["./graphs"],
+                    "graphs": {"agent": "./graphs/agent.py:graph"},
+                    "priority": {"path": "./agent.py:my_priority"},
+                }
+            ),
+            base_image="langchain/langgraph-api",
+        )
+
+
 def test_config_to_docker_nodejs_internal_docker_tag():
     graphs = {"agent": "./graphs/agent.js:graph"}
     actual_docker_stdin, additional_contexts = config_to_docker(
@@ -1731,6 +1837,38 @@ def test_config_to_docker_uv_lock_honors_root_workspace_sources():
         )
 
 
+def test_config_to_docker_uv_lock_rewrites_priority_path():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        tmpdir_path = pathlib.Path(tmpdir)
+        project_root, config_path = _write_uv_lock_workspace(
+            tmpdir_path,
+            agent_sources="[tool.uv.sources]\nshared = { workspace = true }",
+        )
+        (project_root / "libs" / "shared" / "src" / "shared" / "priority.py").touch()
+
+        config = validate_config(
+            {
+                "python_version": "3.11",
+                "graphs": {
+                    "agent": "../../apps/agent/src/agent/graph.py:graph",
+                },
+                "source": {"kind": "uv", "root": "../..", "package": "agent"},
+                "priority": {
+                    "path": "../../libs/shared/src/shared/priority.py:priority"
+                },
+            }
+        )
+        docker, _ = config_to_docker(
+            config_path, config, base_image="langchain/langgraph-api:0.2.47"
+        )
+
+        assert (
+            "ENV LANGGRAPH_PRIORITY="
+            """'{"path": "/deps/workspace/libs/shared/src/shared/priority.py:priority"}'"""
+            in docker
+        )
+
+
 def test_config_to_docker_uv_lock_ignores_unrelated_workspace_package_sources():
     with tempfile.TemporaryDirectory() as tmpdir:
         tmpdir_path = pathlib.Path(tmpdir)

libs/langgraph/langgraph/_internal/_config.py

@@ -29,7 +29,19 @@ from langgraph._internal._constants import (
     NS_SEP,
 )
 
-DEFAULT_RECURSION_LIMIT = int(getenv("LANGGRAPH_DEFAULT_RECURSION_LIMIT", "10007"))
+
+class _DefaultRecursionLimit(int):
+    """The limit `ensure_config` fills in when none is given.
+
+    `merge_configs` skips values of this type rather than any equal value, so an
+    explicit limit that happens to equal the default still applies. It checks the
+    type, not identity, because copying or pickling a config rebuilds the value.
+    """
+
+
+DEFAULT_RECURSION_LIMIT = _DefaultRecursionLimit(
+    getenv("LANGGRAPH_DEFAULT_RECURSION_LIMIT", "10007")
+)
 DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT = int(
     getenv("LANGGRAPH_DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT", "5000")
 )
@@ -182,7 +194,7 @@ def merge_configs(*configs: RunnableConfig | None) -> RunnableConfig:
                     base.get("callbacks"), cast(Callbacks, value)
                 )
             elif key == "recursion_limit":
-                if config["recursion_limit"] != DEFAULT_RECURSION_LIMIT:
+                if not isinstance(config["recursion_limit"], _DefaultRecursionLimit):
                     base["recursion_limit"] = config["recursion_limit"]
             else:
                 base[key] = config[key]  # type: ignore[literal-required]

libs/langgraph/langgraph/pregel/remote.py

@@ -861,11 +861,6 @@ class RemoteGraph(PregelProtocol):
                     )
                 yield {"type": mode, "ns": ns, "data": chunk.data, "interrupts": ints}
             elif subgraphs:
-                if NS_SEP in chunk.event:
-                    mode, ns_ = chunk.event.split(NS_SEP, 1)
-                    ns = tuple(ns_.split(NS_SEP))
-                else:
-                    mode, ns = chunk.event, ()
                 if req_single:
                     yield ns, chunk.data
                 else:
@@ -1016,11 +1011,6 @@ class RemoteGraph(PregelProtocol):
                     )
                 yield {"type": mode, "ns": ns, "data": chunk.data, "interrupts": ints}
             elif subgraphs:
-                if NS_SEP in chunk.event:
-                    mode, ns_ = chunk.event.split(NS_SEP, 1)
-                    ns = tuple(ns_.split(NS_SEP))
-                else:
-                    mode, ns = chunk.event, ()
                 if req_single:
                     yield ns, chunk.data
                 else:

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.16"
+version = "1.2.17"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"

libs/langgraph/tests/test_config_async.py

@@ -1,11 +1,19 @@
+from collections.abc import Callable
+
 import pytest
 from langchain_core.callbacks import AsyncCallbackManager, BaseCallbackHandler
 
-from langgraph._internal._config import get_async_callback_manager_for_config
+from langgraph._internal._config import (
+    DEFAULT_RECURSION_LIMIT,
+    get_async_callback_manager_for_config,
+)
 from langgraph.graph import StateGraph
+from langgraph.pregel import Pregel
 
 pytestmark = pytest.mark.anyio
 
+EXPLICIT_DEFAULT_LIMIT = int(DEFAULT_RECURSION_LIMIT)
+
 
 def test_new_async_manager_includes_tags() -> None:
     config = {"callbacks": None}
@@ -114,3 +122,27 @@ async def test_with_config_tags_preserved_on_invoke() -> None:
     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"
+
+
+@pytest.mark.parametrize(
+    "rebind",
+    [
+        lambda g: g.with_config(recursion_limit=EXPLICIT_DEFAULT_LIMIT),
+        lambda g: g.with_config({"recursion_limit": EXPLICIT_DEFAULT_LIMIT}),
+        lambda g: g.with_config(
+            {"recursion_limit": 1}, recursion_limit=EXPLICIT_DEFAULT_LIMIT
+        ),
+        lambda g: g.with_config(recursion_limit=EXPLICIT_DEFAULT_LIMIT).with_config(
+            tags=["t"]
+        ),
+    ],
+    ids=["kwarg", "config", "kwarg_over_config", "kept_by_later_with_config"],
+)
+def test_with_config_applies_recursion_limit_equal_to_default(
+    rebind: Callable[[Pregel], Pregel],
+) -> None:
+    builder = StateGraph(dict)
+    builder.add_node("node", lambda state: state)
+    builder.add_edge("__start__", "node")
+    graph = builder.compile().with_config(recursion_limit=EXPLICIT_DEFAULT_LIMIT + 1)
+    assert rebind(graph).config["recursion_limit"] == EXPLICIT_DEFAULT_LIMIT

libs/langgraph/tests/test_remote_graph.py

@@ -819,6 +819,50 @@ async def test_astream():
     ]
 
 
+_CALLER_CONFIG: RunnableConfig = {
+    "configurable": {"thread_id": "thread_1", "checkpoint_ns": "outer:123"}
+}
+_SUBGRAPH_PARTS = [
+    StreamPart(event="values|sub:abc", data={"chunk": "inner"}),
+    StreamPart(event="values", data={"chunk": "root"}),
+]
+_CALLER_PREFIXED = [
+    (("outer:123", "sub:abc"), {"chunk": "inner"}),
+    (("outer:123",), {"chunk": "root"}),
+]
+
+
+def test_stream_subgraphs_keeps_caller_namespace():
+    mock_sync_client = MagicMock()
+    mock_sync_client.runs.stream.return_value = _SUBGRAPH_PARTS
+    remote_pregel = RemoteGraph("test_graph_id", sync_client=mock_sync_client)
+
+    stream_parts = list(
+        remote_pregel.stream(
+            {"input": "data"}, _CALLER_CONFIG, stream_mode="values", subgraphs=True
+        )
+    )
+
+    assert stream_parts == _CALLER_PREFIXED
+
+
+async def test_astream_subgraphs_keeps_caller_namespace():
+    mock_async_client = MagicMock()
+    async_iter = MagicMock()
+    async_iter.__aiter__.return_value = _SUBGRAPH_PARTS
+    mock_async_client.runs.stream.return_value = async_iter
+    remote_pregel = RemoteGraph("test_graph_id", client=mock_async_client)
+
+    stream_parts = [
+        part
+        async for part in remote_pregel.astream(
+            {"input": "data"}, _CALLER_CONFIG, stream_mode="values", subgraphs=True
+        )
+    ]
+
+    assert stream_parts == _CALLER_PREFIXED
+
+
 def test_invoke():
     # set up test
     mock_sync_client = MagicMock()

libs/langgraph/tests/test_utils.py

@@ -1,4 +1,6 @@
+import copy
 import functools
+import pickle
 import sys
 import uuid
 from collections.abc import Callable
@@ -25,6 +27,7 @@ from pydantic import BaseModel, Field
 from typing_extensions import NotRequired, Required, TypedDict
 
 from langgraph._internal._config import (
+    DEFAULT_RECURSION_LIMIT,
     _is_not_empty,
     _merge_callbacks,
     ensure_config,
@@ -620,6 +623,30 @@ def test_merge_configs_merges_metadata_lc_versions() -> None:
     assert merged["metadata"]["lc_agent_name"] == "agent"
 
 
+@pytest.mark.parametrize(
+    "rebuild",
+    [
+        lambda config: config,
+        copy.deepcopy,
+        lambda config: pickle.loads(pickle.dumps(config)),
+    ],
+    ids=["as_is", "deepcopy", "pickle"],
+)
+def test_merge_configs_skips_the_injected_default_recursion_limit(
+    rebuild: Callable[[RunnableConfig], RunnableConfig],
+) -> None:
+    merged = merge_configs({"recursion_limit": 50}, rebuild(ensure_config()))
+    assert merged["recursion_limit"] == 50
+
+
+def test_merge_configs_applies_an_explicit_recursion_limit_equal_to_default() -> None:
+    explicit_default = int(DEFAULT_RECURSION_LIMIT)
+    merged = merge_configs(
+        {"recursion_limit": 50}, {"recursion_limit": explicit_default}
+    )
+    assert merged["recursion_limit"] == explicit_default
+
+
 def test_ensure_config_merges_metadata_lc_versions() -> None:
     a = {
         "metadata": {

libs/langgraph/uv.lock

@@ -1451,7 +1451,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.16"
+version = "1.2.17"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1919,7 +1919,7 @@ requires-dist = [
     { name = "langchain-core", specifier = ">=1.4.0,<2" },
     { name = "langchain-protocol", specifier = ">=0.0.15" },
     { name = "orjson", specifier = ">=3.11.5" },
-    { name = "websockets", specifier = ">=14,<17" },
+    { name = "websockets", specifier = ">=14" },
 ]
 
 [package.metadata.requires-dev]

libs/prebuilt/uv.lock

@@ -286,7 +286,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.16"
+version = "1.2.17"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -614,7 +614,7 @@ requires-dist = [
     { name = "langchain-core", specifier = ">=1.4.0,<2" },
     { name = "langchain-protocol", specifier = ">=0.0.15" },
     { name = "orjson", specifier = ">=3.11.5" },
-    { name = "websockets", specifier = ">=14,<17" },
+    { name = "websockets", specifier = ">=14" },
 ]
 
 [package.metadata.requires-dev]

libs/sdk-py/CHANGELOG.md

@@ -34,6 +34,14 @@
 - **`thread.output`** — awaitable that resolves to the terminal thread state
   `values` dict after the run lifecycle completes.
 
+- **`Priority` (beta)** — register an async function with `@priority.assign`
+  and name the object in the `priority` section of `langgraph.json`. The
+  server calls the function with a `PriorityContext` at the first run on a
+  thread with no settings and stores the `PrioritySettings` it returns
+  (`priority`, `fairness_group`, `fairness_weight`) on the thread. Requires
+  `langgraph-api` 0.18.0 or later and a `langgraph-cli` that supports the
+  section; older versions of either ignore it.
+
 ### Changed
 
 - `client.threads.stream()` now accepts `transport="sse"` (default) or

libs/sdk-py/langgraph_sdk/__init__.py

@@ -2,14 +2,19 @@ from langgraph_sdk.auth import Auth
 from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
 from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
+from langgraph_sdk.priority import Priority
+from langgraph_sdk.priority.types import PriorityContext, PrioritySettings
 
-__version__ = "0.4.8"
+__version__ = "0.4.9"
 
 __all__ = [
     "Auth",
     "DecryptResult",
     "Encryption",
     "EncryptionContext",
+    "Priority",
+    "PriorityContext",
+    "PrioritySettings",
     "get_client",
     "get_sync_client",
 ]

libs/sdk-py/langgraph_sdk/priority/__init__.py

@@ -0,0 +1,155 @@
+"""Custom run priority for LangGraph.
+
+.. warning::
+    This API is in beta and may change in future versions.
+
+This module lets developers decide the priority settings of each thread with
+an async function that the server calls at the first run on a thread with no
+settings. The function is registered on an object that `langgraph.json` names,
+and runs server-side.
+"""
+
+from __future__ import annotations
+
+import functools
+import inspect
+import typing
+import warnings
+from collections.abc import Awaitable, Callable
+
+from langgraph_sdk.encryption import DuplicateHandlerError, LangGraphBetaWarning
+from langgraph_sdk.priority import types
+
+_AssignHandler = Callable[
+    [types.PriorityContext], Awaitable[types.PrioritySettings | None]
+]
+_AssignHandlerT = typing.TypeVar("_AssignHandlerT", bound=_AssignHandler)
+
+
+@functools.lru_cache(maxsize=1)
+def _warn_priority_beta() -> None:
+    warnings.warn(
+        "The Priority API is in beta and may change in future versions.",
+        LangGraphBetaWarning,
+        stacklevel=4,
+    )
+
+
+def _validate_handler(fn: typing.Callable) -> None:
+    """Validate that the assign handler is async and takes one parameter.
+
+    Raises:
+        TypeError: If the handler is not an async function or has wrong parameter count
+    """
+    if not inspect.iscoroutinefunction(fn):
+        raise TypeError(
+            f"Priority assign handler must be an async function, got {type(fn)}"
+        )
+
+    sig = inspect.signature(fn)
+    params = [
+        p
+        for p in sig.parameters.values()
+        if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)
+    ]
+    if len(params) != 1:
+        raise TypeError(
+            "Priority assign handler must accept exactly 1 parameter (ctx), "
+            f"got {len(params)}"
+        )
+
+
+class Priority:
+    """Decide the priority settings of your LangGraph application's threads.
+
+    .. warning::
+        This API is in beta and may change in future versions.
+
+    To use, create a separate Python file and add the path to the file to your
+    LangGraph API configuration file (`langgraph.json`). Within that file, create
+    an instance of the Priority class and register an assign function with
+    `@priority.assign`.
+
+    Example `langgraph.json` file:
+
+    ```json
+    {
+      "dependencies": ["."],
+      "graphs": {
+        "agent": "./my_agent/agent.py:graph"
+      },
+      "env": ".env",
+      "priority": {
+        "path": "./priority.py:priority"
+      }
+    }
+    ```
+
+    The server calls the assign function at the first run on a thread with no
+    settings, which can come long after the thread was created, and stores the
+    settings it returns on the thread. Later runs on the thread use the stored
+    settings. The function can be called more than once for one thread, for
+    example after it raises, and only one result is stored, so keep it free of
+    side effects.
+
+    ???+ example "Basic Usage"
+
+        ```python
+        from langgraph_sdk import Priority, PriorityContext, PrioritySettings
+
+        priority = Priority()
+
+        @priority.assign
+        async def assign(ctx: PriorityContext) -> PrioritySettings | None:
+            return {
+                "priority": 3 if "premium" in ctx.permissions else 0,
+                # Stored unencrypted, so an opaque ID rather than personal data.
+                "fairness_group": ctx.user["org_id"],
+            }
+        ```
+    """
+
+    __slots__ = ("_assign_handler",)
+
+    types = types
+    """Reference to priority type definitions.
+
+    Provides access to the type definitions used by the priority system:
+    PriorityContext and PrioritySettings.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the Priority instance."""
+        _warn_priority_beta()
+        self._assign_handler: _AssignHandler | None = None
+
+    def assign(self, fn: _AssignHandlerT) -> _AssignHandlerT:
+        """Register the function that assigns a thread's priority settings.
+
+        The function receives a `PriorityContext` and returns `PrioritySettings`,
+        or `None` for the defaults.
+
+        Example:
+            ```python
+            @priority.assign
+            async def assign(ctx: PriorityContext) -> PrioritySettings | None:
+                if ctx.cron_id is not None:
+                    return {"priority": -2}
+                return None
+            ```
+
+        Args:
+            fn: The assign handler function
+
+        Returns:
+            The registered handler function
+
+        Raises:
+            DuplicateHandlerError: If an assign handler is already registered
+            TypeError: If the handler has an invalid signature
+        """
+        if self._assign_handler is not None:
+            raise DuplicateHandlerError("Priority assign handler already registered")
+        _validate_handler(fn)
+        self._assign_handler = fn
+        return fn

libs/sdk-py/langgraph_sdk/priority/types.py

@@ -0,0 +1,67 @@
+"""Priority types for LangGraph.
+
+This module defines the context the server passes to the function registered
+with `@priority.assign`, and the settings that function returns.
+"""
+
+from __future__ import annotations
+
+import typing
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from langgraph_sdk.auth.types import BaseUser
+
+
+@dataclass(frozen=True, slots=True)
+class PriorityContext:
+    """Context passed to the `@priority.assign` function at the first run on a thread with no settings."""
+
+    user: BaseUser
+    """The user from the deployment's auth, custom auth or LangSmith auth. Without auth, `user.identity` is empty."""
+
+    permissions: Sequence[str]
+    """The user's permissions from the same auth. Without auth, it's empty."""
+
+    thread_id: str | None
+    """The thread's ID, or `None` when the run creates its thread without being given an ID."""
+
+    thread_metadata: dict[str, typing.Any]
+    """The thread's metadata."""
+
+    assistant_id: str
+    """The ID of the assistant the run uses."""
+
+    graph_id: str
+    """The ID of the graph the assistant runs."""
+
+    assistant_metadata: dict[str, typing.Any]
+    """The assistant's metadata."""
+
+    run_metadata: dict[str, typing.Any]
+    """The run's metadata."""
+
+    cron_id: str | None
+    """The ID of the cron that started the run, or `None` when no cron started it."""
+
+
+class PrioritySettings(typing.TypedDict, total=False):
+    """Priority settings returned by the `@priority.assign` function.
+
+    Every key is optional: a missing key or a key set to `None` takes its
+    default, and returning `None` instead of a dict gives every key its default.
+    A value outside its range, or an empty `fairness_group`, is an error.
+    """
+
+    priority: int | None
+    """An integer from -5 to 5, where 5 runs first and -5 runs last. Defaults to 0."""
+
+    fairness_group: str | None
+    """The thread's fairness group, a non-empty string. Defaults to `"default"`.
+
+    The server stores it unencrypted, even when the deployment uses custom
+    encryption, so use an opaque ID rather than personal data.
+    """
+
+    fairness_weight: int | None
+    """The weight of the thread's fairness group, an integer from 1 to 100. Defaults to 1."""

libs/sdk-py/pyproject.toml

@@ -16,7 +16,7 @@ dependencies = [
     "orjson>=3.11.5",
     "langchain-protocol>=0.0.15",
     "langchain-core>=1.4.0,<2",
-    "websockets>=14,<17",
+    "websockets>=14",
 ]
 
 [tool.hatch.version]

libs/sdk-py/tests/test_priority.py

@@ -0,0 +1,128 @@
+import dataclasses
+
+import pytest
+
+import langgraph_sdk
+from langgraph_sdk import Priority, PriorityContext, PrioritySettings
+from langgraph_sdk.encryption import DuplicateHandlerError, LangGraphBetaWarning
+from langgraph_sdk.priority import _warn_priority_beta
+
+
+class _User:
+    is_authenticated = True
+    display_name = "user-1"
+    identity = "user-1"
+    permissions: tuple[str, ...] = ()
+
+    def __getitem__(self, key):
+        return getattr(self, key)
+
+    def __contains__(self, key):
+        return hasattr(self, key)
+
+    def __iter__(self):
+        return iter(("identity", "display_name", "is_authenticated", "permissions"))
+
+
+def _context(thread_id: str | None) -> PriorityContext:
+    return PriorityContext(
+        user=_User(),
+        permissions=[],
+        thread_id=thread_id,
+        thread_metadata={},
+        assistant_id="assistant-1",
+        graph_id="agent",
+        assistant_metadata={"tier": "gold"},
+        run_metadata={},
+        cron_id=None,
+    )
+
+
+async def _no_params() -> PrioritySettings | None:
+    return None
+
+
+async def _two_params(_ctx: PriorityContext, _extra: object) -> PrioritySettings | None:
+    return None
+
+
+def test_priority_and_its_types_are_exported():
+    assert {"Priority", "PriorityContext", "PrioritySettings"} <= set(
+        langgraph_sdk.__all__
+    )
+    assert Priority.types.PriorityContext is PriorityContext
+    assert Priority.types.PrioritySettings is PrioritySettings
+
+
+def test_construction_warns_that_priority_is_beta():
+    # The warning fires once per process, so an earlier test may have used it up.
+    _warn_priority_beta.cache_clear()
+    with pytest.warns(LangGraphBetaWarning, match="Priority API is in beta"):
+        Priority()
+
+
+def test_assign_registers_and_returns_the_function():
+    priority = Priority()
+
+    async def assign(_ctx: PriorityContext) -> PrioritySettings | None:
+        return {"priority": 3, "fairness_group": "org-42"}
+
+    assert priority.assign(assign) is assign
+    assert priority._assign_handler is assign
+
+
+def test_assign_rejects_a_sync_function():
+    priority = Priority()
+
+    def assign(_ctx: PriorityContext) -> PrioritySettings | None:
+        return None
+
+    with pytest.raises(TypeError, match="must be an async function"):
+        priority.assign(assign)  # ty: ignore[invalid-argument-type]
+    assert priority._assign_handler is None
+
+
+@pytest.mark.parametrize("fn", [_no_params, _two_params])
+def test_assign_rejects_the_wrong_number_of_parameters(fn):
+    priority = Priority()
+
+    with pytest.raises(TypeError, match="must accept exactly 1 parameter"):
+        priority.assign(fn)
+    assert priority._assign_handler is None
+
+
+def test_assign_rejects_a_second_function():
+    priority = Priority()
+
+    @priority.assign
+    async def first(_ctx: PriorityContext) -> PrioritySettings | None:
+        return None
+
+    with pytest.raises(DuplicateHandlerError):
+
+        @priority.assign
+        async def second(_ctx: PriorityContext) -> PrioritySettings | None:
+            return None
+
+    assert priority._assign_handler is first
+
+
+def test_context_accepts_no_thread_id():
+    # `make lint` type-checks this call too, so `thread_id` must accept None there.
+    assert _context(thread_id=None).thread_id is None
+
+
+def test_context_is_frozen():
+    ctx = _context(thread_id="thread-1")
+
+    with pytest.raises(dataclasses.FrozenInstanceError):
+        ctx.thread_id = "thread-2"  # ty: ignore[invalid-assignment]
+
+
+def test_every_settings_key_is_optional():
+    assert PrioritySettings.__optional_keys__ == {
+        "priority",
+        "fairness_group",
+        "fairness_weight",
+    }
+    assert PrioritySettings.__required_keys__ == frozenset()

libs/sdk-py/uv.lock

@@ -301,7 +301,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.16"
+version = "1.2.17"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -525,7 +525,7 @@ requires-dist = [
     { name = "langchain-core", specifier = ">=1.4.0,<2" },
     { name = "langchain-protocol", specifier = ">=0.0.15" },
     { name = "orjson", specifier = ">=3.11.5" },
-    { name = "websockets", specifier = ">=14,<17" },
+    { name = "websockets", specifier = ">=14" },
 ]
 
 [package.metadata.requires-dev]