switch generics

initial pass at generics for remote graph
context in remote
2026-10-02 14:35:18 +02:00 · 2026-03-12 11:17:20 -04:00 · 2026-03-12 11:11:25 -04:00 · 2026-03-12 10:52:10 -04:00 · 2026-03-11 21:32:42 -07:00 · 2026-03-11 15:33:32 -07:00
26 changed files with 2296 additions and 147 deletions
@@ -29,7 +29,7 @@ from langgraph_cli.exec import Runner, subp_exec
 from langgraph_cli.host_backend import HostBackendClient, HostBackendError
 from langgraph_cli.progress import Progress
 from langgraph_cli.templates import TEMPLATE_HELP_STRING, create_new
-from langgraph_cli.util import warn_non_wolfi_distro
+from langgraph_cli.util import format_deployments_table, warn_non_wolfi_distro
 from langgraph_cli.version import __version__

 RESERVED_ENV_VARS = frozenset(
@@ -287,6 +287,22 @@ 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_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"]),
@@ -295,7 +311,67 @@ OPT_ENGINE_RUNTIME_MODE = click.option(
 )


-@click.group()
+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.version_option(version=__version__, prog_name="LangGraph CLI")
 def cli():
    pass
@@ -593,72 +669,89 @@ def build(
        )


-@click.option(
-    "--api-key",
-    envvar="LANGGRAPH_HOST_API_KEY",
-    help=(
-        "API key. Can also be set via LANGGRAPH_HOST_API_KEY, "
-        "LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable or .env file."
-    ),
-)
-@click.option(
-    "--name",
-    envvar="LANGSMITH_DEPLOYMENT_NAME",
-    help=(
-        "Deployment name. Can also be set via LANGSMITH_DEPLOYMENT_NAME "
-        "environment variable or .env file. Defaults to current directory name "
-        "if --deployment-id is not provided."
-    ),
-)
-@click.option(
-    "--deployment-id",
-    help=(
-        "ID of an existing deployment to update. If omitted, "
-        "--name is used to find or create the deployment."
-    ),
-)
-@click.option(
-    "--deployment-type",
-    type=click.Choice(["dev", "prod"]),
-    default="dev",
-    show_default=True,
-    help="Deployment type (used when creating a new deployment).",
-)
-@click.option(
-    "--no-wait",
-    is_flag=True,
-    default=False,
-    help="Skip waiting for deployment status.",
-)
-@OPT_VERBOSE
-@click.option(
-    "--host-url",
-    envvar="LANGGRAPH_HOST_URL",
-    default="https://api.host.langchain.com",
-    hidden=True,
-)
-@click.option("--image-name", hidden=True)
-@click.option("--image-tag", default="latest", hidden=True)
-@click.option(
-    "--config",
-    "-c",
-    default=DEFAULT_CONFIG,
-    hidden=True,
-    type=click.Path(
-        exists=True,
-        file_okay=True,
-        dir_okay=False,
-        resolve_path=True,
-        path_type=pathlib.Path,
-    ),
-)
-@click.option("--pull/--no-pull", default=True, hidden=True)
-@click.option("--base-image", hidden=True)
-@click.option("--install-command", hidden=True)
-@click.option("--build-command", hidden=True)
-@click.option("--api-version", type=str, hidden=True)
-@click.argument("docker_build_args", nargs=-1, type=click.UNPROCESSED)
-@cli.command(
+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,
+            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,
+            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,
    help=(
        "[Beta] Build and deploy a LangGraph image to LangSmith Deployments.\n\n"
        "This command is in beta and under active development. "
@@ -667,10 +760,26 @@ def build(
        "is located). This command also accepts build flags (--base-image, "
        "--pull, etc.). See 'langgraph build --help' for details."
    ),
-    context_settings=dict(ignore_unknown_options=True),
+    context_settings=dict(ignore_unknown_options=True, allow_extra_args=True),
+    invoke_without_command=True,  # allow `deploy` click group to execute without command
 )
+@_deploy_base_options(include_docker_args=False, validate_config_path=False)
+@click.pass_context
@log_command
-def deploy(
+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(
    config: pathlib.Path,
    pull: bool,
    verbose: bool,
@@ -1050,6 +1159,115 @@ 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:
+        resolved_api_key = click.prompt("Host 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:
+    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 = HostBackendClient(
+                client._base_url, client._api_key, 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 current_client: current_client.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 current_client: current_client.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.

@@ -39,10 +39,14 @@ class HostBackendClient:
        )

    def _request(
-        self, method: str, path: str, payload: dict[str, Any] | None = None
+        self,
+        method: str,
+        path: str,
+        payload: dict[str, Any] | None = None,
+        params: dict[str, Any] | None = None,
    ) -> Any:
        try:
-            resp = self._client.request(method, path, json=payload)
+            resp = self._client.request(method, path, json=payload, params=params)
            resp.raise_for_status()
        except httpx.HTTPStatusError as err:
            detail = err.response.text or str(err.response.status_code)
@@ -65,12 +69,19 @@ 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", f"/v2/deployments?name_contains={name_contains}")
+    def list_deployments(self, name_contains: str = "") -> dict[str, Any]:
+        return self._request(
+            "GET",
+            "/v2/deployments",
+            params={"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",
@@ -1,3 +1,5 @@
+from collections.abc import Sequence
+
 import click


@@ -23,3 +25,35 @@ 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)
@@ -9,6 +9,7 @@ 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
@@ -287,6 +288,238 @@ 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()
@@ -1784,7 +1784,9 @@ 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
@@ -135,6 +135,28 @@ 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}
@@ -1,6 +1,11 @@
 from unittest.mock import patch

-from langgraph_cli.util import clean_empty_lines, warn_non_wolfi_distro
+from langgraph_cli.util import (
+    _extract_deployment_url,
+    clean_empty_lines,
+    format_deployments_table,
+    warn_non_wolfi_distro,
+)


 def test_clean_empty_lines():
@@ -186,3 +191,36 @@ 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
@@ -0,0 +1,67 @@
+# RESUME Writes Stripping: Complete Flow Reference
+
+## Legend
+
+| Column | Meaning |
+|---|---|
+| **Level** | P = Parent, S = Subgraph |
+| **`is_replaying`** | `CONFIG_KEY_CHECKPOINT_ID` key exists in `config[CONF]` (line 249) |
+| **`__enter__` via** | Which branch loads the checkpoint: **ckpt_id** (explicit checkpoint_id in checkpoint_config), **replay_state** (parent's ReplayState), **latest** (fetch most recent) |
+| **`RESUMING`** | Value of `CONFIG_KEY_RESUMING` in configurable (set by parent for subgraphs, absent for outer graph) |
+| **`is_resuming`** | Computed at line 633 — controls whether to "proceed past previous checkpoint" |
+| **`in_map`** | `replaying_from_checkpoint_map` — subgraph's ns found in checkpoint_map |
+| **Strip?** | Are RESUME pending writes stripped? (line 662-671) |
+
+## Setup
+
+```
+Parent:  START → executor (subgraph, checkpointer=True) → END
+Subgraph: START → step_a → ask_1 (interrupt) → ask_2 (interrupt) → END
+```
+
+## The Table
+
+| # | Scenario | Level | User call | `__enter__` via | `is_replaying` | `RESUMING` | `is_resuming` | `in_map` | Strip? | Why correct |
+|---|---|---|---|---|---|---|---|---|---|---|
+| 1 | **Fresh run** | P | `invoke({"v":[]}, cfg)` | latest (None) | False | _(absent)_ | False | — | N/A | No checkpoint yet, no writes to strip |
+| 1 | | S | _(Send from parent)_ | latest (None) | True¹ | False | False | False | N/A | No checkpoint yet |
+| 2 | **Resume single interrupt** | P | `invoke(Cmd(resume="a"), cfg)` | latest | False | _(absent)_ | True | — | No | Resuming — keep RESUME writes for interrupt() to return answer |
+| 2 | | S | _(Send)_ | latest | True¹ | True | True | False | No | `RESUMING=True` → keep. interrupt() returns "a", node completes |
+| 3 | **Resume 1st of 2 interrupts** | P | `invoke(Cmd(resume="a1"), cfg)` | latest | False | _(absent)_ | True | — | No | Keep RESUME writes — ask_1's answer must survive |
+| 3 | | S | _(Send)_ | latest | True¹ | True | True | False | **No** | ask_1 gets "a1" from RESUME write. ask_2 has no RESUME write → interrupt() re-fires. Correct. |
+| 4 | **Replay parent ckpt** (parent was mid-subgraph) | P | `invoke(None, parent_hist_cfg)` | ckpt_id | True | _(absent)_ | True | — | **Yes** | Replaying — strip stale RESUME writes so interrupts re-fire |
+| 4 | | S | _(Send)_ | replay_state² | True¹ | _(popped)_³ | False | False | **Yes** | `is_replaying=T`, `RESUMING` absent → strip. Subgraph replays cleanly |
+| 5 | **Time-travel to subgraph ckpt** (THE BUG) | P | `invoke(None, sub_cfg)` | ckpt_id⁴ | True | _(absent)_ | True | — | **Yes** | Parent replays from historical checkpoint |
+| 5 | | S | _(Send)_ | **ckpt_id**⁵ | True¹ | **True** | **True** | **True** | **Yes** ✨ | `in_map=True` overrides `RESUMING=True` → force strip. THE FIX. |
+| 5 | | S _(without fix)_ | _(Send)_ | ckpt_id⁵ | True¹ | **True** | **True** | _(no check)_ | **No** ❌ | BUG: `RESUMING=True` prevents strip → stale RESUME values → interrupt() doesn't re-fire |
+| 6 | **Fork from subgraph ckpt** | P | `invoke(None, update_state(sub_cfg,...))` | ckpt_id | True | _(absent)_ | True | — | **Yes** | Same as case 5 — fork creates new ckpt, but checkpoint_map still resolves |
+| 6 | | S | _(Send)_ | ckpt_id⁵ | True¹ | True | True | **True** | **Yes** ✨ | Same fix applies |
+| 7 | **Resume after case 5 re-interrupts** | P | `invoke(Cmd(resume="a2"), cfg)` | latest | False | _(absent)_ | True | — | No | Normal resume — keep RESUME writes |
+| 7 | | S | _(Send)_ | latest | True¹ | True | True | False⁶ | **No** | ask_2 gets "a2" from fresh RESUME write. Correct. |
+
+## Footnotes
+
+**¹** `is_replaying` is always `True` for subgraphs on tick 1 because `_algo.py` sets `CONFIG_KEY_CHECKPOINT_ID: None` — the key exists (even with `None` value), so `key in dict` is `True`. After tick 1, line 563 sets `is_replaying = False`.
+
+**²** `replay_state` branch: parent passed `CONFIG_KEY_REPLAY_STATE = ReplayState(parent_ckpt_id)`. The subgraph uses `replay_state.get_checkpoint()` which does `checkpointer.list(before=parent_ckpt_id, limit=1)` to find the subgraph's checkpoint from before the replay point.
+
+**³** The `replay_state` branch in `__enter__` (line 1158) explicitly pops `CONFIG_KEY_RESUMING` from config. This makes `is_resuming = False` in `_first()` because for nested graphs the fallback (`self.input is None or input_is_command`) is False (input is a Send arg).
+
+**⁴** Parent `__init__` clears `checkpoint_ns → ""` and `checkpoint_id → None` (line 273-277), then resolves `""` from checkpoint_map → gets `parent_checkpoint_id` onto `checkpoint_config` (line 278-290).
+
+**⁵** Subgraph `__init__` resolves its namespace (e.g. `"executor:task_id"`) from checkpoint_map → gets `subgraph_checkpoint_id` onto `checkpoint_config`. This is why the new first branch in `__enter__` (line 1141) fires — `checkpoint_config` has a truthy `checkpoint_id`.
+
+**⁶** After case 5 completes/re-interrupts and user resumes, the config is a normal thread config with no checkpoint_map entry for the subgraph. `in_map` is False, so normal resume logic applies.
+
+## The core tension (case 5)
+
+The parent **can't distinguish** these cases when propagating flags to subgraphs:
+
+| Parent sees | What's actually happening | Subgraph should strip RESUME? |
+|---|---|---|
+| `input=None`, has checkpoint | Resume after interrupt | Yes (replaying) |
+| `input=None`, has checkpoint | Resume after interrupt | No (resuming) |
+| `input=Command(resume=...)` | Active resume | No (resuming) |
+| `input=None`, has checkpoint | Time-travel to subgraph | Yes (replaying) |
+
+The **only** distinguishing signal at the subgraph level is whether its namespace appears in `checkpoint_map`.
@@ -647,13 +647,24 @@ 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.
-        # 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)
+        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)
+            )
        ):
            self.checkpoint_pending_writes = [
                w for w in self.checkpoint_pending_writes if w[1] != RESUME
@@ -1127,9 +1138,15 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
    def __enter__(self) -> Self:
        if not self.checkpointer:
            saved = None
-        elif self.is_nested and (
-            replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE)
-        ):
+        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.
            saved = replay_state.get_checkpoint(
                self.config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, ""),
                self.checkpointer,
@@ -1141,9 +1158,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
            self.config[CONF].pop(CONFIG_KEY_RESUMING, None)
        else:
            # Normal case: fetch the most recent checkpoint for this
-            # graph/thread. If a specific checkpoint_id is in the config,
-            # fetch that exact checkpoint; otherwise fetch the latest one.
-            # Returns None on first invocation (no checkpoints exist yet).
+            # graph/thread. Returns None on first invocation.
            saved = self.checkpointer.get_tuple(self.checkpoint_config)

        if saved is None:
@@ -1322,9 +1337,15 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
    async def __aenter__(self) -> Self:
        if not self.checkpointer:
            saved = None
-        elif self.is_nested and (
-            replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE)
-        ):
+        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.
            saved = await replay_state.aget_checkpoint(
                self.config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, ""),
                self.checkpointer,
@@ -1336,9 +1357,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
            self.config[CONF].pop(CONFIG_KEY_RESUMING, None)
        else:
            # Normal case: fetch the most recent checkpoint for this
-            # graph/thread. If a specific checkpoint_id is in the config,
-            # fetch that exact checkpoint; otherwise fetch the latest one.
-            # Returns None on first invocation (no checkpoints exist yet).
+            # graph/thread. Returns None on first invocation.
            saved = await self.checkpointer.aget_tuple(self.checkpoint_config)

        if saved is None:
@@ -2456,7 +2456,7 @@ class Pregel(
        debug: bool | None = None,
        version: Literal["v2"],
        **kwargs: Unpack[DeprecatedKwargs],
-    ) -> Iterator[StreamPart[OutputT, StateT]]: ...
+    ) -> Iterator[StreamPart[StateT, OutputT]]: ...

    @overload
    def stream(
@@ -2787,7 +2787,7 @@ class Pregel(
        debug: bool | None = None,
        version: Literal["v2"],
        **kwargs: Unpack[DeprecatedKwargs],
-    ) -> AsyncIterator[StreamPart[OutputT, StateT]]: ...
+    ) -> AsyncIterator[StreamPart[StateT, OutputT]]: ...

    @overload
    def astream(
@@ -3194,7 +3194,7 @@ class Pregel(
        durability: Durability | None = None,
        version: Literal["v2"],
        **kwargs: Any,
-    ) -> list[StreamPart[OutputT, StateT]]: ...
+    ) -> list[StreamPart[StateT, OutputT]]: ...

    @overload
    def invoke(
@@ -3364,7 +3364,7 @@ class Pregel(
        durability: Durability | None = None,
        version: Literal["v2"],
        **kwargs: Any,
-    ) -> list[StreamPart[OutputT, StateT]]: ...
+    ) -> list[StreamPart[StateT, OutputT]]: ...

    @overload
    async def ainvoke(
@@ -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[OutputT, StateT]]: ...
+    ) -> Iterator[StreamPart[StateT, OutputT]]: ...

    @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[OutputT, StateT]]: ...
+    ) -> AsyncIterator[StreamPart[StateT, OutputT]]: ...

    @overload
    @abstractmethod
@@ -5,6 +5,7 @@ from collections.abc import AsyncIterator, Iterator, Sequence
 from dataclasses import asdict
 from typing import (
    Any,
+    Generic,
    Literal,
    cast,
    overload,
@@ -65,6 +66,7 @@ from langgraph.types import (
    StreamMode,
    StreamPart,
 )
+from langgraph.typing import ContextT, InputT, OutputT, StateT

 logger = logging.getLogger(__name__)

@@ -108,7 +110,10 @@ class RemoteException(Exception):
    pass


-class RemoteGraph(PregelProtocol):
+class RemoteGraph(
+    PregelProtocol[StateT, ContextT, InputT, OutputT],
+    Generic[StateT, ContextT, InputT, OutputT],
+):
    """The `RemoteGraph` class is a client implementation for calling remote
    APIs that implement the LangGraph Server API specification.

@@ -688,9 +693,10 @@ class RemoteGraph(PregelProtocol):
    @overload
    def stream(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        stream_mode: StreamMode | list[StreamMode] | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
@@ -699,14 +705,15 @@ class RemoteGraph(PregelProtocol):
        params: QueryParamTypes | None = None,
        version: Literal["v2"],
        **kwargs: Any,
-    ) -> Iterator[StreamPart]: ...
+    ) -> Iterator[StreamPart[StateT, OutputT]]: ...

    @overload
    def stream(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        stream_mode: StreamMode | list[StreamMode] | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
@@ -719,9 +726,10 @@ class RemoteGraph(PregelProtocol):

    def stream(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        stream_mode: StreamMode | list[StreamMode] | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
@@ -769,6 +777,7 @@ 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,9 +848,10 @@ class RemoteGraph(PregelProtocol):
    @overload
    def astream(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        stream_mode: StreamMode | list[StreamMode] | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
@@ -850,14 +860,15 @@ class RemoteGraph(PregelProtocol):
        params: QueryParamTypes | None = None,
        version: Literal["v2"],
        **kwargs: Any,
-    ) -> AsyncIterator[StreamPart]: ...
+    ) -> AsyncIterator[StreamPart[StateT, OutputT]]: ...

    @overload
    def astream(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        stream_mode: StreamMode | list[StreamMode] | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
@@ -870,9 +881,10 @@ class RemoteGraph(PregelProtocol):

    async def astream(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        stream_mode: StreamMode | list[StreamMode] | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
@@ -920,6 +932,7 @@ class RemoteGraph(PregelProtocol):
            input=input,
            command=command,
            config=sanitized_config,
+            context=context,
            stream_mode=stream_modes,
            interrupt_before=interrupt_before,
            interrupt_after=interrupt_after,
@@ -1006,23 +1019,25 @@ class RemoteGraph(PregelProtocol):
    @overload
    def invoke(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
        headers: dict[str, str] | None = None,
        params: QueryParamTypes | None = None,
        version: Literal["v2"],
        **kwargs: Any,
-    ) -> GraphOutput[dict[str, Any]]: ...
+    ) -> GraphOutput[OutputT]: ...

    @overload
    def invoke(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
        headers: dict[str, str] | None = None,
@@ -1033,9 +1048,10 @@ class RemoteGraph(PregelProtocol):

    def invoke(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
        headers: dict[str, str] | None = None,
@@ -1061,6 +1077,7 @@ 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,
@@ -1084,23 +1101,25 @@ class RemoteGraph(PregelProtocol):
    @overload
    async def ainvoke(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
        headers: dict[str, str] | None = None,
        params: QueryParamTypes | None = None,
        version: Literal["v2"],
        **kwargs: Any,
-    ) -> GraphOutput[dict[str, Any]]: ...
+    ) -> GraphOutput[OutputT]: ...

    @overload
    async def ainvoke(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
        headers: dict[str, str] | None = None,
@@ -1111,9 +1130,10 @@ class RemoteGraph(PregelProtocol):

    async def ainvoke(
        self,
-        input: dict[str, Any] | Any,
+        input: InputT | Command | None,
        config: RunnableConfig | None = None,
        *,
+        context: ContextT | None = None,
        interrupt_before: All | Sequence[str] | None = None,
        interrupt_after: All | Sequence[str] | None = None,
        headers: dict[str, str] | None = None,
@@ -1139,6 +1159,7 @@ 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,
@@ -335,7 +335,7 @@ StreamPart = TypeAliasType(
    | CheckpointStreamPart[StateT]
    | TasksStreamPart
    | DebugStreamPart[StateT],
-    type_params=(OutputT, StateT),
+    type_params=(StateT, OutputT),
 )
 """A discriminated union of all v2 stream part types.

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "langgraph"
-version = "1.1.0"
+version = "1.1.1"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"
@@ -1,5 +1,6 @@
 import re
 import sys
+from dataclasses import dataclass
 from typing import Annotated
 from unittest.mock import AsyncMock, MagicMock

@@ -10,6 +11,7 @@ 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
@@ -908,6 +910,188 @@ 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.skip(
    "Unskip this test to manually test the LangSmith Deployment integration"
 )
@@ -1116,6 +1116,631 @@ 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)
 # ---------------------------------------------------------------------------
@@ -1051,6 +1051,552 @@ 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)
 # ---------------------------------------------------------------------------
@@ -8,7 +8,8 @@ from pydantic import BaseModel
 from typing_extensions import TypedDict

 from langgraph.graph import StateGraph
-from langgraph.types import Command
+from langgraph.pregel.remote import RemoteGraph
+from langgraph.types import Command, GraphOutput, StreamPart


 def test_typed_dict_state() -> None:
@@ -159,3 +160,75 @@ def test_add_node_with_explicit_input_schema() -> None:
    # because it violates the principles of contravariance
    workflow.add_node("a_narrow", a, input_schema=ANarrow)  # type: ignore[arg-type]
    workflow.add_node("b_narrow", b, input_schema=BNarrow)  # type: ignore[arg-type]
+
+
+@pytest.mark.skip("Purely for type checking")
+def test_remote_graph_generics_typed_dict() -> None:
+    """RemoteGraph parameterized with TypedDict should propagate types."""
+
+    class MyState(TypedDict):
+        messages: list[str]
+
+    rg: RemoteGraph[MyState, None, MyState, MyState] = RemoteGraph(
+        "test", url="http://localhost:8123"
+    )
+
+    # v2 invoke should return GraphOutput[MyState]
+    result: GraphOutput[MyState] = rg.invoke({"messages": ["hi"]}, version="v2")
+    _val: MyState = result.value
+
+    # v1 invoke should return dict[str, Any] | Any
+    _v1_result: dict[str, Any] | Any = rg.invoke({"messages": ["hi"]})
+
+    # v2 stream should yield StreamPart[MyState, MyState]
+    for part in rg.stream({"messages": ["hi"]}, version="v2"):
+        _part: StreamPart[MyState, MyState] = part
+
+    # input should accept the state type
+    rg.invoke({"messages": ["hi"]}, version="v2")
+
+    # input should also accept Command
+    rg.invoke(Command(), version="v2")
+
+    # input should also accept None
+    rg.invoke(None, version="v2")
+
+
+@pytest.mark.skip("Purely for type checking")
+def test_remote_graph_generics_pydantic() -> None:
+    """RemoteGraph parameterized with Pydantic model should propagate types."""
+
+    class PydanticState(BaseModel):
+        messages: list[str]
+
+    rg: RemoteGraph[PydanticState, None, PydanticState, PydanticState] = RemoteGraph(
+        "test", url="http://localhost:8123"
+    )
+
+    result: GraphOutput[PydanticState] = rg.invoke(
+        PydanticState(messages=["hi"]), version="v2"
+    )
+    _val: PydanticState = result.value
+
+
+@pytest.mark.skip("Purely for type checking")
+def test_remote_graph_separate_input_output() -> None:
+    """RemoteGraph with different input/output schemas."""
+
+    class InputState(TypedDict):
+        query: str
+
+    class OutputState(TypedDict):
+        answer: str
+
+    class FullState(InputState, OutputState): ...
+
+    rg: RemoteGraph[FullState, None, InputState, OutputState] = RemoteGraph(
+        "test", url="http://localhost:8123"
+    )
+
+    result: GraphOutput[OutputState] = rg.invoke({"query": "hi"}, version="v2")
+    _val: OutputState = result.value
+
+    # wrong input type should fail type checking
+    rg.invoke({"answer": "wrong"}, version="v2")  # type: ignore[call-overload]
@@ -1367,7 +1367,7 @@ wheels = [

 [[package]]
 name = "langgraph"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "." }
 dependencies = [
    { name = "langchain-core" },
@@ -3615,21 +3615,19 @@ wheels = [

 [[package]]
 name = "tornado"
-version = "6.5.4"
+version = "6.5.5"
 source = { registry = "https://pypi.org/simple" }
-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" }
+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" }
 wheels = [
-    { 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" },
+    { 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" },
 ]

 [[package]]
@@ -268,7 +268,7 @@ wheels = [

 [[package]]
 name = "langgraph"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "../langgraph" }
 dependencies = [
    { name = "langchain-core" },
@@ -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.10"
+__version__ = "0.3.11"

 __all__ = ["Auth", "Encryption", "EncryptionContext", "get_client", "get_sync_client"]
@@ -4,10 +4,11 @@ from __future__ import annotations

 import warnings
 from collections.abc import Mapping, Sequence
-from datetime import datetime
+from datetime import datetime, tzinfo
 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,
@@ -70,6 +71,7 @@ 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,
@@ -84,7 +86,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.
+                Schedules are interpreted in UTC unless a timezone is specified.
            input: The input to the graph.
            metadata: Metadata to assign to the cron job runs.
            config: The configuration for the assistant.
@@ -100,6 +102,7 @@ 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.
@@ -152,6 +155,7 @@ 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,
@@ -184,6 +188,7 @@ 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,
@@ -197,7 +202,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.
+                Schedules are interpreted in UTC unless a timezone is specified.
            input: The input to the graph.
            metadata: Metadata to assign to the cron job runs.
            config: The configuration for the assistant.
@@ -215,6 +220,7 @@ 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.
@@ -268,6 +274,7 @@ 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,
@@ -324,6 +331,7 @@ 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,
@@ -336,7 +344,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.
+                Schedules are interpreted in UTC unless a timezone is specified.
            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.
@@ -350,6 +358,7 @@ 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.
@@ -384,6 +393,7 @@ 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,
@@ -6,13 +6,17 @@ import functools
 import os
 import re
 from collections.abc import Mapping
-from typing import Any, cast
+from datetime import tzinfo
+from typing import TYPE_CHECKING, 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())
@@ -125,6 +129,35 @@ 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}

@@ -4,9 +4,10 @@ from __future__ import annotations

 import warnings
 from collections.abc import Mapping, Sequence
-from datetime import datetime
+from datetime import datetime, tzinfo
 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,
@@ -64,6 +65,7 @@ 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,
@@ -78,7 +80,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.
+                Schedules are interpreted in UTC unless a timezone is specified.
            input: The input to the graph.
            metadata: Metadata to assign to the cron job runs.
            config: The configuration for the assistant.
@@ -92,6 +94,7 @@ 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.
@@ -144,6 +147,7 @@ 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,
@@ -174,6 +178,7 @@ 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,
@@ -187,7 +192,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.
+                Schedules are interpreted in UTC unless a timezone is specified.
            input: The input to the graph.
            metadata: Metadata to assign to the cron job runs.
            config: The configuration for the assistant.
@@ -205,6 +210,7 @@ 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.
@@ -259,6 +265,7 @@ 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,
@@ -313,6 +320,7 @@ 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,
@@ -325,7 +333,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.
+                Schedules are interpreted in UTC unless a timezone is specified.
            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.
@@ -339,6 +347,7 @@ 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.
@@ -373,6 +382,7 @@ 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,
@@ -385,6 +385,8 @@ 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
@@ -406,6 +408,8 @@ 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
@@ -482,6 +486,7 @@ CronSelectField = Literal[
    "thread_id",
    "end_time",
    "schedule",
+    "timezone",
    "created_at",
    "updated_at",
    "user_id",
@@ -265,7 +265,7 @@ wheels = [

 [[package]]
 name = "langgraph"
-version = "1.1.0"
+version = "1.1.1"
 source = { editable = "../langgraph" }
 dependencies = [
    { name = "langchain-core" },
Author	SHA1	Message	Date
Sydney Runkle	576fa6db20	switch generics	2026-03-12 11:17:20 -04:00
Sydney Runkle	e65c221ada	initial pass at generics for remote graph	2026-03-12 11:11:25 -04:00
Sydney Runkle	04076c4bb5	context in remote	2026-03-12 10:52:10 -04:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	b7b052e66c	chore(deps): bump tornado from 6.5.4 to 6.5.5 in /libs/langgraph (#7122 ) Bumps [tornado](https://github.com/tornadoweb/tornado) from 6.5.4 to 6.5.5. <details> <summary>Changelog</summary> <p><em>Sourced from <a href="https://github.com/tornadoweb/tornado/blob/master/docs/releases.rst">tornado's changelog</a>.</em></p> <blockquote> <h1>Release notes</h1> <p>.. toctree:: :maxdepth: 2</p> <p>releases/v6.5.5 releases/v6.5.4 releases/v6.5.3 releases/v6.5.2 releases/v6.5.1 releases/v6.5.0 releases/v6.4.2 releases/v6.4.1 releases/v6.4.0 releases/v6.3.3 releases/v6.3.2 releases/v6.3.1 releases/v6.3.0 releases/v6.2.0 releases/v6.1.0 releases/v6.0.4 releases/v6.0.3 releases/v6.0.2 releases/v6.0.1 releases/v6.0.0 releases/v5.1.1 releases/v5.1.0 releases/v5.0.2 releases/v5.0.1 releases/v5.0.0 releases/v4.5.3 releases/v4.5.2 releases/v4.5.1 releases/v4.5.0 releases/v4.4.3 releases/v4.4.2 releases/v4.4.1 releases/v4.4.0 releases/v4.3.0 releases/v4.2.1 releases/v4.2.0 releases/v4.1.0 releases/v4.0.2 releases/v4.0.1 releases/v4.0.0 releases/v3.2.2 releases/v3.2.1 releases/v3.2.0 releases/v3.1.1</p> <!-- raw HTML omitted --> </blockquote> <p>... (truncated)</p> </details> <details> <summary>Commits</summary> <ul> <li><a href="https://github.com/tornadoweb/tornado/commit/7d6465056ceb7a054b3f64cf1c18271753b10482"><code>7d64650</code></a> Merge pull request <a href="https://redirect.github.com/tornadoweb/tornado/issues/3586">#3586</a> from bdarnell/update-cibw</li> <li><a href="https://github.com/tornadoweb/tornado/commit/d05d59b8080a0d5d6a260994c7aad7049209d345"><code>d05d59b</code></a> build: Bump cibuildwheel to 3.4.0</li> <li><a href="https://github.com/tornadoweb/tornado/commit/c2f46732b0ad14bf0db4219c96a945f4b60205f5"><code>c2f4673</code></a> Merge pull request <a href="https://redirect.github.com/tornadoweb/tornado/issues/3585">#3585</a> from bdarnell/release-655</li> <li><a href="https://github.com/tornadoweb/tornado/commit/e5f1aa4b6fa2c16b29024830227838fcb0c79b6f"><code>e5f1aa4</code></a> Release notes and version bump for v6.5.5</li> <li><a href="https://github.com/tornadoweb/tornado/commit/78a046f99f89977dfc8ff5a1fe16d298afbeeaca"><code>78a046f</code></a> httputil: Add CRLF to _FORBIDDEN_HEADER_CHARS_RE</li> <li><a href="https://github.com/tornadoweb/tornado/commit/24a2d96ea115f663b223887deb0060f13974c104"><code>24a2d96</code></a> web: Validate characters in all cookie attributes.</li> <li><a href="https://github.com/tornadoweb/tornado/commit/119a195e290c43ad2d63a2cf012c29d43d6ed839"><code>119a195</code></a> httputil: Add limits on multipart form data parsing</li> <li>See full diff in <a href="https://github.com/tornadoweb/tornado/compare/v6.5.4...v6.5.5">compare view</a></li> </ul> </details> <br /> [![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=tornado&package-manager=uv&previous-version=6.5.4&new-version=6.5.5)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores) Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`. [//]: # (dependabot-automerge-start) [//]: # (dependabot-automerge-end) --- <details> <summary>Dependabot commands and options</summary> <br /> You can trigger Dependabot actions by commenting on this PR: - `@dependabot rebase` will rebase this PR - `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it - `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency - `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself) - `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself) - `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself) You can disable automated security fix PRs for this repo from the [Security Alerts page](https://github.com/langchain-ai/langgraph/network/alerts). </details> Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-03-11 21:32:42 -07:00
hari-dhanushkodiandGitHub	d595f11b43	feat(cli): add deploy list and deploy delete subcommands (#7118 ) ### Summary This PR introduces a subcommand implementation that allows `langgraph deploy list` and `langgraph deploy delete` subcommands. #### `langgraph deploy list` ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 langgraph % langgraph deploy list --help ⎈ gke_langchain-test-387119_us-west1_langgraph-cloud-us-west1 Usage: langgraph deploy list [OPTIONS] [Beta] List LangSmith Deployments. Options: --name-contains TEXT Only show deployments whose names contain this value. --api-key TEXT API key. Can also be set via LANGGRAPH_HOST_API_KEY, LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable or .env file. --help Show this message and exit. ``` Output example: ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 cli % langgraph deploy list Deployment ID Deployment Name Deployment URL ------------------------------------ -------------------------- ------------------------------------------------------------------------------------ a40d6567-87c0-485a-a23d-94309a7d4519 ht-andrew-test-04 - 9da26acb-d0c9-4af0-af9e-f3fe8dfe85bc ht-anirudh-deployment-test https://ht-anirudh-deployment-test-428af4737f8a533cb2b107587eb8f38f.us.langgraph.app/ ``` #### `langgraph deploy delete` ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 langgraph % langgraph deploy delete --help ⎈ gke_langchain-test-387119_us-west1_langgraph-cloud-us-west1 Usage: langgraph deploy delete [OPTIONS] DEPLOYMENT_ID [Beta] Delete a LangSmith Deployment. Options: --force Delete without prompting for confirmation. --api-key TEXT API key. Can also be set via LANGGRAPH_HOST_API_KEY, LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable or .env file. --help Show this message and exit. ``` Output example: ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 cli % langgraph deploy delete a40d6567-87c0-485a-a23d-94309a7d4519 Are you sure you want to delete deployment ID a40d6567-87c0-485a-a23d-94309a7d4519? (Y/n): Y Host API key: Deleted deployment a40d6567-87c0-485a-a23d-94309a7d4519. ``` ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 cli % langgraph deploy delete a40d6567-87c0-485a-a23d-94309a7d4519 Are you sure you want to delete deployment ID a40d6567-87c0-485a-a23d-94309a7d4519? (Y/n): n Aborted! ``` ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 cli % langgraph deploy delete 9da26acb-d0c9-4af0-af9e-f3fe8dfe85bc --force Host API key: Deleted deployment 9da26acb-d0c9-4af0-af9e-f3fe8dfe85bc. ```	2026-03-11 15:33:32 -07:00
Sydney RunkleandGitHub	ed540155e3	release(langgraph): 1.1.1 (#7120 )	2026-03-11 18:10:09 -04:00
Sydney RunkleandGitHub	f78892d462	fix: replay bug, direct to subgraphs (#7115 ) ## Summary Fixes two bugs when time-traveling to a subgraph checkpoint (`graph.invoke(None, sub_config)`): 1. Wrong checkpoint loaded: The old `__enter__` checked `replay_state` before explicit `checkpoint_id` for nested graphs. When time-traveling, `checkpoint_map` resolves a specific checkpoint_id for the target subgraph, but `replay_state.get_checkpoint()` would find an earlier checkpoint (via `list(before=parent_ckpt_id)`), causing the subgraph to re-execute from the beginning (step_a, ask_1, etc. all re-ran). Fix: check for explicit `checkpoint_id` first. 2. Stale RESUME writes kept: The parent sets `RESUMING=True` on subgraph configs — it can't distinguish time-travel from normal resume. The subgraph sees `RESUMING=True` and preserves old RESUME writes, so `interrupt()` returns cached answers instead of re-firing. Fix: check whether the subgraph's own namespace appears in `checkpoint_map` (it only does during time-travel — normally the map only has ancestor entries). When present, force-strip RESUME writes. ## Test plan Tests cover time travel (replay + fork) at different interrupt points for both single-nested and double-nested subgraphs, including cases where the middle subgraph itself has interrupts. All tests have sync and async variants. ### Single nested (2 levels: parent → executor subgraph) - `test_subgraph_time_travel_to_first_interrupt` — time travel to 1st interrupt checkpoint; verifies step_a doesn't re-run, ask_1 re-fires (replay + fork) - `test_subgraph_time_travel_to_second_interrupt` — time travel to 2nd interrupt checkpoint; verifies step_a and ask_1 don't re-run, ask_2 re-fires (replay + fork) - `test_subgraph_time_travel_after_completion` — replay from final parent checkpoint after full completion; verifies no nodes re-run and all state values preserved ### Double nested (3 levels: parent → outer → inner subgraph) - `test_3_levels_deep_time_travel_to_first_interrupt` — time travel to innermost checkpoint at 1st interrupt (replay + fork) - `test_3_levels_deep_time_travel_to_second_interrupt` — time travel to innermost checkpoint at 2nd interrupt (replay + fork) - `test_3_levels_deep_time_travel_to_middle_subgraph` — time travel to middle-level subgraph checkpoint (replay + fork) - `test_3_levels_deep_middle_has_interrupts` — middle subgraph has its own `interrupt()` calls (in `pre` node) plus an inner subgraph with interrupts; time travel to middle checkpoint at each interrupt point (replay + fork at both the middle's own interrupt and the inner's interrupt)	2026-03-11 21:59:45 +00:00
Andrew NguonlyandGitHub	7488cf2448	fix: Revert "feat(cli): Add `deploy list` and `deploy delete` subcommands" (#7116 ) Reverts langchain-ai/langgraph#7106 This PR broke the `langgraph deploy` command flags.	2026-03-11 10:56:31 -07:00
Andrew NguonlyandGitHub	e77201cbb1	feat(cli): Add `deploy list` and `deploy delete` subcommands (#7106 ) ### Summary This PR introduces a subcommand implementation that allows `langgraph deploy list` and `langgraph deploy delete` subcommands. #### `langgraph deploy list` ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 langgraph % langgraph deploy list --help ⎈ gke_langchain-test-387119_us-west1_langgraph-cloud-us-west1 Usage: langgraph deploy list [OPTIONS] [Beta] List LangSmith Deployments. Options: --name-contains TEXT Only show deployments whose names contain this value. --api-key TEXT API key. Can also be set via LANGGRAPH_HOST_API_KEY, LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable or .env file. --help Show this message and exit. ``` Output example: ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 cli % langgraph deploy list Deployment ID Deployment Name Deployment URL ------------------------------------ -------------------------- ------------------------------------------------------------------------------------ a40d6567-87c0-485a-a23d-94309a7d4519 ht-andrew-test-04 - 9da26acb-d0c9-4af0-af9e-f3fe8dfe85bc ht-anirudh-deployment-test https://ht-anirudh-deployment-test-428af4737f8a533cb2b107587eb8f38f.us.langgraph.app ``` #### `langgraph deploy delete` ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 langgraph % langgraph deploy delete --help ⎈ gke_langchain-test-387119_us-west1_langgraph-cloud-us-west1 Usage: langgraph deploy delete [OPTIONS] DEPLOYMENT_ID [Beta] Delete a LangSmith Deployment. Options: --force Delete without prompting for confirmation. --api-key TEXT API key. Can also be set via LANGGRAPH_HOST_API_KEY, LANGSMITH_API_KEY, or LANGCHAIN_API_KEY environment variable or .env file. --help Show this message and exit. ``` Output example: ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 cli % langgraph deploy delete a40d6567-87c0-485a-a23d-94309a7d4519 Are you sure you want to delete deployment ID a40d6567-87c0-485a-a23d-94309a7d4519? (Y/n): Y Host API key: Deleted deployment a40d6567-87c0-485a-a23d-94309a7d4519. ``` ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 cli % langgraph deploy delete a40d6567-87c0-485a-a23d-94309a7d4519 Are you sure you want to delete deployment ID a40d6567-87c0-485a-a23d-94309a7d4519? (Y/n): n Aborted! ``` ```bash (env) andrewnguonly@Andrew-Nguonly-KC23X90J02 cli % langgraph deploy delete 9da26acb-d0c9-4af0-af9e-f3fe8dfe85bc --force Host API key: Deleted deployment 9da26acb-d0c9-4af0-af9e-f3fe8dfe85bc. ```	2026-03-11 09:12:17 -07:00
William FHandGitHub	acae5e23b0	chore(sdk-py): cron tz support (#7108 )	2026-03-10 17:38:02 -07:00