done

exciting!

relnotes preview:

# LangGraph 1.1.0 Release Notes

## Type-Safe Streaming & Invoke

LangGraph 1.1 introduces `version="v2"` — a new opt-in streaming format
that brings full type safety to `stream()`, `astream()`, `invoke()`, and
`ainvoke()`.

### What's changing

**v1 (default, unchanged):** `stream()` yields bare tuples like
`(stream_mode, data)` or just `data`. `invoke()` returns a plain `dict`.
Interrupts are mixed into the output dict under `"__interrupt__"`.

**v2 (opt-in):** `stream()` yields strongly-typed `StreamPart` dicts
with `type`, `ns`, `data`, and (for values) `interrupts` fields.
`invoke()` returns a `GraphOutput` object with `.value` and
`.interrupts` attributes. When your state schema is a Pydantic model or
dataclass, outputs are automatically coerced to the correct type.

### `invoke()` / `ainvoke()` with `version="v2"`

```python
from langgraph.types import GraphOutput

result = graph.invoke({"input": "hello"}, version="v2")

# result is a GraphOutput, not a dict
assert isinstance(result, GraphOutput)
result.value       # your output — dict, Pydantic model, or dataclass
result.interrupts  # tuple[Interrupt, ...], empty if none occurred
```

With a non-`"values"` stream mode, `invoke(..., stream_mode="updates",
version="v2")` returns `list[StreamPart]` instead of `list[tuple]`.

### `stream()` / `astream()` with `version="v2"`

```python
for part in graph.stream({"input": "hello"}, version="v2"):
    if part["type"] == "values":
        part["data"]        # OutputT — full state
        part["interrupts"]  # tuple[Interrupt, ...]
    elif part["type"] == "updates":
        part["data"]        # dict[str, Any]
    elif part["type"] == "messages":
        part["data"]        # tuple[BaseMessage, dict]
    elif part["type"] == "custom":
        part["data"]        # Any
    elif part["type"] == "tasks":
        part["data"]        # TaskPayload | TaskResultPayload
    elif part["type"] == "debug":
        part["data"]        # DebugPayload
```

Each stream mode has its own `TypedDict` — `ValuesStreamPart`,
`UpdatesStreamPart`, `MessagesStreamPart`, `CustomStreamPart`,
`CheckpointStreamPart`, `TasksStreamPart`, `DebugStreamPart` — all
importable from `langgraph.types`. The union type `StreamPart` is a
discriminated union on `part["type"]`, enabling full type narrowing in
editors and type checkers.

### Pydantic & dataclass output coercion

When your graph's state schema is a Pydantic model or dataclass,
`version="v2"` automatically coerces outputs to the declared type:

```python
from pydantic import BaseModel

class MyState(BaseModel):
    answer: str
    count: int

graph = StateGraph(MyState)
# ... build graph ...
compiled = graph.compile()

result = compiled.invoke({"answer": "", "count": 0}, version="v2")
assert isinstance(result.value, MyState)  # not a dict!
```

### Backward compatibility

- **Default is still `version="v1"`** — existing code works without
changes.
- To make migration easier, `GraphOutput` supports old-style best-effort
access to graph values and interrupts. Dict-style access
(`result["key"]`, `"key" in result`, `result["__interrupt__"]`) still
works and delegates to `result.value` / `result.interrupts` under the
hood. However, this is **deprecated** and emits a
`LangGraphDeprecatedSinceV11` warning. It will be removed in v3.0 —
migrate to `result.value` and `result.interrupts` at your convenience.

```python
result = graph.invoke({"input": "hello"}, version="v2")

# Old style — still works, but deprecated
result["input"]          # delegates to result.value["input"]
result["__interrupt__"]  # delegates to result.interrupts
"input" in result        # delegates to "input" in result.value

# New style — preferred
result.value["input"]
result.interrupts
```

## Migration Guide

1. **No action required** — `version="v1"` remains the default. All
existing code continues to work.
2. **Adopt v2 incrementally** — Add `version="v2"` to individual
`invoke()`/`stream()` calls to get typed outputs.
3. **Use typed imports** — Import `GraphOutput`, `StreamPart`, and
individual part types from `langgraph.types` for type-safe code.

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.14"
+__version__ = "0.4.15"

libs/cli/langgraph_cli/cli.py

@@ -1,14 +1,23 @@
 """CLI entrypoint for LangGraph API server."""
 
+import base64
+import copy
+import json as json_mod
 import os
 import pathlib
+import platform
+import re
 import shutil
 import sys
+import tempfile
+import time
 from collections.abc import Callable, Sequence
+from contextlib import contextmanager
 
 import click
 import click.exceptions
 from click import secho
+from dotenv import dotenv_values
 
 import langgraph_cli.config
 import langgraph_cli.docker
@@ -17,11 +26,131 @@ from langgraph_cli.config import Config
 from langgraph_cli.constants import DEFAULT_CONFIG, DEFAULT_PORT
 from langgraph_cli.docker import DockerCapabilities
 from langgraph_cli.exec import Runner, subp_exec
+from langgraph_cli.host_backend import HostBackendClient, HostBackendError
 from langgraph_cli.progress import Progress
 from langgraph_cli.templates import TEMPLATE_HELP_STRING, create_new
 from langgraph_cli.util import warn_non_wolfi_distro
 from langgraph_cli.version import __version__
 
+RESERVED_ENV_VARS = frozenset(
+    [
+        # LANGCHAIN_RESERVED_ENV_VARS from host-backend
+        "LANGCHAIN_TRACING_V2",
+        "LANGSMITH_TRACING_V2",
+        "LANGCHAIN_ENDPOINT",
+        "LANGCHAIN_PROJECT",
+        "LANGSMITH_PROJECT",
+        "LANGSMITH_LANGGRAPH_GIT_REPO",
+        "LANGGRAPH_GIT_REPO_PATH",
+        "LANGCHAIN_API_KEY",
+        "LANGSMITH_CONTROL_PLANE_API_KEY",
+        "POSTGRES_URI",
+        "POSTGRES_PASSWORD",
+        "DATABASE_URI",
+        "LANGSMITH_LANGGRAPH_GIT_REF",
+        "LANGSMITH_LANGGRAPH_GIT_REF_SHA",
+        "LANGGRAPH_AUTH_TYPE",
+        "LANGSMITH_AUTH_ENDPOINT",
+        "LANGSMITH_TENANT_ID",
+        "LANGSMITH_AUTH_VERIFY_TENANT_ID",
+        "LANGSMITH_HOST_PROJECT_ID",
+        "LANGSMITH_HOST_PROJECT_NAME",
+        "LANGSMITH_HOST_REVISION_ID",
+        "LOG_JSON",
+        "LOG_DICT_TRACEBACKS",
+        "REDIS_URI",
+        "LANGCHAIN_CALLBACKS_BACKGROUND",
+        "DD_TRACE_PSYCOPG_ENABLED",
+        "DD_TRACE_REDIS_ENABLED",
+        "LANGSMITH_DEPLOYMENT_NAME",
+        "LANGGRAPH_CLOUD_LICENSE_KEY",
+        # ALLOWED_SELF_HOSTED_ENV_VARS (rejected for non-self-hosted)
+        "LANGSMITH_API_KEY",
+        "LANGSMITH_ENDPOINT",
+        "POSTGRES_URI_CUSTOM",
+        "REDIS_URI_CUSTOM",
+        "PATH",
+        "PORT",
+        "MOUNT_PREFIX",
+        "LSD_ENV",
+        "LSD_DD_API_KEY",
+        "LSD_DD_ENDPOINT",
+        "LSD_DEPLOYMENT_TYPE",
+    ]
+)
+
+_API_KEY_ENV_NAMES = (
+    "LANGGRAPH_HOST_API_KEY",
+    "LANGSMITH_API_KEY",
+    "LANGCHAIN_API_KEY",
+)
+
+_DEPLOYMENT_NAME_ENV = "LANGSMITH_DEPLOYMENT_NAME"
+
+
+def _parse_env_from_config(
+    config_json: dict, config_path: pathlib.Path
+) -> dict[str, str]:
+    """Resolve env vars from langgraph.json 'env' field or a .env fallback."""
+    env_field = config_json.get("env")
+    # validate_config_file will default env to {}
+    if isinstance(env_field, dict) and env_field:
+        return {str(k): str(v) for k, v in env_field.items()}
+    if isinstance(env_field, str):
+        env_path = (config_path.parent / env_field).resolve()
+        if not env_path.exists():
+            click.secho(
+                f"Warning: env file '{env_field}' specified in langgraph.json not found.",
+                fg="yellow",
+            )
+            return {}
+    else:
+        env_path = pathlib.Path.cwd() / ".env"
+    return {k: v for k, v in dotenv_values(env_path).items() if v is not None}
+
+
+def _secrets_from_env(
+    env_vars: dict[str, str],
+) -> list[dict[str, str]]:
+    """Convert env dict to secrets list, filtering reserved vars with warnings."""
+    secrets: list[dict[str, str]] = []
+    for name, value in env_vars.items():
+        if name in RESERVED_ENV_VARS:
+            click.secho(f"   Skipping reserved env var: {name}", fg="yellow")
+            continue
+        if not value:
+            continue
+        secrets.append({"name": name, "value": value})
+    return secrets
+
+
+_TERMINAL_STATUSES = frozenset(
+    [
+        "DEPLOYED",
+        "CREATE_FAILED",
+        "BUILD_FAILED",
+        "DEPLOY_FAILED",
+        "SKIPPED",
+    ]
+)
+
+
+@contextmanager
+def _docker_config_for_token(registry_host: str, token: str):
+    """Create a temporary Docker config with only the push token.
+
+    Yields the path to a temporary config directory that can be passed
+    to ``docker --config <path>`` so that system credential helpers
+    (e.g. gcloud) don't interfere with the push token.
+    """
+    auth_b64 = base64.b64encode(f"oauth2accesstoken:{token}".encode()).decode()
+    config_data = {"auths": {registry_host: {"auth": auth_b64}}}
+    with tempfile.TemporaryDirectory() as tmpdir:
+        with open(os.path.join(tmpdir, "config.json"), "w") as f:
+            json_mod.dump(config_data, f)
+        yield tmpdir
+
+
 OPT_DOCKER_COMPOSE = click.option(
     "--docker-compose",
     "-d",
@@ -158,6 +287,13 @@ OPT_API_VERSION = click.option(
     help="API server version to use for the base image. If unspecified, the latest version will be used.",
 )
 
+OPT_ENGINE_RUNTIME_MODE = click.option(
+    "--engine-runtime-mode",
+    type=click.Choice(["combined_queue_worker", "distributed"]),
+    default="combined_queue_worker",
+    help="Runtime mode. 'distributed' uses separate executor and orchestrator containers.",
+)
+
 
 @click.group()
 @click.version_option(version=__version__, prog_name="LangGraph CLI")
@@ -176,6 +312,7 @@ def cli():
 @OPT_WATCH
 @OPT_POSTGRES_URI
 @OPT_API_VERSION
+@OPT_ENGINE_RUNTIME_MODE
 @click.option(
     "--image",
     type=str,
@@ -210,6 +347,7 @@ def up(
     debugger_base_url: str | None,
     postgres_uri: str | None,
     api_version: str | None,
+    engine_runtime_mode: str,
     image: str | None,
     base_image: str | None,
 ):
@@ -233,6 +371,7 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE
             debugger_base_url=debugger_base_url,
             postgres_uri=postgres_uri,
             api_version=api_version,
+            engine_runtime_mode=engine_runtime_mode,
             image=image,
             base_image=base_image,
         )
@@ -304,6 +443,9 @@ def _build(
     passthrough: Sequence[str] = (),
     install_command: str | None = None,
     build_command: str | None = None,
+    docker_command: Sequence[str] | None = None,
+    extra_flags: Sequence[str] = (),
+    verbose: bool = True,
 ):
     # pull latest images
     if pull:
@@ -312,7 +454,7 @@ def _build(
                 "docker",
                 "pull",
                 langgraph_cli.config.docker_tag(config_json, base_image, api_version),
-                verbose=True,
+                verbose=verbose,
             )
         )
     set("Building...")
@@ -334,7 +476,9 @@ def _build(
     else:
         build_context = str(config.parent)
 
-    # apply config
+    # Deep copy to avoid mutating the caller's config (config_to_docker
+    # rewrites graph paths to container-internal paths in place).
+    config_json = copy.deepcopy(config_json)
     stdin, additional_contexts = langgraph_cli.config.config_to_docker(
         config_path=config,
         config=config_json,
@@ -348,15 +492,16 @@ def _build(
     if additional_contexts:
         for k, v in additional_contexts.items():
             args.extend(["--build-context", f"{k}={v}"])
+    cmd = tuple(docker_command) if docker_command else ("docker", "build")
     runner.run(
         subp_exec(
-            "docker",
-            "build",
+            *cmd,
             *args,
+            *extra_flags,
             *passthrough,
             build_context,
             input=stdin,
-            verbose=True,
+            verbose=verbose,
         )
     )
 
@@ -383,6 +528,7 @@ def _build(
     "\n    --base-image langchain/langgraph-server:0.2  # Pin to a minor version (Python)",
 )
 @OPT_API_VERSION
+@OPT_ENGINE_RUNTIME_MODE
 @click.option(
     "--install-command",
     help="Custom install command to run from the build context root. If not provided, auto-detects based on package manager files.",
@@ -404,6 +550,7 @@ def build(
     docker_build_args: Sequence[str],
     base_image: str | None,
     api_version: str | None,
+    engine_runtime_mode: str,
     pull: bool,
     tag: str,
     install_command: str | None,
@@ -426,12 +573,17 @@ def build(
             raise click.UsageError("Docker not installed") from None
         config_json = langgraph_cli.config.validate_config_file(config)
         warn_non_wolfi_distro(config_json)
+        effective_base_image = base_image
+        if engine_runtime_mode == "distributed" and not base_image:
+            effective_base_image = langgraph_cli.config.default_base_image(
+                config_json, engine_runtime_mode=engine_runtime_mode
+            )
         _build(
             runner,
             set,
             config,
             config_json,
-            base_image,
+            effective_base_image,
             api_version,
             pull,
             tag,
@@ -441,6 +593,489 @@ def build(
         )
 
 
+@click.option(
+    "--api-key",
+    envvar="LANGGRAPH_HOST_API_KEY",
+    help=(
+        "API key. Can also be set via LANGGRAPH_HOST_API_KEY, "
+        "LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable or .env file."
+    ),
+)
+@click.option(
+    "--name",
+    envvar="LANGSMITH_DEPLOYMENT_NAME",
+    help=(
+        "Deployment name. Can also be set via LANGSMITH_DEPLOYMENT_NAME "
+        "environment variable or .env file. Defaults to current directory name "
+        "if --deployment-id is not provided."
+    ),
+)
+@click.option(
+    "--deployment-id",
+    help=(
+        "ID of an existing deployment to update. If omitted, "
+        "--name is used to find or create the deployment."
+    ),
+)
+@click.option(
+    "--deployment-type",
+    type=click.Choice(["dev", "prod"]),
+    default="dev",
+    show_default=True,
+    help="Deployment type (used when creating a new deployment).",
+)
+@click.option(
+    "--no-wait",
+    is_flag=True,
+    default=False,
+    help="Skip waiting for deployment status.",
+)
+@OPT_VERBOSE
+@click.option(
+    "--host-url",
+    envvar="LANGGRAPH_HOST_URL",
+    default="https://api.host.langchain.com",
+    hidden=True,
+)
+@click.option("--image-name", hidden=True)
+@click.option("--image-tag", default="latest", hidden=True)
+@click.option(
+    "--config",
+    "-c",
+    default=DEFAULT_CONFIG,
+    hidden=True,
+    type=click.Path(
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        resolve_path=True,
+        path_type=pathlib.Path,
+    ),
+)
+@click.option("--pull/--no-pull", default=True, hidden=True)
+@click.option("--base-image", hidden=True)
+@click.option("--install-command", hidden=True)
+@click.option("--build-command", hidden=True)
+@click.option("--api-version", type=str, hidden=True)
+@click.argument("docker_build_args", nargs=-1, type=click.UNPROCESSED)
+@cli.command(
+    help=(
+        "[Beta] Build and deploy a LangGraph image to LangSmith Deployments.\n\n"
+        "This command is in beta and under active development. "
+        "Expect frequent updates and improvements.\n\n"
+        "Run from the root of your LangGraph project (where langgraph.json "
+        "is located). This command also accepts build flags (--base-image, "
+        "--pull, etc.). See 'langgraph build --help' for details."
+    ),
+    context_settings=dict(ignore_unknown_options=True),
+)
+@log_command
+def deploy(
+    config: pathlib.Path,
+    pull: bool,
+    verbose: bool,
+    api_version: str | None,
+    host_url: str | None,
+    api_key: str | None,
+    deployment_id: str | None,
+    deployment_type: str,
+    name: str | None,
+    image_name: str | None,
+    image_tag: str,
+    base_image: str | None,
+    install_command: str | None,
+    build_command: str | None,
+    no_wait: bool,
+    docker_build_args: Sequence[str],
+):
+    click.secho(
+        "Note: 'langgraph deploy' is in beta. Expect frequent updates and improvements.",
+        fg="yellow",
+    )
+    click.echo()
+    config_json = langgraph_cli.config.validate_config_file(config)
+    warn_non_wolfi_distro(config_json)
+
+    env_vars = _parse_env_from_config(config_json, config)
+
+    if not api_key:
+        for key_name in _API_KEY_ENV_NAMES:
+            val = env_vars.get(key_name) or os.environ.get(key_name)
+            if val:
+                api_key = val
+                break
+    if not api_key:
+        api_key = click.prompt("Host API key", hide_input=True)
+
+    if not deployment_id and not name:
+        name = env_vars.get(_DEPLOYMENT_NAME_ENV)
+    if not deployment_id and not name:
+        default_name = _normalize_image_name(pathlib.Path.cwd().name)
+        name = click.prompt("Deployment name", default=default_name)
+
+    secrets = _secrets_from_env(env_vars)
+
+    # Use buildx to cross-compile for amd64 when running on a non-x86_64 host
+    # (e.g. Apple Silicon). On amd64 hosts, plain docker build is sufficient.
+    needs_buildx = platform.machine() != "x86_64"
+    local_tag = f"langgraph-deploy-tmp:{int(time.time())}"
+
+    with Runner() as runner:
+        if shutil.which("docker") is None:
+            raise click.UsageError(
+                "Docker is required but not installed.\n"
+                "Install Docker Desktop: https://docs.docker.com/get-docker/\n\n"
+                "Remote builds (no Docker required) are coming in a future update."
+            )
+        if needs_buildx:
+            try:
+                runner.run(subp_exec("docker", "buildx", "version", collect=True))
+            except click.exceptions.Exit:
+                raise click.UsageError(
+                    "Docker Buildx is required but not installed.\n"
+                    "Your machine architecture ("
+                    + platform.machine()
+                    + ") requires Buildx to cross-compile images for linux/amd64.\n"
+                    "Install Buildx: https://docs.docker.com/build/install-buildx/\n\n"
+                    "Remote builds (no Docker required) are coming in a future update."
+                ) from None
+
+        def log_step(message: str) -> None:
+            click.secho(message, fg="cyan")
+
+        client = HostBackendClient(host_url, api_key)
+        step = 1
+        needs_creation = False
+
+        if deployment_id:
+            log_step(f"{step}. Using deployment {deployment_id}")
+            try:
+                client.get_deployment(deployment_id)
+            except HostBackendError as err:
+                if (
+                    err.status_code == 403
+                    and "requires workspace specification" in err.message
+                ):
+                    click.secho(
+                        "Your API key is org-scoped and requires a workspace ID.",
+                        fg="yellow",
+                    )
+                    click.secho(
+                        "Find your workspace ID in LangSmith under Settings > Workspaces.",
+                        fg="yellow",
+                    )
+                    tenant_id = click.prompt("Workspace ID")
+                    client = HostBackendClient(host_url, api_key, tenant_id=tenant_id)
+                    client.get_deployment(deployment_id)
+                else:
+                    raise
+            step += 1
+        else:
+            log_step(f"{step}. Looking up deployment '{name}'")
+            try:
+                existing = client.list_deployments(name_contains=name)
+            except HostBackendError as err:
+                if (
+                    err.status_code == 403
+                    and "requires workspace specification" in err.message
+                ):
+                    click.secho(
+                        "Your API key is org-scoped and requires a workspace ID.",
+                        fg="yellow",
+                    )
+                    click.secho(
+                        "Find your workspace ID in LangSmith under Settings > Workspaces.",
+                        fg="yellow",
+                    )
+                    tenant_id = click.prompt("Workspace ID")
+                    client = HostBackendClient(host_url, api_key, tenant_id=tenant_id)
+                    existing = client.list_deployments(name_contains=name)
+                else:
+                    raise
+            found_id = None
+            if isinstance(existing, dict):
+                for dep in existing.get("resources", []):
+                    if isinstance(dep, dict) and dep.get("name") == name:
+                        found_id = dep.get("id")
+                        break
+            if found_id:
+                deployment_id = str(found_id)
+                click.secho(
+                    f"   Found existing deployment (ID: {deployment_id})",
+                    fg="green",
+                )
+            else:
+                needs_creation = True
+                click.secho(
+                    "   No deployment found. Will create after build.", fg="yellow"
+                )
+            step += 1
+
+        # -- Step: Build image --
+        log_step(f"{step}. Building image")
+        if needs_buildx:
+            build_flags: list[str] = [
+                "--platform",
+                "linux/amd64",
+                "--load",
+            ]
+            if not verbose:
+                build_flags.append("--progress=quiet")
+            with Progress(message="Building...", elapsed=not verbose):
+                _build(
+                    runner,
+                    lambda _msg: None,
+                    config,
+                    config_json,
+                    base_image,
+                    api_version,
+                    pull,
+                    local_tag,
+                    docker_build_args,
+                    install_command,
+                    build_command,
+                    docker_command=("docker", "buildx", "build"),
+                    extra_flags=build_flags,
+                    verbose=verbose,
+                )
+        else:
+            with Progress(message="Building...", elapsed=not verbose):
+                _build(
+                    runner,
+                    lambda _msg: None,
+                    config,
+                    config_json,
+                    base_image,
+                    api_version,
+                    pull,
+                    local_tag,
+                    docker_build_args,
+                    install_command,
+                    build_command,
+                    verbose=verbose,
+                )
+        step += 1
+
+        if needs_creation:
+            log_step(f"{step}. Creating deployment '{name}'")
+            payload = {
+                "name": name,
+                "source": "internal_docker",
+                "source_config": {"deployment_type": deployment_type},
+                "source_revision_config": {},
+                "secrets": secrets,
+            }
+            created = client.create_deployment(payload)
+            created_id = created.get("id") if isinstance(created, dict) else None
+            if not isinstance(created_id, str) or not created_id:
+                raise HostBackendError(
+                    "POST /v2/deployments succeeded but response missing a valid 'id'"
+                )
+            deployment_id = created_id
+            click.secho(f"   Deployment ID: {deployment_id}", fg="green")
+            step += 1
+
+        # -- Step: Get push token and authenticate --
+        log_step(f"{step}. Requesting push token")
+        try:
+            push_data = client.request_push_token(deployment_id)
+        except HostBackendError as err:
+            if (
+                err.status_code == 400
+                and "only available for 'internal_docker' source deployments"
+                in err.message
+            ):
+                raise click.ClickException(
+                    f"Deployment '{deployment_id}' was not created by 'langgraph deploy' "
+                    "and cannot be updated with this command.\n"
+                    "Please create a new deployment by running 'langgraph deploy' "
+                    "without --deployment-id, or use a different --name."
+                ) from None
+            raise
+        deployment_token = push_data.get("token")
+        registry_url = push_data.get("registry_url")
+        if not deployment_token or not registry_url:
+            raise click.ClickException(
+                "Push token response missing token or registry_url"
+            )
+        step += 1
+
+        normalized_registry = registry_url.rstrip("/")
+        if "://" in normalized_registry:
+            normalized_registry = normalized_registry.split("//", 1)[1]
+        repo_seed = image_name or name or config.parent.name
+        repo_name = _normalize_image_name(repo_seed)
+        tag_value = _normalize_image_tag(image_tag)
+        remote_image = f"{normalized_registry}/{repo_name}:{tag_value}"
+
+        registry_host = normalized_registry.split("/")[0]
+
+        # Use a clean Docker config with only the push token so that
+        # system credential helpers (e.g. gcloud) don't interfere.
+        with _docker_config_for_token(registry_host, deployment_token) as cfg:
+            log_step(f"{step}. Logging into {registry_host}")
+            token_input = (
+                deployment_token
+                if deployment_token.endswith("\n")
+                else f"{deployment_token}\n"
+            )
+            runner.run(
+                subp_exec(
+                    "docker",
+                    "--config",
+                    cfg,
+                    "login",
+                    "-u",
+                    "oauth2accesstoken",
+                    "--password-stdin",
+                    registry_host,
+                    input=token_input,
+                    verbose=verbose,
+                )
+            )
+            step += 1
+
+            # -- Step: Tag and push --
+            log_step(f"{step}. Pushing image {remote_image}")
+            runner.run(
+                subp_exec(
+                    "docker",
+                    "tag",
+                    local_tag,
+                    remote_image,
+                    verbose=verbose,
+                )
+            )
+            max_push_retries = 3
+            for attempt in range(max_push_retries):
+                try:
+                    with Progress(message="Pushing...", elapsed=not verbose):
+                        runner.run(
+                            subp_exec(
+                                "docker",
+                                "--config",
+                                cfg,
+                                "push",
+                                remote_image,
+                                verbose=verbose,
+                            )
+                        )
+                    break
+                except click.exceptions.Exit:
+                    if attempt < max_push_retries - 1:
+                        click.secho(
+                            f"   Push failed, retrying (attempt {attempt + 2} of {max_push_retries})...",
+                            fg="yellow",
+                        )
+                    else:
+                        raise
+        step += 1
+
+        # -- Step: Update deployment --
+        log_step(f"{step}. Updating deployment {deployment_id}")
+        updated = client.update_deployment(deployment_id, remote_image, secrets=secrets)
+        tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None
+        if tenant_id:
+            status_url = (
+                f"https://smith.langchain.com/o/{tenant_id}"
+                f"/host/deployments/{deployment_id}"
+            )
+            click.secho(f"   View status: {status_url}", fg="cyan")
+
+        if no_wait:
+            click.secho("   Deployment updated", fg="green")
+            return
+
+        # -- Poll revision status --
+        revisions_resp = client.list_revisions(deployment_id, limit=1)
+        resources = (
+            revisions_resp.get("resources", [])
+            if isinstance(revisions_resp, dict)
+            else []
+        )
+        if not resources:
+            click.secho("   Deployment updated", fg="green")
+            return
+
+        revision_id = str(resources[0]["id"])
+        last_status = ""
+
+        deadline = time.time() + 300
+        with Progress(message="Deploying...", elapsed=True) as set_progress:
+            while time.time() < deadline:
+                rev = client.get_revision(deployment_id, revision_id)
+                status = (
+                    rev.get("status", "UNKNOWN") if isinstance(rev, dict) else "UNKNOWN"
+                )
+                if status != last_status:
+                    last_status = status
+                    # pause spinner so we can avoid conflict when writing status
+                    set_progress("")
+                    click.secho(f"   Status: {status}", fg="cyan")
+                    if status in _TERMINAL_STATUSES:
+                        break
+                    set_progress(f"{status}...")
+                time.sleep(1)
+            else:
+                set_progress("")
+
+        dep_info = client.get_deployment(deployment_id)
+        custom_url = None
+        if isinstance(dep_info, dict):
+            sc = dep_info.get("source_config")
+            if isinstance(sc, dict):
+                custom_url = sc.get("custom_url")
+
+        if last_status == "DEPLOYED":
+            click.secho("   Deployment successful!", fg="green")
+            if custom_url:
+                click.secho(f"   URL: {custom_url}", fg="green")
+        elif last_status in ("BUILD_FAILED", "DEPLOY_FAILED", "CREATE_FAILED"):
+            click.secho(f"   Deployment failed: {last_status}", fg="red")
+            raise click.exceptions.Exit(1)
+        else:
+            click.secho(
+                f"   Timed out waiting for deployment (last status: {last_status}).",
+                fg="yellow",
+            )
+            if custom_url:
+                click.secho(
+                    f"   Check status at: {custom_url}",
+                    fg="yellow",
+                )
+            else:
+                click.secho(
+                    "   Check status in the LangSmith Deployments dashboard.",
+                    fg="yellow",
+                )
+
+
+def _normalize_image_name(value: str | None) -> str:
+    """Sanitize a deployment/directory name into a valid Docker repository name.
+
+    Docker repository names must be lowercase and may only contain
+    [a-z0-9._-].  Invalid characters are replaced with hyphens.
+    """
+    if not value:
+        return "app"
+    slug = re.sub(r"[^a-z0-9._-]+", "-", value.lower()).strip("-.")
+    return slug or "app"
+
+
+def _normalize_image_tag(value: str) -> str:
+    """Validate and return a Docker image tag.
+
+    Tags may only contain [A-Za-z0-9_.-].  Defaults to "latest" when empty.
+    """
+    if not value:
+        value = "latest"
+    if not re.fullmatch(r"[A-Za-z0-9_.-]+", value):
+        raise click.UsageError(
+            "Image tag may only contain characters A-Z, a-z, 0-9, '_', '-', '.'"
+        )
+    return value
+
+
 def _get_docker_ignore_content() -> str:
     """Return the content of a .dockerignore file.
 
@@ -518,6 +1153,7 @@ tests
     "\n    --base-image langchain/langgraph-server:0.2  # Pin to a minor version (Python)",
 )
 @OPT_API_VERSION
+@OPT_ENGINE_RUNTIME_MODE
 @log_command
 def dockerfile(
     save_path: str,
@@ -525,6 +1161,7 @@ def dockerfile(
     add_docker_compose: bool,
     base_image: str | None = None,
     api_version: str | None = None,
+    engine_runtime_mode: str = "combined_queue_worker",
 ) -> None:
     save_path = pathlib.Path(save_path).absolute()
     secho(f"🔍 Validating configuration at path: {config}", fg="yellow")
@@ -532,11 +1169,17 @@ def dockerfile(
     warn_non_wolfi_distro(config_json)
     secho("✅ Configuration validated!", fg="green")
 
+    effective_base_image = base_image
+    if engine_runtime_mode == "distributed" and not base_image:
+        effective_base_image = langgraph_cli.config.default_base_image(
+            config_json, engine_runtime_mode=engine_runtime_mode
+        )
+
     secho(f"📝 Generating Dockerfile at {save_path}", fg="yellow")
     dockerfile, additional_contexts = langgraph_cli.config.config_to_docker(
         config_path=config,
         config=config_json,
-        base_image=base_image,
+        base_image=effective_base_image,
         api_version=api_version,
     )
     with open(str(save_path), "w", encoding="utf-8") as f:
@@ -807,6 +1450,7 @@ def prepare_args_and_stdin(
     debugger_base_url: str | None = None,
     postgres_uri: str | None = None,
     api_version: str | None = None,
+    engine_runtime_mode: str = "combined_queue_worker",
     # Like "my-tag" (if you already built it locally)
     image: str | None = None,
     # Like "langchain/langgraphjs-api" or "langchain/langgraph-api
@@ -820,9 +1464,10 @@ def prepare_args_and_stdin(
         debugger_port=debugger_port,
         debugger_base_url=debugger_base_url,
         postgres_uri=postgres_uri,
-        image=image,  # Pass image to compose YAML generator
+        image=image,
         base_image=base_image,
         api_version=api_version,
+        engine_runtime_mode=engine_runtime_mode,
     )
     args = [
         "--project-directory",
@@ -840,6 +1485,7 @@ def prepare_args_and_stdin(
         base_image=langgraph_cli.config.default_base_image(config),
         api_version=api_version,
         image=image,
+        engine_runtime_mode=engine_runtime_mode,
     )
     return args, stdin
 
@@ -858,6 +1504,7 @@ def prepare(
     debugger_base_url: str | None = None,
     postgres_uri: str | None = None,
     api_version: str | None = None,
+    engine_runtime_mode: str = "combined_queue_worker",
     image: str | None = None,
     base_image: str | None = None,
 ) -> tuple[list[str], str]:
@@ -874,6 +1521,20 @@ def prepare(
                 verbose=verbose,
             )
         )
+        if engine_runtime_mode == "distributed":
+            executor_base = langgraph_cli.config.default_base_image(
+                config_json, engine_runtime_mode="distributed"
+            )
+            runner.run(
+                subp_exec(
+                    "docker",
+                    "pull",
+                    langgraph_cli.config.docker_tag(
+                        config_json, executor_base, api_version
+                    ),
+                    verbose=verbose,
+                )
+            )
 
     args, stdin = prepare_args_and_stdin(
         capabilities=capabilities,
@@ -886,6 +1547,7 @@ def prepare(
         debugger_base_url=debugger_base_url or f"http://127.0.0.1:{port}",
         postgres_uri=postgres_uri,
         api_version=api_version,
+        engine_runtime_mode=engine_runtime_mode,
         image=image,
         base_image=base_image,
     )

libs/cli/langgraph_cli/config.py

@@ -1,3 +1,4 @@
+import copy
 import json
 import os
 import pathlib
@@ -1232,11 +1233,15 @@ def node_config_to_docker(
     return os.linesep.join(docker_file_contents), {}
 
 
-def default_base_image(config: Config) -> str:
+def default_base_image(
+    config: Config, engine_runtime_mode: str = "combined_queue_worker"
+) -> str:
     if config.get("base_image"):
         return config["base_image"]
     if config.get("node_version") and not config.get("python_version"):
         return "langchain/langgraphjs-api"
+    if engine_runtime_mode == "distributed":
+        return "langchain/langgraph-executor"
     return "langchain/langgraph-api"
 
 
@@ -1329,6 +1334,7 @@ def config_to_compose(
     api_version: str | None = None,
     image: str | None = None,
     watch: bool = False,
+    engine_runtime_mode: str = "combined_queue_worker",
 ) -> str:
     base_image = base_image or default_base_image(config)
 
@@ -1362,6 +1368,11 @@ def config_to_compose(
 """
 
     else:
+        # Save a pristine copy before config_to_docker mutates graph paths
+        config_snapshot = (
+            copy.deepcopy(config) if engine_runtime_mode == "distributed" else None
+        )
+
         dockerfile, additional_contexts = config_to_docker(
             config_path=config_path,
             config=config,
@@ -1379,7 +1390,7 @@ def config_to_compose(
             additional_contexts:
 {additional_contexts_str}"""
 
-        return f"""
+        result = f"""
 {textwrap.indent(env_vars_str, "            ")}
         {env_file_str}
         pull_policy: build
@@ -1389,3 +1400,60 @@ def config_to_compose(
 {textwrap.indent(dockerfile, "                ")}
         {watch_str}
 """
+
+        if engine_runtime_mode == "distributed":
+            executor_base_image = default_base_image(
+                config_snapshot, engine_runtime_mode="distributed"
+            )
+            executor_dockerfile, executor_additional_contexts = config_to_docker(
+                config_path=config_path,
+                config=config_snapshot,
+                base_image=executor_base_image,
+                api_version=api_version,
+                escape_variables=True,
+            )
+
+            executor_additional_contexts_str = "\n".join(
+                f"                    - {name}: {path}"
+                for name, path in executor_additional_contexts.items()
+            )
+            if executor_additional_contexts_str:
+                executor_additional_contexts_str = f"""
+                additional_contexts:
+{executor_additional_contexts_str}"""
+
+            postgres_uri = "postgres://postgres:postgres@langgraph-postgres:5432/postgres?sslmode=disable"
+            result += f"""    langgraph-orchestrator:
+        image: langchain/langgraph-orchestrator-licensed:latest
+        depends_on:
+            langgraph-api:
+                condition: service_healthy
+            langgraph-postgres:
+                condition: service_healthy
+        environment:
+            DATABASE_URI: {postgres_uri}
+            EXECUTOR_TARGET: langgraph-executor:8188
+        {env_file_str}
+    langgraph-executor:
+        depends_on:
+            langgraph-postgres:
+                condition: service_healthy
+            langgraph-api:
+                condition: service_healthy
+        entrypoint: ["sh", "/storage/executor_entrypoint.sh"]
+        environment:
+            DATABASE_URI: {postgres_uri}
+            REDIS_URI: redis://langgraph-redis:6379
+            EXECUTOR_GRPC_PORT: "8188"
+            ENGINE_GRPC_ADDRESS: "langgraph-orchestrator:50054"
+            LSD_GRPC_SERVER_ADDRESS: "localhost:50050"
+            LANGGRAPH_HTTP: ""
+        {env_file_str}
+        pull_policy: build
+        build:
+            context: .{executor_additional_contexts_str}
+            dockerfile_inline: |
+{textwrap.indent(executor_dockerfile, "                ")}
+"""
+
+        return result

libs/cli/langgraph_cli/docker.py

@@ -149,6 +149,7 @@ def compose_as_dict(
     base_image: str | None = None,
     # API version of the base image
     api_version: str | None = None,
+    engine_runtime_mode: str = "combined_queue_worker",
 ) -> dict:
     """Create a docker compose file as a dictionary in YML style."""
     if postgres_uri is None:
@@ -207,15 +208,19 @@ def compose_as_dict(
         )["langgraph-debugger"]
 
     # Add langgraph-api service
+    api_environment = {
+        "REDIS_URI": "redis://langgraph-redis:6379",
+        "POSTGRES_URI": postgres_uri,
+    }
+    if engine_runtime_mode == "distributed":
+        api_environment["N_JOBS_PER_WORKER"] = '"0"'
+
     services["langgraph-api"] = {
         "ports": [f'"{port}:8000"'],
         "depends_on": {
             "langgraph-redis": {"condition": "service_healthy"},
         },
-        "environment": {
-            "REDIS_URI": "redis://langgraph-redis:6379",
-            "POSTGRES_URI": postgres_uri,
-        },
+        "environment": api_environment,
     }
     if image:
         services["langgraph-api"]["image"] = image
@@ -255,6 +260,7 @@ def compose(
     image: str | None = None,
     base_image: str | None = None,
     api_version: str | None = None,
+    engine_runtime_mode: str = "combined_queue_worker",
 ) -> str:
     """Create a docker compose file as a string."""
     compose_content = compose_as_dict(
@@ -266,6 +272,7 @@ def compose(
         image=image,
         base_image=base_image,
         api_version=api_version,
+        engine_runtime_mode=engine_runtime_mode,
     )
     compose_str = dict_to_yaml(compose_content)
     return compose_str

libs/cli/langgraph_cli/host_backend.py

@@ -0,0 +1,107 @@
+"""HTTP client for LangGraph host backend deployments."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import click
+import httpx
+
+
+class HostBackendError(click.ClickException):
+    """Raised when the host backend returns an error response."""
+
+    def __init__(self, message: str, status_code: int | None = None):
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class HostBackendClient:
+    """Minimal JSON HTTP client for the host backend deployment service."""
+
+    def __init__(self, base_url: str, api_key: str, tenant_id: str | None = None):
+        if not base_url:
+            raise click.UsageError("Host backend URL is required")
+        transport = httpx.HTTPTransport(retries=3)
+        headers: dict[str, str] = {
+            "X-Api-Key": api_key,
+            "Accept": "application/json",
+        }
+        if tenant_id:
+            headers["X-Tenant-ID"] = tenant_id
+        self._base_url = base_url.rstrip("/")
+        self._api_key = api_key
+        self._client = httpx.Client(
+            base_url=self._base_url,
+            headers=headers,
+            transport=transport,
+            timeout=30,
+        )
+
+    def _request(
+        self, method: str, path: str, payload: dict[str, Any] | None = None
+    ) -> Any:
+        try:
+            resp = self._client.request(method, path, json=payload)
+            resp.raise_for_status()
+        except httpx.HTTPStatusError as err:
+            detail = err.response.text or str(err.response.status_code)
+            raise HostBackendError(
+                f"{method} {path} failed with status {err.response.status_code}: {detail}",
+                status_code=err.response.status_code,
+            ) from None
+        except httpx.TransportError as err:
+            raise HostBackendError(str(err)) from None
+
+        if not resp.content:
+            return None
+        try:
+            return resp.json()
+        except ValueError as err:
+            raise HostBackendError(
+                f"Failed to decode response from {path}: {err}"
+            ) from None
+
+    def create_deployment(self, payload: dict[str, Any]) -> dict[str, Any]:
+        return self._request("POST", "/v2/deployments", payload)
+
+    def list_deployments(self, name_contains: str) -> dict[str, Any]:
+        return self._request("GET", f"/v2/deployments?name_contains={name_contains}")
+
+    def get_deployment(self, deployment_id: str) -> dict[str, Any]:
+        return self._request("GET", f"/v2/deployments/{deployment_id}")
+
+    def request_push_token(self, deployment_id: str) -> dict[str, Any]:
+        return self._request(
+            "POST",
+            f"/v2/deployments/{deployment_id}/push-token",
+        )
+
+    def update_deployment(
+        self,
+        deployment_id: str,
+        image_uri: str,
+        secrets: list[dict[str, str]] | None = None,
+    ) -> dict[str, Any]:
+        payload: dict[str, Any] = {
+            "source_revision_config": {"image_uri": image_uri},
+        }
+        if secrets is not None:
+            payload["secrets"] = secrets
+        return self._request(
+            "PATCH",
+            f"/v2/deployments/{deployment_id}",
+            payload,
+        )
+
+    def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]:
+        return self._request(
+            "GET",
+            f"/v2/deployments/{deployment_id}/revisions?limit={limit}",
+        )
+
+    def get_revision(self, deployment_id: str, revision_id: str) -> dict[str, Any]:
+        return self._request(
+            "GET",
+            f"/v2/deployments/{deployment_id}/revisions/{revision_id}",
+        )

libs/cli/langgraph_cli/progress.py

@@ -12,8 +12,12 @@ class Progress:
         while True:
             yield from "|/-\\"
 
-    def __init__(self, *, message=""):
+    def __init__(self, *, message="", elapsed: bool = False):
         self.message = message
+        self._base_message = message
+        self._show_elapsed = elapsed
+        # use this to make sure we don't kill thread when we set msg to ""
+        self._stop = threading.Event()
         self.spinner_generator = self.spinning_cursor()
 
     def spinner_iteration(self):
@@ -29,9 +33,23 @@ class Progress:
         )
         sys.stdout.flush()
 
+    def _format_elapsed(self, seconds: float) -> str:
+        mins, secs = divmod(int(seconds), 60)
+        if mins:
+            return f"{self._base_message} ({mins}m {secs:02d}s)"
+        return f"{self._base_message} ({secs}s)"
+
     def spinner_task(self):
-        while self.message:
+        start = time.monotonic()
+        while not self._stop.is_set():
+            if not self.message:
+                time.sleep(self.delay)
+                continue
+            if self._show_elapsed:
+                self.message = self._format_elapsed(time.monotonic() - start)
             message = self.message
+            if not message:
+                continue
             sys.stdout.write(next(self.spinner_generator) + " " + message)
             sys.stdout.flush()
             time.sleep(self.delay)
@@ -50,21 +68,22 @@ class Progress:
 
             def set_message(message):
                 self.message = message
-                if not message:
-                    self.thread.join()
+                self._base_message = message or self._base_message
 
             return set_message
         else:
 
             def set_message(message):
-                sys.stderr.write(message + "\n")
-                sys.stderr.flush()
+                if message:
+                    sys.stderr.write(message + "\n")
+                    sys.stderr.flush()
 
             return set_message
 
     def __exit__(self, exception, value, tb):
         if sys.stdout.isatty():
             self.message = ""
+            self._stop.set()
             try:
                 self.thread.join()
             finally:

libs/cli/pyproject.toml

@@ -13,7 +13,9 @@ license = "MIT"
 license-files = ['LICENSE']
 dependencies = [
     "click>=8.1.7",
+    "httpx>=0.24.0",
     "langgraph-sdk>=0.1.0 ; python_version >= '3.11'",
+    "python-dotenv>=0.8.0",
 ]
 [tool.hatch.version]
 path = "langgraph_cli/__init__.py"
@@ -21,7 +23,6 @@ path = "langgraph_cli/__init__.py"
 inmem = [
     "langgraph-api>=0.5.35,<0.8.0 ; python_version >= '3.11'",
     "langgraph-runtime-inmem>=0.7 ; python_version >= '3.11'",
-    "python-dotenv>=0.8.0",
 ]
 
 [project.urls]

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

@@ -822,3 +822,139 @@ def test_prepare_args_and_stdin_with_api_version_and_image() -> None:
     # When image is provided, api_version should be ignored for the image
     # but the stdin should not contain a build section (since image is provided)
     assert "pull_policy: build" not in actual_stdin
+
+
+def test_dockerfile_command_distributed_mode() -> None:
+    """Test the 'dockerfile' command with --engine-runtime-mode distributed."""
+    runner = CliRunner()
+    config_content = {
+        "python_version": "3.11",
+        "graphs": {"agent": "agent.py:graph"},
+        "dependencies": ["."],
+    }
+
+    with temporary_config_folder(config_content) as temp_dir:
+        save_path = temp_dir / "Dockerfile"
+        agent_path = temp_dir / "agent.py"
+        agent_path.touch()
+
+        result = runner.invoke(
+            cli,
+            [
+                "dockerfile",
+                str(save_path),
+                "--config",
+                str(temp_dir / "config.json"),
+                "--engine-runtime-mode",
+                "distributed",
+            ],
+        )
+
+        assert result.exit_code == 0, result.output
+        assert "✅ Created: Dockerfile" in result.output
+
+        assert save_path.exists()
+        with open(save_path) as f:
+            dockerfile = f.read()
+            assert "FROM langchain/langgraph-executor:3.11" in dockerfile
+
+
+def test_dockerfile_command_combined_mode() -> None:
+    """Test the 'dockerfile' command with --engine-runtime-mode combined_queue_worker."""
+    runner = CliRunner()
+    config_content = {
+        "python_version": "3.11",
+        "graphs": {"agent": "agent.py:graph"},
+        "dependencies": ["."],
+    }
+
+    with temporary_config_folder(config_content) as temp_dir:
+        save_path = temp_dir / "Dockerfile"
+        agent_path = temp_dir / "agent.py"
+        agent_path.touch()
+
+        result = runner.invoke(
+            cli,
+            [
+                "dockerfile",
+                str(save_path),
+                "--config",
+                str(temp_dir / "config.json"),
+                "--engine-runtime-mode",
+                "combined_queue_worker",
+            ],
+        )
+
+        assert result.exit_code == 0, result.output
+        assert save_path.exists()
+        with open(save_path) as f:
+            dockerfile = f.read()
+            assert "FROM langchain/langgraph-api:3.11" in dockerfile
+
+
+def test_dockerfile_command_distributed_with_explicit_base_image() -> None:
+    """Test distributed mode with explicit --base-image overrides executor default."""
+    runner = CliRunner()
+    config_content = {
+        "python_version": "3.11",
+        "graphs": {"agent": "agent.py:graph"},
+        "dependencies": ["."],
+    }
+
+    with temporary_config_folder(config_content) as temp_dir:
+        save_path = temp_dir / "Dockerfile"
+        agent_path = temp_dir / "agent.py"
+        agent_path.touch()
+
+        result = runner.invoke(
+            cli,
+            [
+                "dockerfile",
+                str(save_path),
+                "--config",
+                str(temp_dir / "config.json"),
+                "--engine-runtime-mode",
+                "distributed",
+                "--base-image",
+                "my-custom-executor:latest",
+            ],
+        )
+
+        assert result.exit_code == 0, result.output
+        assert save_path.exists()
+        with open(save_path) as f:
+            dockerfile = f.read()
+            assert "FROM my-custom-executor:latest" in dockerfile
+
+
+def test_prepare_args_and_stdin_distributed_mode() -> None:
+    """Test prepare_args_and_stdin with distributed mode includes all services."""
+    config_path = pathlib.Path(__file__).parent / "langgraph.json"
+    config = validate_config(
+        Config(dependencies=["."], graphs={"agent": "agent.py:graph"})
+    )
+    port = 8000
+
+    actual_args, actual_stdin = prepare_args_and_stdin(
+        capabilities=DEFAULT_DOCKER_CAPABILITIES,
+        config_path=config_path,
+        config=config,
+        docker_compose=None,
+        port=port,
+        watch=False,
+        engine_runtime_mode="distributed",
+    )
+
+    # API service should use langgraph-api base image
+    assert "FROM langchain/langgraph-api:" in actual_stdin
+
+    # Distributed mode sets N_JOBS_PER_WORKER=0 on the API service
+    assert 'N_JOBS_PER_WORKER: "0"' in actual_stdin
+
+    # Orchestrator service present
+    assert "langgraph-orchestrator:" in actual_stdin
+
+    # Executor service present with correct base image
+    assert "langgraph-executor:" in actual_stdin
+    assert "FROM langchain/langgraph-executor:" in actual_stdin
+    assert "executor_entrypoint.sh" in actual_stdin

libs/cli/tests/unit_tests/test_config.py

@@ -13,6 +13,7 @@ from langgraph_cli.config import (
     _get_pip_cleanup_lines,
     config_to_compose,
     config_to_docker,
+    default_base_image,
     docker_tag,
     has_disallowed_build_command_content,
     validate_config,
@@ -1695,6 +1696,194 @@ def test_config_to_compose_with_api_version():
     assert "FROM langchain/langgraphjs-api:0.2.74-node20" in actual_compose_str
 
 
+def test_default_base_image_combined_mode():
+    """Test default_base_image returns langgraph-api for combined_queue_worker mode."""
+    config = validate_config(
+        {
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+        }
+    )
+    assert default_base_image(config) == "langchain/langgraph-api"
+    assert (
+        default_base_image(config, engine_runtime_mode="combined_queue_worker")
+        == "langchain/langgraph-api"
+    )
+
+
+def test_default_base_image_distributed_mode():
+    """Test default_base_image returns langgraph-executor for distributed mode."""
+    config = validate_config(
+        {
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+        }
+    )
+    assert (
+        default_base_image(config, engine_runtime_mode="distributed")
+        == "langchain/langgraph-executor"
+    )
+
+
+def test_default_base_image_distributed_with_explicit_base():
+    """Test default_base_image returns explicit base_image even in distributed mode."""
+    config = validate_config(
+        {
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "base_image": "my-custom-image:latest",
+        }
+    )
+    assert (
+        default_base_image(config, engine_runtime_mode="distributed")
+        == "my-custom-image:latest"
+    )
+
+
+def test_default_base_image_nodejs():
+    """Test default_base_image returns langgraphjs-api for Node.js config."""
+    config = validate_config(
+        {
+            "node_version": "20",
+            "graphs": {"agent": "./agent.js:graph"},
+        }
+    )
+    assert default_base_image(config) == "langchain/langgraphjs-api"
+
+
+def test_config_to_docker_executor_base_image():
+    """Test config_to_docker with executor base image for distributed mode."""
+    graphs = {"agent": "./agent.py:graph"}
+    config = validate_config({"dependencies": ["."], "graphs": graphs})
+    actual_docker_stdin, _ = config_to_docker(
+        PATH_TO_CONFIG,
+        config,
+        base_image="langchain/langgraph-executor",
+    )
+    assert "FROM langchain/langgraph-executor:3.11" in actual_docker_stdin
+    assert "LANGSERVE_GRAPHS=" in actual_docker_stdin
+
+
+def test_config_to_compose_distributed_mode():
+    """Test config_to_compose with engine_runtime_mode='distributed'."""
+    graphs = {"agent": "./agent.py:graph"}
+    actual_compose_stdin = config_to_compose(
+        PATH_TO_CONFIG,
+        validate_config({"dependencies": ["."], "graphs": graphs}),
+        "langchain/langgraph-api",
+        engine_runtime_mode="distributed",
+    )
+
+    # API service uses langchain/langgraph-api base image
+    assert "FROM langchain/langgraph-api:3.11" in actual_compose_stdin
+
+    # Orchestrator service is present
+    assert "langgraph-orchestrator:" in actual_compose_stdin
+    assert "EXECUTOR_TARGET: langgraph-executor:8188" in actual_compose_stdin
+
+    # Executor service is present with correct base image
+    assert "langgraph-executor:" in actual_compose_stdin
+    assert "FROM langchain/langgraph-executor:3.11" in actual_compose_stdin
+    assert 'entrypoint: ["sh", "/storage/executor_entrypoint.sh"]' in actual_compose_stdin
+
+    # Executor has required environment variables
+    assert "EXECUTOR_GRPC_PORT:" in actual_compose_stdin
+    assert "ENGINE_GRPC_ADDRESS:" in actual_compose_stdin
+    assert "LSD_GRPC_SERVER_ADDRESS:" in actual_compose_stdin
+    assert 'LANGGRAPH_HTTP: ""' in actual_compose_stdin
+    assert "REDIS_URI: redis://langgraph-redis:6379" in actual_compose_stdin
+
+
+def test_config_to_compose_distributed_mode_with_env_file():
+    """Test config_to_compose distributed mode propagates env_file to all services."""
+    graphs = {"agent": "./agent.py:graph"}
+    actual_compose_stdin = config_to_compose(
+        PATH_TO_CONFIG,
+        validate_config({"dependencies": ["."], "graphs": graphs, "env": ".env"}),
+        "langchain/langgraph-api",
+        engine_runtime_mode="distributed",
+    )
+
+    # env_file should appear multiple times: API, orchestrator, executor
+    env_file_count = actual_compose_stdin.count("env_file: .env")
+    assert env_file_count == 3, (
+        f"Expected env_file to appear 3 times (api, orchestrator, executor), "
+        f"got {env_file_count}"
+    )
+
+
+def test_config_to_compose_distributed_mode_generates_two_dockerfiles():
+    """Test that distributed mode generates separate Dockerfiles for API and executor."""
+    graphs = {"agent": "./agent.py:graph"}
+    actual_compose_stdin = config_to_compose(
+        PATH_TO_CONFIG,
+        validate_config({"dependencies": ["."], "graphs": graphs}),
+        "langchain/langgraph-api",
+        engine_runtime_mode="distributed",
+    )
+
+    # Should contain two different FROM lines
+    from_lines = [
+        line.strip()
+        for line in actual_compose_stdin.splitlines()
+        if line.strip().startswith("FROM ")
+    ]
+    assert len(from_lines) == 2
+    assert "FROM langchain/langgraph-api:3.11" in from_lines[0]
+    assert "FROM langchain/langgraph-executor:3.11" in from_lines[1]
+
+
+def test_config_to_compose_combined_mode_no_orchestrator():
+    """Test that combined_queue_worker mode does NOT generate orchestrator/executor."""
+    graphs = {"agent": "./agent.py:graph"}
+    actual_compose_stdin = config_to_compose(
+        PATH_TO_CONFIG,
+        validate_config({"dependencies": ["."], "graphs": graphs}),
+        "langchain/langgraph-api",
+        engine_runtime_mode="combined_queue_worker",
+    )
+    assert "langgraph-orchestrator:" not in actual_compose_stdin
+    assert "langgraph-executor:" not in actual_compose_stdin
+
+
+def test_config_to_compose_default_mode_no_orchestrator():
+    """Test that default mode (no engine_runtime_mode) has no orchestrator/executor."""
+    graphs = {"agent": "./agent.py:graph"}
+    actual_compose_stdin = config_to_compose(
+        PATH_TO_CONFIG,
+        validate_config({"dependencies": ["."], "graphs": graphs}),
+        "langchain/langgraph-api",
+    )
+    assert "langgraph-orchestrator:" not in actual_compose_stdin
+    assert "langgraph-executor:" not in actual_compose_stdin
+
+
+def test_config_to_compose_distributed_executor_gets_correct_paths():
+    """Test that executor Dockerfile gets correct host paths despite API Dockerfile
+    mutation. This validates the deep copy fix in config_to_compose -- without it,
+    the executor's config_to_docker call would see already-mutated container paths
+    from the API's config_to_docker call, causing FileNotFoundError."""
+    graphs = {"agent": "./agent.py:graph"}
+    actual_compose_stdin = config_to_compose(
+        PATH_TO_CONFIG,
+        validate_config({"dependencies": ["."], "graphs": graphs}),
+        "langchain/langgraph-api",
+        engine_runtime_mode="distributed",
+    )
+
+    # Both API and executor Dockerfiles should contain valid LANGSERVE_GRAPHS
+    # referencing container paths (not host paths). If the deep copy was missing,
+    # the executor Dockerfile would fail to generate or have wrong paths.
+    from_lines = [
+        line.strip()
+        for line in actual_compose_stdin.splitlines()
+        if "LANGSERVE_GRAPHS=" in line.strip()
+    ]
+    assert len(from_lines) == 2, (
+        f"Expected 2 LANGSERVE_GRAPHS lines (api + executor), got {len(from_lines)}"
+    )
+
+
 class TestHasDisallowedBuildCommandContent:
     """Tests for has_disallowed_build_command_content."""
 

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -0,0 +1,134 @@
+import base64
+import json
+import os
+
+import click
+import pytest
+
+from langgraph_cli.cli import (
+    _docker_config_for_token,
+    _normalize_image_name,
+    _normalize_image_tag,
+    _parse_env_from_config,
+)
+
+
+class TestDockerConfigForToken:
+    def test_creates_config_json(self):
+        with _docker_config_for_token("us-docker.pkg.dev", "my-token") as cfg:
+            config_path = os.path.join(cfg, "config.json")
+            assert os.path.isfile(config_path)
+            with open(config_path) as f:
+                data = json.load(f)
+            expected_auth = base64.b64encode(b"oauth2accesstoken:my-token").decode()
+            assert data == {"auths": {"us-docker.pkg.dev": {"auth": expected_auth}}}
+
+    def test_tempdir_cleaned_up(self):
+        with _docker_config_for_token("registry.example.com", "tok") as cfg:
+            assert os.path.isdir(cfg)
+        assert not os.path.exists(cfg)
+
+    def test_different_registries(self):
+        with _docker_config_for_token("gcr.io", "token123") as cfg:
+            with open(os.path.join(cfg, "config.json")) as f:
+                data = json.load(f)
+            assert "gcr.io" in data["auths"]
+
+
+class TestNormalizeImageName:
+    def test_simple_name(self):
+        assert _normalize_image_name("myapp") == "myapp"
+
+    def test_uppercase_lowered(self):
+        assert _normalize_image_name("MyApp") == "myapp"
+
+    def test_special_chars_replaced(self):
+        assert _normalize_image_name("my app!@#v2") == "my-app-v2"
+
+    def test_dots_and_hyphens_kept(self):
+        assert _normalize_image_name("my-app.v2") == "my-app.v2"
+
+    def test_leading_trailing_stripped(self):
+        assert _normalize_image_name("--my-app..") == "my-app"
+
+    def test_empty_string_returns_app(self):
+        assert _normalize_image_name("") == "app"
+
+    def test_none_returns_app(self):
+        assert _normalize_image_name(None) == "app"
+
+    def test_all_invalid_chars_returns_app(self):
+        assert _normalize_image_name("!!!") == "app"
+
+
+class TestNormalizeImageTag:
+    def test_valid_tag(self):
+        assert _normalize_image_tag("v1.2.3") == "v1.2.3"
+
+    def test_empty_defaults_to_latest(self):
+        assert _normalize_image_tag("") == "latest"
+
+    def test_alphanumeric_and_special(self):
+        assert _normalize_image_tag("my_tag-1.0") == "my_tag-1.0"
+
+    def test_invalid_chars_raises(self):
+        with pytest.raises(click.UsageError, match="Image tag may only contain"):
+            _normalize_image_tag("v1.0:bad")
+
+    def test_spaces_raises(self):
+        with pytest.raises(click.UsageError, match="Image tag may only contain"):
+            _normalize_image_tag("has space")
+
+
+class TestParseEnvFromConfig:
+    def test_env_dict(self, tmp_path):
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({"env": {"FOO": "bar", "NUM": 42}}, config_path)
+        assert result == {"FOO": "bar", "NUM": "42"}
+
+    def test_env_string_dotenv_file(self, tmp_path):
+        env_file = tmp_path / "my.env"
+        env_file.write_text("KEY1=val1\nKEY2=val2\n")
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({"env": "my.env"}, config_path)
+        assert result == {"KEY1": "val1", "KEY2": "val2"}
+
+    def test_env_missing_falls_back_to_dotenv(self, tmp_path, monkeypatch):
+        env_file = tmp_path / ".env"
+        env_file.write_text("DEFAULT_KEY=default_val\n")
+        monkeypatch.chdir(tmp_path)
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({}, config_path)
+        assert result == {"DEFAULT_KEY": "default_val"}
+
+    def test_env_empty_dict_falls_back_to_dotenv(self, tmp_path, monkeypatch):
+        """validate_config defaults env to {}, should still fall back to .env."""
+        env_file = tmp_path / ".env"
+        env_file.write_text("MY_KEY=my_val\n")
+        monkeypatch.chdir(tmp_path)
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({"env": {}}, config_path)
+        assert result == {"MY_KEY": "my_val"}
+
+    def test_env_missing_no_dotenv_returns_empty(self, tmp_path, monkeypatch):
+        monkeypatch.chdir(tmp_path)
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({}, config_path)
+        assert result == {}
+
+    def test_env_dotenv_filters_none_values(self, tmp_path):
+        # Lines like "KEY=" produce empty string, lines like "KEY" produce None
+        env_file = tmp_path / "test.env"
+        env_file.write_text("GOOD=value\nEMPTY=\n")
+        config_path = tmp_path / "langgraph.json"
+        config_path.touch()
+        result = _parse_env_from_config({"env": "test.env"}, config_path)
+        assert "GOOD" in result
+        assert result["GOOD"] == "value"
+        # EMPTY= gives empty string, not None, so it should be present
+        assert result["EMPTY"] == ""

libs/cli/tests/unit_tests/test_docker.py

@@ -368,6 +368,61 @@ services:
     assert clean_empty_lines(actual_compose_str) == expected_compose_str
 
 
+def test_compose_distributed_mode_with_custom_db():
+    """Test compose with engine_runtime_mode='distributed' adds N_JOBS_PER_WORKER=0."""
+    port = 8123
+    custom_postgres_uri = "custom_postgres_uri"
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        postgres_uri=custom_postgres_uri,
+        engine_runtime_mode="distributed",
+    )
+    expected_compose_str = f"""services:
+    langgraph-redis:
+        image: redis:6
+        healthcheck:
+            test: redis-cli ping
+            interval: 5s
+            timeout: 1s
+            retries: 5
+    langgraph-api:
+        ports:
+            - "{port}:8000"
+        depends_on:
+            langgraph-redis:
+                condition: service_healthy
+        environment:
+            REDIS_URI: redis://langgraph-redis:6379
+            POSTGRES_URI: {custom_postgres_uri}
+            N_JOBS_PER_WORKER: "0\""""
+    assert clean_empty_lines(actual_compose_str) == expected_compose_str
+
+
+def test_compose_distributed_mode_with_default_db():
+    """Test compose distributed mode with default DB includes N_JOBS_PER_WORKER=0."""
+    port = 8123
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        engine_runtime_mode="distributed",
+    )
+    assert 'N_JOBS_PER_WORKER: "0"' in actual_compose_str
+    assert "langgraph-postgres:" in actual_compose_str
+    assert "langgraph-redis:" in actual_compose_str
+
+
+def test_compose_combined_mode_has_no_n_jobs():
+    """Test compose with default combined_queue_worker mode does NOT set N_JOBS_PER_WORKER."""
+    port = 8123
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        engine_runtime_mode="combined_queue_worker",
+    )
+    assert "N_JOBS_PER_WORKER" not in actual_compose_str
+
+
 @pytest.mark.parametrize(
     "input_str,expected",
     [

libs/cli/tests/unit_tests/test_host_backend.py

@@ -0,0 +1,162 @@
+import httpx
+import pytest
+
+from langgraph_cli.host_backend import HostBackendClient, HostBackendError
+
+
+@pytest.fixture
+def mock_transport():
+    return httpx.MockTransport(lambda req: httpx.Response(200, json={"ok": True}))
+
+
+@pytest.fixture
+def client(mock_transport):
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=mock_transport,
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    return c
+
+
+def test_constructor_strips_trailing_slash():
+    c = HostBackendClient("https://api.example.com/", "key")
+    assert str(c._client.base_url) == "https://api.example.com"
+
+
+def test_constructor_empty_url_raises():
+    with pytest.raises(Exception, match="Host backend URL is required"):
+        HostBackendClient("", "key")
+
+
+def test_request_sends_headers():
+    def handler(req: httpx.Request) -> httpx.Response:
+        assert req.headers["x-api-key"] == "test-key"
+        assert req.headers["accept"] == "application/json"
+        return httpx.Response(200, json={"ok": True})
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    result = c._request("GET", "/test")
+    assert result == {"ok": True}
+
+
+def test_request_sends_json_payload():
+    def handler(req: httpx.Request) -> httpx.Response:
+        assert req.headers["content-type"] == "application/json"
+        assert req.content == b'{"key":"value"}'
+        return httpx.Response(200, json={"created": True})
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    result = c._request("POST", "/test", {"key": "value"})
+    assert result == {"created": True}
+
+
+def test_request_empty_body_returns_none():
+    transport = httpx.MockTransport(lambda req: httpx.Response(200, content=b""))
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=transport,
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    assert c._request("DELETE", "/test") is None
+
+
+def test_request_http_error_raises():
+    transport = httpx.MockTransport(lambda req: httpx.Response(404, text="not found"))
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=transport,
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    with pytest.raises(HostBackendError, match="404"):
+        c._request("GET", "/missing")
+
+
+def test_request_invalid_json_raises():
+    transport = httpx.MockTransport(
+        lambda req: httpx.Response(200, content=b"not json")
+    )
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=transport,
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    with pytest.raises(HostBackendError, match="Failed to decode"):
+        c._request("GET", "/bad-json")
+
+
+def test_request_transport_error_raises():
+    def handler(req: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("connection refused")
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    with pytest.raises(HostBackendError, match="connection refused"):
+        c._request("GET", "/test")
+
+
+def test_create_deployment(client):
+    result = client.create_deployment({"name": "my-deploy"})
+    assert result == {"ok": True}
+
+
+def test_get_deployment(client):
+    result = client.get_deployment("dep-123")
+    assert result == {"ok": True}
+
+
+def test_list_deployments(client):
+    result = client.list_deployments("my-app")
+    assert result == {"ok": True}
+
+
+def test_request_push_token(client):
+    result = client.request_push_token("dep-123")
+    assert result == {"ok": True}
+
+
+def test_update_deployment(client):
+    result = client.update_deployment(
+        "dep-123", "image:latest", secrets=[{"name": "KEY", "value": "val"}]
+    )
+    assert result == {"ok": True}
+
+
+def test_update_deployment_no_secrets(client):
+    result = client.update_deployment("dep-123", "image:latest")
+    assert result == {"ok": True}
+
+
+def test_list_revisions(client):
+    result = client.list_revisions("dep-123", limit=5)
+    assert result == {"ok": True}
+
+
+def test_get_revision(client):
+    result = client.get_revision("dep-123", "rev-456")
+    assert result == {"ok": True}

libs/cli/uv.lock

@@ -983,7 +983,9 @@ name = "langgraph-cli"
 source = { editable = "." }
 dependencies = [
     { name = "click" },
+    { name = "httpx" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11'" },
+    { name = "python-dotenv" },
 ]
 
 [package.optional-dependencies]
@@ -1021,10 +1023,11 @@ test = [
 [package.metadata]
 requires-dist = [
     { name = "click", specifier = ">=8.1.7" },
+    { name = "httpx", specifier = ">=0.24.0" },
     { name = "langgraph-api", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.5.35,<0.8.0" },
     { name = "langgraph-runtime-inmem", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.7" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11'", specifier = ">=0.1.0" },
-    { name = "python-dotenv", marker = "extra == 'inmem'", specifier = ">=0.8.0" },
+    { name = "python-dotenv", specifier = ">=0.8.0" },
 ]
 provides-extras = ["inmem"]
 

libs/langgraph/langgraph/_internal/_constants.py

@@ -41,6 +41,9 @@ CONFIG_KEY_CACHE = sys.intern("__pregel_cache")
 # holds a `BaseCache` made available to subgraphs
 CONFIG_KEY_RESUMING = sys.intern("__pregel_resuming")
 # holds a boolean indicating if subgraphs should resume from a previous checkpoint
+CONFIG_KEY_REPLAY_STATE = sys.intern("__pregel_replay_state")
+# holds a ReplayState tracking the parent checkpoint_id upper bound and which
+# subgraph namespaces have already loaded their pre-replay checkpoint
 CONFIG_KEY_TASK_ID = sys.intern("__pregel_task_id")
 # holds the task ID for the current task
 CONFIG_KEY_THREAD_ID = sys.intern("thread_id")
@@ -98,6 +101,7 @@ RESERVED = {
     CONFIG_KEY_STREAM,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_RESUMING,
+    CONFIG_KEY_REPLAY_STATE,
     CONFIG_KEY_TASK_ID,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_ID,

libs/langgraph/langgraph/_internal/_replay.py

@@ -0,0 +1,90 @@
+"""Replay state for subgraph checkpoint loading during time-travel."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from langgraph._internal._constants import NS_END
+
+if TYPE_CHECKING:
+    from langchain_core.runnables import RunnableConfig
+    from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointTuple
+
+
+class ReplayState:
+    """Tracks which subgraphs have already loaded their pre-replay checkpoint.
+
+    During a parent replay, each subgraph's first invocation should restore the
+    checkpoint from before the replay point. Subsequent invocations of the same
+    subgraph (e.g. in a loop) should use normal checkpoint loading so they pick
+    up freshly created checkpoints.
+
+    The single `ReplayState` instance is shared by reference across all derived
+    configs within one parent execution.
+    """
+
+    __slots__ = ("checkpoint_id", "_visited_ns")
+
+    def __init__(self, checkpoint_id: str) -> None:
+        self.checkpoint_id = checkpoint_id
+        # DO NOT CHANGE THIS VARIABLE – it may need to be rehydrated
+        # in other runtimes
+        self._visited_ns: set[str] = set()
+
+    def _is_first_visit(self, checkpoint_ns: str) -> bool:
+        """Return True the first time a subgraph namespace is seen.
+
+        The task-id suffix is stripped so that the same logical subgraph
+        (e.g. ``"sub_node"``) is recognized across loop iterations even
+        though each iteration has a different task id.
+        """
+        # "sub_node:task_id" -> "sub_node"
+        stable_ns = (
+            checkpoint_ns.rsplit(NS_END, 1)[0]
+            if NS_END in checkpoint_ns
+            else checkpoint_ns
+        )
+        if stable_ns in self._visited_ns:
+            return False
+        self._visited_ns.add(stable_ns)
+        return True
+
+    def get_checkpoint(
+        self,
+        checkpoint_ns: str,
+        checkpointer: BaseCheckpointSaver,
+        checkpoint_config: RunnableConfig,
+    ) -> CheckpointTuple | None:
+        """Load the right checkpoint for a subgraph during replay.
+
+        On the first call for a given subgraph namespace, returns the latest
+        checkpoint created *before* the replay point. On subsequent calls
+        (e.g. the same subgraph in a later loop iteration), falls back to
+        normal latest-checkpoint loading.
+        """
+        if self._is_first_visit(checkpoint_ns):
+            for saved in checkpointer.list(
+                checkpoint_config,
+                before={"configurable": {"checkpoint_id": self.checkpoint_id}},
+                limit=1,
+            ):
+                return saved
+            return None
+        return checkpointer.get_tuple(checkpoint_config)
+
+    async def aget_checkpoint(
+        self,
+        checkpoint_ns: str,
+        checkpointer: BaseCheckpointSaver,
+        checkpoint_config: RunnableConfig,
+    ) -> CheckpointTuple | None:
+        """Async version of `get_checkpoint`."""
+        if self._is_first_visit(checkpoint_ns):
+            async for saved in checkpointer.alist(
+                checkpoint_config,
+                before={"configurable": {"checkpoint_id": self.checkpoint_id}},
+                limit=1,
+            ):
+                return saved
+            return None
+        return await checkpointer.aget_tuple(checkpoint_config)

libs/langgraph/langgraph/advanced_graph/__init__.py

@@ -0,0 +1,25 @@
+from langgraph.advanced_graph.state import (
+    AdvancedStateGraph,
+    AnyOfCondition,
+    ChannelCondition,
+    CompiledGraphEngine,
+    Context,
+    GraphRunHandler,
+    TimerCondition,
+    any_of,
+    channel_condition,
+    timer_condition,
+)
+
+__all__ = (
+    "AdvancedStateGraph",
+    "AnyOfCondition",
+    "ChannelCondition",
+    "Context",
+    "CompiledGraphEngine",
+    "GraphRunHandler",
+    "TimerCondition",
+    "any_of",
+    "channel_condition",
+    "timer_condition",
+)

libs/langgraph/langgraph/advanced_graph/evolve-extend-langgraph.md

@@ -0,0 +1,130 @@
+# Evolve/Extend LangGraph with next level of orchestration
+
+## LangGraph Today: A Strong Foundation with Creative Innovation
+
+LangGraph is already an exceptional orchestration framework. It has introduced a number of creative features that no other workflow engine on the market has even attempted.
+
+**First-class streaming.** No workflow engine has ever integrated streaming as seamlessly as LangGraph. Streaming is not an afterthought bolted on top — it is woven into the core execution model, allowing every node, every tool call, and every LLM interaction to emit incremental output naturally.
+
+**Flexible durability modes.** LangGraph defaults to asynchronous execution and supports sync and "exit" modes as well. This is a significant departure from traditional workflow engines, which typically only offer synchronous execution. The ability to choose a durability mode gives developers fine-grained control over the trade-off between persistence guarantees and execution speed.
+
+**Reusable checkpoints.** The checkpoint system allows state to be captured at any point during graph execution and freely replayed, forked, or resumed later. This enables powerful patterns like time-travel debugging, human-in-the-loop approval flows, and long-running conversations that can be picked up exactly where they left off.
+
+**Double texting.** LangGraph natively handles the real-world scenario where a user sends a new message while a previous one is still being processed — a problem most orchestration frameworks simply ignore.
+
+Beyond these innovative features, LangGraph provides solid support for the foundational workflow execution patterns that developers rely on daily. Sequential execution, or loops and conditional branching. Basic parallelism is also well supported: when multiple LLM calls or tool invocations are independent of each other, they can run concurrently to avoid the latency cost of sequential execution, and their results are merged back into the shared state for downstream processing.
+
+LangGraph also offers a simple and intuitive mechanism for human-in-the-loop interactions, allowing a graph to pause execution and wait for user input before continuing.
+
+Combined with the broader LangChain ecosystem, these have made LangGraph a significant success in the market.
+
+## Emerging Gaps: What LangGraph Struggles to Support
+
+As adoption has grown and use cases have become more sophisticated, we have discovered an increasing number of scenarios and design patterns that LangGraph cannot support well today.
+
+**Complex sub-agent coordination.** A main agent often needs to manage multiple sub-agents, but the coordination involved is far more nuanced than simply launching a batch of sub-agents, waiting for all of them to finish, and then moving on. In practice, a main agent may launch a sub-agent, continue doing other work, spawn additional sub-agents later, wait selectively for certain results, retry with a different strategy if one sub-agent fails, or dynamically decide what to do next based on partial results that arrive at unpredictable times.
+
+LangGraph today lacks the coordination primitives to express this. The current parallelism model groups multiple nodes into a single superstep — all of them execute concurrently, but _all_ must complete before the graph can advance to the next step. There is no way for one node to proceed independently while others are still running, and no built-in mechanism for selective waiting, partial result handling, or dynamic task spawning mid-execution.
+
+Sub-agents also cannot simply be modeled as subgraphs, because subgraphs today execute within the same run. They cannot be scaled up independently — if a sub-agent is resource-intensive, there is no straightforward way to run it on a separate machine. Ideally, launching a sub-agent should be(or opt in) as simple as dispatching it for distributed execution across multiple machines.
+
+**Concurrent input and output (e.g. audio agents).** Audio agents also present a particularly clear example of a pattern LangGraph cannot express today. In a voice interaction, speech input and speech output may happen simultaneously — the agent should be able to process a previous utterance, continue receiving new audio input, and produce output all at the same time. These three activities should not be mutually exclusive.
+
+The closest workaround today is double texting, but it has a fundamental flaw: when a new audio input arrives, the previous one is interrupted and canceled rather than being allowed to gracefully complete. The workflow code itself should have the control to decide whether to stop running.
+
+LangGraph is, at its core, a general-purpose workflow engine. Although we focus primarily on agent development, none of the primitives it offers are exclusive to agents or dedicated solely to agentic use cases. Conversely, there is nothing that a general-purpose workflow engine provides that we can safely assume agent development will _never_ need. 
+
+The difference is probably only priority. For example, durable timer where a step can sleep for hours, days or months before resuming. Traditional workflow engines — those built for general microservice orchestration(which doesn't need streaming) -- they may need durable timers. In the agent development world today, most agents are still relatively simple. There are not yet many scenarios that require a step to wait for hours or days before proceeding.
+
+
+## Deriving What's Needed from First Principles
+
+Before jumping to solutions, it is worth stepping back and asking a fundamental question: what is an orchestration engine, and what do users expect it to provide?
+
+At its most fundamental level, a workflow engine's value proposition is making a long-running process execute reliably. If a machine crashes, execution should smoothly fail over to another machine and resume from the last point where it was interrupted — not start over from the beginning. So we can reason about what is needed by asking: what would a developer do if they had to build a long-running process _without_ a workflow engine?
+
+Starting from the simple. A developer could write a simple `main` function — a single-threaded program, just like everyone writes when they first learn to code. It would have `if/else` branches, `for` loops, and maybe it would wait for command-line input. Many early agent use cases look exactly like this: execute a sequence of steps, make decisions along the way, loop when necessary.
+
+But if that machine crashes, you probably do not want the process to start over from scratch. You want it to resume from the last step that completed successfully. And if a step fails, you might want it to retry automatically before giving up.
+
+LangGraph handles this case very well. 
+
+There is an important constraint worth calling out explicitly: LangGraph requires the developer to organize their code into **nodes**, which serve as the boundaries at which checkpoints can be taken. This is a constraint shared by every workflow engine — it is simply not feasible to persist a checkpoint after every single line of arbitrary code.
+
+### From Single-Threaded to Concurrent: Where the Model Breaks Down
+
+But as product requirements grow more complex, a single-threaded program is no longer sufficient. The process becomes multi-threaded or multi-process. And in a multi-threaded program, each thread executes independently — when one thread finishes a step and moves on to its next step, it does not need to wait for another thread to finish _its_ current step first.
+
+This is precisely why LangGraph's superstep restriction feels awkward in practice. In the superstep model, all concurrently executing nodes must complete before any of them can advance. But that is not how independent threads work. Each thread should be able to progress at its own pace, checkpoint its own state, and move to its next step without being blocked by unrelated work happening in parallel.
+
+Multiple threads and processes do, however, need to coordinate with each other. In concurrent programming, channels are an essential primitive precisely because they provide a safe, structured way for threads to communicate and synchronize without relying on shared mutable memory — avoiding data races and deadlocks. In some cases, threads may use locking for coordination, but the preferred approach is message passing through channels.
+
+NOTE: "channel" is overloaded term here as it's also an internal term within current LangGraph pregel algorithm. 
+
+LangGraph already has a mechanism that is closely related: `interrupt`. A run can be interrupted, and then another run can resume it. If we look at this through the lens of channels, `interrupt` is essentially a **channel with size 0** — a synchronous rendezvous point where one side blocks until the other side is ready. 
+
+The natural extension:
+
+1. **Variable-size channels.** The channel buffer size should be configurable — size 0 for synchronous handoff (like `interrupt` today), size N for buffered communication where the sender can proceed without waiting, and unbounded for fully asynchronous fire-and-forget messaging.
+2. **Channels across boundaries.** Channels should not be limited to communication between separate runs. Nodes within the same graph should also be able to send and receive through channels — mirroring the way both multi-process communication (between runs) and multi-thread communication (between nodes within a run) work in ordinary concurrent programs.
+3. **Node-level blocking, not run-level pausing.** When a node waits on a channel (i.e. `interrupt`), only that node should block — the rest of the graph should continue executing. Today, `interrupt` pauses the entire run. In a concurrent program, when one thread blocks on a channel read, the other threads keep running. The same should be true: an interrupt should suspend the individual node, not halt the whole run.
+
+## Summmary of all extension opportunity
+
+### P1: urgently needed
+#### Remove the Superstep Restriction
+
+Today, when multiple nodes execute in parallel, they are grouped into a superstep. All nodes in a superstep must complete before any downstream node can begin. This means that even if `b1` finishes quickly and its successor `b11` is ready to run, it must wait for `b2` to finish first.
+
+With the superstep restriction removed, each parallel branch progresses independently. As soon as a node completes, its downstream successor can begin immediately — regardless of what is happening in other branches.
+
+**Current behavior (superstep model):**
+
+```
+Step 1:  a
+Step 2:  b1, b2        ← both must finish before step 3
+Step 3:  b11, b22      ← both start together
+```
+
+Even if `b1` finishes in 1 second and `b2` takes 30 seconds, `b11` cannot start until `b2` is done.
+
+**Proposed behavior (independent branches):**
+
+```
+Branch 1:  a → b1 → b11 → ...
+Branch 2:  a → b2 → b22 → ...
+```
+
+Each branch advances at its own pace. `b1` finishing triggers `b11` immediately, without waiting for `b2`.
+
+
+No API change is needed from the user's perspective — the graph definition stays the same. The change is in the execution semantics: the engine no longer forces all parallel nodes to synchronize at each step boundary. Each branch is checkpointed independently, so if `b1 → b11` completes while `b2` is still running, `b11`'s result is already persisted.
+
+This is necessary for the next one -- Light-weight Interrupt: Only Block the Current Node. Because we want to let other nodes continue to run while a node is waiting on something.
+
+#### Light-weight Interrupt -- wait_for API: Only Block the Current Node Until Channel Has Enough Messages
+
+Today, `interrupt` pauses the entire run. Every node stops, and nothing can proceed until the interrupt is resolved externally. This is the right behavior for a simple single-threaded workflow, but it breaks down when multiple branches are executing concurrently — one branch needing input should not freeze all the others.
+
+The proposed change has three parts:
+
+1. **Named channels.** A graph can declare named channels as coordination points. These are distinct from the graph's state — they are message-passing primitives, not shared memory.
+2. **`wait_for` blocks only the current node.** When a node calls `wait_for`, it suspends itself and waits for messages on the specified channel. All other nodes in the graph continue executing normally.
+3. A channel can be published from both external and internal
+
+The `wait_for` call takes a channel name and optionally a count `N`, meaning "wait until N messages have arrived on this channel before resuming."
+
+**Prototype:**
+
+See [test_sub_agents.py](../libs/langgraph/tests/advanced-graph/test_sub_agents.py)
+
+
+### P2: likely needed
+#### subGraph redesign
+#### durable timers
+#### more flexiable waiting conditions on interrupts
+#### locking on state fields
+
+
+### P3: future needed or nice to have
+#### RPC

libs/langgraph/langgraph/advanced_graph/state.py

@@ -0,0 +1,445 @@
+from __future__ import annotations
+
+import asyncio
+import inspect
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Any, Generic, TypeVar, cast
+
+from langgraph.types import Command, Send
+
+StateT = TypeVar("StateT")
+
+
+@dataclass(frozen=True)
+class _ChannelSpec:
+    typ: Any
+
+
+@dataclass(frozen=True)
+class ChannelCondition:
+    channel: str
+    n: int = 1
+
+
+@dataclass(frozen=True)
+class TimerCondition:
+    seconds: float
+
+
+@dataclass(frozen=True)
+class AnyOfCondition:
+    conditions: tuple[WaitCondition, ...]
+
+
+WaitCondition = ChannelCondition | TimerCondition
+
+
+class AdvancedStateGraph(Generic[StateT]):
+    """Experimental in-memory graph engine with async channels."""
+
+    def __init__(self, state_schema: type[StateT]) -> None:
+        self.state_schema = state_schema
+        self._nodes: dict[str, Callable[..., Any]] = {}
+        self._async_channels: dict[str, _ChannelSpec] = {}
+        self._entry_point: str | None = None
+        self._finish_point: str | None = None
+
+    def add_node(
+        self,
+        name_or_node: str | Callable[..., Any],
+        node: Callable[..., Any] | None = None,
+    ) -> str:
+        if node is None:
+            if not callable(name_or_node):
+                raise TypeError("add_node() expects a callable when name is omitted")
+            node_name = _infer_node_name(name_or_node)
+            node_fn = name_or_node
+        else:
+            if not isinstance(name_or_node, str):
+                raise TypeError("add_node() expects a string node name")
+            node_name = name_or_node
+            node_fn = node
+
+        if node_name in self._nodes:
+            raise ValueError(f"Node `{node_name}` already exists")
+        self._nodes[node_name] = node_fn
+        return node_name
+
+    def add_async_channel(self, name: str, typ: Any) -> None:
+        if name in self._async_channels:
+            raise ValueError(f"Channel `{name}` already exists")
+        self._async_channels[name] = _ChannelSpec(typ=typ)
+
+    def set_entry_point(self, name_or_node: str | Callable[..., Any]) -> None:
+        self._entry_point = self._resolve_node_name(name_or_node)
+
+    def set_finish_point(self, name_or_node: str | Callable[..., Any]) -> None:
+        self._finish_point = self._resolve_node_name(name_or_node)
+
+    def add_entry_node(self, node: Callable[..., Any]) -> str:
+        node_name = self.add_node(node)
+        self.set_entry_point(node_name)
+        return node_name
+
+    def add_finish_node(self, node: Callable[..., Any]) -> str:
+        node_name = self.add_node(node)
+        self.set_finish_point(node_name)
+        return node_name
+
+    def _resolve_node_name(self, name_or_node: str | Callable[..., Any]) -> str:
+        if isinstance(name_or_node, str):
+            return name_or_node
+        node_name = _infer_node_name(name_or_node)
+        if node_name not in self._nodes:
+            self._nodes[node_name] = name_or_node
+        return node_name
+
+    def compile(self) -> CompiledGraphEngine[StateT]:
+        if self._entry_point is None:
+            raise ValueError("Entry point is not set")
+        if self._finish_point is None:
+            raise ValueError("Finish point is not set")
+        if self._entry_point not in self._nodes:
+            raise ValueError(f"Entry point node `{self._entry_point}` does not exist")
+        if self._finish_point not in self._nodes:
+            raise ValueError(f"Finish point node `{self._finish_point}` does not exist")
+        return CompiledGraphEngine(
+            nodes=dict(self._nodes),
+            async_channels=dict(self._async_channels),
+            entry_point=self._entry_point,
+            finish_point=self._finish_point,
+        )
+
+
+class CompiledGraphEngine(Generic[StateT]):
+    """Executable runtime for `AdvancedStateGraph`."""
+
+    def __init__(
+        self,
+        *,
+        nodes: dict[str, Callable[..., Any]],
+        async_channels: dict[str, _ChannelSpec],
+        entry_point: str,
+        finish_point: str,
+    ) -> None:
+        self._nodes = nodes
+        self._async_channels = async_channels
+        self._entry_point = entry_point
+        self._finish_point = finish_point
+
+    async def ainvoke(self, initial_state: StateT) -> StateT:
+        handler = await self.astart(initial_state)
+        return await handler
+
+    async def astart(self, initial_state: StateT) -> GraphRunHandler[StateT]:
+        run = _GraphEngineRun(
+            nodes=self._nodes,
+            async_channel_specs=self._async_channels,
+            entry_point=self._entry_point,
+            finish_point=self._finish_point,
+        )
+        task = asyncio.create_task(run.run(initial_state))
+        return GraphRunHandler(run=run, task=task)
+
+
+class Context:
+    """Per-run context injected into advanced graph nodes."""
+
+    def __init__(self, run: _GraphEngineRun) -> None:
+        self._run = run
+
+    async def wait_for(self, target: WaitCondition | AnyOfCondition) -> Any:
+        return await self._run.wait_for(target)
+
+    def publish_to_channel(self, channel: str, value: Any) -> None:
+        self._run.publish_nowait(channel, value)
+
+    async def apublish_to_channel(self, channel: str, value: Any) -> None:
+        await self._run.publish(channel, value)
+
+
+class GraphRunHandler(Generic[StateT]):
+    """Handle for an active in-memory run."""
+
+    def __init__(self, *, run: _GraphEngineRun, task: asyncio.Task[StateT]) -> None:
+        self._run = run
+        self._task = task
+
+    async def apublish_to_channel(self, channel: str, value: Any) -> None:
+        if self._task.done():
+            raise RuntimeError("Run has already completed")
+        await self._run.publish(channel, value)
+
+    async def aresult(self) -> StateT:
+        return await self._task
+
+    def __await__(self) -> Any:
+        return self._task.__await__()
+
+
+class _GraphEngineRun:
+    def __init__(
+        self,
+        *,
+        nodes: dict[str, Callable[..., Any]],
+        async_channel_specs: dict[str, _ChannelSpec],
+        entry_point: str,
+        finish_point: str,
+    ) -> None:
+        self._nodes = nodes
+        self._entry_point = entry_point
+        self._finish_point = finish_point
+        self._async_channels: dict[str, asyncio.Queue[Any]] = {
+            name: asyncio.Queue() for name, _spec in async_channel_specs.items()
+        }
+        self._tasks: set[asyncio.Task[list[Send]]] = set()
+        self._finished = False
+        self._state: Any = None
+        self.context = Context(self)
+
+    async def run(self, initial_state: StateT) -> StateT:
+        self._state = initial_state
+        self._schedule(Send(self._entry_point, initial_state))
+        try:
+            while self._tasks and not self._finished:
+                done, _ = await asyncio.wait(
+                    self._tasks, return_when=asyncio.FIRST_COMPLETED
+                )
+                for task in done:
+                    self._tasks.remove(task)
+                    exc = task.exception()
+                    if exc is not None:
+                        await self._cancel_all_tasks()
+                        raise exc
+                    sends = task.result()
+                    for send in sends:
+                        self._schedule(send)
+            if self._finished:
+                await self._cancel_all_tasks()
+            return cast(StateT, self._state)
+        finally:
+            await self._cancel_all_tasks()
+
+    async def publish(self, channel: str, value: Any) -> None:
+        queue = self._get_async_channel(channel)
+        await queue.put(value)
+
+    def publish_nowait(self, channel: str, value: Any) -> None:
+        queue = self._get_async_channel(channel)
+        queue.put_nowait(value)
+
+    async def wait_for(self, target: WaitCondition | AnyOfCondition) -> Any:
+        if isinstance(target, ChannelCondition):
+            value = await self._wait_for_channel_values(target.channel, n=target.n)
+            return {
+                "condition": "channel",
+                "channel": target.channel,
+                "value": value,
+            }
+        if isinstance(target, TimerCondition):
+            await asyncio.sleep(target.seconds)
+            return {"condition": "timer", "seconds": target.seconds}
+        if isinstance(target, AnyOfCondition):
+            return await self._wait_for_any_of(target)
+        raise ValueError(f"Unsupported wait condition type: {type(target)!r}")
+
+    async def _wait_for_channel_values(self, channel: str, n: int) -> Any:
+        if n < 1:
+            raise ValueError("wait_for count `n` must be >= 1")
+        queue = self._get_async_channel(channel)
+        if n == 1:
+            return await queue.get()
+        values: list[Any] = []
+        for _ in range(n):
+            values.append(await queue.get())
+        return values
+
+    async def _wait_for_any_of(self, condition: AnyOfCondition) -> Any:
+        if not condition.conditions:
+            raise ValueError("any_of() requires at least one condition")
+
+        tasks = [
+            asyncio.create_task(self.wait_for(inner_condition))
+            for inner_condition in condition.conditions
+        ]
+        done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+        for task in pending:
+            task.cancel()
+        await asyncio.gather(*pending, return_exceptions=True)
+        first = done.pop()
+        return first.result()
+
+    def _get_async_channel(self, channel: str) -> asyncio.Queue[Any]:
+        if channel not in self._async_channels:
+            raise ValueError(f"Unknown channel `{channel}`")
+        return self._async_channels[channel]
+
+    def _schedule(self, send: Send) -> None:
+        if self._finished:
+            return
+        task: asyncio.Task[list[Send]] = asyncio.create_task(self._execute_send(send))
+        self._tasks.add(task)
+
+    async def _cancel_all_tasks(self) -> None:
+        if not self._tasks:
+            return
+        to_cancel = list(self._tasks)
+        for task in to_cancel:
+            task.cancel()
+        await asyncio.gather(*to_cancel, return_exceptions=True)
+        self._tasks.clear()
+
+    async def _execute_send(self, send: Send) -> list[Send]:
+        node_name = _resolve_target_name(send.node)
+        if node_name not in self._nodes:
+            raise ValueError(f"Unknown node `{node_name}`")
+        node = self._nodes[node_name]
+
+        result = _invoke_node(node, self.context, send.arg)
+        if inspect.isawaitable(result):
+            result = await result
+
+        if isinstance(result, Command):
+            self._apply_update(result.update)
+            next_sends = _normalize_goto(result.goto, default_arg=self._state)
+        else:
+            self._apply_update(result)
+            next_sends = _normalize_result_to_sends(result, default_arg=self._state)
+
+        if node_name == self._finish_point:
+            self._finished = True
+            return []
+        return next_sends
+
+    def _apply_update(self, update: Any) -> None:
+        if update is None:
+            return
+        if isinstance(update, Mapping):
+            if isinstance(self._state, Mapping):
+                # Keep semantics simple: in-place update for mapping-like state.
+                cast(dict[str, Any], self._state).update(update)
+            return
+        if isinstance(update, Sequence) and not isinstance(update, (str, bytes)):
+            pairs = list(update)
+            if all(
+                isinstance(item, tuple) and len(item) == 2 and isinstance(item[0], str)
+                for item in pairs
+            ):
+                if isinstance(self._state, Mapping):
+                    cast(dict[str, Any], self._state).update(
+                        cast(dict[str, Any], pairs)
+                    )
+                return
+
+
+def _normalize_result_to_sends(result: Any, *, default_arg: Any) -> list[Send]:
+    if result is None:
+        return []
+    if isinstance(result, Send):
+        return [result]
+    if callable(result):
+        return [Send(_infer_node_name(result), default_arg)]
+    if isinstance(result, str):
+        return [Send(result, default_arg)]
+    if isinstance(result, Sequence) and not isinstance(result, (str, bytes)):
+        sends: list[Send] = []
+        for item in result:
+            if isinstance(item, Send):
+                sends.append(item)
+            elif callable(item):
+                sends.append(Send(_infer_node_name(item), default_arg))
+            elif isinstance(item, str):
+                sends.append(Send(item, default_arg))
+        return sends
+    return []
+
+
+def _normalize_goto(goto: Any, *, default_arg: Any) -> list[Send]:
+    if not goto:
+        return []
+    if isinstance(goto, Send):
+        return [goto]
+    if callable(goto):
+        return [Send(_infer_node_name(goto), default_arg)]
+    if isinstance(goto, str):
+        return [Send(goto, default_arg)]
+    if isinstance(goto, Sequence):
+        sends: list[Send] = []
+        for item in goto:
+            if isinstance(item, Send):
+                sends.append(item)
+            elif callable(item):
+                sends.append(Send(_infer_node_name(item), default_arg))
+            elif isinstance(item, str):
+                sends.append(Send(item, default_arg))
+        return sends
+    return []
+
+
+def channel_condition(channel: str, n: int = 1) -> ChannelCondition:
+    if n < 1:
+        raise ValueError("channel_condition `n` must be >= 1")
+    return ChannelCondition(channel=channel, n=n)
+
+
+def timer_condition(
+    timeout: float | timedelta | None = None,
+    *,
+    seconds: float | None = None,
+    minutes: float | None = None,
+) -> TimerCondition:
+    if timeout is not None and (seconds is not None or minutes is not None):
+        raise ValueError(
+            "Provide either `timeout` or named `seconds`/`minutes`, not both"
+        )
+
+    if isinstance(timeout, timedelta):
+        resolved_seconds = timeout.total_seconds()
+    elif isinstance(timeout, (int, float)):
+        resolved_seconds = float(timeout)
+    else:
+        resolved_seconds = 0.0
+        if seconds is not None:
+            resolved_seconds += float(seconds)
+        if minutes is not None:
+            resolved_seconds += float(minutes) * 60.0
+
+    if resolved_seconds <= 0:
+        raise ValueError("timer_condition must be greater than 0 seconds")
+    return TimerCondition(seconds=resolved_seconds)
+
+
+def any_of(*conditions: WaitCondition) -> AnyOfCondition:
+    if not conditions:
+        raise ValueError("any_of() requires at least one condition")
+    return AnyOfCondition(conditions=tuple(conditions))
+
+
+def _infer_node_name(node: Callable[..., Any]) -> str:
+    node_name = getattr(node, "__name__", "")
+    if not node_name or node_name == "<lambda>":
+        raise ValueError("Cannot infer node name from anonymous callable")
+    return node_name
+
+
+def _resolve_target_name(target: Any) -> str:
+    if isinstance(target, str):
+        return target
+    if callable(target):
+        return _infer_node_name(target)
+    raise ValueError(f"Unsupported node target type: {type(target)!r}")
+
+
+def _invoke_node(node: Callable[..., Any], ctx: Context, state: Any) -> Any:
+    try:
+        params = list(inspect.signature(node).parameters.values())
+    except (TypeError, ValueError):
+        params = []
+
+    if len(params) >= 2:
+        return node(ctx, state)
+    if len(params) == 1:
+        return node(state)
+    return node()

libs/langgraph/langgraph/graph/message.py

@@ -191,8 +191,14 @@ def add_messages(
     if not isinstance(right, list):
         right = [right]  # type: ignore[assignment]
     # coerce to message
-    left = [message_chunk_to_message(m) for m in convert_to_messages(left)]
-    right = [message_chunk_to_message(m) for m in convert_to_messages(right)]
+    left = [
+        message_chunk_to_message(cast(BaseMessageChunk, m))
+        for m in convert_to_messages(left)
+    ]
+    right = [
+        message_chunk_to_message(cast(BaseMessageChunk, m))
+        for m in convert_to_messages(right)
+    ]
     # assign missing ids
     for m in left:
         if m.id is None:

libs/langgraph/langgraph/pregel/_algo.py

@@ -220,7 +220,6 @@ def apply_writes(
     tasks: Iterable[WritesProtocol],
     get_next_version: GetNextVersion | None,
     trigger_to_nodes: Mapping[str, Sequence[str]],
-    available_channels: set[str] | None = None,
 ) -> set[str]:
     """Apply writes from a set of tasks (usually the tasks from a Pregel step)
     to the checkpoint and channels, and return managed values writes to be applied
@@ -267,18 +266,6 @@ def apply_writes(
             None,
         )
 
-    # Sync available_channels with channel's actual availability state.
-    # Returns True if the channel is available (for callers that also need
-    # to update updated_channels).
-    def _track(chan: str) -> bool:
-        avail = channels[chan].is_available()
-        if available_channels is not None:
-            if avail:
-                available_channels.add(chan)
-            else:
-                available_channels.discard(chan)
-        return avail
-
     # Consume all channels that were read
     for chan in {
         chan
@@ -288,7 +275,6 @@ def apply_writes(
     }:
         if channels[chan].consume() and next_version is not None:
             checkpoint["channel_versions"][chan] = next_version
-        _track(chan)
 
     # Group writes by channel
     pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)
@@ -310,28 +296,18 @@ def apply_writes(
             if channels[chan].update(vals) and next_version is not None:
                 checkpoint["channel_versions"][chan] = next_version
                 # unavailable channels can't trigger tasks, so don't add them
-                if _track(chan):
+                if channels[chan].is_available():
                     updated_channels.add(chan)
-            else:
-                _track(chan)
 
     # Channels that weren't updated in this step are notified of a new step
     if bump_step:
-        candidates = (
-            available_channels - updated_channels
-            if available_channels is not None
-            else (
-                chan
-                for chan in channels
-                if channels[chan].is_available() and chan not in updated_channels
-            )
-        )
-        for chan in candidates:
-            if channels[chan].update(EMPTY_SEQ) and next_version is not None:
-                checkpoint["channel_versions"][chan] = next_version
-                # unavailable channels can't trigger tasks, so don't add them
-                if _track(chan):
-                    updated_channels.add(chan)
+        for chan in channels:
+            if channels[chan].is_available() and chan not in updated_channels:
+                if channels[chan].update(EMPTY_SEQ) and next_version is not None:
+                    checkpoint["channel_versions"][chan] = next_version
+                    # unavailable channels can't trigger tasks, so don't add them
+                    if channels[chan].is_available():
+                        updated_channels.add(chan)
 
     # If this is (tentatively) the last superstep, notify all channels of finish
     if bump_step and updated_channels.isdisjoint(trigger_to_nodes):
@@ -339,10 +315,8 @@ def apply_writes(
             if channels[chan].finish() and next_version is not None:
                 checkpoint["channel_versions"][chan] = next_version
                 # unavailable channels can't trigger tasks, so don't add them
-                if _track(chan):
+                if channels[chan].is_available():
                     updated_channels.add(chan)
-            else:
-                _track(chan)
 
     # Return managed values writes to be applied externally
     return updated_channels
@@ -520,7 +494,7 @@ PUSH_TRIGGER = (PUSH,)
 
 
 class _TaskIDFn(Protocol):
-    def __call__(self, namespace: bytes, *parts: str) -> str:
+    def __call__(self, namespace: bytes, *parts: str | bytes) -> str:
         pass
 
 
@@ -1191,37 +1165,32 @@ def _proc_input(
     return val
 
 
-def _uuid5_str(namespace: bytes, *parts: str) -> str:
+def _uuid5_str(namespace: bytes, *parts: str | bytes) -> str:
     """Generate a UUID from the SHA-1 hash of a namespace and str parts."""
 
     sha = sha1(namespace, usedforsecurity=False)
-    sha.update(b"".join(p.encode() for p in parts))
+    sha.update(b"".join(p.encode() if isinstance(p, str) else p for p in parts))
     hex = sha.hexdigest()
     return f"{hex[:8]}-{hex[8:12]}-{hex[12:16]}-{hex[16:20]}-{hex[20:32]}"
 
 
-def _xxhash_str(namespace: bytes, *parts: str) -> str:
+def _xxhash_str(namespace: bytes, *parts: str | bytes) -> str:
     """Generate a UUID from the XXH3 hash of a namespace and str parts."""
-    hex = xxh3_128_hexdigest(namespace + b"".join(p.encode() for p in parts))
+    hex = xxh3_128_hexdigest(
+        namespace + b"".join(p.encode() if isinstance(p, str) else p for p in parts)
+    )
     return f"{hex[:8]}-{hex[8:12]}-{hex[12:16]}-{hex[16:20]}-{hex[20:32]}"
 
 
-def task_path_str(tup: str | int | tuple | list) -> str:
+def task_path_str(tup: str | int | tuple) -> str:
     """Generate a string representation of the task path."""
-    if isinstance(tup, (tuple, list)):
-        parts: list[str] = []
-        for x in tup:
-            if isinstance(x, int):
-                parts.append(f"{x:010d}")
-            elif isinstance(x, (tuple, list)):
-                parts.append(task_path_str(x))
-            else:
-                parts.append(str(x))
-        return f"~{', '.join(parts)}"
-    elif isinstance(tup, int):
-        return f"{tup:010d}"
-    else:
-        return str(tup)
+    return (
+        f"~{', '.join(task_path_str(x) for x in tup)}"
+        if isinstance(tup, (tuple, list))
+        else f"{tup:010d}"
+        if isinstance(tup, int)
+        else str(tup)
+    )
 
 
 LAZY_ATOMIC_COUNTER_LOCK = threading.Lock()

libs/langgraph/langgraph/pregel/_loop.py

@@ -42,6 +42,7 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_ID,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_NS,
+    CONFIG_KEY_REPLAY_STATE,
     CONFIG_KEY_RESUME_MAP,
     CONFIG_KEY_RESUMING,
     CONFIG_KEY_SCRATCHPAD,
@@ -58,6 +59,7 @@ from langgraph._internal._constants import (
     RESUME,
     TASKS,
 )
+from langgraph._internal._replay import ReplayState
 from langgraph._internal._scratchpad import PregelScratchpad
 from langgraph._internal._typing import EMPTY_SEQ, MISSING
 from langgraph.channels.base import BaseChannel
@@ -152,7 +154,7 @@ class PregelLoop:
     input_keys: str | Sequence[str]
     output_keys: str | Sequence[str]
     stream_keys: str | Sequence[str]
-    skip_done_tasks: bool
+    is_replaying: bool
     is_nested: bool
     manager: None | AsyncParentRunManager | ParentRunManager
     interrupt_after: All | Sequence[str]
@@ -180,7 +182,6 @@ class PregelLoop:
     _migrate_checkpoint: Callable[[Checkpoint], None] | None
     submit: Submit
     channels: Mapping[str, BaseChannel]
-    _available_channels: set[str]
     managed: ManagedValueMapping
     checkpoint: Checkpoint
     checkpoint_id_saved: str
@@ -245,7 +246,7 @@ class PregelLoop:
         self.interrupt_before = interrupt_before
         self.manager = manager
         self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {})
-        self.skip_done_tasks = CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
+        self.is_replaying = CONFIG_KEY_CHECKPOINT_ID in config[CONF]
         self._migrate_checkpoint = migrate_checkpoint
         self.trigger_to_nodes = trigger_to_nodes
         self.retry_policy = retry_policy
@@ -452,7 +453,7 @@ class PregelLoop:
             # save the new task
             self.tasks[pushed.id] = pushed
             # match any pending writes to the new task
-            if self.skip_done_tasks:
+            if not self.is_replaying:
                 self._match_writes({pushed.id: pushed})
             # return the new task, to be started if not run before
             return pushed
@@ -516,7 +517,7 @@ class PregelLoop:
             return False
 
         # if there are pending writes from a previous loop, apply them
-        if self.skip_done_tasks and self.checkpoint_pending_writes:
+        if not self.is_replaying and self.checkpoint_pending_writes:
             self._match_writes(self.tasks)
 
         # before execution, check if we should interrupt
@@ -546,7 +547,6 @@ class PregelLoop:
             self.tasks.values(),
             self.checkpointer_get_next_version,
             self.trigger_to_nodes,
-            available_channels=self._available_channels,
         )
         # produce values output
         if not self.updated_channels.isdisjoint(
@@ -559,8 +559,8 @@ class PregelLoop:
             )
         # clear pending writes
         self.checkpoint_pending_writes.clear()
-        # "not skip_done_tasks" only applies to first tick after resuming
-        self.skip_done_tasks = True
+        # only replay (re-execute) done tasks on the first tick
+        self.is_replaying = False
         # save checkpoint
         self._put_checkpoint({"source": "loop"})
         # after execution, check if we should interrupt
@@ -620,15 +620,21 @@ class PregelLoop:
     def _first(
         self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None
     ) -> set[str] | None:
-        # resuming from previous checkpoint requires
-        # - finding a previous checkpoint
-        # - receiving None input (outer graph) or RESUMING flag (subgraph)
+        # Resuming from a previous checkpoint requires two things:
+        # 1. A prior checkpoint exists (channel_versions is non-empty)
+        # 2. The input signals continuation (not a fresh run with new input)
+        # For subgraphs, the parent explicitly sets CONFIG_KEY_RESUMING.
+        # For the outer graph, we infer from the input:
+        #   - None input: resume after interrupt (invoke(None, config))
+        #   - Command input: any Command operates on existing state
+        #   - Same run_id: re-entry into an ongoing run (e.g. stream reconnect)
         configurable = self.config.get(CONF, {})
+        input_is_command = isinstance(self.input, Command)
         is_resuming = bool(self.checkpoint["channel_versions"]) and bool(
             configurable.get(
                 CONFIG_KEY_RESUMING,
                 self.input is None
-                or isinstance(self.input, Command)
+                or input_is_command
                 or (
                     not self.is_nested
                     and self.config.get("metadata", {}).get("run_id")
@@ -637,9 +643,25 @@ class PregelLoop:
             )
         )
 
+        # When replaying from a specific checkpoint, drop cached RESUME
+        # writes so that interrupt() calls re-fire instead of returning
+        # stale values. But if we're actively resuming, keep them —
+        # multi-interrupt scenarios need previously resolved values preserved.
+        # We check two conditions because resume signals arrive differently:
+        # - Command(resume=...): the outer graph receives resume via input
+        # - CONFIG_KEY_RESUMING: child subgraphs receive it via config from
+        #   the parent (their input is a Send arg, not a Command)
+        if self.is_replaying and not (
+            (input_is_command and cast(Command, self.input).resume is not None)
+            or configurable.get(CONFIG_KEY_RESUMING, False)
+        ):
+            self.checkpoint_pending_writes = [
+                w for w in self.checkpoint_pending_writes if w[1] != RESUME
+            ]
+
         # map command to writes
-        if isinstance(self.input, Command):
-            if (resume := self.input.resume) is not None:
+        if input_is_command:
+            if (resume := cast(Command, self.input).resume) is not None:
                 if not self.checkpointer:
                     raise RuntimeError(
                         "Cannot use Command(resume=...) without checkpointer"
@@ -659,7 +681,7 @@ class PregelLoop:
 
             writes: defaultdict[str, list[tuple[str, Any]]] = defaultdict(list)
             # group writes by task ID
-            for tid, c, v in map_command(cmd=self.input):
+            for tid, c, v in map_command(cmd=cast(Command, self.input)):
                 if not (c == RESUME and resume_is_map):
                     writes[tid].append((c, v))
             if not writes and not resume_is_map:
@@ -677,7 +699,6 @@ class PregelLoop:
                 [PregelTaskWrites((), INPUT, null_writes, [])],
                 self.checkpointer_get_next_version,
                 self.trigger_to_nodes,
-                available_channels=self._available_channels,
             )
             if updated_channels is not None:
                 updated_channels.update(null_updated_channels)
@@ -720,17 +741,36 @@ class PregelLoop:
                 ],
                 self.checkpointer_get_next_version,
                 self.trigger_to_nodes,
-                available_channels=self._available_channels,
             )
             # save input checkpoint
             self.updated_channels = updated_channels
             self._put_checkpoint({"source": "input"})
         elif CONFIG_KEY_RESUMING not in configurable:
             raise EmptyInputError(f"Received no input for {input_keys}")
-        # update config
+        # Propagate resuming and replaying flags to subgraphs.
         if not self.is_nested:
+            # Pass the resolved before-bound checkpoint ID so subgraphs can
+            # find their corresponding checkpoint without re-fetching the
+            # parent. For forks (source=update), use the fork's parent
+            # checkpoint ID since the fork was created after the subgraph's
+            # checkpoints from the original execution.
+            replay_state: ReplayState | None = None
+            if self.is_replaying:
+                replay_checkpoint_id = self.checkpoint["id"]
+                if (
+                    self.checkpoint_metadata.get("source") == "update"
+                    and self.prev_checkpoint_config
+                ):
+                    replay_checkpoint_id = self.prev_checkpoint_config[CONF].get(
+                        CONFIG_KEY_CHECKPOINT_ID, replay_checkpoint_id
+                    )
+                replay_state = ReplayState(replay_checkpoint_id)
             self.config = patch_configurable(
-                self.config, {CONFIG_KEY_RESUMING: is_resuming}
+                self.config,
+                {
+                    CONFIG_KEY_RESUMING: is_resuming,
+                    CONFIG_KEY_REPLAY_STATE: replay_state,
+                },
             )
         # set flag
         self.status = "pending"
@@ -848,7 +888,6 @@ class PregelLoop:
                     self.tasks.values(),
                     self.checkpointer_get_next_version,
                     self.trigger_to_nodes,
-                    available_channels=self._available_channels,
                 )
                 if not updated_channels.isdisjoint(
                     (self.output_keys,)
@@ -1086,10 +1125,27 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
     # context manager
 
     def __enter__(self) -> Self:
-        if self.checkpointer:
-            saved = self.checkpointer.get_tuple(self.checkpoint_config)
-        else:
+        if not self.checkpointer:
             saved = None
+        elif self.is_nested and (
+            replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE)
+        ):
+            saved = replay_state.get_checkpoint(
+                self.config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, ""),
+                self.checkpointer,
+                self.checkpoint_config,
+            )
+            # Clear RESUMING so _first re-applies input instead of resuming.
+            # This recreates ephemeral routing channels so nodes trigger
+            # naturally via version comparison.
+            self.config[CONF].pop(CONFIG_KEY_RESUMING, None)
+        else:
+            # Normal case: fetch the most recent checkpoint for this
+            # graph/thread. If a specific checkpoint_id is in the config,
+            # fetch that exact checkpoint; otherwise fetch the latest one.
+            # Returns None on first invocation (no checkpoints exist yet).
+            saved = self.checkpointer.get_tuple(self.checkpoint_config)
+
         if saved is None:
             saved = CheckpointTuple(
                 self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1114,14 +1170,10 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             if saved.pending_writes is not None
             else []
         )
-
         self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
         self.channels, self.managed = channels_from_checkpoint(
             self.specs, self.checkpoint
         )
-        self._available_channels: set[str] = {
-            k for k, v in self.channels.items() if v.is_available()
-        }
         self.stack.push(self._suppress_interrupt)
         self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1
@@ -1268,10 +1320,27 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
     # context manager
 
     async def __aenter__(self) -> Self:
-        if self.checkpointer:
-            saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
-        else:
+        if not self.checkpointer:
             saved = None
+        elif self.is_nested and (
+            replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE)
+        ):
+            saved = await replay_state.aget_checkpoint(
+                self.config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, ""),
+                self.checkpointer,
+                self.checkpoint_config,
+            )
+            # Clear RESUMING so _first re-applies input instead of resuming.
+            # This recreates ephemeral routing channels so nodes trigger
+            # naturally via version comparison.
+            self.config[CONF].pop(CONFIG_KEY_RESUMING, None)
+        else:
+            # Normal case: fetch the most recent checkpoint for this
+            # graph/thread. If a specific checkpoint_id is in the config,
+            # fetch that exact checkpoint; otherwise fetch the latest one.
+            # Returns None on first invocation (no checkpoints exist yet).
+            saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
+
         if saved is None:
             saved = CheckpointTuple(
                 self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1296,16 +1365,12 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             if saved.pending_writes is not None
             else []
         )
-
         self.submit = await self.stack.enter_async_context(
             AsyncBackgroundExecutor(self.config)
         )
         self.channels, self.managed = channels_from_checkpoint(
             self.specs, self.checkpoint
         )
-        self._available_channels: set[str] = {
-            k for k, v in self.channels.items() if v.is_available()
-        }
         self.stack.push(self._suppress_interrupt)
         self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/advanced-graph/test_sub_agents.py

@@ -0,0 +1,219 @@
+import asyncio
+from dataclasses import dataclass
+from typing import Any, Literal
+
+import pytest
+from typing_extensions import TypedDict
+
+from langgraph.advanced_graph import (
+    AdvancedStateGraph,
+    Context,
+    any_of,
+    channel_condition,
+    timer_condition,
+)
+from langgraph.constants import END, START
+from langgraph.graph import StateGraph
+from langgraph.types import Command, Send
+
+pytestmark = pytest.mark.anyio
+
+
+class MainAgentState(TypedDict):
+    input: str
+    output: list[str]
+    done: str | None
+
+
+class SubAgentState(TypedDict):
+    input: str
+    output: str
+
+
+@dataclass(frozen=True)
+class Decision:
+    type: Literal["end", "sub_agent", "tool"]
+    sub_agent: str | None = None
+    tool: str | None = None
+    complete: str | None = None
+
+
+class MockPlanner:
+    def __init__(self) -> None:
+        self.responses: list[list[Decision]] = []
+        self._idx = 0
+
+    async def ainvoke(self, _: MainAgentState) -> list[Decision]:
+        if self._idx >= len(self.responses):
+            return []
+        response = self.responses[self._idx]
+        self._idx += 1
+        return response
+
+
+def build_sub_agent() -> Any:
+    # Sub-agent uses the regular/simple StateGraph API.
+    sub_agent = StateGraph(SubAgentState)
+
+    async def research_node(state: SubAgentState) -> dict[str, str]:
+        # Intentionally slower than timer_condition(seconds=1) to validate timer path.
+        await asyncio.sleep(5)
+        return {"output": f"research sub agent completed for: {state['input']}"}
+
+    sub_agent.add_node("research_node", research_node)
+    sub_agent.add_edge(START, "research_node")
+    sub_agent.add_edge("research_node", END)
+    return sub_agent.compile()
+
+
+def build_main_agent(planner: MockPlanner, sub_agent: Any) -> Any:
+    async def llm_node(state: MainAgentState) -> Command:
+        # Planner decides whether to call a tool, spawn a sub-agent, or finish.
+        decisions = await planner.ainvoke(state)
+        sends: list[Send] = []
+        for decision in decisions:
+            if decision.type == "end":
+                # NOTE: this can be simplified further in the future with a dedicated
+                # complete primitive, instead of routing to a finish node manually.
+                return Command(
+                    goto=Send(
+                        order_food_node,
+                        {
+                            "state": state,
+                            "complete": decision.complete or "order flow completed",
+                        },
+                    )
+                )
+            if decision.type == "sub_agent" and decision.sub_agent:
+                sends.append(Send("sub_agent_node", decision.sub_agent))
+            if decision.type == "tool" and decision.tool:
+                sends.append(Send("tool_node", decision.tool))
+        # Keep the main loop responsive: wait for one inbound message and continue.
+        sends.append(Send("wait_node", state))
+        return Command(goto=sends)
+
+    async def wait_node(ctx: Context, state: MainAgentState) -> Command:
+        # Lightweight interrupt: only this node blocks for the next relevant signal.
+        event = await ctx.wait_for(
+            any_of(
+                channel_condition("tool_completion_channel"),
+                channel_condition("subagent_completion_channel"),
+                channel_condition("user_input_channel"),
+                timer_condition(seconds=1),
+            )
+        )
+        if event["condition"] == "channel":
+            channel = event["channel"]
+            payload = event["value"]
+            if channel == "tool_completion_channel":
+                state["output"].append(f"tool: {payload}")
+            elif channel == "subagent_completion_channel":
+                state["output"].append(f"sub_agent: {payload}")
+            elif channel == "user_input_channel":
+                state["output"].append(f"user_input: {payload}")
+            # State changed -> ask planner what to do next.
+            return Command(goto=Send("llm_node", state))
+        else:
+            state["output"].append("timer: no updates yet")
+            # No meaningful state change -> keep waiting without calling planner.
+            return Command(goto=Send("wait_node", state))
+
+    async def tool_node(ctx: Context, tool_input: str) -> None:
+        await asyncio.sleep(0.1)
+        # Fire-and-forget style completion: publish result to inbox and exit.
+        # (i.e., just complete without explicitly going to a next node)
+        ctx.publish_to_channel(
+            "tool_completion_channel",
+            f"tool completed for: {tool_input}",
+        )
+
+    async def sub_agent_node(ctx: Context, sub_agent_input: str) -> None:
+        # Sub-agent remains a regular StateGraph, compiled independently.
+        sub_agent_output = await sub_agent.ainvoke(
+            {"input": sub_agent_input, "output": ""}
+        )
+        # Same pattern as tool node: publish result and complete current node.
+        ctx.publish_to_channel(
+            "subagent_completion_channel",
+            sub_agent_output["output"],
+        )
+
+    async def order_food_node(payload: dict[str, Any]) -> dict[str, Any]:
+        state = payload["state"]
+        complete_message = payload["complete"]
+        return {
+            "done": complete_message,
+            "output": [*state["output"], f"order_food: {complete_message}"],
+        }
+
+    advanced_flow = AdvancedStateGraph(MainAgentState)
+    # Default behavior is an unbounded async channel like Rust channel
+    advanced_flow.add_async_channel("tool_completion_channel", str)
+    advanced_flow.add_async_channel("subagent_completion_channel", str)
+    advanced_flow.add_async_channel("user_input_channel", str)
+    # nodes are the same as in the regular StateGraph API
+    advanced_flow.add_entry_node(llm_node)
+    advanced_flow.add_node(wait_node)
+    advanced_flow.add_node(tool_node)
+    advanced_flow.add_node(sub_agent_node)
+    advanced_flow.add_finish_node(order_food_node)
+
+    return advanced_flow.compile()
+
+
+async def test_async_sub_graph() -> None:
+    planner = MockPlanner()
+    sub_agent = build_sub_agent()
+    main_agent = build_main_agent(planner, sub_agent)
+
+    planner.responses = [
+        [
+            # First planner pass triggers one slow sub-agent.
+            Decision(type="sub_agent", sub_agent="research lunch options"),
+            Decision(type="tool", tool="slack_tool"),
+        ],
+        # After user input.
+        [],
+        # After tool completion.
+        [],
+        # After first sub-agent completion, planner decides to run second research.
+        [Decision(type="sub_agent", sub_agent="find vegetarian fallback")],
+        # After second sub-agent completion, planner decides to end.
+        [Decision(type="end", complete="order submitted")],
+    ]
+
+    handler = await main_agent.astart(
+        {"input": "help me get something for lunch", "output": [], "done": None}
+    )
+
+    # External input can be injected while graph execution is in progress.
+    await asyncio.sleep(0.01)
+    await handler.apublish_to_channel("user_input_channel", "No spicy food please")
+    result = await handler.aresult()
+
+    assert result["input"] == "help me get something for lunch"
+    assert result["done"] == "order submitted"
+
+    output = result["output"]
+    assert output.count("timer: no updates yet") >= 3
+    assert "user_input: No spicy food please" in output
+    assert "tool: tool completed for: slack_tool" in output
+    assert (
+        "sub_agent: research sub agent completed for: research lunch options" in output
+    )
+    assert (
+        "sub_agent: research sub agent completed for: find vegetarian fallback" in output
+    )
+    assert output[-1] == "order_food: order submitted"
+
+    first_sub_idx = output.index(
+        "sub_agent: research sub agent completed for: research lunch options"
+    )
+    second_sub_idx = output.index(
+        "sub_agent: research sub agent completed for: find vegetarian fallback"
+    )
+    order_food_idx = output.index("order_food: order submitted")
+    assert first_sub_idx < second_sub_idx < order_food_idx
+    assert planner._idx == len(planner.responses)
+    import json
+    print(json.dumps(result, ensure_ascii=False, indent=2))

libs/langgraph/uv.lock

@@ -1367,7 +1367,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.0.10"
+version = "1.1.0"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1689,23 +1689,25 @@ name = "langgraph-cli"
 source = { editable = "../cli" }
 dependencies = [
     { name = "click", marker = "python_full_version < '3.14'" },
+    { name = "httpx", marker = "python_full_version < '3.14'" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11' and python_full_version < '3.14'" },
+    { name = "python-dotenv", marker = "python_full_version < '3.14'" },
 ]
 
 [package.optional-dependencies]
 inmem = [
     { name = "langgraph-api", marker = "python_full_version >= '3.11' and python_full_version < '3.14'" },
     { name = "langgraph-runtime-inmem", marker = "python_full_version >= '3.11' and python_full_version < '3.14'" },
-    { name = "python-dotenv", marker = "python_full_version < '3.14'" },
 ]
 
 [package.metadata]
 requires-dist = [
     { name = "click", specifier = ">=8.1.7" },
+    { name = "httpx", specifier = ">=0.24.0" },
     { name = "langgraph-api", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.5.35,<0.8.0" },
     { name = "langgraph-runtime-inmem", marker = "python_full_version >= '3.11' and extra == 'inmem'", specifier = ">=0.7" },
     { name = "langgraph-sdk", marker = "python_full_version >= '3.11'", specifier = ">=0.1.0" },
-    { name = "python-dotenv", marker = "extra == 'inmem'", specifier = ">=0.8.0" },
+    { name = "python-dotenv", specifier = ">=0.8.0" },
 ]
 provides-extras = ["inmem"]
 
@@ -1826,16 +1828,16 @@ dev = [
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },
     { name = "pytest-watch" },
-    { name = "ruff", specifier = "==0.15.1" },
+    { name = "ruff", specifier = "==0.15.5" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.17" },
+    { name = "ty", specifier = "==0.0.21" },
 ]
 lint = [
     { name = "codespell" },
     { name = "mypy", specifier = "==1.19.1" },
-    { name = "ruff", specifier = "==0.15.1" },
+    { name = "ruff", specifier = "==0.15.5" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.17" },
+    { name = "ty", specifier = "==0.0.21" },
 ]
 test = [
     { name = "pytest" },

libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py

@@ -638,18 +638,15 @@ def create_react_agent(
             messages = (
                 _get_state_value(state, "llm_input_messages")
             ) or _get_state_value(state, "messages")
+            error_msg = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
         else:
             messages = _get_state_value(state, "messages")
+            error_msg = (
+                f"Expected input to call_model to have 'messages' key, but got {state}"
+            )
 
         if messages is None:
-            if pre_model_hook is not None:
-                raise ValueError(
-                    f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
-                )
-            else:
-                raise ValueError(
-                    f"Expected input to call_model to have 'messages' key, but got {state}"
-                )
+            raise ValueError(error_msg)
 
         _validate_chat_history(messages)
         # we're passing messages under `messages` key, as this is expected by the prompt

libs/prebuilt/uv.lock

@@ -268,7 +268,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.0.10"
+version = "1.1.0"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -599,16 +599,16 @@ dev = [
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },
     { name = "pytest-watch" },
-    { name = "ruff", specifier = "==0.15.1" },
+    { name = "ruff", specifier = "==0.15.5" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.17" },
+    { name = "ty", specifier = "==0.0.21" },
 ]
 lint = [
     { name = "codespell" },
     { name = "mypy", specifier = "==1.19.1" },
-    { name = "ruff", specifier = "==0.15.1" },
+    { name = "ruff", specifier = "==0.15.5" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.17" },
+    { name = "ty", specifier = "==0.0.21" },
 ]
 test = [
     { name = "pytest" },

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.9"
+__version__ = "0.3.10"
 
 __all__ = ["Auth", "Encryption", "EncryptionContext", "get_client", "get_sync_client"]

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

@@ -464,7 +464,7 @@ class CronClient:
             ```
 
         """
-        payload = {
+        payload: dict[str, Any] = {
             "assistant_id": assistant_id,
             "thread_id": thread_id,
             "enabled": enabled,

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

@@ -134,7 +134,7 @@ class StoreClient:
                 raise ValueError(
                     f"Invalid namespace label '{label}'. Namespace labels cannot contain periods ('.')."
                 )
-        get_params = {"namespace": ".".join(namespace), "key": key}
+        get_params: dict[str, Any] = {"namespace": ".".join(namespace), "key": key}
         if refresh_ttl is not None:
             get_params["refresh_ttl"] = refresh_ttl
         if params:

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

@@ -451,7 +451,7 @@ class SyncCronClient:
             ]
             ```
         """
-        payload = {
+        payload: dict[str, Any] = {
             "assistant_id": assistant_id,
             "thread_id": thread_id,
             "enabled": enabled,

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

@@ -134,7 +134,7 @@ class SyncStoreClient:
                     f"Invalid namespace label '{label}'. Namespace labels cannot contain periods ('.')."
                 )
 
-        query_params = {"key": key, "namespace": ".".join(namespace)}
+        query_params: dict[str, Any] = {"key": key, "namespace": ".".join(namespace)}
         if refresh_ttl is not None:
             query_params["refresh_ttl"] = refresh_ttl
         if params:

libs/sdk-py/langgraph_sdk/cache.py

@@ -0,0 +1,62 @@
+"""Key/value cache for use inside LangGraph deployments.
+
+Thin wrapper around ``langgraph_api.cache``.
+Values must be JSON-serializable (dicts, lists, strings, numbers, booleans,
+``None``).
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from typing import Any
+
+try:
+    from langgraph_api.cache import (  # type: ignore[unresolved-import]
+        cache_get as _cache_get,
+    )
+    from langgraph_api.cache import (  # type: ignore[unresolved-import]
+        cache_set as _cache_set,
+    )
+except ImportError:
+    _cache_get = None
+    _cache_set = None
+
+
+__all__ = [
+    "cache_get",
+    "cache_set",
+]
+
+
+async def cache_get(key: str) -> Any | None:
+    """Get a value from the cache.
+
+    Returns the deserialized value, or ``None`` if the key is missing or expired.
+
+    Requires Agent Server runtime version 0.7.29 or later.
+    """
+    if _cache_get is None:
+        raise RuntimeError(
+            "Cache is only available server-side within the LangGraph Agent Server "
+            "(https://docs.langchain.com/langsmith/deployments)."
+        )
+    return await _cache_get(key)
+
+
+async def cache_set(key: str, value: Any, *, ttl: timedelta | None = None) -> None:
+    """Set a value in the cache.
+
+    Args:
+        key: The cache key.
+        value: The value to cache (must be JSON-serializable).
+        ttl: Optional time-to-live. Capped at 1 day; ``None`` or zero
+            defaults to 1 day.
+
+    Requires Agent Server runtime version 0.7.29 or later.
+    """
+    if _cache_set is None:
+        raise RuntimeError(
+            "Cache is only available server-side within the LangGraph Agent Server "
+            "(https://docs.langchain.com/langsmith/deployments)."
+        )
+    await _cache_set(key, value, ttl)

libs/sdk-py/pyproject.toml

@@ -30,10 +30,10 @@ test = [
     "pytest-watch",
 ]
 lint = [
-    "ruff==0.15.1",
+    "ruff==0.15.5",
     "codespell",
     "mypy==1.19.1",
-    "ty==0.0.17",
+    "ty==0.0.21",
     "starlette",
 ]
 dev = [

libs/sdk-py/uv.lock

@@ -134,11 +134,11 @@ wheels = [
 
 [[package]]
 name = "codespell"
-version = "2.4.1"
+version = "2.4.2"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/15/e0/709453393c0ea77d007d907dd436b3ee262e28b30995ea1aa36c6ffbccaf/codespell-2.4.1.tar.gz", hash = "sha256:299fcdcb09d23e81e35a671bbe746d5ad7e8385972e65dbb833a2eaac33c01e5", size = 344740, upload-time = "2025-01-28T18:52:39.411Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/2d/9d/1d0903dff693160f893ca6abcabad545088e7a2ee0a6deae7c24e958be69/codespell-2.4.2.tar.gz", hash = "sha256:3c33be9ae34543807f088aeb4832dfad8cb2dae38da61cac0a7045dd376cfdf3", size = 352058, upload-time = "2026-03-05T18:10:42.936Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/20/01/b394922252051e97aab231d416c86da3d8a6d781eeadcdca1082867de64e/codespell-2.4.1-py3-none-any.whl", hash = "sha256:3dadafa67df7e4a3dbf51e0d7315061b80d265f9552ebd699b3dd6834b47e425", size = 344501, upload-time = "2025-01-28T18:52:37.057Z" },
+    { url = "https://files.pythonhosted.org/packages/42/a1/52fa05533e95fe45bcc09bcf8a503874b1c08f221a4e35608017e0938f55/codespell-2.4.2-py3-none-any.whl", hash = "sha256:97e0c1060cf46bd1d5db89a936c98db8c2b804e1fdd4b5c645e82a1ec6b1f886", size = 353715, upload-time = "2026-03-05T18:10:41.398Z" },
 ]
 
 [[package]]
@@ -265,7 +265,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.0.10"
+version = "1.1.0"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -498,16 +498,16 @@ dev = [
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },
     { name = "pytest-watch" },
-    { name = "ruff", specifier = "==0.15.1" },
+    { name = "ruff", specifier = "==0.15.5" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.17" },
+    { name = "ty", specifier = "==0.0.21" },
 ]
 lint = [
     { name = "codespell" },
     { name = "mypy", specifier = "==1.19.1" },
-    { name = "ruff", specifier = "==0.15.1" },
+    { name = "ruff", specifier = "==0.15.5" },
     { name = "starlette" },
-    { name = "ty", specifier = "==0.0.17" },
+    { name = "ty", specifier = "==0.0.21" },
 ]
 test = [
     { name = "pytest" },
@@ -1119,27 +1119,27 @@ wheels = [
 
 [[package]]
 name = "ruff"
-version = "0.15.1"
+version = "0.15.5"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/04/dc/4e6ac71b511b141cf626357a3946679abeba4cf67bc7cc5a17920f31e10d/ruff-0.15.1.tar.gz", hash = "sha256:c590fe13fb57c97141ae975c03a1aedb3d3156030cabd740d6ff0b0d601e203f", size = 4540855, upload-time = "2026-02-12T23:09:09.998Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/77/9b/840e0039e65fcf12758adf684d2289024d6140cde9268cc59887dc55189c/ruff-0.15.5.tar.gz", hash = "sha256:7c3601d3b6d76dce18c5c824fc8d06f4eef33d6df0c21ec7799510cde0f159a2", size = 4574214, upload-time = "2026-03-05T20:06:34.946Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/23/bf/e6e4324238c17f9d9120a9d60aa99a7daaa21204c07fcd84e2ef03bb5fd1/ruff-0.15.1-py3-none-linux_armv6l.whl", hash = "sha256:b101ed7cf4615bda6ffe65bdb59f964e9f4a0d3f85cbf0e54f0ab76d7b90228a", size = 10367819, upload-time = "2026-02-12T23:09:03.598Z" },
-    { url = "https://files.pythonhosted.org/packages/b3/ea/c8f89d32e7912269d38c58f3649e453ac32c528f93bb7f4219258be2e7ed/ruff-0.15.1-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:939c995e9277e63ea632cc8d3fae17aa758526f49a9a850d2e7e758bfef46602", size = 10798618, upload-time = "2026-02-12T23:09:22.928Z" },
-    { url = "https://files.pythonhosted.org/packages/5e/0f/1d0d88bc862624247d82c20c10d4c0f6bb2f346559d8af281674cf327f15/ruff-0.15.1-py3-none-macosx_11_0_arm64.whl", hash = "sha256:1d83466455fdefe60b8d9c8df81d3c1bbb2115cede53549d3b522ce2bc703899", size = 10148518, upload-time = "2026-02-12T23:08:58.339Z" },
-    { url = "https://files.pythonhosted.org/packages/f5/c8/291c49cefaa4a9248e986256df2ade7add79388fe179e0691be06fae6f37/ruff-0.15.1-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a9457e3c3291024866222b96108ab2d8265b477e5b1534c7ddb1810904858d16", size = 10518811, upload-time = "2026-02-12T23:09:31.865Z" },
-    { url = "https://files.pythonhosted.org/packages/c3/1a/f5707440e5ae43ffa5365cac8bbb91e9665f4a883f560893829cf16a606b/ruff-0.15.1-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:92c92b003e9d4f7fbd33b1867bb15a1b785b1735069108dfc23821ba045b29bc", size = 10196169, upload-time = "2026-02-12T23:09:17.306Z" },
-    { url = "https://files.pythonhosted.org/packages/2a/ff/26ddc8c4da04c8fd3ee65a89c9fb99eaa5c30394269d424461467be2271f/ruff-0.15.1-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1fe5c41ab43e3a06778844c586251eb5a510f67125427625f9eb2b9526535779", size = 10990491, upload-time = "2026-02-12T23:09:25.503Z" },
-    { url = "https://files.pythonhosted.org/packages/fc/00/50920cb385b89413f7cdb4bb9bc8fc59c1b0f30028d8bccc294189a54955/ruff-0.15.1-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:66a6dd6df4d80dc382c6484f8ce1bcceb55c32e9f27a8b94c32f6c7331bf14fb", size = 11843280, upload-time = "2026-02-12T23:09:19.88Z" },
-    { url = "https://files.pythonhosted.org/packages/5d/6d/2f5cad8380caf5632a15460c323ae326f1e1a2b5b90a6ee7519017a017ca/ruff-0.15.1-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:6a4a42cbb8af0bda9bcd7606b064d7c0bc311a88d141d02f78920be6acb5aa83", size = 11274336, upload-time = "2026-02-12T23:09:14.907Z" },
-    { url = "https://files.pythonhosted.org/packages/a3/1d/5f56cae1d6c40b8a318513599b35ea4b075d7dc1cd1d04449578c29d1d75/ruff-0.15.1-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4ab064052c31dddada35079901592dfba2e05f5b1e43af3954aafcbc1096a5b2", size = 11137288, upload-time = "2026-02-12T23:09:07.475Z" },
-    { url = "https://files.pythonhosted.org/packages/cd/20/6f8d7d8f768c93b0382b33b9306b3b999918816da46537d5a61635514635/ruff-0.15.1-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:5631c940fe9fe91f817a4c2ea4e81f47bee3ca4aa646134a24374f3c19ad9454", size = 11070681, upload-time = "2026-02-12T23:08:55.43Z" },
-    { url = "https://files.pythonhosted.org/packages/9a/67/d640ac76069f64cdea59dba02af2e00b1fa30e2103c7f8d049c0cff4cafd/ruff-0.15.1-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:68138a4ba184b4691ccdc39f7795c66b3c68160c586519e7e8444cf5a53e1b4c", size = 10486401, upload-time = "2026-02-12T23:09:27.927Z" },
-    { url = "https://files.pythonhosted.org/packages/65/3d/e1429f64a3ff89297497916b88c32a5cc88eeca7e9c787072d0e7f1d3e1e/ruff-0.15.1-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:518f9af03bfc33c03bdb4cb63fabc935341bb7f54af500f92ac309ecfbba6330", size = 10197452, upload-time = "2026-02-12T23:09:12.147Z" },
-    { url = "https://files.pythonhosted.org/packages/78/83/e2c3bade17dad63bf1e1c2ffaf11490603b760be149e1419b07049b36ef2/ruff-0.15.1-py3-none-musllinux_1_2_i686.whl", hash = "sha256:da79f4d6a826caaea95de0237a67e33b81e6ec2e25fc7e1993a4015dffca7c61", size = 10693900, upload-time = "2026-02-12T23:09:34.418Z" },
-    { url = "https://files.pythonhosted.org/packages/a1/27/fdc0e11a813e6338e0706e8b39bb7a1d61ea5b36873b351acee7e524a72a/ruff-0.15.1-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:3dd86dccb83cd7d4dcfac303ffc277e6048600dfc22e38158afa208e8bf94a1f", size = 11227302, upload-time = "2026-02-12T23:09:36.536Z" },
-    { url = "https://files.pythonhosted.org/packages/f6/58/ac864a75067dcbd3b95be5ab4eb2b601d7fbc3d3d736a27e391a4f92a5c1/ruff-0.15.1-py3-none-win32.whl", hash = "sha256:660975d9cb49b5d5278b12b03bb9951d554543a90b74ed5d366b20e2c57c2098", size = 10462555, upload-time = "2026-02-12T23:09:29.899Z" },
-    { url = "https://files.pythonhosted.org/packages/e0/5e/d4ccc8a27ecdb78116feac4935dfc39d1304536f4296168f91ed3ec00cd2/ruff-0.15.1-py3-none-win_amd64.whl", hash = "sha256:c820fef9dd5d4172a6570e5721704a96c6679b80cf7be41659ed439653f62336", size = 11599956, upload-time = "2026-02-12T23:09:01.157Z" },
-    { url = "https://files.pythonhosted.org/packages/2a/07/5bda6a85b220c64c65686bc85bd0bbb23b29c62b3a9f9433fa55f17cda93/ruff-0.15.1-py3-none-win_arm64.whl", hash = "sha256:5ff7d5f0f88567850f45081fac8f4ec212be8d0b963e385c3f7d0d2eb4899416", size = 10874604, upload-time = "2026-02-12T23:09:05.515Z" },
+    { url = "https://files.pythonhosted.org/packages/47/20/5369c3ce21588c708bcbe517a8fbe1a8dfdb5dfd5137e14790b1da71612c/ruff-0.15.5-py3-none-linux_armv6l.whl", hash = "sha256:4ae44c42281f42e3b06b988e442d344a5b9b72450ff3c892e30d11b29a96a57c", size = 10478185, upload-time = "2026-03-05T20:06:29.093Z" },
+    { url = "https://files.pythonhosted.org/packages/44/ed/e81dd668547da281e5dce710cf0bc60193f8d3d43833e8241d006720e42b/ruff-0.15.5-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:6edd3792d408ebcf61adabc01822da687579a1a023f297618ac27a5b51ef0080", size = 10859201, upload-time = "2026-03-05T20:06:32.632Z" },
+    { url = "https://files.pythonhosted.org/packages/c4/8f/533075f00aaf19b07c5cd6aa6e5d89424b06b3b3f4583bfa9c640a079059/ruff-0.15.5-py3-none-macosx_11_0_arm64.whl", hash = "sha256:89f463f7c8205a9f8dea9d658d59eff49db05f88f89cc3047fb1a02d9f344010", size = 10184752, upload-time = "2026-03-05T20:06:40.312Z" },
+    { url = "https://files.pythonhosted.org/packages/66/0e/ba49e2c3fa0395b3152bad634c7432f7edfc509c133b8f4529053ff024fb/ruff-0.15.5-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:ba786a8295c6574c1116704cf0b9e6563de3432ac888d8f83685654fe528fd65", size = 10534857, upload-time = "2026-03-05T20:06:19.581Z" },
+    { url = "https://files.pythonhosted.org/packages/59/71/39234440f27a226475a0659561adb0d784b4d247dfe7f43ffc12dd02e288/ruff-0.15.5-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:fd4b801e57955fe9f02b31d20375ab3a5c4415f2e5105b79fb94cf2642c91440", size = 10309120, upload-time = "2026-03-05T20:06:00.435Z" },
+    { url = "https://files.pythonhosted.org/packages/f5/87/4140aa86a93df032156982b726f4952aaec4a883bb98cb6ef73c347da253/ruff-0.15.5-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:391f7c73388f3d8c11b794dbbc2959a5b5afe66642c142a6effa90b45f6f5204", size = 11047428, upload-time = "2026-03-05T20:05:51.867Z" },
+    { url = "https://files.pythonhosted.org/packages/5a/f7/4953e7e3287676f78fbe85e3a0ca414c5ca81237b7575bdadc00229ac240/ruff-0.15.5-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:8dc18f30302e379fe1e998548b0f5e9f4dff907f52f73ad6da419ea9c19d66c8", size = 11914251, upload-time = "2026-03-05T20:06:22.887Z" },
+    { url = "https://files.pythonhosted.org/packages/77/46/0f7c865c10cf896ccf5a939c3e84e1cfaeed608ff5249584799a74d33835/ruff-0.15.5-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:1cc6e7f90087e2d27f98dc34ed1b3ab7c8f0d273cc5431415454e22c0bd2a681", size = 11333801, upload-time = "2026-03-05T20:05:57.168Z" },
+    { url = "https://files.pythonhosted.org/packages/d3/01/a10fe54b653061585e655f5286c2662ebddb68831ed3eaebfb0eb08c0a16/ruff-0.15.5-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c1cb7169f53c1ddb06e71a9aebd7e98fc0fea936b39afb36d8e86d36ecc2636a", size = 11206821, upload-time = "2026-03-05T20:06:03.441Z" },
+    { url = "https://files.pythonhosted.org/packages/7a/0d/2132ceaf20c5e8699aa83da2706ecb5c5dcdf78b453f77edca7fb70f8a93/ruff-0.15.5-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:9b037924500a31ee17389b5c8c4d88874cc6ea8e42f12e9c61a3d754ff72f1ca", size = 11133326, upload-time = "2026-03-05T20:06:25.655Z" },
+    { url = "https://files.pythonhosted.org/packages/72/cb/2e5259a7eb2a0f87c08c0fe5bf5825a1e4b90883a52685524596bfc93072/ruff-0.15.5-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:65bb414e5b4eadd95a8c1e4804f6772bbe8995889f203a01f77ddf2d790929dd", size = 10510820, upload-time = "2026-03-05T20:06:37.79Z" },
+    { url = "https://files.pythonhosted.org/packages/ff/20/b67ce78f9e6c59ffbdb5b4503d0090e749b5f2d31b599b554698a80d861c/ruff-0.15.5-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:d20aa469ae3b57033519c559e9bc9cd9e782842e39be05b50e852c7c981fa01d", size = 10302395, upload-time = "2026-03-05T20:05:54.504Z" },
+    { url = "https://files.pythonhosted.org/packages/5f/e5/719f1acccd31b720d477751558ed74e9c88134adcc377e5e886af89d3072/ruff-0.15.5-py3-none-musllinux_1_2_i686.whl", hash = "sha256:15388dd28c9161cdb8eda68993533acc870aa4e646a0a277aa166de9ad5a8752", size = 10754069, upload-time = "2026-03-05T20:06:06.422Z" },
+    { url = "https://files.pythonhosted.org/packages/c3/9c/d1db14469e32d98f3ca27079dbd30b7b44dbb5317d06ab36718dee3baf03/ruff-0.15.5-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:b30da330cbd03bed0c21420b6b953158f60c74c54c5f4c1dabbdf3a57bf355d2", size = 11304315, upload-time = "2026-03-05T20:06:10.867Z" },
+    { url = "https://files.pythonhosted.org/packages/28/3a/950367aee7c69027f4f422059227b290ed780366b6aecee5de5039d50fa8/ruff-0.15.5-py3-none-win32.whl", hash = "sha256:732e5ee1f98ba5b3679029989a06ca39a950cced52143a0ea82a2102cb592b74", size = 10551676, upload-time = "2026-03-05T20:06:13.705Z" },
+    { url = "https://files.pythonhosted.org/packages/b8/00/bf077a505b4e649bdd3c47ff8ec967735ce2544c8e4a43aba42ee9bf935d/ruff-0.15.5-py3-none-win_amd64.whl", hash = "sha256:821d41c5fa9e19117616c35eaa3f4b75046ec76c65e7ae20a333e9a8696bc7fe", size = 11678972, upload-time = "2026-03-05T20:06:45.379Z" },
+    { url = "https://files.pythonhosted.org/packages/fe/4e/cd76eca6db6115604b7626668e891c9dd03330384082e33662fb0f113614/ruff-0.15.5-py3-none-win_arm64.whl", hash = "sha256:b498d1c60d2fe5c10c45ec3f698901065772730b411f164ae270bb6bfcc4740b", size = 10965572, upload-time = "2026-03-05T20:06:16.984Z" },
 ]
 
 [[package]]
@@ -1220,26 +1220,26 @@ wheels = [
 
 [[package]]
 name = "ty"
-version = "0.0.17"
+version = "0.0.21"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/66/c3/41ae6346443eedb65b96761abfab890a48ce2aa5a8a27af69c5c5d99064d/ty-0.0.17.tar.gz", hash = "sha256:847ed6c120913e280bf9b54d8eaa7a1049708acb8824ad234e71498e8ad09f97", size = 5167209, upload-time = "2026-02-13T13:26:36.835Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/ee/20/2ba8fd9493c89c41dfe9dbb73bc70a28b28028463bc0d2897ba8be36230a/ty-0.0.21.tar.gz", hash = "sha256:a4c2ba5d67d64df8fcdefd8b280ac1149d24a73dbda82fa953a0dff9d21400ed", size = 5297967, upload-time = "2026-03-06T01:57:13.809Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/c0/01/0ef15c22a1c54b0f728ceff3f62d478dbf8b0dcf8ff7b80b954f79584f3e/ty-0.0.17-py3-none-linux_armv6l.whl", hash = "sha256:64a9a16555cc8867d35c2647c2f1afbd3cae55f68fd95283a574d1bb04fe93e0", size = 10192793, upload-time = "2026-02-13T13:27:13.943Z" },
-    { url = "https://files.pythonhosted.org/packages/0f/2c/f4c322d9cded56edc016b1092c14b95cf58c8a33b4787316ea752bb9418e/ty-0.0.17-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:eb2dbd8acd5c5a55f4af0d479523e7c7265a88542efe73ed3d696eb1ba7b6454", size = 10051977, upload-time = "2026-02-13T13:26:57.741Z" },
-    { url = "https://files.pythonhosted.org/packages/4c/a5/43746c1ff81e784f5fc303afc61fe5bcd85d0fcf3ef65cb2cef78c7486c7/ty-0.0.17-py3-none-macosx_11_0_arm64.whl", hash = "sha256:f18f5fd927bc628deb9ea2df40f06b5f79c5ccf355db732025a3e8e7152801f6", size = 9564639, upload-time = "2026-02-13T13:26:42.781Z" },
-    { url = "https://files.pythonhosted.org/packages/d6/b8/280b04e14a9c0474af574f929fba2398b5e1c123c1e7735893b4cd73d13c/ty-0.0.17-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:5383814d1d7a5cc53b3b07661856bab04bb2aac7a677c8d33c55169acdaa83df", size = 10061204, upload-time = "2026-02-13T13:27:00.152Z" },
-    { url = "https://files.pythonhosted.org/packages/2a/d7/493e1607d8dfe48288d8a768a2adc38ee27ef50e57f0af41ff273987cda0/ty-0.0.17-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:9c20423b8744b484f93e7bf2ef8a9724bca2657873593f9f41d08bd9f83444c9", size = 10013116, upload-time = "2026-02-13T13:26:34.543Z" },
-    { url = "https://files.pythonhosted.org/packages/80/ef/22f3ed401520afac90dbdf1f9b8b7755d85b0d5c35c1cb35cf5bd11b59c2/ty-0.0.17-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:e6f5b1aba97db9af86517b911674b02f5bc310750485dc47603a105bd0e83ddd", size = 10533623, upload-time = "2026-02-13T13:26:31.449Z" },
-    { url = "https://files.pythonhosted.org/packages/75/ce/744b15279a11ac7138832e3a55595706b4a8a209c9f878e3ab8e571d9032/ty-0.0.17-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:488bce1a9bea80b851a97cd34c4d2ffcd69593d6c3f54a72ae02e5c6e47f3d0c", size = 11069750, upload-time = "2026-02-13T13:26:48.638Z" },
-    { url = "https://files.pythonhosted.org/packages/f2/be/1133c91f15a0e00d466c24f80df486d630d95d1b2af63296941f7473812f/ty-0.0.17-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:8df66b91ec84239420985ec215e7f7549bfda2ac036a3b3c065f119d1c06825a", size = 10870862, upload-time = "2026-02-13T13:26:54.715Z" },
-    { url = "https://files.pythonhosted.org/packages/3e/4a/a2ed209ef215b62b2d3246e07e833081e07d913adf7e0448fc204be443d6/ty-0.0.17-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:002139e807c53002790dfefe6e2f45ab0e04012e76db3d7c8286f96ec121af8f", size = 10628118, upload-time = "2026-02-13T13:26:45.439Z" },
-    { url = "https://files.pythonhosted.org/packages/b3/0c/87476004cb5228e9719b98afffad82c3ef1f84334bde8527bcacba7b18cb/ty-0.0.17-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:6c4e01f05ce82e5d489ab3900ca0899a56c4ccb52659453780c83e5b19e2b64c", size = 10038185, upload-time = "2026-02-13T13:27:02.693Z" },
-    { url = "https://files.pythonhosted.org/packages/46/4b/98f0b3ba9aef53c1f0305519536967a4aa793a69ed72677b0a625c5313ac/ty-0.0.17-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:2b226dd1e99c0d2152d218c7e440150d1a47ce3c431871f0efa073bbf899e881", size = 10047644, upload-time = "2026-02-13T13:27:05.474Z" },
-    { url = "https://files.pythonhosted.org/packages/93/e0/06737bb80aa1a9103b8651d2eb691a7e53f1ed54111152be25f4a02745db/ty-0.0.17-py3-none-musllinux_1_2_i686.whl", hash = "sha256:8b11f1da7859e0ad69e84b3c5ef9a7b055ceed376a432fad44231bdfc48061c2", size = 10231140, upload-time = "2026-02-13T13:27:10.844Z" },
-    { url = "https://files.pythonhosted.org/packages/7c/79/e2a606bd8852383ba9abfdd578f4a227bd18504145381a10a5f886b4e751/ty-0.0.17-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:c04e196809ff570559054d3e011425fd7c04161529eb551b3625654e5f2434cb", size = 10718344, upload-time = "2026-02-13T13:26:51.66Z" },
-    { url = "https://files.pythonhosted.org/packages/c5/2d/2663984ac11de6d78f74432b8b14ba64d170b45194312852b7543cf7fd56/ty-0.0.17-py3-none-win32.whl", hash = "sha256:305b6ed150b2740d00a817b193373d21f0767e10f94ac47abfc3b2e5a5aec809", size = 9672932, upload-time = "2026-02-13T13:27:08.522Z" },
-    { url = "https://files.pythonhosted.org/packages/de/b5/39be78f30b31ee9f5a585969930c7248354db90494ff5e3d0756560fb731/ty-0.0.17-py3-none-win_amd64.whl", hash = "sha256:531828267527aee7a63e972f54e5eee21d9281b72baf18e5c2850c6b862add83", size = 10542138, upload-time = "2026-02-13T13:27:17.084Z" },
-    { url = "https://files.pythonhosted.org/packages/40/b7/f875c729c5d0079640c75bad2c7e5d43edc90f16ba242f28a11966df8f65/ty-0.0.17-py3-none-win_arm64.whl", hash = "sha256:de9810234c0c8d75073457e10a84825b9cd72e6629826b7f01c7a0b266ae25b1", size = 10023068, upload-time = "2026-02-13T13:26:39.637Z" },
+    { url = "https://files.pythonhosted.org/packages/36/70/edf38bb37517531681d1c37f5df64744e5ad02673c02eb48447eae4bea08/ty-0.0.21-py3-none-linux_armv6l.whl", hash = "sha256:7bdf2f572378de78e1f388d24691c89db51b7caf07cf90f2bfcc1d6b18b70a76", size = 10299222, upload-time = "2026-03-06T01:57:16.64Z" },
+    { url = "https://files.pythonhosted.org/packages/72/62/0047b0bd19afeefbc7286f20a5f78a2aa39f92b4d89853f0d7185ab89edc/ty-0.0.21-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:7e9613994610431ab8625025bd2880dbcb77c5c9fabdd21134cda12d840a529d", size = 10130513, upload-time = "2026-03-06T01:57:29.93Z" },
+    { url = "https://files.pythonhosted.org/packages/a2/20/0b93a9e91aaed23155780258cdfdb4726ef68b6985378ac069bc427291a0/ty-0.0.21-py3-none-macosx_11_0_arm64.whl", hash = "sha256:56d3b198b64dd0a19b2b66e257deaed2ecea568e722ae5352f3c6fb62027f89d", size = 9605425, upload-time = "2026-03-06T01:57:27.115Z" },
+    { url = "https://files.pythonhosted.org/packages/ea/fd/9945e2fa2996a1287b1e1d7ce050e97e1f420233b271e770934bfa0880a0/ty-0.0.21-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:d23d2c34f7a77d974bb08f0860ef700addc8a683d81a0319f71c08f87506cfd0", size = 10108298, upload-time = "2026-03-06T01:57:35.429Z" },
+    { url = "https://files.pythonhosted.org/packages/52/e7/4ec52fcb15f3200826c9f048472c062549a05b0d1ef0b51f32d527b513c4/ty-0.0.21-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:56b01fd2519637a4ca88344f61c96225f540c98ff18bca321d4eaa7bb0f7aa2f", size = 10121556, upload-time = "2026-03-06T01:57:03.242Z" },
+    { url = "https://files.pythonhosted.org/packages/ee/c0/ad457be2a8abea0f25549598bd098554540ced66229488daa0d558dad3c8/ty-0.0.21-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:e9de7e11c63c6afc40f3e9ba716374add171aee7fabc70b5146a510705c6d41b", size = 10603264, upload-time = "2026-03-06T01:56:52.134Z" },
+    { url = "https://files.pythonhosted.org/packages/f8/5b/2ecc7a2175243a4bcb72f5298ae41feabbb93b764bb0dc45722f3752c2c2/ty-0.0.21-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:62f7f5b235c4f7876db305c36997aea07b7af29b1a068f373d0e2547e25f32ff", size = 11196428, upload-time = "2026-03-06T01:57:32.94Z" },
+    { url = "https://files.pythonhosted.org/packages/37/f5/aff507d6a901f328ef96a298032b0c11aaaf950a146ed7dd3b5bf2cd3acf/ty-0.0.21-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:ee8399f7c453a425291e6688efe430cfae7ab0ac4ffd50eba9f872bf878b54f6", size = 10866355, upload-time = "2026-03-06T01:56:57.831Z" },
+    { url = "https://files.pythonhosted.org/packages/be/30/822bbcb92d55b65989aa7ed06d9585f28ade9c9447369194ed4b0fb3b5b9/ty-0.0.21-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:210e7568c9f886c4d01308d751949ee714ad7ad9d7d928d2ba90d329dd880367", size = 10738177, upload-time = "2026-03-06T01:57:11.256Z" },
+    { url = "https://files.pythonhosted.org/packages/57/cc/46e7991b6469e93ac2c7e533a028983e402485580150ac864c56352a3a82/ty-0.0.21-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:53508e345b11569f78b21ba8e2b4e61df38a9754947fb3cd9f2ef574367338fb", size = 10079158, upload-time = "2026-03-06T01:57:00.516Z" },
+    { url = "https://files.pythonhosted.org/packages/15/c2/0bbdadfbd008240f8f1a87dc877433cb3884436097926107ccf06e618199/ty-0.0.21-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:553e43571f4a35604c36cfd07d8b61a5eb7a714e3c67f8c4ff2cf674fefbaef9", size = 10150535, upload-time = "2026-03-06T01:57:08.815Z" },
+    { url = "https://files.pythonhosted.org/packages/c5/b5/2dbdb7b57b5362200ef0a39738ebd31331726328336def0143ac097ee59d/ty-0.0.21-py3-none-musllinux_1_2_i686.whl", hash = "sha256:666f6822e3b9200abfa7e95eb0ddd576460adb8d66b550c0ad2c70abc84a2048", size = 10319803, upload-time = "2026-03-06T01:57:19.106Z" },
+    { url = "https://files.pythonhosted.org/packages/72/84/70e52c0b7abc7c2086f9876ef454a73b161d3125315536d8d7e911c94ca4/ty-0.0.21-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:a0854d008347ce4a5fb351af132f660a390ab2a1163444d075251d43e6f74b9b", size = 10826239, upload-time = "2026-03-06T01:57:21.727Z" },
+    { url = "https://files.pythonhosted.org/packages/a1/8a/1f72480fd013bbc6cd1929002abbbcde9a0b08ead6a15154de9d7f7fa37e/ty-0.0.21-py3-none-win32.whl", hash = "sha256:bef3ab4c7b966bcc276a8ac6c11b63ba222d21355b48d471ea782c4104eee4e0", size = 9693196, upload-time = "2026-03-06T01:57:24.126Z" },
+    { url = "https://files.pythonhosted.org/packages/8d/f8/1104808b875c26c640e536945753a78562d606bef4e241d9dbf3d92477f6/ty-0.0.21-py3-none-win_amd64.whl", hash = "sha256:a709d576e5bea84b745d43058d8b9cd4f27f74a0b24acb4b0cbb7d3d41e0d050", size = 10668660, upload-time = "2026-03-06T01:56:55.06Z" },
+    { url = "https://files.pythonhosted.org/packages/1b/b8/25e0adc404bbf986977657b25318991f93097b49f8aea640d93c0b0db68e/ty-0.0.21-py3-none-win_arm64.whl", hash = "sha256:f72047996598ac20553fb7e21ba5741e3c82dee4e9eadf10d954551a5fe09391", size = 10104161, upload-time = "2026-03-06T01:57:06.072Z" },
 ]
 
 [[package]]