done

libs/cli/langgraph_cli/cli.py

@@ -13,7 +13,6 @@ import tempfile
 import time
 from collections.abc import Callable, Sequence
 from contextlib import contextmanager
-from datetime import datetime, timezone
 
 import click
 import click.exceptions
@@ -27,11 +26,10 @@ 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.helpers import format_log_entry, level_fg, resolve_deployment_id
 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 format_deployments_table, warn_non_wolfi_distro
+from langgraph_cli.util import warn_non_wolfi_distro
 from langgraph_cli.version import __version__
 
 RESERVED_ENV_VARS = frozenset(
@@ -289,33 +287,6 @@ OPT_API_VERSION = click.option(
     help="API server version to use for the base image. If unspecified, the latest version will be used.",
 )
 
-OPT_HOST_API_KEY = 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."
-    ),
-)
-
-
-OPT_HOST_DEPLOYMENT_NAME = click.option(
-    "--name",
-    envvar=_DEPLOYMENT_NAME_ENV,
-    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."
-    ),
-)
-
-OPT_HOST_URL = click.option(
-    "--host-url",
-    envvar="LANGGRAPH_HOST_URL",
-    default="https://api.host.langchain.com",
-    hidden=True,
-)
-
 OPT_ENGINE_RUNTIME_MODE = click.option(
     "--engine-runtime-mode",
     type=click.Choice(["combined_queue_worker", "distributed"]),
@@ -324,67 +295,7 @@ OPT_ENGINE_RUNTIME_MODE = click.option(
 )
 
 
-class NestedHelpGroup(click.Group):
-    """Click group that shows one level of nested subcommands in top-level help."""
-
-    def format_commands(
-        self, ctx: click.Context, formatter: click.HelpFormatter
-    ) -> None:
-        command_entries: list[tuple[str, click.Command]] = []
-        # Collect the top-level commands first, then append one level of nested
-        # subcommands using names like "deploy list" so they show up in the
-        # top-level help output.
-        for command_name in self.list_commands(ctx):
-            command = self.get_command(ctx, command_name)
-            if command is None or command.hidden:
-                continue
-            command_entries.append((command_name, command))
-            if isinstance(command, click.Group):
-                # Build a child context so Click resolves the subcommands the same
-                # way it would for the nested group itself.
-                sub_ctx = click.Context(command, info_name=command_name, parent=ctx)
-                for subcommand_name in command.list_commands(sub_ctx):
-                    subcommand = command.get_command(sub_ctx, subcommand_name)
-                    if subcommand is None or subcommand.hidden:
-                        continue
-                    command_entries.append(
-                        (f"{command_name} {subcommand_name}", subcommand)
-                    )
-
-        # Compute the available width for help text up front so we can truncate
-        # descriptions before handing them to Click. That keeps each command on
-        # a single line instead of allowing wrapped descriptions.
-        command_width = max((len(name) for name, _ in command_entries), default=0)
-        help_width = max(formatter.width - command_width - 6, 10)
-        rows = [
-            (name, command.get_short_help_str(help_width))
-            for name, command in command_entries
-        ]
-
-        if rows:
-            # Render the flattened command list using Click's standard
-            # definition-list formatter so alignment stays consistent with the
-            # rest of the CLI help output.
-            with formatter.section("Commands"):
-                formatter.write_dl(rows)
-
-
-class DeployGroup(NestedHelpGroup):
-    """Group that treats leading '-' args as passthrough docker flags."""
-
-    def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
-        result = super().parse_args(ctx, args)
-        if ctx._protected_args and ctx._protected_args[0].startswith("-"):
-            # Click stores the would-be subcommand in _protected_args; if it looks
-            # like an option (e.g. --build-arg) treat it as passthrough docker
-            # args instead of insisting on a nested command.
-            ctx.args = [*ctx._protected_args, *ctx.args]
-            ctx._protected_args = []
-            return ctx.args
-        return result
-
-
-@click.group(cls=NestedHelpGroup)
+@click.group()
 @click.version_option(version=__version__, prog_name="LangGraph CLI")
 def cli():
     pass
@@ -682,109 +593,84 @@ def build(
         )
 
 
-def _deploy_base_options(
-    func: Callable | None = None,
-    *,
-    include_docker_args: bool = True,
-    validate_config_path: bool = True,
-):
-    """Apply shared deploy flags.
-
-    The group shares most options but should not consume subcommands, so the
-    docker build args are only attached when requested.
-    """
-
-    def _apply(target: Callable) -> Callable:
-        decorators = [
-            OPT_HOST_API_KEY,
-            OPT_HOST_DEPLOYMENT_NAME,
-            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,
-            OPT_HOST_URL,
-            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=validate_config_path,
-                    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),
-        ]
-        if include_docker_args:
-            # Only attach build args to the default command; on the group they
-            # would capture subcommand names like `list` before Click resolves
-            # them, making those subcommands unreachable.
-            decorators.append(
-                click.argument("docker_build_args", nargs=-1, type=click.UNPROCESSED)
-            )
-        for decorator in reversed(decorators):
-            target = decorator(target)
-        return target
-
-    return _apply(func) if func is not None else _apply
-
-
-@cli.group(
-    cls=DeployGroup,
+@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, "
-        "--config, --pull, etc.). See 'langgraph build --help' for details."
+        "--pull, etc.). See 'langgraph build --help' for details."
     ),
-    context_settings=dict(ignore_unknown_options=True, allow_extra_args=True),
-    invoke_without_command=True,  # allow `deploy` click group to execute without command
+    context_settings=dict(ignore_unknown_options=True),
 )
-@_deploy_base_options(include_docker_args=False, validate_config_path=False)
-@click.pass_context
 @log_command
-def deploy(ctx: click.Context, **_: object):
-    # We register deploy as both a group and a command here.
-    # if we detect no subcommand, we run _deploy (basically run langgraph deploy as a top level command)
-    # otherwise, we return None here and click will proceed to actually run the subcommand (list or delete)
-    if ctx.invoked_subcommand is not None:
-        return
-    docker_build_args = tuple(ctx.args)
-    ctx.args = []  # Prevent Click from re-processing passthrough args later.
-    return ctx.forward(_deploy, docker_build_args=docker_build_args)
-
-
-@_deploy_base_options()
-@click.command(context_settings=dict(ignore_unknown_options=True))
-def _deploy(
+def deploy(
     config: pathlib.Path,
     pull: bool,
     verbose: bool,
@@ -812,6 +698,15 @@ def _deploy(
 
     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:
@@ -848,21 +743,55 @@ def _deploy(
         def log_step(message: str) -> None:
             click.secho(message, fg="cyan")
 
-        client = _create_host_backend_client(host_url, api_key, env_vars=env_vars)
+        client = HostBackendClient(host_url, api_key)
         step = 1
         needs_creation = False
 
         if deployment_id:
             log_step(f"{step}. Using deployment {deployment_id}")
-            _call_host_backend_with_optional_tenant(
-                client, lambda c: c.get_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}'")
-            existing = _call_host_backend_with_optional_tenant(
-                client, lambda c: c.list_deployments(name_contains=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", []):
@@ -1121,122 +1050,6 @@ def _deploy(
                 )
 
 
-def _create_host_backend_client(
-    host_url: str | None,
-    api_key: str | None,
-    env_vars: dict[str, str] | None = None,
-) -> HostBackendClient:
-    if env_vars is None:
-        env_vars = _parse_env_from_config({}, pathlib.Path.cwd() / DEFAULT_CONFIG)
-    resolved_api_key = api_key
-    if not resolved_api_key:
-        for key_name in _API_KEY_ENV_NAMES:
-            val = env_vars.get(key_name)
-            if val:
-                resolved_api_key = val
-                break
-            val = os.environ.get(key_name)
-            if val:
-                resolved_api_key = val
-                break
-    if not resolved_api_key:
-        click.secho(
-            "No LangSmith API key found. Create one at Settings > API Keys in LangSmith.",
-            fg="yellow",
-        )
-        resolved_api_key = click.prompt("Enter LangSmith API key", hide_input=True)
-    return HostBackendClient(host_url, resolved_api_key)
-
-
-def _call_host_backend_with_optional_tenant(
-    client: HostBackendClient,
-    operation: Callable[[HostBackendClient], object],
-) -> object:
-    """Run *operation*, prompting for a workspace ID on org-scoped 403s.
-
-    On success the original *client* is returned as-is.  If the user is
-    prompted for a workspace ID, the tenant header is set on *client*
-    in-place so all subsequent calls through the same instance are
-    tenant-aware.
-    """
-    try:
-        return operation(client)
-    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._client.headers["X-Tenant-ID"] = tenant_id
-            return operation(client)
-        raise
-
-
-@OPT_HOST_API_KEY
-@OPT_HOST_URL
-@click.option(
-    "--name-contains",
-    default="",
-    help="Only show deployments whose names contain this value.",
-)
-@deploy.command("list", help="[Beta] List LangSmith Deployments.")
-def deploy_list(api_key: str | None, host_url: str | None, name_contains: str) -> None:
-    client = _create_host_backend_client(host_url, api_key)
-    response = _call_host_backend_with_optional_tenant(
-        client,
-        lambda c: c.list_deployments(name_contains=name_contains),
-    )
-    resources = response.get("resources", []) if isinstance(response, dict) else []
-    deployments = [item for item in resources if isinstance(item, dict)]
-    if not deployments:
-        click.echo("No deployments found.")
-        return
-    click.echo(format_deployments_table(deployments))
-
-
-@OPT_HOST_API_KEY
-@OPT_HOST_URL
-@click.option(
-    "--force",
-    is_flag=True,
-    default=False,
-    help="Delete without prompting for confirmation.",
-)
-@click.argument("deployment_id")
-@deploy.command(
-    "delete",
-    help=(
-        "[Beta] Delete a LangSmith Deployment.\n\n"
-        "Use the `deploy list` command to list deployment IDs."
-    ),
-)
-def deploy_delete(
-    api_key: str | None, host_url: str | None, force: bool, deployment_id: str
-) -> None:
-    if not force:
-        response = click.prompt(
-            click.style(
-                f"Are you sure you want to delete deployment ID {deployment_id}? (Y/n)",
-                fg="yellow",
-            ),
-            default="Y",
-            show_default=False,
-        )
-        if response.strip().lower() not in {"y", "yes"}:
-            raise click.Abort()
-    client = _create_host_backend_client(host_url, api_key)
-    _call_host_backend_with_optional_tenant(
-        client,
-        lambda c: c.delete_deployment(deployment_id),
-    )
-    click.secho(f"Deleted deployment {deployment_id}.", fg="green")
-
-
 def _normalize_image_name(value: str | None) -> str:
     """Sanitize a deployment/directory name into a valid Docker repository name.
 
@@ -1263,180 +1076,6 @@ def _normalize_image_tag(value: str) -> str:
     return value
 
 
-@OPT_HOST_API_KEY
-@OPT_HOST_DEPLOYMENT_NAME
-@click.option(
-    "--deployment-id",
-    help="Deployment ID. If omitted, --name is used to find the deployment.",
-)
-@click.option(
-    "--type",
-    "log_type",
-    type=click.Choice(["deploy", "build"]),
-    default="deploy",
-    show_default=True,
-    help=(
-        "Log stream to fetch: 'deploy' shows agent server runtime logs; "
-        "'build' shows build logs (for deployments built remotely)."
-    ),
-)
-@click.option(
-    "--revision-id",
-    help="Specific revision ID. For build logs, defaults to latest revision.",
-)
-@click.option(
-    "--level",
-    type=click.Choice(
-        ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], case_sensitive=False
-    ),
-    help="Filter by log level.",
-)
-@click.option(
-    "--limit",
-    type=int,
-    default=100,
-    show_default=True,
-    help="Max log entries to fetch.",
-)
-@click.option(
-    "--query",
-    "-q",
-    help="Search string filter.",
-)
-@click.option(
-    "--start-time",
-    help="ISO8601 start time (e.g. 2026-03-08T00:00:00Z).",
-)
-@click.option(
-    "--end-time",
-    help="ISO8601 end time. (e.g. 2026-03-08T00:00:00Z)",
-)
-@click.option(
-    "--follow",
-    "-f",
-    is_flag=True,
-    default=False,
-    help="Continuously poll for new logs.",
-)
-@OPT_HOST_URL
-@deploy.command(
-    "logs",
-    help=(
-        "[Beta] Fetch LangSmith Deployment logs. Use 'deploy' for agent runtime "
-        "logs, or 'build' for remote build logs."
-    ),
-)
-@log_command
-def deploy_logs(
-    api_key: str | None,
-    name: str | None,
-    deployment_id: str | None,
-    log_type: str,
-    revision_id: str | None,
-    level: str | None,
-    limit: int,
-    query: str | None,
-    start_time: str | None,
-    end_time: str | None,
-    follow: bool,
-    host_url: str,
-):
-    env_vars = _parse_env_from_config({}, pathlib.Path.cwd() / DEFAULT_CONFIG)
-    client = _create_host_backend_client(host_url, api_key, env_vars=env_vars)
-    if not deployment_id and not name:
-        name = env_vars.get(_DEPLOYMENT_NAME_ENV)
-    dep_id = _call_host_backend_with_optional_tenant(
-        client, lambda c: resolve_deployment_id(c, deployment_id, name)
-    )
-
-    if log_type == "build" and not revision_id:
-        revisions_resp = client.list_revisions(dep_id, limit=1)
-        resources = (
-            revisions_resp.get("resources", [])
-            if isinstance(revisions_resp, dict)
-            else []
-        )
-        if not resources:
-            raise click.ClickException(
-                "No revisions found for this deployment. Cannot fetch build logs."
-            )
-        revision_id = str(resources[0]["id"])
-        click.secho(f"Using latest revision: {revision_id}", fg="cyan")
-
-    payload: dict = {"limit": limit, "order": "desc"}
-    if level:
-        payload["level"] = level.upper()
-    if query:
-        payload["query"] = query
-    if start_time:
-        payload["start_time"] = start_time
-    if end_time:
-        payload["end_time"] = end_time
-
-    def _fetch(request_payload: dict) -> list[dict]:
-        if log_type == "build":
-            resp = client.get_build_logs(dep_id, revision_id, request_payload)
-        else:
-            resp = client.get_deploy_logs(dep_id, request_payload, revision_id)
-
-        if isinstance(resp, dict):
-            return resp.get("logs", [])
-        return []
-
-    def _print_entries(entries: list[dict], *, reverse: bool = False) -> None:
-        iterable = reversed(entries) if reverse else entries
-        for entry in iterable:
-            line = format_log_entry(entry)
-            fg = level_fg(entry.get("level", ""))
-            click.secho(line, fg=fg)
-
-    def _fetch_and_print(request_payload: dict, *, reverse: bool = False) -> list[dict]:
-        entries = _fetch(request_payload)
-        _print_entries(entries, reverse=reverse)
-        return entries
-
-    def _fetch_and_print_new(request_payload: dict, seen_ids: set[str]) -> list[dict]:
-        entries = _fetch(request_payload)
-        new = [e for e in entries if e.get("id", "") not in seen_ids]
-        if new:
-            _print_entries(new)
-            seen_ids.update(e.get("id", "") for e in new)
-        return new
-
-    # initial log fetch will be newest -> oldest, so we need to reverse
-    entries = _fetch_and_print(payload, reverse=True)
-
-    if not follow:
-        if not entries:
-            click.secho("No log entries found.", fg="yellow")
-        return
-
-    payload["order"] = "asc"
-    seen_ids: set[str] = {e.get("id", "") for e in entries if e.get("id")}
-
-    def _update_start_time(ts) -> None:
-        if ts is None:
-            return
-        if isinstance(ts, (int, float)):
-            dt = datetime.fromtimestamp(ts / 1000, tz=timezone.utc)
-            payload["start_time"] = dt.isoformat()
-        else:
-            payload["start_time"] = str(ts)
-
-    if entries:
-        # entries are in descending order here, so index 0 is the newest log
-        _update_start_time(entries[0].get("timestamp"))
-
-    try:
-        while True:
-            time.sleep(2)
-            new_entries = _fetch_and_print_new(payload, seen_ids)
-            if new_entries:
-                _update_start_time(new_entries[-1].get("timestamp"))
-    except KeyboardInterrupt:
-        click.echo("\nStopped.")
-
-
 def _get_docker_ignore_content() -> str:
     """Return the content of a .dockerignore file.
 

libs/cli/langgraph_cli/helpers.py

@@ -1,59 +0,0 @@
-"""Helpers for the ``langgraph logs`` CLI command."""
-
-from __future__ import annotations
-
-from datetime import datetime, timezone
-
-import click
-
-from langgraph_cli.host_backend import HostBackendClient
-
-
-def resolve_deployment_id(
-    client: HostBackendClient,
-    deployment_id: str | None,
-    name: str | None,
-) -> str:
-    """Resolve a deployment ID from --deployment-id or --name."""
-    if deployment_id:
-        return deployment_id
-    if not name:
-        raise click.UsageError("Either --deployment-id or --name is required.")
-    existing = client.list_deployments(name_contains=name)
-    if isinstance(existing, dict):
-        for dep in existing.get("resources", []):
-            if isinstance(dep, dict) and dep.get("name") == name:
-                found_id = dep.get("id")
-                if found_id:
-                    return str(found_id)
-    raise click.ClickException(f"Deployment '{name}' not found.")
-
-
-def format_timestamp(ts) -> str:
-    """Convert a timestamp (epoch ms or string) to a readable string."""
-    if isinstance(ts, (int, float)):
-        dt = datetime.fromtimestamp(ts / 1000, tz=timezone.utc)
-        return dt.strftime("%Y-%m-%d %H:%M:%S")
-    return str(ts) if ts else ""
-
-
-def format_log_entry(entry: dict) -> str:
-    """Format a single log entry for display."""
-    ts = format_timestamp(entry.get("timestamp", ""))
-    level = entry.get("level", "")
-    message = entry.get("message", "")
-    if ts and level:
-        return f"[{ts}] [{level}] {message}"
-    elif ts:
-        return f"[{ts}] {message}"
-    return message
-
-
-def level_fg(level: str) -> str | None:
-    """Return click color for a log level."""
-    level_upper = level.upper() if level else ""
-    if level_upper in {"ERROR", "CRITICAL"}:
-        return "red"
-    if level_upper == "WARNING":
-        return "yellow"
-    return None

libs/cli/langgraph_cli/host_backend.py

@@ -19,12 +19,7 @@ class HostBackendError(click.ClickException):
 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,
-    ):
+    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)
@@ -35,6 +30,7 @@ class HostBackendClient:
         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,
@@ -43,14 +39,10 @@ class HostBackendClient:
         )
 
     def _request(
-        self,
-        method: str,
-        path: str,
-        payload: dict[str, Any] | None = None,
-        params: dict[str, Any] | None = None,
+        self, method: str, path: str, payload: dict[str, Any] | None = None
     ) -> Any:
         try:
-            resp = self._client.request(method, path, json=payload, params=params)
+            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)
@@ -73,19 +65,12 @@ class HostBackendClient:
     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",
-            "/v2/deployments",
-            params={"name_contains": name_contains},
-        )
+    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 delete_deployment(self, deployment_id: str) -> None:
-        return self._request("DELETE", f"/v2/deployments/{deployment_id}")
-
     def request_push_token(self, deployment_id: str) -> dict[str, Any]:
         return self._request(
             "POST",
@@ -120,24 +105,3 @@ class HostBackendClient:
             "GET",
             f"/v2/deployments/{deployment_id}/revisions/{revision_id}",
         )
-
-    def get_build_logs(
-        self, project_id: str, revision_id: str, payload: dict[str, Any]
-    ) -> Any:
-        return self._request(
-            "POST",
-            f"/v1/projects/{project_id}/revisions/{revision_id}/build_logs",
-            payload,
-        )
-
-    def get_deploy_logs(
-        self,
-        project_id: str,
-        payload: dict[str, Any],
-        revision_id: str | None = None,
-    ) -> Any:
-        if revision_id:
-            path = f"/v1/projects/{project_id}/revisions/{revision_id}/deploy_logs"
-        else:
-            path = f"/v1/projects/{project_id}/deploy_logs"
-        return self._request("POST", path, payload)

libs/cli/langgraph_cli/util.py

@@ -1,5 +1,3 @@
-from collections.abc import Sequence
-
 import click
 
 
@@ -25,35 +23,3 @@ def warn_non_wolfi_distro(config_json: dict) -> None:
             fg="yellow",
         )
         click.secho("")  # Empty line for better readability
-
-
-def _extract_deployment_url(deployment: dict[str, object]) -> str:
-    source_config = deployment.get("source_config")
-    if isinstance(source_config, dict):
-        custom_url = source_config.get("custom_url")
-        if isinstance(custom_url, str) and custom_url:
-            return custom_url
-    return "-"
-
-
-def format_deployments_table(deployments: Sequence[dict[str, object]]) -> str:
-    headers = ("Deployment ID", "Deployment Name", "Deployment URL")
-    rows = [
-        (
-            str(deployment.get("id", "-") or "-"),
-            str(deployment.get("name", "-") or "-"),
-            _extract_deployment_url(deployment),
-        )
-        for deployment in deployments
-    ]
-    widths = [
-        max(len(headers[index]), *(len(row[index]) for row in rows))
-        for index in range(len(headers))
-    ]
-
-    def format_row(row: Sequence[str]) -> str:
-        return "  ".join(value.ljust(widths[index]) for index, value in enumerate(row))
-
-    lines = [format_row(headers), format_row(tuple("-" * width for width in widths))]
-    lines.extend(format_row(row) for row in rows)
-    return "\n".join(lines)

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

@@ -9,7 +9,6 @@ from pathlib import Path
 
 from click.testing import CliRunner
 
-import langgraph_cli.cli as cli_module
 from langgraph_cli.cli import cli, prepare_args_and_stdin
 from langgraph_cli.config import Config, _get_pip_cleanup_lines, validate_config
 from langgraph_cli.docker import DEFAULT_POSTGRES_URI, DockerCapabilities, Version
@@ -288,238 +287,6 @@ def test_version_option() -> None:
     )
 
 
-def test_top_level_help_shows_deploy_subcommands() -> None:
-    runner = CliRunner()
-
-    result = runner.invoke(cli, ["--help"])
-
-    assert result.exit_code == 0, result.output
-    assert "deploy" in result.output
-    assert "deploy list" in result.output
-    assert "deploy delete" in result.output
-    assert "[Beta] List LangSmith Deployments." in result.output
-
-
-def test_top_level_help_truncates_command_descriptions_to_single_line() -> None:
-    runner = CliRunner()
-
-    result = runner.invoke(cli, ["--help"])
-
-    assert result.exit_code == 0, result.output
-    lines = result.output.splitlines()
-    deploy_line = next(line for line in lines if line.strip().startswith("deploy"))
-    deploy_list_line = next(
-        line for line in lines if line.strip().startswith("deploy list")
-    )
-
-    assert not lines[lines.index(deploy_line) + 1].startswith("               ")
-    assert "..." in deploy_line
-    assert "[Beta] List LangSmith Deployments." in deploy_list_line
-
-
-def test_deploy_list_command(monkeypatch) -> None:
-    runner = CliRunner()
-    captured: dict[str, str] = {}
-
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            captured["host_url"] = host_url
-            captured["api_key"] = api_key
-            captured["tenant_id"] = tenant_id or ""
-
-        def list_deployments(self, name_contains: str = ""):
-            captured["name_contains"] = name_contains
-            return {
-                "resources": [
-                    {
-                        "id": "dep-123",
-                        "name": "alpha",
-                        "source_config": {"custom_url": "https://alpha.example.com"},
-                    },
-                    {
-                        "id": "dep-456",
-                        "name": "beta",
-                        "source_config": {"custom_url": "https://beta.example.com"},
-                    },
-                ]
-            }
-
-    monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient)
-
-    result = runner.invoke(
-        cli,
-        [
-            "deploy",
-            "list",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-            "--name-contains",
-            "alp",
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert captured == {
-        "host_url": "https://api.example.com",
-        "api_key": "test-key",
-        "tenant_id": "",
-        "name_contains": "alp",
-    }
-    assert "Deployment ID" in result.output
-    assert "Deployment Name" in result.output
-    assert "Deployment URL" in result.output
-    assert "dep-123" in result.output
-    assert "https://beta.example.com" in result.output
-
-
-def test_deploy_list_command_no_results(monkeypatch) -> None:
-    runner = CliRunner()
-
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            pass
-
-        def list_deployments(self, name_contains: str = ""):
-            return {"resources": []}
-
-    monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient)
-
-    result = runner.invoke(
-        cli,
-        [
-            "deploy",
-            "list",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert result.output.strip() == "No deployments found."
-
-
-def test_deploy_delete_command(monkeypatch) -> None:
-    runner = CliRunner()
-    captured: dict[str, str] = {}
-
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            captured["host_url"] = host_url
-            captured["api_key"] = api_key
-            captured["tenant_id"] = tenant_id or ""
-
-        def delete_deployment(self, deployment_id: str):
-            captured["deployment_id"] = deployment_id
-            return None
-
-    monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient)
-
-    result = runner.invoke(
-        cli,
-        [
-            "deploy",
-            "delete",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-            "dep-123",
-        ],
-        input="y\n",
-    )
-
-    assert result.exit_code == 0, result.output
-    assert captured == {
-        "host_url": "https://api.example.com",
-        "api_key": "test-key",
-        "tenant_id": "",
-        "deployment_id": "dep-123",
-    }
-    assert (
-        "Are you sure you want to delete deployment ID dep-123? (Y/n):" in result.output
-    )
-    assert result.output.strip().endswith("Deleted deployment dep-123.")
-
-
-def test_deploy_delete_command_cancelled(monkeypatch) -> None:
-    runner = CliRunner()
-    deleted = False
-
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            pass
-
-        def delete_deployment(self, deployment_id: str):
-            nonlocal deleted
-            deleted = True
-            return None
-
-    monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient)
-
-    result = runner.invoke(
-        cli,
-        [
-            "deploy",
-            "delete",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-            "dep-123",
-        ],
-        input="n\n",
-    )
-
-    assert result.exit_code == 1, result.output
-    assert not deleted
-    assert "Aborted!" in result.output
-
-
-def test_deploy_delete_command_force(monkeypatch) -> None:
-    runner = CliRunner()
-    captured: dict[str, str] = {}
-
-    class FakeClient:
-        def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None):
-            captured["host_url"] = host_url
-            captured["api_key"] = api_key
-            captured["tenant_id"] = tenant_id or ""
-
-        def delete_deployment(self, deployment_id: str):
-            captured["deployment_id"] = deployment_id
-            return None
-
-    monkeypatch.setattr(cli_module, "HostBackendClient", FakeClient)
-
-    result = runner.invoke(
-        cli,
-        [
-            "deploy",
-            "delete",
-            "--force",
-            "--api-key",
-            "test-key",
-            "--host-url",
-            "https://api.example.com",
-            "dep-123",
-        ],
-    )
-
-    assert result.exit_code == 0, result.output
-    assert "Are you sure you want to delete deployment ID dep-123?" not in result.output
-    assert captured == {
-        "host_url": "https://api.example.com",
-        "api_key": "test-key",
-        "tenant_id": "",
-        "deployment_id": "dep-123",
-    }
-    assert result.output.strip() == "Deleted deployment dep-123."
-
-
 def test_dockerfile_command_basic() -> None:
     """Test the 'dockerfile' command with basic configuration."""
     runner = CliRunner()

libs/cli/tests/unit_tests/test_config.py

@@ -1784,9 +1784,7 @@ def test_config_to_compose_distributed_mode():
     # 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
-    )
+    assert 'entrypoint: ["sh", "/storage/executor_entrypoint.sh"]' in actual_compose_stdin
 
     # Executor has required environment variables
     assert "EXECUTOR_GRPC_PORT:" in actual_compose_stdin

libs/cli/tests/unit_tests/test_host_backend.py

@@ -135,28 +135,6 @@ def test_list_deployments(client):
     assert result == {"ok": True}
 
 
-def test_list_deployments_sends_query_params():
-    def handler(req: httpx.Request) -> httpx.Response:
-        assert req.url.path == "/v2/deployments"
-        assert req.url.params["name_contains"] == "my app"
-        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.list_deployments("my app")
-    assert result == {"ok": True}
-
-
-def test_delete_deployment(client):
-    result = client.delete_deployment("dep-123")
-    assert result == {"ok": True}
-
-
 def test_request_push_token(client):
     result = client.request_push_token("dep-123")
     assert result == {"ok": True}
@@ -182,41 +160,3 @@ def test_list_revisions(client):
 def test_get_revision(client):
     result = client.get_revision("dep-123", "rev-456")
     assert result == {"ok": True}
-
-
-def test_get_build_logs(client):
-    result = client.get_build_logs("proj-1", "rev-1", {"limit": 10})
-    assert result == {"ok": True}
-
-
-def test_get_deploy_logs_all_revisions():
-    def handler(req: httpx.Request) -> httpx.Response:
-        assert "/v1/projects/proj-1/deploy_logs" in str(req.url)
-        assert "/revisions/" not in str(req.url)
-        return httpx.Response(200, json={"logs": [{"message": "running"}]})
-
-    c = HostBackendClient("https://api.example.com", "key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=httpx.MockTransport(handler),
-        headers={"X-Api-Key": "key", "Accept": "application/json"},
-        timeout=30,
-    )
-    result = c.get_deploy_logs("proj-1", {"limit": 10})
-    assert result == {"logs": [{"message": "running"}]}
-
-
-def test_get_deploy_logs_specific_revision():
-    def handler(req: httpx.Request) -> httpx.Response:
-        assert "/v1/projects/proj-1/revisions/rev-2/deploy_logs" in str(req.url)
-        return httpx.Response(200, json={"logs": []})
-
-    c = HostBackendClient("https://api.example.com", "key")
-    c._client = httpx.Client(
-        base_url="https://api.example.com",
-        transport=httpx.MockTransport(handler),
-        headers={"X-Api-Key": "key", "Accept": "application/json"},
-        timeout=30,
-    )
-    result = c.get_deploy_logs("proj-1", {"limit": 10}, revision_id="rev-2")
-    assert result == {"logs": []}

libs/cli/tests/unit_tests/test_logs_helpers.py

@@ -1,58 +0,0 @@
-from langgraph_cli.helpers import format_log_entry, format_timestamp, level_fg
-
-
-class TestFormatTimestamp:
-    def test_epoch_ms(self):
-        assert format_timestamp(1773119644012) == "2026-03-10 05:14:04"
-
-    def test_string_passthrough(self):
-        assert format_timestamp("2026-03-08T00:00:00Z") == "2026-03-08T00:00:00Z"
-
-    def test_empty(self):
-        assert format_timestamp("") == ""
-
-    def test_none(self):
-        assert format_timestamp(None) == ""
-
-
-class TestFormatLogEntry:
-    def test_full_entry_epoch(self):
-        entry = {"timestamp": 1773119644012, "level": "ERROR", "message": "boom"}
-        result = format_log_entry(entry)
-        assert result == "[2026-03-10 05:14:04] [ERROR] boom"
-
-    def test_full_entry_string(self):
-        entry = {
-            "timestamp": "2026-03-08T12:00:00Z",
-            "level": "ERROR",
-            "message": "boom",
-        }
-        assert format_log_entry(entry) == "[2026-03-08T12:00:00Z] [ERROR] boom"
-
-    def test_no_level(self):
-        entry = {"timestamp": "2026-03-08T12:00:00Z", "message": "hello"}
-        assert format_log_entry(entry) == "[2026-03-08T12:00:00Z] hello"
-
-    def test_no_timestamp(self):
-        entry = {"message": "bare message"}
-        assert format_log_entry(entry) == "bare message"
-
-    def test_empty_entry(self):
-        assert format_log_entry({}) == ""
-
-
-class TestLevelFg:
-    def test_error(self):
-        assert level_fg("ERROR") == "red"
-
-    def test_error_lowercase(self):
-        assert level_fg("error") == "red"
-
-    def test_warning(self):
-        assert level_fg("WARNING") == "yellow"
-
-    def test_info_returns_none(self):
-        assert level_fg("INFO") is None
-
-    def test_empty_returns_none(self):
-        assert level_fg("") is None

libs/cli/tests/unit_tests/test_util.py

@@ -1,11 +1,6 @@
 from unittest.mock import patch
 
-from langgraph_cli.util import (
-    _extract_deployment_url,
-    clean_empty_lines,
-    format_deployments_table,
-    warn_non_wolfi_distro,
-)
+from langgraph_cli.util import clean_empty_lines, warn_non_wolfi_distro
 
 
 def test_clean_empty_lines():
@@ -191,36 +186,3 @@ def test_warn_non_wolfi_distro_does_not_modify_config():
     warn_non_wolfi_distro(config_copy)
 
     assert config_copy == original_config  # Config should remain unchanged
-
-
-def test_extract_deployment_url_uses_custom_url():
-    deployment = {"source_config": {"custom_url": "https://example.com/custom"}}
-    assert _extract_deployment_url(deployment) == "https://example.com/custom"
-
-
-def test_extract_deployment_url_defaults_to_dash():
-    assert _extract_deployment_url({"id": "dep-123"}) == "-"
-
-
-def test_format_deployments_table():
-    output = format_deployments_table(
-        [
-            {
-                "id": "dep-123",
-                "name": "alpha",
-                "source_config": {"custom_url": "https://alpha.example.com"},
-            },
-            {
-                "id": "dep-456",
-                "name": "beta",
-                "url": "https://beta.example.com",
-            },
-        ]
-    )
-    assert "Deployment ID" in output
-    assert "Deployment Name" in output
-    assert "Deployment URL" in output
-    assert "dep-123" in output
-    assert "alpha" in output
-    assert "https://alpha.example.com" in output
-    assert "dep-456" in output

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/pregel/_loop.py

@@ -647,24 +647,13 @@ class PregelLoop:
         # 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.
-        if self.is_replaying and (
-            # Time-travel to a subgraph checkpoint: the parent sets
-            # RESUMING=True (it can't distinguish time-travel from resume),
-            # so we check if this subgraph's own ns is in checkpoint_map.
-            # Normally the map only has ancestor entries (_algo.py); the
-            # subgraph's own entry only appears via get_state(subgraphs=True).
-            (
-                self.is_nested
-                and configurable.get(CONFIG_KEY_CHECKPOINT_NS, "")
-                in configurable.get(CONFIG_KEY_CHECKPOINT_MAP, {})
-            )
-            or not (
-                # Outer graph: resume arrives as Command(resume=...)
-                (input_is_command and cast(Command, self.input).resume is not None)
-                # Subgraphs: resume arrives via config flag from parent
-                # (subgraph input is a Send arg, not a Command)
-                or configurable.get(CONFIG_KEY_RESUMING, False)
-            )
+        # 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
@@ -1138,15 +1127,9 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
     def __enter__(self) -> Self:
         if not self.checkpointer:
             saved = None
-        elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
-            # Explicit checkpoint_id requested — fetch that exact checkpoint.
-            # This covers both normal replay and subgraphs resolved via
-            # checkpoint_map during time-travel.
-            saved = self.checkpointer.get_tuple(self.checkpoint_config)
-        elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
-            # Subgraph replay: the parent is replaying and passed us a
-            # replay_state with its checkpoint_id. Look up our checkpoint
-            # from the parent's checkpoint_map instead of fetching latest.
+        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,
@@ -1158,7 +1141,9 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             self.config[CONF].pop(CONFIG_KEY_RESUMING, None)
         else:
             # Normal case: fetch the most recent checkpoint for this
-            # graph/thread. Returns None on first invocation.
+            # 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:
@@ -1337,15 +1322,9 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
     async def __aenter__(self) -> Self:
         if not self.checkpointer:
             saved = None
-        elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
-            # Explicit checkpoint_id requested — fetch that exact checkpoint.
-            # This covers both normal replay and subgraphs resolved via
-            # checkpoint_map during time-travel.
-            saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
-        elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
-            # Subgraph replay: the parent is replaying and passed us a
-            # replay_state with its checkpoint_id. Look up our checkpoint
-            # from the parent's checkpoint_map instead of fetching latest.
+        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,
@@ -1357,7 +1336,9 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             self.config[CONF].pop(CONFIG_KEY_RESUMING, None)
         else:
             # Normal case: fetch the most recent checkpoint for this
-            # graph/thread. Returns None on first invocation.
+            # 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:

libs/langgraph/langgraph/pregel/_messages.py

@@ -25,7 +25,7 @@ except ImportError:
     _StreamingCallbackHandler = object  # type: ignore
 
 T = TypeVar("T")
-Meta = tuple[tuple[str, ...], dict[str, Any] | None]
+Meta = tuple[tuple[str, ...], dict[str, Any]]
 
 
 def _state_values(obj: Any) -> Sequence[Any]:
@@ -56,7 +56,6 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
         subgraphs: bool,
         *,
         parent_ns: tuple[str, ...] | None = None,
-        dedupe_metadata: bool = False,
     ) -> None:
         """Configure the handler to stream messages from LLMs and nodes.
 
@@ -85,10 +84,8 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
         self.stream = stream
         self.subgraphs = subgraphs
         self.metadata: dict[UUID, Meta] = {}
-        self.emitted_metadata: set[UUID] = set()
         self.seen: set[int | str] = set()
         self.parent_ns = parent_ns
-        self.dedupe_metadata = dedupe_metadata
 
     def _emit(self, meta: Meta, message: BaseMessage, *, dedupe: bool = False) -> None:
         if dedupe and message.id in self.seen:
@@ -158,10 +155,6 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
         if not isinstance(chunk, ChatGenerationChunk):
             return
         if meta := self.metadata.get(run_id):
-            if self.dedupe_metadata and run_id in self.emitted_metadata:
-                meta = (meta[0], None)
-            else:
-                self.emitted_metadata.add(run_id)
             self._emit(meta, chunk.message)
 
     def on_llm_end(
@@ -177,7 +170,6 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
                 gen = response.generations[0][0]
                 if isinstance(gen, ChatGeneration):
                     self._emit(meta, gen.message, dedupe=True)
-        self.emitted_metadata.discard(run_id)
         self.metadata.pop(run_id, None)
 
     def on_llm_error(
@@ -188,7 +180,6 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
         parent_run_id: UUID | None = None,
         **kwargs: Any,
     ) -> Any:
-        self.emitted_metadata.discard(run_id)
         self.metadata.pop(run_id, None)
 
     def on_chain_start(

libs/langgraph/langgraph/pregel/main.py

@@ -2456,7 +2456,7 @@ class Pregel(
         debug: bool | None = None,
         version: Literal["v2"],
         **kwargs: Unpack[DeprecatedKwargs],
-    ) -> Iterator[StreamPart[StateT, OutputT]]: ...
+    ) -> Iterator[StreamPart[OutputT, StateT]]: ...
 
     @overload
     def stream(
@@ -2614,7 +2614,6 @@ class Pregel(
                         stream.put,
                         subgraphs,
                         parent_ns=tuple(ns_.split(NS_SEP)) if ns_ else None,
-                        dedupe_metadata="compact" in stream_modes,
                     )
                 )
 
@@ -2788,7 +2787,7 @@ class Pregel(
         debug: bool | None = None,
         version: Literal["v2"],
         **kwargs: Unpack[DeprecatedKwargs],
-    ) -> AsyncIterator[StreamPart[StateT, OutputT]]: ...
+    ) -> AsyncIterator[StreamPart[OutputT, StateT]]: ...
 
     @overload
     def astream(
@@ -2966,7 +2965,6 @@ class Pregel(
                         stream_put,
                         subgraphs,
                         parent_ns=tuple(ns_.split(NS_SEP)) if ns_ else None,
-                        dedupe_metadata="compact" in stream_modes,
                     )
                 )
 
@@ -3196,7 +3194,7 @@ class Pregel(
         durability: Durability | None = None,
         version: Literal["v2"],
         **kwargs: Any,
-    ) -> list[StreamPart[StateT, OutputT]]: ...
+    ) -> list[StreamPart[OutputT, StateT]]: ...
 
     @overload
     def invoke(
@@ -3366,7 +3364,7 @@ class Pregel(
         durability: Durability | None = None,
         version: Literal["v2"],
         **kwargs: Any,
-    ) -> list[StreamPart[StateT, OutputT]]: ...
+    ) -> list[StreamPart[OutputT, StateT]]: ...
 
     @overload
     async def ainvoke(

libs/langgraph/langgraph/pregel/protocol.py

@@ -117,7 +117,7 @@ class PregelProtocol(Runnable[InputT, Any], Generic[StateT, ContextT, InputT, Ou
         interrupt_after: All | Sequence[str] | None = None,
         subgraphs: bool = False,
         version: Literal["v2"],
-    ) -> Iterator[StreamPart[StateT, OutputT]]: ...
+    ) -> Iterator[StreamPart[OutputT, StateT]]: ...
 
     @overload
     @abstractmethod
@@ -161,7 +161,7 @@ class PregelProtocol(Runnable[InputT, Any], Generic[StateT, ContextT, InputT, Ou
         interrupt_after: All | Sequence[str] | None = None,
         subgraphs: bool = False,
         version: Literal["v2"],
-    ) -> AsyncIterator[StreamPart[StateT, OutputT]]: ...
+    ) -> AsyncIterator[StreamPart[OutputT, StateT]]: ...
 
     @overload
     @abstractmethod

libs/langgraph/langgraph/pregel/remote.py

@@ -31,7 +31,6 @@ from langgraph_sdk.client import (
 )
 from langgraph_sdk.schema import (
     Checkpoint,
-    Context,
     QueryParamTypes,
     ThreadState,
 )
@@ -109,45 +108,6 @@ class RemoteException(Exception):
     pass
 
 
-def _restore_message_metadata(
-    data: Any, metadata_by_message_id: dict[str, dict[str, Any]]
-) -> Any:
-    """Restore deduplicated message metadata using the message id as cache key."""
-    if not (isinstance(data, list) and len(data) == 2):
-        return data
-    message, metadata = data
-    if not isinstance(message, dict):
-        return data
-    message_id = message.get("id")
-    if isinstance(message_id, str):
-        if isinstance(metadata, dict):
-            metadata_by_message_id[message_id] = metadata
-        else:
-            metadata = metadata_by_message_id.get(message_id)
-    return (message, metadata)
-
-
-def _merge_values_patch(
-    ns: tuple[str, ...],
-    mode: str,
-    data: Any,
-    values_by_ns: dict[tuple[str, ...], dict[str, Any]],
-) -> tuple[str, Any]:
-    """Merge `values-patch` events back into full values snapshots."""
-    if mode != "values-patch" or not isinstance(data, dict):
-        return mode, data
-    values = data.get("values")
-    if not isinstance(values, dict):
-        return "values", values if values is not None else {}
-    merged = dict(values_by_ns.get(ns, {}))
-    merged.update(values)
-    for key in data.get("deleted_keys", ()):
-        if isinstance(key, str):
-            merged.pop(key, None)
-    values_by_ns[ns] = merged
-    return "values", merged
-
-
 class RemoteGraph(PregelProtocol):
     """The `RemoteGraph` class is a client implementation for calling remote
     APIs that implement the LangGraph Server API specification.
@@ -731,7 +691,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         stream_mode: StreamMode | list[StreamMode] | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
@@ -748,7 +707,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         stream_mode: StreamMode | list[StreamMode] | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
@@ -764,7 +722,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         stream_mode: StreamMode | list[StreamMode] | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
@@ -777,7 +734,7 @@ class RemoteGraph(PregelProtocol):
         """Create a run and stream the results.
 
         This method calls `POST /threads/{thread_id}/runs/stream` if a `thread_id`
-        is specified in the `configurable` field of the config or
+        is speciffed in the `configurable` field of the config or
         `POST /runs/stream` otherwise.
 
         Args:
@@ -805,8 +762,6 @@ class RemoteGraph(PregelProtocol):
         else:
             command = None
         thread_id = sanitized_config.get("configurable", {}).pop("thread_id", None)
-        message_metadata_by_id: dict[str, dict[str, Any]] = {}
-        values_by_ns: dict[tuple[str, ...], dict[str, Any]] = {}
 
         for chunk in sync_client.runs.stream(
             thread_id=thread_id,
@@ -814,7 +769,6 @@ class RemoteGraph(PregelProtocol):
             input=input,
             command=command,
             config=sanitized_config,
-            context=context,
             stream_mode=stream_modes,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
@@ -839,11 +793,6 @@ class RemoteGraph(PregelProtocol):
             if caller_ns := (config or {}).get(CONF, {}).get(CONFIG_KEY_CHECKPOINT_NS):
                 caller_ns = tuple(caller_ns.split(NS_SEP))
                 ns = caller_ns + ns
-            mode, data = _merge_values_patch(ns, mode, chunk.data, values_by_ns)
-            if mode != chunk.event:
-                chunk = chunk._replace(data=data)
-            elif data is not chunk.data:
-                chunk = chunk._replace(data=data)
             # stream to parent stream
             if stream is not None and mode in stream.modes:
                 stream((ns, mode, chunk.data))
@@ -861,9 +810,7 @@ class RemoteGraph(PregelProtocol):
                 continue
 
             if chunk.event.startswith("messages"):
-                chunk = chunk._replace(
-                    data=_restore_message_metadata(chunk.data, message_metadata_by_id)
-                )
+                chunk = chunk._replace(data=tuple(chunk.data))
 
             # emit chunk
             if version == "v2":
@@ -875,6 +822,11 @@ class RemoteGraph(PregelProtocol):
                     )
                 yield {"type": mode, "ns": ns, "data": chunk.data, "interrupts": ints}
             elif subgraphs:
+                if NS_SEP in chunk.event:
+                    mode, ns_ = chunk.event.split(NS_SEP, 1)
+                    ns = tuple(ns_.split(NS_SEP))
+                else:
+                    mode, ns = chunk.event, ()
                 if req_single:
                     yield ns, chunk.data
                 else:
@@ -890,7 +842,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         stream_mode: StreamMode | list[StreamMode] | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
@@ -907,7 +858,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         stream_mode: StreamMode | list[StreamMode] | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
@@ -923,7 +873,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         stream_mode: StreamMode | list[StreamMode] | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
@@ -936,7 +885,7 @@ class RemoteGraph(PregelProtocol):
         """Create a run and stream the results.
 
         This method calls `POST /threads/{thread_id}/runs/stream` if a `thread_id`
-        is specified in the `configurable` field of the config or
+        is speciffed in the `configurable` field of the config or
         `POST /runs/stream` otherwise.
 
         Args:
@@ -964,8 +913,6 @@ class RemoteGraph(PregelProtocol):
         else:
             command = None
         thread_id = sanitized_config.get("configurable", {}).pop("thread_id", None)
-        message_metadata_by_id: dict[str, dict[str, Any]] = {}
-        values_by_ns: dict[tuple[str, ...], dict[str, Any]] = {}
 
         async for chunk in client.runs.stream(
             thread_id=thread_id,
@@ -973,7 +920,6 @@ class RemoteGraph(PregelProtocol):
             input=input,
             command=command,
             config=sanitized_config,
-            context=context,
             stream_mode=stream_modes,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
@@ -998,11 +944,6 @@ class RemoteGraph(PregelProtocol):
             if caller_ns := (config or {}).get(CONF, {}).get(CONFIG_KEY_CHECKPOINT_NS):
                 caller_ns = tuple(caller_ns.split(NS_SEP))
                 ns = caller_ns + ns
-            mode, data = _merge_values_patch(ns, mode, chunk.data, values_by_ns)
-            if mode != chunk.event:
-                chunk = chunk._replace(data=data)
-            elif data is not chunk.data:
-                chunk = chunk._replace(data=data)
             # stream to parent stream
             if stream is not None and mode in stream.modes:
                 stream((ns, mode, chunk.data))
@@ -1020,9 +961,7 @@ class RemoteGraph(PregelProtocol):
                 continue
 
             if chunk.event.startswith("messages"):
-                chunk = chunk._replace(
-                    data=_restore_message_metadata(chunk.data, message_metadata_by_id)
-                )
+                chunk = chunk._replace(data=tuple(chunk.data))
 
             # emit chunk
             if version == "v2":
@@ -1034,6 +973,11 @@ class RemoteGraph(PregelProtocol):
                     )
                 yield {"type": mode, "ns": ns, "data": chunk.data, "interrupts": ints}
             elif subgraphs:
+                if NS_SEP in chunk.event:
+                    mode, ns_ = chunk.event.split(NS_SEP, 1)
+                    ns = tuple(ns_.split(NS_SEP))
+                else:
+                    mode, ns = chunk.event, ()
                 if req_single:
                     yield ns, chunk.data
                 else:
@@ -1065,7 +1009,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         headers: dict[str, str] | None = None,
@@ -1080,7 +1023,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         headers: dict[str, str] | None = None,
@@ -1094,7 +1036,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         headers: dict[str, str] | None = None,
@@ -1120,7 +1061,6 @@ class RemoteGraph(PregelProtocol):
         for chunk in self.stream(  # type: ignore[misc, call-overload]
             input,
             config=config,
-            context=context,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
             headers=headers,
@@ -1147,7 +1087,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         headers: dict[str, str] | None = None,
@@ -1162,7 +1101,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         headers: dict[str, str] | None = None,
@@ -1176,7 +1114,6 @@ class RemoteGraph(PregelProtocol):
         input: dict[str, Any] | Any,
         config: RunnableConfig | None = None,
         *,
-        context: Context | None = None,
         interrupt_before: All | Sequence[str] | None = None,
         interrupt_after: All | Sequence[str] | None = None,
         headers: dict[str, str] | None = None,
@@ -1202,7 +1139,6 @@ class RemoteGraph(PregelProtocol):
         async for chunk in self.astream(  # type: ignore[misc, call-overload]
             input,
             config=config,
-            context=context,
             interrupt_before=interrupt_before,
             interrupt_after=interrupt_after,
             headers=headers,

libs/langgraph/langgraph/types.py

@@ -116,14 +116,7 @@ def ensure_valid_checkpointer(checkpointer: Checkpointer) -> Checkpointer:
 
 
 StreamMode = Literal[
-    "values",
-    "updates",
-    "checkpoints",
-    "tasks",
-    "debug",
-    "messages",
-    "custom",
-    "compact",
+    "values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"
 ]
 """How the stream method should emit outputs.
 
@@ -282,14 +275,13 @@ class MessagesStreamPart(TypedDict):
     """Stream part emitted for `stream_mode="messages"`.
 
     `data` is a 2-tuple of `(message, metadata)` where `message` is a
-    `BaseMessage` (e.g. `AIMessageChunk`) and `metadata` is either a dict containing
-    keys like `langgraph_step`, `langgraph_node`, `langgraph_triggers`, etc. or
-    `None` for deduplicated follow-up chunks when `stream_mode` includes `"compact"`.
+    `BaseMessage` (e.g. `AIMessageChunk`) and `metadata` is a dict containing
+    keys like `langgraph_step`, `langgraph_node`, `langgraph_triggers`, etc.
     """
 
     type: Literal["messages"]
     ns: tuple[str, ...]
-    data: tuple[AnyMessage, dict[str, Any] | None]
+    data: tuple[AnyMessage, dict[str, Any]]
 
 
 class CustomStreamPart(TypedDict):
@@ -343,7 +335,7 @@ StreamPart = TypeAliasType(
     | CheckpointStreamPart[StateT]
     | TasksStreamPart
     | DebugStreamPart[StateT],
-    type_params=(StateT, OutputT),
+    type_params=(OutputT, StateT),
 )
 """A discriminated union of all v2 stream part types.
 
@@ -354,7 +346,7 @@ async for part in graph.astream(input, version="v2"):
     if part["type"] == "values":
         part["data"]  # OutputT — full state (pydantic/dataclass/dict)
     elif part["type"] == "messages":
-        part["data"]  # tuple[BaseMessage, dict | None] — (message, metadata)
+        part["data"]  # tuple[BaseMessage, dict] — (message, metadata)
     elif part["type"] == "custom":
         part["data"]  # Any — user-defined
 ```

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.1.2"
+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/tests/test_remote_graph.py

@@ -1,6 +1,5 @@
 import re
 import sys
-from dataclasses import dataclass
 from typing import Annotated
 from unittest.mock import AsyncMock, MagicMock
 
@@ -11,7 +10,6 @@ from langchain_core.runnables import RunnableConfig
 from langchain_core.runnables.graph import Edge as DrawableEdge
 from langchain_core.runnables.graph import Node as DrawableNode
 from langgraph_sdk.schema import StreamPart
-from pydantic import BaseModel
 from typing_extensions import TypedDict
 
 from langgraph.errors import GraphInterrupt
@@ -882,80 +880,6 @@ def test_stream_sanitizes_thread_id():
     assert not passed_config["configurable"]
 
 
-def test_stream_restores_messages_and_merges_values_patch():
-    mock_sync_client = MagicMock()
-    mock_sync_client.runs.stream.return_value = [
-        StreamPart(
-            event="messages|tools:call_1",
-            data=[
-                {"id": "msg-1", "type": "AIMessageChunk", "content": "hel"},
-                {
-                    "langgraph_checkpoint_ns": "tools:call_1",
-                    "langgraph_node": "agent",
-                },
-            ],
-        ),
-        StreamPart(
-            event="messages|tools:call_1",
-            data=[
-                {"id": "msg-1", "type": "AIMessageChunk", "content": "lo"},
-                None,
-            ],
-        ),
-        StreamPart(
-            event="values|tools:call_1",
-            data={"messages": [{"type": "human", "content": "hi"}], "count": 1},
-        ),
-        StreamPart(
-            event="values-patch|tools:call_1",
-            data={"values": {"count": 2}, "deleted_keys": ["messages"]},
-        ),
-    ]
-    remote_pregel = RemoteGraph("test_graph_id", sync_client=mock_sync_client)
-
-    parts = list(
-        remote_pregel.stream(
-            {"input": "data"},
-            config={"configurable": {"thread_id": "thread_1"}},
-            stream_mode=["messages", "values", "compact"],
-            subgraphs=True,
-            version="v2",
-        )
-    )
-
-    message_parts = [part for part in parts if part["type"] == "messages"]
-    assert message_parts[0]["data"][1] == {
-        "langgraph_checkpoint_ns": "tools:call_1",
-        "langgraph_node": "agent",
-    }
-    assert message_parts[1]["data"][1] == {
-        "langgraph_checkpoint_ns": "tools:call_1",
-        "langgraph_node": "agent",
-    }
-    value_parts = [part for part in parts if part["type"] == "values"]
-    assert value_parts == [
-        {
-            "type": "values",
-            "ns": ("tools:call_1",),
-            "data": {"messages": [{"type": "human", "content": "hi"}], "count": 1},
-            "interrupts": (),
-        },
-        {
-            "type": "values",
-            "ns": ("tools:call_1",),
-            "data": {"count": 2},
-            "interrupts": (),
-        },
-    ]
-    _, kwargs = mock_sync_client.runs.stream.call_args
-    assert set(kwargs["stream_mode"]) == {
-        "messages-tuple",
-        "values",
-        "compact",
-        "updates",
-    }
-
-
 @pytest.mark.anyio
 async def test_ainvoke():
     # set up test
@@ -984,265 +908,6 @@ async def test_ainvoke():
     assert result == {"messages": [{"type": "human", "content": "world"}]}
 
 
-def test_stream_context():
-    """Test that context is passed through to the SDK client in stream."""
-    mock_sync_client = MagicMock()
-    mock_sync_client.runs.stream.return_value = [
-        StreamPart(event="values", data={"chunk": "data1"}),
-    ]
-
-    remote_pregel = RemoteGraph(
-        "test_graph_id",
-        sync_client=mock_sync_client,
-    )
-
-    config = {"configurable": {"thread_id": "thread_1"}}
-    context = {"model_name": "anthropic", "user_id": "123"}
-    stream_parts = list(
-        remote_pregel.stream(
-            {"input": "data"},
-            config,
-            context=context,
-            stream_mode="values",
-        )
-    )
-
-    assert stream_parts == [{"chunk": "data1"}]
-    _, kwargs = mock_sync_client.runs.stream.call_args
-    assert kwargs["context"] == {"model_name": "anthropic", "user_id": "123"}
-
-
-def test_stream_context_none():
-    """Test that context defaults to None when not provided."""
-    mock_sync_client = MagicMock()
-    mock_sync_client.runs.stream.return_value = [
-        StreamPart(event="values", data={"chunk": "data1"}),
-    ]
-
-    remote_pregel = RemoteGraph(
-        "test_graph_id",
-        sync_client=mock_sync_client,
-    )
-
-    config = {"configurable": {"thread_id": "thread_1"}}
-    list(remote_pregel.stream({"input": "data"}, config, stream_mode="values"))
-
-    _, kwargs = mock_sync_client.runs.stream.call_args
-    assert kwargs["context"] is None
-
-
-@pytest.mark.anyio
-async def test_astream_context():
-    """Test that context is passed through to the SDK client in astream."""
-    mock_async_client = MagicMock()
-    async_iter = MagicMock()
-    async_iter.__aiter__.return_value = [
-        StreamPart(event="values", data={"chunk": "data1"}),
-    ]
-    mock_async_client.runs.stream.return_value = async_iter
-
-    remote_pregel = RemoteGraph(
-        "test_graph_id",
-        client=mock_async_client,
-    )
-
-    config = {"configurable": {"thread_id": "thread_1"}}
-    context = {"model_name": "anthropic"}
-    chunks = []
-    async for chunk in remote_pregel.astream(
-        {"input": "data"},
-        config,
-        context=context,
-        stream_mode="values",
-    ):
-        chunks.append(chunk)
-
-    assert chunks == [{"chunk": "data1"}]
-    _, kwargs = mock_async_client.runs.stream.call_args
-    assert kwargs["context"] == {"model_name": "anthropic"}
-
-
-def test_invoke_context():
-    """Test that context is passed through to the SDK client in invoke."""
-    mock_sync_client = MagicMock()
-    mock_sync_client.runs.stream.return_value = [
-        StreamPart(event="values", data={"result": "done"}),
-    ]
-
-    remote_pregel = RemoteGraph(
-        "test_graph_id",
-        sync_client=mock_sync_client,
-    )
-
-    config = {"configurable": {"thread_id": "thread_1"}}
-    context = {"model_name": "openai"}
-    result = remote_pregel.invoke({"input": "data"}, config, context=context)
-
-    assert result == {"result": "done"}
-    _, kwargs = mock_sync_client.runs.stream.call_args
-    assert kwargs["context"] == {"model_name": "openai"}
-
-
-@pytest.mark.anyio
-async def test_ainvoke_context():
-    """Test that context is passed through to the SDK client in ainvoke."""
-    mock_async_client = MagicMock()
-    async_iter = MagicMock()
-    async_iter.__aiter__.return_value = [
-        StreamPart(event="values", data={"result": "done"}),
-    ]
-    mock_async_client.runs.stream.return_value = async_iter
-
-    remote_pregel = RemoteGraph(
-        "test_graph_id",
-        client=mock_async_client,
-    )
-
-    config = {"configurable": {"thread_id": "thread_1"}}
-    context = {"user_id": "456"}
-    result = await remote_pregel.ainvoke({"input": "data"}, config, context=context)
-
-    assert result == {"result": "done"}
-    _, kwargs = mock_async_client.runs.stream.call_args
-    assert kwargs["context"] == {"user_id": "456"}
-
-
-def test_stream_context_dataclass():
-    """Test that a dataclass context is passed through to the SDK client."""
-
-    @dataclass
-    class MyContext:
-        model_name: str
-        user_id: str
-
-    mock_sync_client = MagicMock()
-    mock_sync_client.runs.stream.return_value = [
-        StreamPart(event="values", data={"chunk": "data1"}),
-    ]
-
-    remote_pregel = RemoteGraph(
-        "test_graph_id",
-        sync_client=mock_sync_client,
-    )
-
-    config = {"configurable": {"thread_id": "thread_1"}}
-    ctx = MyContext(model_name="anthropic", user_id="123")
-    list(
-        remote_pregel.stream(
-            {"input": "data"}, config, context=ctx, stream_mode="values"
-        )
-    )
-
-    _, kwargs = mock_sync_client.runs.stream.call_args
-    assert kwargs["context"] == ctx
-
-
-def test_stream_context_base_model():
-    """Test that a BaseModel context is passed through to the SDK client."""
-
-    class MyContext(BaseModel):
-        model_name: str
-        user_id: str
-
-    mock_sync_client = MagicMock()
-    mock_sync_client.runs.stream.return_value = [
-        StreamPart(event="values", data={"chunk": "data1"}),
-    ]
-
-    remote_pregel = RemoteGraph(
-        "test_graph_id",
-        sync_client=mock_sync_client,
-    )
-
-    config = {"configurable": {"thread_id": "thread_1"}}
-    ctx = MyContext(model_name="anthropic", user_id="123")
-    list(
-        remote_pregel.stream(
-            {"input": "data"}, config, context=ctx, stream_mode="values"
-        )
-    )
-
-    _, kwargs = mock_sync_client.runs.stream.call_args
-    assert kwargs["context"] == ctx
-
-
-@pytest.mark.anyio
-async def test_astream_restores_messages_and_merges_values_patch():
-    mock_async_client = MagicMock()
-    async_iter = MagicMock()
-    async_iter.__aiter__.return_value = [
-        StreamPart(
-            event="messages|tools:call_1",
-            data=[
-                {"id": "msg-1", "type": "AIMessageChunk", "content": "hel"},
-                {
-                    "langgraph_checkpoint_ns": "tools:call_1",
-                    "langgraph_node": "agent",
-                },
-            ],
-        ),
-        StreamPart(
-            event="messages|tools:call_1",
-            data=[
-                {"id": "msg-1", "type": "AIMessageChunk", "content": "lo"},
-                None,
-            ],
-        ),
-        StreamPart(
-            event="values|tools:call_1",
-            data={"messages": [{"type": "human", "content": "hi"}], "count": 1},
-        ),
-        StreamPart(
-            event="values-patch|tools:call_1",
-            data={"values": {"count": 2}, "deleted_keys": ["messages"]},
-        ),
-    ]
-    mock_async_client.runs.stream.return_value = async_iter
-    remote_pregel = RemoteGraph("test_graph_id", client=mock_async_client)
-
-    parts = []
-    async for part in remote_pregel.astream(
-        {"input": "data"},
-        config={"configurable": {"thread_id": "thread_1"}},
-        stream_mode=["messages", "values", "compact"],
-        subgraphs=True,
-        version="v2",
-    ):
-        parts.append(part)
-
-    message_parts = [part for part in parts if part["type"] == "messages"]
-    assert message_parts[0]["data"][1] == {
-        "langgraph_checkpoint_ns": "tools:call_1",
-        "langgraph_node": "agent",
-    }
-    assert message_parts[1]["data"][1] == {
-        "langgraph_checkpoint_ns": "tools:call_1",
-        "langgraph_node": "agent",
-    }
-    value_parts = [part for part in parts if part["type"] == "values"]
-    assert value_parts == [
-        {
-            "type": "values",
-            "ns": ("tools:call_1",),
-            "data": {"messages": [{"type": "human", "content": "hi"}], "count": 1},
-            "interrupts": (),
-        },
-        {
-            "type": "values",
-            "ns": ("tools:call_1",),
-            "data": {"count": 2},
-            "interrupts": (),
-        },
-    ]
-    _, kwargs = mock_async_client.runs.stream.call_args
-    assert set(kwargs["stream_mode"]) == {
-        "messages-tuple",
-        "values",
-        "compact",
-        "updates",
-    }
-
-
 @pytest.mark.skip(
     "Unskip this test to manually test the LangSmith Deployment integration"
 )

libs/langgraph/tests/test_stream_v2.py

@@ -90,25 +90,6 @@ def _make_messages_graph() -> StateGraph[
     return builder
 
 
-def _make_streaming_messages_graph() -> StateGraph[
-    MessagesState, None, MessagesState, MessagesState
-]:
-    model = FakeChatModel(messages=[AIMessage(content="hello world", id="ai-1")])
-
-    def call_model(state: MessagesState) -> dict[str, Any]:
-        streamed = model.stream(state["messages"])
-        message = next(streamed)
-        for chunk in streamed:
-            message += chunk
-        return {"messages": message}
-
-    builder = StateGraph(MessagesState, input_schema=MessagesState)
-    builder.add_node("call_model", call_model)
-    builder.add_edge(START, "call_model")
-    builder.add_edge("call_model", END)
-    return builder
-
-
 def _make_custom_graph() -> Any:
     @entrypoint()
     def graph(inputs: Any, *, writer: StreamWriter) -> Any:
@@ -183,26 +164,6 @@ class TestV1BackwardsCompat:
             ns, _data = chunk
             assert isinstance(ns, tuple)
 
-    def test_stream_v1_messages_keep_metadata_on_every_chunk(self) -> None:
-        graph = _make_streaming_messages_graph().compile()
-        chunks = list(graph.stream(_MSG_INPUT, stream_mode="messages"))
-        metadata = [meta for _message, meta in chunks]
-        assert len(metadata) >= 3
-        assert all(isinstance(meta, dict) for meta in metadata)
-
-    def test_stream_v1_messages_compact_dedupes_metadata(self) -> None:
-        graph = _make_streaming_messages_graph().compile()
-        chunks = list(graph.stream(_MSG_INPUT, stream_mode=["messages", "compact"]))
-        metadata = [
-            meta
-            for mode, payload in chunks
-            if mode == "messages"
-            for _message, meta in [payload]
-        ]
-        assert len(metadata) >= 3
-        assert isinstance(metadata[0], dict)
-        assert all(meta is None for meta in metadata[1:])
-
 
 # --- v2 sync stream ---
 
@@ -244,26 +205,6 @@ class TestV2Stream:
             assert isinstance(metadata, dict)
             assert "langgraph_node" in metadata
 
-    def test_messages_streaming_compact_dedupes_metadata(self) -> None:
-        graph = _make_streaming_messages_graph().compile()
-        chunks = list(
-            graph.stream(
-                _MSG_INPUT,
-                stream_mode=["messages", "compact"],
-                version="v2",
-            )
-        )
-        msg_chunks = [c for c in chunks if c["type"] == "messages"]
-        assert len(msg_chunks) >= 3
-        first_message, first_metadata = msg_chunks[0]["data"]
-        assert isinstance(first_message, BaseMessage)
-        assert isinstance(first_metadata, dict)
-        assert "langgraph_node" in first_metadata
-        for chunk in msg_chunks[1:]:
-            message, metadata = chunk["data"]
-            assert isinstance(message, BaseMessage)
-            assert metadata is None
-
     def test_custom(self) -> None:
         graph = _make_custom_graph()
         chunks = list(graph.stream({"key": "val"}, stream_mode="custom", version="v2"))
@@ -603,29 +544,6 @@ class TestV2StreamAsync:
             assert isinstance(metadata, dict)
             assert "langgraph_node" in metadata
 
-    @NEEDS_CONTEXTVARS
-    @pytest.mark.anyio
-    async def test_messages_streaming_compact_dedupes_metadata(self) -> None:
-        graph = _make_streaming_messages_graph().compile()
-        chunks = [
-            c
-            async for c in graph.astream(
-                _MSG_INPUT,
-                stream_mode=["messages", "compact"],
-                version="v2",
-            )
-        ]
-        msg_chunks = [c for c in chunks if c["type"] == "messages"]
-        assert len(msg_chunks) >= 3
-        first_message, first_metadata = msg_chunks[0]["data"]
-        assert isinstance(first_message, BaseMessage)
-        assert isinstance(first_metadata, dict)
-        assert "langgraph_node" in first_metadata
-        for chunk in msg_chunks[1:]:
-            message, metadata = chunk["data"]
-            assert isinstance(message, BaseMessage)
-            assert metadata is None
-
     @NEEDS_CONTEXTVARS
     @pytest.mark.anyio
     async def test_custom(self) -> None:
@@ -1211,7 +1129,7 @@ _OutputT = TypeVar("_OutputT")
 _StateT = TypeVar("_StateT")
 
 
-def _check_type_narrowing(part: StreamPart[_StateT, _OutputT]) -> None:
+def _check_type_narrowing(part: StreamPart[_OutputT, _StateT]) -> None:
     """Compile-time type narrowing checks — never called at runtime."""
     if part["type"] == "values":
         assert_type(part, ValuesStreamPart[_OutputT])

libs/langgraph/tests/test_time_travel.py

@@ -1116,631 +1116,6 @@ def test_subgraph_replay_from_subgraph_checkpoint(
     assert "post" in final_result["value"]
 
 
-def test_subgraph_time_travel_to_first_interrupt(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to a subgraph checkpoint at the FIRST interrupt.
-
-    Architecture:
-      Parent:    START --> executor (subgraph, checkpointer=True) --> END
-      Executor:  START --> step_a --> ask_1 (interrupt) --> ask_2 (interrupt) --> END
-
-    Flow: run through both interrupts, then time travel back to the subgraph
-    checkpoint captured at the first interrupt. ask_1 should re-fire,
-    step_a should NOT re-run. Then resume through both interrupts with new answers.
-    """
-
-    called: list[str] = []
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    executor = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("executor", executor)
-        .add_edge(START, "executor")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until first interrupt (ask_1)
-    result = graph.invoke({"value": []}, config)
-    assert "__interrupt__" in result
-    assert result["__interrupt__"][0].value == "Question 1?"
-
-    # Capture subgraph state at the first interrupt
-    parent_state = graph.get_state(config, subgraphs=True)
-    sub_config_at_first = parent_state.tasks[0].state.config
-
-    # Resume first interrupt
-    result = graph.invoke(Command(resume="answer_1"), config)
-    assert result["__interrupt__"][0].value == "Question 2?"
-
-    # Resume second interrupt to complete
-    result = graph.invoke(Command(resume="answer_2"), config)
-    assert "__interrupt__" not in result
-
-    # --- Scenario 1: Replay from subgraph checkpoint at 1st interrupt ---
-    called.clear()
-    replay_result = graph.invoke(None, sub_config_at_first)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 1?"
-    # step_a should NOT re-run — it was before this checkpoint
-    assert "step_a" not in called
-    # ask_1 re-fires because the interrupt replays
-    assert "ask_1" in called
-
-    # --- Scenario 2: Fork from subgraph checkpoint at 1st interrupt ---
-    called.clear()
-    fork_config = graph.update_state(sub_config_at_first, {"value": ["forked"]})
-    fork_result = graph.invoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 1?"
-    assert "step_a" not in called
-    assert "ask_1" in called
-
-
-def test_subgraph_time_travel_to_second_interrupt(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to a subgraph checkpoint at the SECOND interrupt.
-
-    Architecture:
-      Parent:    START --> executor (subgraph, checkpointer=True) --> END
-      Executor:  START --> step_a --> ask_1 (interrupt) --> ask_2 (interrupt) --> END
-
-    Flow: run through both interrupts resuming each, then time travel back to the
-    subgraph checkpoint at the second interrupt. Only ask_2 should re-fire.
-    Then resume with a new answer and verify state.
-    """
-
-    called: list[str] = []
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    executor = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("executor", executor)
-        .add_edge(START, "executor")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until first interrupt (ask_1)
-    result = graph.invoke({"value": []}, config)
-    assert result["__interrupt__"][0].value == "Question 1?"
-
-    # Resume first interrupt
-    result = graph.invoke(Command(resume="answer_1"), config)
-    assert result["__interrupt__"][0].value == "Question 2?"
-
-    # Capture subgraph state at the second interrupt
-    parent_state = graph.get_state(config, subgraphs=True)
-    sub_config = parent_state.tasks[0].state.config
-
-    # Resume second interrupt to complete the graph
-    result = graph.invoke(Command(resume="answer_2"), config)
-    assert "__interrupt__" not in result
-
-    # --- Scenario 1: Replay from subgraph checkpoint at 2nd interrupt ---
-    called.clear()
-    replay_result = graph.invoke(None, sub_config)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 2?"
-    # step_a and ask_1 should NOT re-run — they were before this checkpoint
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-    # --- Scenario 2: Fork from subgraph checkpoint at 2nd interrupt ---
-    called.clear()
-    fork_config = graph.update_state(sub_config, {"value": ["forked"]})
-    fork_result = graph.invoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 2?"
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-
-def test_subgraph_time_travel_after_completion(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to a subgraph checkpoint AFTER both interrupts are resolved.
-
-    Architecture:
-      Parent:    START --> executor (subgraph, checkpointer=True) --> END
-      Executor:  START --> step_a --> ask_1 (interrupt) --> ask_2 (interrupt) --> END
-
-    After completing the full flow, capture the subgraph's final state checkpoint
-    and replay from it — should be a no-op (no nodes re-run).
-    """
-
-    called: list[str] = []
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    executor = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("executor", executor)
-        .add_edge(START, "executor")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run through both interrupts
-    graph.invoke({"value": []}, config)
-    graph.invoke(Command(resume="answer_1"), config)
-
-    # Before resuming 2nd interrupt, get state history to find the
-    # subgraph checkpoint that will exist after ask_2 completes
-    graph.invoke(Command(resume="answer_2"), config)
-
-    # Get the final parent state — no pending tasks
-    final_state = graph.get_state(config)
-    assert len(final_state.tasks) == 0
-
-    # Replay from the final parent checkpoint — should be a no-op
-    called.clear()
-    replay_result = graph.invoke(None, final_state.config)
-    assert "__interrupt__" not in replay_result
-    assert "step_a" not in called
-    assert "ask_1" not in called
-    assert "ask_2" not in called
-    # All values should be present
-    assert "step_a_done" in replay_result["value"]
-    assert "ask_1:answer_1" in replay_result["value"]
-    assert "ask_2:answer_2" in replay_result["value"]
-
-
-def test_3_levels_deep_time_travel_to_first_interrupt(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to the innermost subgraph checkpoint at the FIRST interrupt.
-
-    Architecture:
-      Parent:  START --> outer (subgraph, checkpointer=True) --> END
-      Outer:   START --> inner (subgraph, checkpointer=True) --> END
-      Inner:   START --> step_a --> ask_1 (interrupt) --> ask_2 (interrupt) --> END
-    """
-
-    called: list[str] = []
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    inner = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    middle = (
-        StateGraph(State)
-        .add_node("inner", inner)
-        .add_edge(START, "inner")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("outer", middle)
-        .add_edge(START, "outer")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until first interrupt
-    result = graph.invoke({"value": []}, config)
-    assert result["__interrupt__"][0].value == "Question 1?"
-
-    # Capture innermost subgraph state at the first interrupt
-    parent_state = graph.get_state(config, subgraphs=True)
-    mid_state = parent_state.tasks[0].state
-    inner_config = mid_state.tasks[0].state.config
-
-    # Resume through both interrupts to complete
-    graph.invoke(Command(resume="answer_1"), config)
-    graph.invoke(Command(resume="answer_2"), config)
-
-    # --- Scenario 1: Replay from innermost checkpoint at 1st interrupt ---
-    called.clear()
-    replay_result = graph.invoke(None, inner_config)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 1?"
-    assert "step_a" not in called
-    assert "ask_1" in called
-
-    # --- Scenario 2: Fork from innermost checkpoint at 1st interrupt ---
-    called.clear()
-    fork_config = graph.update_state(inner_config, {"value": ["forked"]})
-    fork_result = graph.invoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 1?"
-    assert "step_a" not in called
-    assert "ask_1" in called
-
-
-def test_3_levels_deep_time_travel_to_second_interrupt(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to the innermost subgraph checkpoint at the SECOND interrupt.
-
-    Architecture:
-      Parent:  START --> outer (subgraph, checkpointer=True) --> END
-      Outer:   START --> inner (subgraph, checkpointer=True) --> END
-      Inner:   START --> step_a --> ask_1 (interrupt) --> ask_2 (interrupt) --> END
-    """
-
-    called: list[str] = []
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    inner = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    middle = (
-        StateGraph(State)
-        .add_node("inner", inner)
-        .add_edge(START, "inner")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("outer", middle)
-        .add_edge(START, "outer")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until first interrupt
-    graph.invoke({"value": []}, config)
-
-    # Resume first interrupt
-    result = graph.invoke(Command(resume="answer_1"), config)
-    assert result["__interrupt__"][0].value == "Question 2?"
-
-    # Capture innermost subgraph state at the second interrupt
-    parent_state = graph.get_state(config, subgraphs=True)
-    mid_state = parent_state.tasks[0].state
-    inner_config = mid_state.tasks[0].state.config
-
-    # Resume second interrupt to complete
-    graph.invoke(Command(resume="answer_2"), config)
-
-    # --- Scenario 1: Replay from innermost checkpoint at 2nd interrupt ---
-    called.clear()
-    replay_result = graph.invoke(None, inner_config)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 2?"
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-    # --- Scenario 2: Fork from innermost checkpoint at 2nd interrupt ---
-    called.clear()
-    fork_config = graph.update_state(inner_config, {"value": ["forked"]})
-    fork_result = graph.invoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 2?"
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-
-def test_3_levels_deep_time_travel_to_middle_subgraph(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to the MIDDLE-level subgraph checkpoint (not innermost).
-
-    Architecture:
-      Parent:  START --> outer (subgraph, checkpointer=True) --> END
-      Outer:   START --> inner (subgraph, checkpointer=True) --> END
-      Inner:   START --> step_a --> ask_1 (interrupt) --> ask_2 (interrupt) --> END
-
-    After completing the full flow, time travel back to the middle subgraph's
-    checkpoint at the second interrupt. The middle subgraph should replay the
-    inner subgraph from the correct point.
-    """
-
-    called: list[str] = []
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    inner = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    middle = (
-        StateGraph(State)
-        .add_node("inner", inner)
-        .add_edge(START, "inner")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("outer", middle)
-        .add_edge(START, "outer")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until first interrupt
-    graph.invoke({"value": []}, config)
-
-    # Resume first, capture middle config at second interrupt
-    graph.invoke(Command(resume="answer_1"), config)
-    parent_state = graph.get_state(config, subgraphs=True)
-    mid_config = parent_state.tasks[0].state.config
-
-    # Resume second to complete
-    graph.invoke(Command(resume="answer_2"), config)
-
-    # --- Scenario 1: Replay from middle-level subgraph checkpoint ---
-    # The middle subgraph's checkpoint knows about the inner subgraph's state
-    # via checkpoint_map, so the inner replays from the correct point.
-    called.clear()
-    replay_result = graph.invoke(None, mid_config)
-    assert "__interrupt__" in replay_result
-
-    # --- Scenario 2: Fork from middle-level subgraph checkpoint ---
-    called.clear()
-    fork_config = graph.update_state(mid_config, {"value": ["forked"]})
-    fork_result = graph.invoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-
-
-def test_3_levels_deep_middle_has_interrupts(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel when the MIDDLE subgraph itself has interrupts.
-
-    Architecture:
-      Parent:  START --> outer (subgraph, checkpointer=True) --> END
-      Outer:   START --> pre (interrupt) --> inner (subgraph, checkpointer=True) --> END
-      Inner:   START --> step_a --> ask_1 (interrupt) --> END
-
-    Flow: run through both interrupts (pre then ask_1), then time travel back to
-    the middle subgraph checkpoint at each interrupt point.
-    """
-
-    called: list[str] = []
-
-    def pre(state: State) -> State:
-        called.append("pre")
-        answer = interrupt("Pre-question?")
-        return {"value": [f"pre:{answer}"]}
-
-    def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    inner = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    middle = (
-        StateGraph(State)
-        .add_node("pre", pre)
-        .add_node("inner", inner)
-        .add_edge(START, "pre")
-        .add_edge("pre", "inner")
-        .add_edge("inner", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("outer", middle)
-        .add_edge(START, "outer")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until first interrupt (pre in middle subgraph)
-    result = graph.invoke({"value": []}, config)
-    assert result["__interrupt__"][0].value == "Pre-question?"
-
-    # Capture middle subgraph config at the pre interrupt
-    parent_state = graph.get_state(config, subgraphs=True)
-    mid_config_at_pre = parent_state.tasks[0].state.config
-
-    # Resume pre, hits ask_1 in inner subgraph
-    result = graph.invoke(Command(resume="pre_answer"), config)
-    assert result["__interrupt__"][0].value == "Question 1?"
-
-    # Capture middle subgraph config at the ask_1 interrupt
-    parent_state = graph.get_state(config, subgraphs=True)
-    mid_config_at_ask1 = parent_state.tasks[0].state.config
-
-    # Resume ask_1 to complete
-    result = graph.invoke(Command(resume="answer_1"), config)
-    assert "__interrupt__" not in result
-
-    # --- Time travel to middle checkpoint at pre interrupt ---
-    called.clear()
-    replay_result = graph.invoke(None, mid_config_at_pre)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Pre-question?"
-    # pre should re-fire (interrupt replays), but nothing else should run
-    assert "pre" in called
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-    # Fork from middle checkpoint at pre interrupt
-    called.clear()
-    fork_config = graph.update_state(mid_config_at_pre, {"value": ["forked"]})
-    fork_result = graph.invoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Pre-question?"
-    assert "pre" in called
-    assert "step_a" not in called
-
-    # --- Time travel to middle checkpoint at ask_1 interrupt ---
-    called.clear()
-    replay_result = graph.invoke(None, mid_config_at_ask1)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 1?"
-    # pre should NOT re-run (it completed before this checkpoint)
-    assert "pre" not in called
-    # ask_1 re-fires
-    assert "ask_1" in called
-
-    # Fork from middle checkpoint at ask_1 interrupt
-    called.clear()
-    fork_config = graph.update_state(mid_config_at_ask1, {"value": ["forked"]})
-    fork_result = graph.invoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 1?"
-    assert "pre" not in called
-    assert "ask_1" in called
-
-
 # ---------------------------------------------------------------------------
 # Section 6: __copy__ / update_state(None)
 # ---------------------------------------------------------------------------

libs/langgraph/tests/test_time_travel_async.py

@@ -1051,552 +1051,6 @@ async def test_subgraph_interrupt_full_flow_no_sub_checkpointer(
     assert "post" in final_result["value"]
 
 
-@NEEDS_CONTEXTVARS
-async def test_subgraph_time_travel_to_first_interrupt_async(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to a subgraph checkpoint at the FIRST interrupt (async)."""
-
-    called: list[str] = []
-
-    async def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    async def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    async def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    executor = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("executor", executor)
-        .add_edge(START, "executor")
-        .compile(checkpointer=async_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until first interrupt (ask_1)
-    result = await graph.ainvoke({"value": []}, config)
-    assert result["__interrupt__"][0].value == "Question 1?"
-
-    # Capture subgraph state at the first interrupt
-    parent_state = await graph.aget_state(config, subgraphs=True)
-    sub_config_at_first = parent_state.tasks[0].state.config
-
-    # Resume through both interrupts to complete
-    await graph.ainvoke(Command(resume="answer_1"), config)
-    await graph.ainvoke(Command(resume="answer_2"), config)
-
-    # --- Scenario 1: Replay from subgraph checkpoint at 1st interrupt ---
-    called.clear()
-    replay_result = await graph.ainvoke(None, sub_config_at_first)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 1?"
-    assert "step_a" not in called
-    assert "ask_1" in called
-
-    # --- Scenario 2: Fork from subgraph checkpoint at 1st interrupt ---
-    called.clear()
-    fork_config = await graph.aupdate_state(sub_config_at_first, {"value": ["forked"]})
-    fork_result = await graph.ainvoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 1?"
-    assert "step_a" not in called
-    assert "ask_1" in called
-
-
-@NEEDS_CONTEXTVARS
-async def test_subgraph_time_travel_to_second_interrupt_async(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to a subgraph checkpoint at the SECOND interrupt (async)."""
-
-    called: list[str] = []
-
-    async def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    async def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    async def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    executor = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("executor", executor)
-        .add_edge(START, "executor")
-        .compile(checkpointer=async_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until first interrupt
-    await graph.ainvoke({"value": []}, config)
-
-    # Resume first interrupt
-    result = await graph.ainvoke(Command(resume="answer_1"), config)
-    assert result["__interrupt__"][0].value == "Question 2?"
-
-    # Capture subgraph state at the second interrupt
-    parent_state = await graph.aget_state(config, subgraphs=True)
-    sub_config = parent_state.tasks[0].state.config
-
-    # Resume second interrupt to complete
-    await graph.ainvoke(Command(resume="answer_2"), config)
-
-    # --- Scenario 1: Replay from subgraph checkpoint at 2nd interrupt ---
-    called.clear()
-    replay_result = await graph.ainvoke(None, sub_config)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 2?"
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-    # --- Scenario 2: Fork from subgraph checkpoint at 2nd interrupt ---
-    called.clear()
-    fork_config = await graph.aupdate_state(sub_config, {"value": ["forked"]})
-    fork_result = await graph.ainvoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 2?"
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-
-@NEEDS_CONTEXTVARS
-async def test_subgraph_time_travel_after_completion_async(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to a subgraph checkpoint AFTER both interrupts resolved (async)."""
-
-    called: list[str] = []
-
-    async def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    async def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    async def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    executor = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("executor", executor)
-        .add_edge(START, "executor")
-        .compile(checkpointer=async_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    await graph.ainvoke({"value": []}, config)
-    await graph.ainvoke(Command(resume="answer_1"), config)
-    await graph.ainvoke(Command(resume="answer_2"), config)
-
-    final_state = await graph.aget_state(config)
-    assert len(final_state.tasks) == 0
-
-    # Replay from the final parent checkpoint — should be a no-op
-    called.clear()
-    replay_result = await graph.ainvoke(None, final_state.config)
-    assert "__interrupt__" not in replay_result
-    assert "step_a" not in called
-    assert "ask_1" not in called
-    assert "ask_2" not in called
-    assert "step_a_done" in replay_result["value"]
-    assert "ask_1:answer_1" in replay_result["value"]
-    assert "ask_2:answer_2" in replay_result["value"]
-
-
-@NEEDS_CONTEXTVARS
-async def test_3_levels_deep_time_travel_to_first_interrupt_async(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to innermost subgraph checkpoint at FIRST interrupt (async, 3 levels)."""
-
-    called: list[str] = []
-
-    async def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    async def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    async def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    inner = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    middle = (
-        StateGraph(State)
-        .add_node("inner", inner)
-        .add_edge(START, "inner")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("outer", middle)
-        .add_edge(START, "outer")
-        .compile(checkpointer=async_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    result = await graph.ainvoke({"value": []}, config)
-    assert result["__interrupt__"][0].value == "Question 1?"
-
-    parent_state = await graph.aget_state(config, subgraphs=True)
-    mid_state = parent_state.tasks[0].state
-    inner_config = mid_state.tasks[0].state.config
-
-    await graph.ainvoke(Command(resume="answer_1"), config)
-    await graph.ainvoke(Command(resume="answer_2"), config)
-
-    # --- Scenario 1: Replay from innermost checkpoint at 1st interrupt ---
-    called.clear()
-    replay_result = await graph.ainvoke(None, inner_config)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 1?"
-    assert "step_a" not in called
-    assert "ask_1" in called
-
-    # --- Scenario 2: Fork from innermost checkpoint at 1st interrupt ---
-    called.clear()
-    fork_config = await graph.aupdate_state(inner_config, {"value": ["forked"]})
-    fork_result = await graph.ainvoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 1?"
-    assert "step_a" not in called
-    assert "ask_1" in called
-
-
-@NEEDS_CONTEXTVARS
-async def test_3_levels_deep_time_travel_to_second_interrupt_async(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to innermost subgraph checkpoint at SECOND interrupt (async, 3 levels)."""
-
-    called: list[str] = []
-
-    async def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    async def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    async def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    inner = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    middle = (
-        StateGraph(State)
-        .add_node("inner", inner)
-        .add_edge(START, "inner")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("outer", middle)
-        .add_edge(START, "outer")
-        .compile(checkpointer=async_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    await graph.ainvoke({"value": []}, config)
-
-    result = await graph.ainvoke(Command(resume="answer_1"), config)
-    assert result["__interrupt__"][0].value == "Question 2?"
-
-    parent_state = await graph.aget_state(config, subgraphs=True)
-    mid_state = parent_state.tasks[0].state
-    inner_config = mid_state.tasks[0].state.config
-
-    await graph.ainvoke(Command(resume="answer_2"), config)
-
-    # --- Scenario 1: Replay ---
-    called.clear()
-    replay_result = await graph.ainvoke(None, inner_config)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 2?"
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-    # --- Scenario 2: Fork ---
-    called.clear()
-    fork_config = await graph.aupdate_state(inner_config, {"value": ["forked"]})
-    fork_result = await graph.ainvoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 2?"
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-
-@NEEDS_CONTEXTVARS
-async def test_3_levels_deep_time_travel_to_middle_subgraph_async(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel to the MIDDLE-level subgraph checkpoint (async, 3 levels)."""
-
-    called: list[str] = []
-
-    async def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    async def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    async def ask_2(state: State) -> State:
-        called.append("ask_2")
-        answer = interrupt("Question 2?")
-        return {"value": [f"ask_2:{answer}"]}
-
-    inner = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_node("ask_2", ask_2)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "ask_2")
-        .add_edge("ask_2", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    middle = (
-        StateGraph(State)
-        .add_node("inner", inner)
-        .add_edge(START, "inner")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("outer", middle)
-        .add_edge(START, "outer")
-        .compile(checkpointer=async_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    await graph.ainvoke({"value": []}, config)
-    await graph.ainvoke(Command(resume="answer_1"), config)
-
-    parent_state = await graph.aget_state(config, subgraphs=True)
-    mid_config = parent_state.tasks[0].state.config
-
-    await graph.ainvoke(Command(resume="answer_2"), config)
-
-    # --- Scenario 1: Replay from middle-level subgraph checkpoint ---
-    # The middle subgraph's checkpoint knows about the inner subgraph's state
-    # via checkpoint_map, so the inner replays from the correct point.
-    called.clear()
-    replay_result = await graph.ainvoke(None, mid_config)
-    assert "__interrupt__" in replay_result
-
-    # --- Scenario 2: Fork from middle-level subgraph checkpoint ---
-    called.clear()
-    fork_config = await graph.aupdate_state(mid_config, {"value": ["forked"]})
-    fork_result = await graph.ainvoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-
-
-@NEEDS_CONTEXTVARS
-async def test_3_levels_deep_middle_has_interrupts_async(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Time travel when the MIDDLE subgraph itself has interrupts (async)."""
-
-    called: list[str] = []
-
-    async def pre(state: State) -> State:
-        called.append("pre")
-        answer = interrupt("Pre-question?")
-        return {"value": [f"pre:{answer}"]}
-
-    async def step_a(state: State) -> State:
-        called.append("step_a")
-        return {"value": ["step_a_done"]}
-
-    async def ask_1(state: State) -> State:
-        called.append("ask_1")
-        answer = interrupt("Question 1?")
-        return {"value": [f"ask_1:{answer}"]}
-
-    inner = (
-        StateGraph(State)
-        .add_node("step_a", step_a)
-        .add_node("ask_1", ask_1)
-        .add_edge(START, "step_a")
-        .add_edge("step_a", "ask_1")
-        .add_edge("ask_1", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    middle = (
-        StateGraph(State)
-        .add_node("pre", pre)
-        .add_node("inner", inner)
-        .add_edge(START, "pre")
-        .add_edge("pre", "inner")
-        .add_edge("inner", "__end__")
-        .compile(checkpointer=True)
-    )
-
-    graph = (
-        StateGraph(State)
-        .add_node("outer", middle)
-        .add_edge(START, "outer")
-        .compile(checkpointer=async_checkpointer)
-    )
-
-    config = {"configurable": {"thread_id": "1"}}
-
-    # Run until first interrupt (pre in middle subgraph)
-    result = await graph.ainvoke({"value": []}, config)
-    assert result["__interrupt__"][0].value == "Pre-question?"
-
-    # Capture middle subgraph config at the pre interrupt
-    parent_state = await graph.aget_state(config, subgraphs=True)
-    mid_config_at_pre = parent_state.tasks[0].state.config
-
-    # Resume pre, hits ask_1 in inner subgraph
-    result = await graph.ainvoke(Command(resume="pre_answer"), config)
-    assert result["__interrupt__"][0].value == "Question 1?"
-
-    # Capture middle subgraph config at the ask_1 interrupt
-    parent_state = await graph.aget_state(config, subgraphs=True)
-    mid_config_at_ask1 = parent_state.tasks[0].state.config
-
-    # Resume ask_1 to complete
-    result = await graph.ainvoke(Command(resume="answer_1"), config)
-    assert "__interrupt__" not in result
-
-    # --- Time travel to middle checkpoint at pre interrupt ---
-    called.clear()
-    replay_result = await graph.ainvoke(None, mid_config_at_pre)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Pre-question?"
-    assert "pre" in called
-    assert "step_a" not in called
-    assert "ask_1" not in called
-
-    # Fork from middle checkpoint at pre interrupt
-    called.clear()
-    fork_config = await graph.aupdate_state(mid_config_at_pre, {"value": ["forked"]})
-    fork_result = await graph.ainvoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Pre-question?"
-    assert "pre" in called
-    assert "step_a" not in called
-
-    # --- Time travel to middle checkpoint at ask_1 interrupt ---
-    called.clear()
-    replay_result = await graph.ainvoke(None, mid_config_at_ask1)
-    assert "__interrupt__" in replay_result
-    assert replay_result["__interrupt__"][0].value == "Question 1?"
-    assert "pre" not in called
-    assert "ask_1" in called
-
-    # Fork from middle checkpoint at ask_1 interrupt
-    called.clear()
-    fork_config = await graph.aupdate_state(mid_config_at_ask1, {"value": ["forked"]})
-    fork_result = await graph.ainvoke(None, fork_config)
-    assert "__interrupt__" in fork_result
-    assert fork_result["__interrupt__"][0].value == "Question 1?"
-    assert "pre" not in called
-    assert "ask_1" in called
-
-
 # ---------------------------------------------------------------------------
 # Section 6: __copy__ / update_state(None)
 # ---------------------------------------------------------------------------

libs/langgraph/uv.lock

@@ -1367,7 +1367,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.1.2"
+version = "1.1.0"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -3615,19 +3615,21 @@ wheels = [
 
 [[package]]
 name = "tornado"
-version = "6.5.5"
+version = "6.5.4"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/f8/f1/3173dfa4a18db4a9b03e5d55325559dab51ee653763bb8745a75af491286/tornado-6.5.5.tar.gz", hash = "sha256:192b8f3ea91bd7f1f50c06955416ed76c6b72f96779b962f07f911b91e8d30e9", size = 516006, upload-time = "2026-03-10T21:31:02.067Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/37/1d/0a336abf618272d53f62ebe274f712e213f5a03c0b2339575430b8362ef2/tornado-6.5.4.tar.gz", hash = "sha256:a22fa9047405d03260b483980635f0b041989d8bcc9a313f8fe18b411d84b1d7", size = 513632, upload-time = "2025-12-15T19:21:03.836Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/59/8c/77f5097695f4dd8255ecbd08b2a1ed8ba8b953d337804dd7080f199e12bf/tornado-6.5.5-cp39-abi3-macosx_10_9_universal2.whl", hash = "sha256:487dc9cc380e29f58c7ab88f9e27cdeef04b2140862e5076a66fb6bb68bb1bfa", size = 445983, upload-time = "2026-03-10T21:30:44.28Z" },
-    { url = "https://files.pythonhosted.org/packages/ab/5e/7625b76cd10f98f1516c36ce0346de62061156352353ef2da44e5c21523c/tornado-6.5.5-cp39-abi3-macosx_10_9_x86_64.whl", hash = "sha256:65a7f1d46d4bb41df1ac99f5fcb685fb25c7e61613742d5108b010975a9a6521", size = 444246, upload-time = "2026-03-10T21:30:46.571Z" },
-    { url = "https://files.pythonhosted.org/packages/b2/04/7b5705d5b3c0fab088f434f9c83edac1573830ca49ccf29fb83bf7178eec/tornado-6.5.5-cp39-abi3-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:e74c92e8e65086b338fd56333fb9a68b9f6f2fe7ad532645a290a464bcf46be5", size = 447229, upload-time = "2026-03-10T21:30:48.273Z" },
-    { url = "https://files.pythonhosted.org/packages/34/01/74e034a30ef59afb4097ef8659515e96a39d910b712a89af76f5e4e1f93c/tornado-6.5.5-cp39-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:435319e9e340276428bbdb4e7fa732c2d399386d1de5686cb331ec8eee754f07", size = 448192, upload-time = "2026-03-10T21:30:51.22Z" },
-    { url = "https://files.pythonhosted.org/packages/be/00/fe9e02c5a96429fce1a1d15a517f5d8444f9c412e0bb9eadfbe3b0fc55bf/tornado-6.5.5-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:3f54aa540bdbfee7b9eb268ead60e7d199de5021facd276819c193c0fb28ea4e", size = 448039, upload-time = "2026-03-10T21:30:53.52Z" },
-    { url = "https://files.pythonhosted.org/packages/82/9e/656ee4cec0398b1d18d0f1eb6372c41c6b889722641d84948351ae19556d/tornado-6.5.5-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:36abed1754faeb80fbd6e64db2758091e1320f6bba74a4cf8c09cd18ccce8aca", size = 447445, upload-time = "2026-03-10T21:30:55.541Z" },
-    { url = "https://files.pythonhosted.org/packages/5a/76/4921c00511f88af86a33de770d64141170f1cfd9c00311aea689949e274e/tornado-6.5.5-cp39-abi3-win32.whl", hash = "sha256:dd3eafaaeec1c7f2f8fdcd5f964e8907ad788fe8a5a32c4426fbbdda621223b7", size = 448582, upload-time = "2026-03-10T21:30:57.142Z" },
-    { url = "https://files.pythonhosted.org/packages/2c/23/f6c6112a04d28eed765e374435fb1a9198f73e1ec4b4024184f21faeb1ad/tornado-6.5.5-cp39-abi3-win_amd64.whl", hash = "sha256:6443a794ba961a9f619b1ae926a2e900ac20c34483eea67be4ed8f1e58d3ef7b", size = 448990, upload-time = "2026-03-10T21:30:58.857Z" },
-    { url = "https://files.pythonhosted.org/packages/b7/c8/876602cbc96469911f0939f703453c1157b0c826ecb05bdd32e023397d4e/tornado-6.5.5-cp39-abi3-win_arm64.whl", hash = "sha256:2c9a876e094109333f888539ddb2de4361743e5d21eece20688e3e351e4990a6", size = 448016, upload-time = "2026-03-10T21:31:00.43Z" },
+    { url = "https://files.pythonhosted.org/packages/ab/a9/e94a9d5224107d7ce3cc1fab8d5dc97f5ea351ccc6322ee4fb661da94e35/tornado-6.5.4-cp39-abi3-macosx_10_9_universal2.whl", hash = "sha256:d6241c1a16b1c9e4cc28148b1cda97dd1c6cb4fb7068ac1bedc610768dff0ba9", size = 443909, upload-time = "2025-12-15T19:20:48.382Z" },
+    { url = "https://files.pythonhosted.org/packages/db/7e/f7b8d8c4453f305a51f80dbb49014257bb7d28ccb4bbb8dd328ea995ecad/tornado-6.5.4-cp39-abi3-macosx_10_9_x86_64.whl", hash = "sha256:2d50f63dda1d2cac3ae1fa23d254e16b5e38153758470e9956cbc3d813d40843", size = 442163, upload-time = "2025-12-15T19:20:49.791Z" },
+    { url = "https://files.pythonhosted.org/packages/ba/b5/206f82d51e1bfa940ba366a8d2f83904b15942c45a78dd978b599870ab44/tornado-6.5.4-cp39-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:d1cf66105dc6acb5af613c054955b8137e34a03698aa53272dbda4afe252be17", size = 445746, upload-time = "2025-12-15T19:20:51.491Z" },
+    { url = "https://files.pythonhosted.org/packages/8e/9d/1a3338e0bd30ada6ad4356c13a0a6c35fbc859063fa7eddb309183364ac1/tornado-6.5.4-cp39-abi3-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:50ff0a58b0dc97939d29da29cd624da010e7f804746621c78d14b80238669335", size = 445083, upload-time = "2025-12-15T19:20:52.778Z" },
+    { url = "https://files.pythonhosted.org/packages/50/d4/e51d52047e7eb9a582da59f32125d17c0482d065afd5d3bc435ff2120dc5/tornado-6.5.4-cp39-abi3-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e5fb5e04efa54cf0baabdd10061eb4148e0be137166146fff835745f59ab9f7f", size = 445315, upload-time = "2025-12-15T19:20:53.996Z" },
+    { url = "https://files.pythonhosted.org/packages/27/07/2273972f69ca63dbc139694a3fc4684edec3ea3f9efabf77ed32483b875c/tornado-6.5.4-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:9c86b1643b33a4cd415f8d0fe53045f913bf07b4a3ef646b735a6a86047dda84", size = 446003, upload-time = "2025-12-15T19:20:56.101Z" },
+    { url = "https://files.pythonhosted.org/packages/d1/83/41c52e47502bf7260044413b6770d1a48dda2f0246f95ee1384a3cd9c44a/tornado-6.5.4-cp39-abi3-musllinux_1_2_i686.whl", hash = "sha256:6eb82872335a53dd063a4f10917b3efd28270b56a33db69009606a0312660a6f", size = 445412, upload-time = "2025-12-15T19:20:57.398Z" },
+    { url = "https://files.pythonhosted.org/packages/10/c7/bc96917f06cbee182d44735d4ecde9c432e25b84f4c2086143013e7b9e52/tornado-6.5.4-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:6076d5dda368c9328ff41ab5d9dd3608e695e8225d1cd0fd1e006f05da3635a8", size = 445392, upload-time = "2025-12-15T19:20:58.692Z" },
+    { url = "https://files.pythonhosted.org/packages/0c/1a/d7592328d037d36f2d2462f4bc1fbb383eec9278bc786c1b111cbbd44cfa/tornado-6.5.4-cp39-abi3-win32.whl", hash = "sha256:1768110f2411d5cd281bac0a090f707223ce77fd110424361092859e089b38d1", size = 446481, upload-time = "2025-12-15T19:21:00.008Z" },
+    { url = "https://files.pythonhosted.org/packages/d6/6d/c69be695a0a64fd37a97db12355a035a6d90f79067a3cf936ec2b1dc38cd/tornado-6.5.4-cp39-abi3-win_amd64.whl", hash = "sha256:fa07d31e0cd85c60713f2b995da613588aa03e1303d75705dca6af8babc18ddc", size = 446886, upload-time = "2025-12-15T19:21:01.287Z" },
+    { url = "https://files.pythonhosted.org/packages/50/49/8dc3fd90902f70084bd2cd059d576ddb4f8bb44c2c7c0e33a11422acb17e/tornado-6.5.4-cp39-abi3-win_arm64.whl", hash = "sha256:053e6e16701eb6cbe641f308f4c1a9541f91b6261991160391bfc342e8a551a1", size = 445910, upload-time = "2025-12-15T19:21:02.571Z" },
 ]
 
 [[package]]

libs/prebuilt/uv.lock

@@ -268,7 +268,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.1.2"
+version = "1.1.0"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },

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

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

@@ -4,11 +4,10 @@ from __future__ import annotations
 
 import warnings
 from collections.abc import Mapping, Sequence
-from datetime import datetime, tzinfo
+from datetime import datetime
 from typing import Any
 
 from langgraph_sdk._async.http import HttpClient
-from langgraph_sdk._shared.utilities import _resolve_timezone
 from langgraph_sdk.schema import (
     All,
     Config,
@@ -71,7 +70,6 @@ class CronClient:
         multitask_strategy: str | None = None,
         end_time: datetime | None = None,
         enabled: bool | None = None,
-        timezone: str | tzinfo | None = None,
         stream_mode: StreamMode | Sequence[StreamMode] | None = None,
         stream_subgraphs: bool | None = None,
         stream_resumable: bool | None = None,
@@ -86,7 +84,7 @@ class CronClient:
             assistant_id: The assistant ID or graph name to use for the cron job.
                 If using graph name, will default to first assistant created from that graph.
             schedule: The cron schedule to execute this job on.
-                Schedules are interpreted in UTC unless a timezone is specified.
+                Schedules are interpreted in UTC.
             input: The input to the graph.
             metadata: Metadata to assign to the cron job runs.
             config: The configuration for the assistant.
@@ -102,7 +100,6 @@ class CronClient:
                 Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
             end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely.
             enabled: Whether the cron job is enabled or not.
-            timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).
             stream_mode: The stream mode(s) to use.
             stream_subgraphs: Whether to stream output from subgraphs.
             stream_resumable: Whether to persist the stream chunks in order to resume the stream later.
@@ -155,7 +152,6 @@ class CronClient:
             "webhook": webhook,
             "end_time": end_time.isoformat() if end_time else None,
             "enabled": enabled,
-            "timezone": _resolve_timezone(timezone),
             "stream_mode": stream_mode,
             "stream_subgraphs": stream_subgraphs,
             "stream_resumable": stream_resumable,
@@ -188,7 +184,6 @@ class CronClient:
         multitask_strategy: str | None = None,
         end_time: datetime | None = None,
         enabled: bool | None = None,
-        timezone: str | tzinfo | None = None,
         stream_mode: StreamMode | Sequence[StreamMode] | None = None,
         stream_subgraphs: bool | None = None,
         stream_resumable: bool | None = None,
@@ -202,7 +197,7 @@ class CronClient:
             assistant_id: The assistant ID or graph name to use for the cron job.
                 If using graph name, will default to first assistant created from that graph.
             schedule: The cron schedule to execute this job on.
-                Schedules are interpreted in UTC unless a timezone is specified.
+                Schedules are interpreted in UTC.
             input: The input to the graph.
             metadata: Metadata to assign to the cron job runs.
             config: The configuration for the assistant.
@@ -220,7 +215,6 @@ class CronClient:
                 Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
             end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely.
             enabled: Whether the cron job is enabled or not.
-            timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).
             stream_mode: The stream mode(s) to use.
             stream_subgraphs: Whether to stream output from subgraphs.
             stream_resumable: Whether to persist the stream chunks in order to resume the stream later.
@@ -274,7 +268,6 @@ class CronClient:
             "on_run_completed": on_run_completed,
             "end_time": end_time.isoformat() if end_time else None,
             "enabled": enabled,
-            "timezone": _resolve_timezone(timezone),
             "stream_mode": stream_mode,
             "stream_subgraphs": stream_subgraphs,
             "stream_resumable": stream_resumable,
@@ -331,7 +324,6 @@ class CronClient:
         interrupt_after: All | list[str] | None = None,
         on_run_completed: OnCompletionBehavior | None = None,
         enabled: bool | None = None,
-        timezone: str | tzinfo | None = None,
         stream_mode: StreamMode | Sequence[StreamMode] | None = None,
         stream_subgraphs: bool | None = None,
         stream_resumable: bool | None = None,
@@ -344,7 +336,7 @@ class CronClient:
         Args:
             cron_id: The cron ID to update.
             schedule: The cron schedule to execute this job on.
-                Schedules are interpreted in UTC unless a timezone is specified.
+                Schedules are interpreted in UTC.
             end_time: The end date to stop running the cron.
             input: The input to the graph.
             metadata: Metadata to assign to the cron job runs.
@@ -358,7 +350,6 @@ class CronClient:
                 after execution. 'keep' creates a new thread for each execution but does not
                 clean them up.
             enabled: Enable or disable the cron job.
-            timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).
             stream_mode: The stream mode(s) to use.
             stream_subgraphs: Whether to stream output from subgraphs.
             stream_resumable: Whether to persist the stream chunks in order to resume the stream later.
@@ -393,7 +384,6 @@ class CronClient:
             "interrupt_after": interrupt_after,
             "on_run_completed": on_run_completed,
             "enabled": enabled,
-            "timezone": _resolve_timezone(timezone),
             "stream_mode": stream_mode,
             "stream_subgraphs": stream_subgraphs,
             "stream_resumable": stream_resumable,

libs/sdk-py/langgraph_sdk/_shared/utilities.py

@@ -6,17 +6,13 @@ import functools
 import os
 import re
 from collections.abc import Mapping
-from datetime import tzinfo
-from typing import TYPE_CHECKING, Any, cast
+from typing import Any, cast
 
 import httpx
 
 import langgraph_sdk
 from langgraph_sdk.schema import RunCreateMetadata
 
-if TYPE_CHECKING:
-    from zoneinfo import ZoneInfo
-
 RESERVED_HEADERS = ("x-api-key",)
 
 NOT_PROVIDED = cast(None, object())
@@ -129,35 +125,6 @@ def _sse_to_v2_dict(event: str, data: Any) -> dict[str, Any] | None:
     return result
 
 
-def _resolve_timezone(tz: str | tzinfo | ZoneInfo | None) -> str | None:
-    """Convert a timezone argument to an IANA timezone string.
-
-    Accepts:
-        - A string (returned as-is, assumed to be an IANA timezone name)
-        - A ``datetime.tzinfo`` instance (e.g. ``zoneinfo.ZoneInfo("America/New_York")``,
-          ``datetime.timezone.utc``). The ``key`` attribute is used if available,
-          otherwise ``tzname(None)`` is used.
-        - ``None`` (returned as ``None``)
-    """
-    if tz is None or isinstance(tz, str):
-        return tz
-    if isinstance(tz, tzinfo):
-        # ZoneInfo objects have a .key attribute with the IANA name
-        if hasattr(tz, "key"):
-            return tz.key  # type: ignore[union-attr]
-        # Fall back to tzname for fixed-offset timezones like datetime.timezone.utc
-        name = tz.tzname(None)
-        if name is not None:
-            return name
-        raise ValueError(
-            f"Cannot determine timezone name from {tz!r}. "
-            "Use a zoneinfo.ZoneInfo instance or pass a string like 'America/New_York'."
-        )
-    raise TypeError(
-        f"Expected str, datetime.tzinfo, or None for timezone, got {type(tz).__name__}"
-    )
-
-
 def _provided_vals(d: Mapping[str, Any]) -> dict[str, Any]:
     return {k: v for k, v in d.items() if v is not None}
 

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

@@ -4,10 +4,9 @@ from __future__ import annotations
 
 import warnings
 from collections.abc import Mapping, Sequence
-from datetime import datetime, tzinfo
+from datetime import datetime
 from typing import Any
 
-from langgraph_sdk._shared.utilities import _resolve_timezone
 from langgraph_sdk._sync.http import SyncHttpClient
 from langgraph_sdk.schema import (
     All,
@@ -65,7 +64,6 @@ class SyncCronClient:
         multitask_strategy: str | None = None,
         end_time: datetime | None = None,
         enabled: bool | None = None,
-        timezone: str | tzinfo | None = None,
         stream_mode: StreamMode | Sequence[StreamMode] | None = None,
         stream_subgraphs: bool | None = None,
         stream_resumable: bool | None = None,
@@ -80,7 +78,7 @@ class SyncCronClient:
             assistant_id: The assistant ID or graph name to use for the cron job.
                 If using graph name, will default to first assistant created from that graph.
             schedule: The cron schedule to execute this job on.
-                Schedules are interpreted in UTC unless a timezone is specified.
+                Schedules are interpreted in UTC.
             input: The input to the graph.
             metadata: Metadata to assign to the cron job runs.
             config: The configuration for the assistant.
@@ -94,7 +92,6 @@ class SyncCronClient:
                 Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
             end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely.
             enabled: Whether the cron job is enabled. By default, it is considered enabled.
-            timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).
             stream_mode: The stream mode(s) to use.
             stream_subgraphs: Whether to stream output from subgraphs.
             stream_resumable: Whether to persist the stream chunks in order to resume the stream later.
@@ -147,7 +144,6 @@ class SyncCronClient:
             "multitask_strategy": multitask_strategy,
             "end_time": end_time.isoformat() if end_time else None,
             "enabled": enabled,
-            "timezone": _resolve_timezone(timezone),
             "stream_mode": stream_mode,
             "stream_subgraphs": stream_subgraphs,
             "stream_resumable": stream_resumable,
@@ -178,7 +174,6 @@ class SyncCronClient:
         multitask_strategy: str | None = None,
         end_time: datetime | None = None,
         enabled: bool | None = None,
-        timezone: str | tzinfo | None = None,
         stream_mode: StreamMode | Sequence[StreamMode] | None = None,
         stream_subgraphs: bool | None = None,
         stream_resumable: bool | None = None,
@@ -192,7 +187,7 @@ class SyncCronClient:
             assistant_id: The assistant ID or graph name to use for the cron job.
                 If using graph name, will default to first assistant created from that graph.
             schedule: The cron schedule to execute this job on.
-                Schedules are interpreted in UTC unless a timezone is specified.
+                Schedules are interpreted in UTC.
             input: The input to the graph.
             metadata: Metadata to assign to the cron job runs.
             config: The configuration for the assistant.
@@ -210,7 +205,6 @@ class SyncCronClient:
                 Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.
             end_time: The time to stop running the cron job. If not provided, the cron job will run indefinitely.
             enabled: Whether the cron job is enabled. By default, it is considered enabled.
-            timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).
             stream_mode: The stream mode(s) to use.
             stream_subgraphs: Whether to stream output from subgraphs.
             stream_resumable: Whether to persist the stream chunks in order to resume the stream later.
@@ -265,7 +259,6 @@ class SyncCronClient:
             "multitask_strategy": multitask_strategy,
             "end_time": end_time.isoformat() if end_time else None,
             "enabled": enabled,
-            "timezone": _resolve_timezone(timezone),
             "stream_mode": stream_mode,
             "stream_subgraphs": stream_subgraphs,
             "stream_resumable": stream_resumable,
@@ -320,7 +313,6 @@ class SyncCronClient:
         interrupt_after: All | list[str] | None = None,
         on_run_completed: OnCompletionBehavior | None = None,
         enabled: bool | None = None,
-        timezone: str | tzinfo | None = None,
         stream_mode: StreamMode | Sequence[StreamMode] | None = None,
         stream_subgraphs: bool | None = None,
         stream_resumable: bool | None = None,
@@ -333,7 +325,7 @@ class SyncCronClient:
         Args:
             cron_id: The cron ID to update.
             schedule: The cron schedule to execute this job on.
-                Schedules are interpreted in UTC unless a timezone is specified.
+                Schedules are interpreted in UTC.
             end_time: The end date to stop running the cron.
             input: The input to the graph.
             metadata: Metadata to assign to the cron job runs.
@@ -347,7 +339,6 @@ class SyncCronClient:
                 after execution. 'keep' creates a new thread for each execution but does not
                 clean them up.
             enabled: Enable or disable the cron job.
-            timezone: IANA timezone for the cron schedule. Accepts a string (e.g. 'America/New_York') or a ``datetime.tzinfo`` instance (e.g. ``ZoneInfo("America/New_York")``).
             stream_mode: The stream mode(s) to use.
             stream_subgraphs: Whether to stream output from subgraphs.
             stream_resumable: Whether to persist the stream chunks in order to resume the stream later.
@@ -382,7 +373,6 @@ class SyncCronClient:
             "interrupt_after": interrupt_after,
             "on_run_completed": on_run_completed,
             "enabled": enabled,
-            "timezone": _resolve_timezone(timezone),
             "stream_mode": stream_mode,
             "stream_subgraphs": stream_subgraphs,
             "stream_resumable": stream_resumable,

libs/sdk-py/langgraph_sdk/schema.py

@@ -58,7 +58,6 @@ StreamMode = Literal[
     "debug",
     "custom",
     "messages-tuple",
-    "compact",
 ]
 """
 Defines the mode of streaming:
@@ -70,7 +69,6 @@ Defines the mode of streaming:
 - "tasks": Stream task start and finish events.
 - "debug": Stream detailed debug information.
 - "custom": Stream custom events.
-- "compact": Enable compact streaming payloads for other selected modes.
 """
 
 DisconnectMode = Literal["cancel", "continue"]
@@ -387,8 +385,6 @@ class Cron(TypedDict):
     """The end date to stop running the cron."""
     schedule: str
     """The schedule to run, cron format."""
-    timezone: str | None
-    """IANA timezone for the cron schedule (e.g. 'America/New_York'). Defaults to null, which is treated as UTC."""
     created_at: datetime
     """The time the cron was created."""
     updated_at: datetime
@@ -410,8 +406,6 @@ class CronUpdate(TypedDict, total=False):
 
     schedule: str
     """The cron schedule to execute this job on."""
-    timezone: str
-    """IANA timezone for the cron schedule (e.g. 'America/New_York')."""
     end_time: datetime
     """The end date to stop running the cron."""
     input: Input
@@ -488,7 +482,6 @@ CronSelectField = Literal[
     "thread_id",
     "end_time",
     "schedule",
-    "timezone",
     "created_at",
     "updated_at",
     "user_id",
@@ -735,26 +728,6 @@ class ValuesStreamPart(TypedDict):
     """List of interrupts that occurred during this step."""
 
 
-class ValuesPatchPayload(TypedDict):
-    """Incremental patch payload for subgraph `values` updates."""
-
-    values: dict[str, Any]
-    """Only the changed fields since the previous `values` event for this namespace."""
-    deleted_keys: NotRequired[list[str]]
-    """Optional list of keys that were removed from the previous values snapshot."""
-
-
-class ValuesPatchStreamPart(TypedDict):
-    """Stream part emitted for incremental subgraph value patches (`values-patch`)."""
-
-    type: Literal["values-patch"]
-    """Stream part type discriminator."""
-    ns: list[str]
-    """Namespace path of the emitting node (empty for root graph)."""
-    data: ValuesPatchPayload
-    """Incremental state patch for the namespace."""
-
-
 class UpdatesStreamPart(TypedDict):
     """Stream part emitted for `stream_mode="updates"`."""
 
@@ -867,7 +840,6 @@ class MetadataStreamPart(TypedDict):
 
 StreamPartV2 = (
     ValuesStreamPart
-    | ValuesPatchStreamPart
     | UpdatesStreamPart
     | MessagesPartialStreamPart
     | MessagesCompleteStreamPart

libs/sdk-py/tests/test_client_stream.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import json
 from collections.abc import Iterator, Sequence
 from pathlib import Path
 from typing import Any
@@ -9,9 +8,7 @@ import httpx
 import pytest
 from typing_extensions import assert_type
 
-from langgraph_sdk._async.runs import RunsClient
 from langgraph_sdk._shared.utilities import _sse_to_v2_dict
-from langgraph_sdk._sync.runs import SyncRunsClient
 from langgraph_sdk.client import HttpClient, SyncHttpClient
 from langgraph_sdk.schema import (
     CheckpointPayload,
@@ -27,7 +24,6 @@ from langgraph_sdk.schema import (
     TaskResultPayload,
     TasksStreamPart,
     UpdatesStreamPart,
-    ValuesPatchStreamPart,
     ValuesStreamPart,
 )
 from langgraph_sdk.sse import BytesLike, BytesLineDecoder, SSEDecoder
@@ -379,81 +375,6 @@ def test_sse_to_v2_dict_values_with_interrupts() -> None:
     assert "__interrupt__" not in result["data"]
 
 
-def test_sse_to_v2_dict_values_patch() -> None:
-    payload = {"values": {"count": 2}, "deleted_keys": ["stale"]}
-    result = _sse_to_v2_dict("values-patch|tools:call_1", payload)
-    assert result is not None
-    _assert_v2_shape(result)
-    assert result == {
-        "type": "values-patch",
-        "ns": ["tools:call_1"],
-        "data": {"values": {"count": 2}, "deleted_keys": ["stale"]},
-        "interrupts": [],
-    }
-
-
-@pytest.mark.asyncio
-async def test_async_runs_stream_includes_compact_mode():
-    async def handler(request: httpx.Request) -> httpx.Response:
-        assert request.method == "POST"
-        assert request.url.path == "/runs/stream"
-        body = json.loads(request.content)
-        assert body["stream_mode"] == ["values", "compact"]
-        return httpx.Response(
-            200,
-            headers={"Content-Type": "text/event-stream"},
-            content=b"event: end\ndata: null\n\n",
-        )
-
-    transport = httpx.MockTransport(handler)
-    async with httpx.AsyncClient(
-        transport=transport, base_url="https://example.com"
-    ) as client:
-        runs_client = RunsClient(HttpClient(client))
-        parts = [
-            part
-            async for part in runs_client.stream(
-                thread_id=None,
-                assistant_id="agent",
-                input={"messages": []},
-                stream_mode=["values", "compact"],
-            )
-        ]
-
-    assert len(parts) == 1
-    assert parts[0].event == "end"
-    assert parts[0].data is None
-
-
-def test_sync_runs_stream_includes_compact_mode():
-    def handler(request: httpx.Request) -> httpx.Response:
-        assert request.method == "POST"
-        assert request.url.path == "/runs/stream"
-        body = json.loads(request.content)
-        assert body["stream_mode"] == ["values", "compact"]
-        return httpx.Response(
-            200,
-            headers={"Content-Type": "text/event-stream"},
-            content=b"event: end\ndata: null\n\n",
-        )
-
-    transport = httpx.MockTransport(handler)
-    with httpx.Client(transport=transport, base_url="https://example.com") as client:
-        runs_client = SyncRunsClient(SyncHttpClient(client))
-        parts = list(
-            runs_client.stream(
-                thread_id=None,
-                assistant_id="agent",
-                input={"messages": []},
-                stream_mode=["values", "compact"],
-            )
-        )
-
-    assert len(parts) == 1
-    assert parts[0].event == "end"
-    assert parts[0].data is None
-
-
 # --- client-side v2 stream wrapping ---
 
 
@@ -527,8 +448,6 @@ def _check_v2_type_narrowing(part: StreamPartV2) -> None:
     if part["type"] == "values":
         assert_type(part, ValuesStreamPart)
         assert_type(part["data"], dict[str, Any])
-    elif part["type"] == "values-patch":
-        assert_type(part, ValuesPatchStreamPart)
     elif part["type"] == "updates":
         assert_type(part, UpdatesStreamPart)
         assert_type(part["data"], dict[str, Any])

libs/sdk-py/uv.lock

@@ -265,7 +265,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.1.2"
+version = "1.1.0"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },