Compare commits

..
Author SHA1 Message Date
John KennedyandGitHub 428ebe1a01 Merge branch 'main' into langster/remove-stale-debugger-pull 2026-09-25 21:38:30 -07:00
John Kennedyandopen-swe[bot] <open-swe@users.noreply.github.com> 8e05912899 fix(cli): extend API healthcheck startup window
Keep one-second health probes active for slower graph imports so Compose does not stall for the steady-state interval.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
2026-08-20 07:24:09 +00:00
John KennedyandGitHub f899af1c73 Merge branch 'main' into langster/remove-stale-debugger-pull 2026-08-20 00:07:35 -07:00
John Kennedy 5f479f1af5 fix(cli): preserve hosted Studio URL overrides 2026-08-11 09:39:33 -07:00
John Kennedy 33c3edfde2 fix(cli): drop unrelated uv export change 2026-08-11 09:39:24 -07:00
John Kennedyandopen-swe[bot] <open-swe@users.noreply.github.com> 292fd5787c fix(cli): include workspace metadata in uv exports
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
2026-08-10 23:03:38 +00:00
John Kennedyandopen-swe[bot] <open-swe@users.noreply.github.com> b904db211c fix(cli): update integration runner arguments
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
2026-08-10 22:53:15 +00:00
langsmith-fleet[bot] ecee22fa16 fix(cli): remove discontinued local debugger image 2026-08-06 00:01:58 +00:00
20 changed files with 287 additions and 1294 deletions
+1 -3
View File
@@ -32,8 +32,6 @@ def test(config: pathlib.Path, port: int, tag: str, verbose: bool):
docker_compose=None, docker_compose=None,
port=port, port=port,
watch=False, watch=False,
debugger_port=None,
debugger_base_url=f"http://127.0.0.1:{port}",
postgres_uri=None, postgres_uri=None,
api_version=None, api_version=None,
image=tag, image=tag,
@@ -173,5 +171,5 @@ if __name__ == "__main__":
except BaseException: except BaseException:
logger.exception("Test failed") logger.exception("Test failed")
raise raise
logger.info("Test execution finished") logger.info("Test execution finished")
-2
View File
@@ -103,8 +103,6 @@ The CLI uses a `langgraph.json` configuration file with these key settings:
} }
``` ```
Git dependencies should use credential-free URLs. The CLI conservatively scans direct `langgraph.json` dependencies, common Python package files, uv project and lock files, and common Node.js package and lock files for HTTP Git URLs with userinfo. This check is not exhaustive: generated Docker builds can copy other files, including nested requirement or constraint files, into image layers without scanning them. For private dependencies, provide short-lived credentials through your build environment's secret-backed Git credential helper. Do not store credentials in copied files such as `langgraph.json` or `pip_config_file`.
See the [full documentation](https://reference.langchain.com/python/langgraph-cli) for detailed configuration options. See the [full documentation](https://reference.langchain.com/python/langgraph-cli) for detailed configuration options.
## Development ## Development
-3
View File
@@ -48,9 +48,6 @@ def get_anonymized_params(
if kwargs.get("docker_compose"): if kwargs.get("docker_compose"):
params["docker_compose"] = True params["docker_compose"] = True
if kwargs.get("debugger_port"):
params["debugger_port"] = True
if kwargs.get("postgres_uri"): if kwargs.get("postgres_uri"):
params["postgres_uri"] = True params["postgres_uri"] = True
+89 -34
View File
@@ -5,6 +5,7 @@ import pathlib
import shutil import shutil
import sys import sys
from collections.abc import Sequence from collections.abc import Sequence
from urllib.parse import SplitResult, urlencode, urlsplit, urlunsplit
import click import click
import click.exceptions import click.exceptions
@@ -140,17 +141,6 @@ OPT_VERBOSE = click.option(
help="Show more output from the server logs", help="Show more output from the server logs",
) )
OPT_WATCH = click.option("--watch", is_flag=True, help="Restart on file changes") OPT_WATCH = click.option("--watch", is_flag=True, help="Restart on file changes")
OPT_DEBUGGER_PORT = click.option(
"--debugger-port",
type=int,
help="Pull the debugger image locally and serve the UI on specified port",
)
OPT_DEBUGGER_BASE_URL = click.option(
"--debugger-base-url",
type=str,
help="URL used by the debugger to access LangGraph API. Defaults to http://127.0.0.1:[PORT]",
)
OPT_POSTGRES_URI = click.option( OPT_POSTGRES_URI = click.option(
"--postgres-uri", "--postgres-uri",
help="Postgres URI to use for the database. Defaults to launching a local database", help="Postgres URI to use for the database. Defaults to launching a local database",
@@ -242,18 +232,94 @@ cli.add_command(deploy)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _validated_http_url(value: str, option_name: str) -> SplitResult:
try:
parsed = urlsplit(value)
hostname = parsed.hostname
_ = parsed.port
except ValueError as exc:
raise click.UsageError(
f"{option_name} must be a valid HTTP(S) URL without credentials."
) from exc
if (
value != value.strip()
or parsed.scheme not in {"http", "https"}
or not parsed.netloc
or not hostname
or parsed.username is not None
or parsed.password is not None
):
raise click.UsageError(
f"{option_name} must be a valid HTTP(S) URL without credentials."
)
return parsed
def _studio_link(
*,
port: int,
studio_url: str | None,
api_url: str | None,
debugger_base_url: str | None,
) -> str:
if debugger_base_url is not None:
if api_url is not None and api_url != debugger_base_url:
raise click.UsageError(
"--api-url and --debugger-base-url cannot specify different URLs."
)
click.echo(
"Warning: --debugger-base-url is deprecated; use --api-url instead.",
err=True,
)
api_url = debugger_base_url
studio_url = "https://smith.langchain.com" if studio_url is None else studio_url
api_url = f"http://127.0.0.1:{port}" if api_url is None else api_url
studio_parts = _validated_http_url(studio_url, "--studio-url")
_validated_http_url(api_url, "--api-url")
if studio_parts.query or studio_parts.fragment:
raise click.UsageError(
"--studio-url must not include a query string or fragment."
)
studio_path = f"{studio_parts.path.rstrip('/')}/studio/"
return urlunsplit(
studio_parts._replace(
path=studio_path,
query=urlencode({"baseUrl": api_url}),
)
)
@OPT_RECREATE @OPT_RECREATE
@OPT_PULL @OPT_PULL
@OPT_PORT @OPT_PORT
@OPT_DOCKER_COMPOSE @OPT_DOCKER_COMPOSE
@OPT_CONFIG @OPT_CONFIG
@OPT_VERBOSE @OPT_VERBOSE
@OPT_DEBUGGER_PORT
@OPT_DEBUGGER_BASE_URL
@OPT_WATCH @OPT_WATCH
@OPT_POSTGRES_URI @OPT_POSTGRES_URI
@OPT_API_VERSION @OPT_API_VERSION
@OPT_ENGINE_RUNTIME_MODE @OPT_ENGINE_RUNTIME_MODE
@click.option(
"--studio-url",
type=str,
default=None,
help="URL of the LangGraph Studio instance. Defaults to https://smith.langchain.com",
)
@click.option(
"--api-url",
type=str,
default=None,
help="URL that LangGraph Studio uses to access the API. Defaults to http://127.0.0.1:[PORT]",
)
@click.option(
"--debugger-base-url",
type=str,
default=None,
hidden=True,
)
@click.option( @click.option(
"--image", "--image",
type=str, type=str,
@@ -284,14 +350,21 @@ def up(
watch: bool, watch: bool,
wait: bool, wait: bool,
verbose: bool, verbose: bool,
debugger_port: int | None,
debugger_base_url: str | None,
postgres_uri: str | None, postgres_uri: str | None,
api_version: str | None, api_version: str | None,
engine_runtime_mode: str, engine_runtime_mode: str,
studio_url: str | None,
api_url: str | None,
debugger_base_url: str | None,
image: str | None, image: str | None,
base_image: str | None, base_image: str | None,
): ):
studio_link = _studio_link(
port=port,
studio_url=studio_url,
api_url=api_url,
debugger_base_url=debugger_base_url,
)
click.secho("Starting LangGraph API server...", fg="green") click.secho("Starting LangGraph API server...", fg="green")
click.secho( click.secho(
"""For local dev, requires env var LANGSMITH_API_KEY with access to LangSmith Deployment. """For local dev, requires env var LANGSMITH_API_KEY with access to LangSmith Deployment.
@@ -308,8 +381,6 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE
pull=pull, pull=pull,
watch=watch, watch=watch,
verbose=verbose, verbose=verbose,
debugger_port=debugger_port,
debugger_base_url=debugger_base_url,
postgres_uri=postgres_uri, postgres_uri=postgres_uri,
api_version=api_version, api_version=api_version,
engine_runtime_mode=engine_runtime_mode, engine_runtime_mode=engine_runtime_mode,
@@ -337,20 +408,12 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE
if "unpacking to docker.io" in line: if "unpacking to docker.io" in line:
set("Starting...") set("Starting...")
elif "Application startup complete" in line: elif "Application startup complete" in line:
debugger_origin = (
f"http://localhost:{debugger_port}"
if debugger_port
else "https://smith.langchain.com"
)
debugger_base_url_query = (
debugger_base_url or f"http://127.0.0.1:{port}"
)
set("") set("")
sys.stdout.write( sys.stdout.write(
f"""Ready! f"""Ready!
- API: http://localhost:{port} - API: http://localhost:{port}
- Docs: http://localhost:{port}/docs - Docs: http://localhost:{port}/docs
- LangGraph Studio: {debugger_origin}/studio/?baseUrl={debugger_base_url_query} - LangGraph Studio: {studio_link}
""" """
) )
sys.stdout.flush() sys.stdout.flush()
@@ -935,8 +998,6 @@ def prepare_args_and_stdin(
docker_compose: pathlib.Path | None, docker_compose: pathlib.Path | None,
port: int, port: int,
watch: bool, watch: bool,
debugger_port: int | None = None,
debugger_base_url: str | None = None,
postgres_uri: str | None = None, postgres_uri: str | None = None,
api_version: str | None = None, api_version: str | None = None,
engine_runtime_mode: str = "combined_queue_worker", engine_runtime_mode: str = "combined_queue_worker",
@@ -950,8 +1011,6 @@ def prepare_args_and_stdin(
stdin = langgraph_cli.docker.compose( stdin = langgraph_cli.docker.compose(
capabilities, capabilities,
port=port, port=port,
debugger_port=debugger_port,
debugger_base_url=debugger_base_url,
postgres_uri=postgres_uri, postgres_uri=postgres_uri,
image=image, image=image,
base_image=base_image, base_image=base_image,
@@ -989,8 +1048,6 @@ def prepare(
pull: bool, pull: bool,
watch: bool, watch: bool,
verbose: bool, verbose: bool,
debugger_port: int | None = None,
debugger_base_url: str | None = None,
postgres_uri: str | None = None, postgres_uri: str | None = None,
api_version: str | None = None, api_version: str | None = None,
engine_runtime_mode: str = "combined_queue_worker", engine_runtime_mode: str = "combined_queue_worker",
@@ -1032,8 +1089,6 @@ def prepare(
docker_compose=docker_compose, docker_compose=docker_compose,
port=port, port=port,
watch=watch, watch=watch,
debugger_port=debugger_port,
debugger_base_url=debugger_base_url or f"http://127.0.0.1:{port}",
postgres_uri=postgres_uri, postgres_uri=postgres_uri,
api_version=api_version, api_version=api_version,
engine_runtime_mode=engine_runtime_mode, engine_runtime_mode=engine_runtime_mode,
+3 -87
View File
@@ -6,7 +6,6 @@ import re
import shlex import shlex
import textwrap import textwrap
from collections import Counter from collections import Counter
from collections.abc import Iterable
from typing import Literal, NamedTuple from typing import Literal, NamedTuple
import click import click
@@ -37,10 +36,6 @@ DISALLOWED_BUILD_COMMAND_CHARS = [
# This blocks background execution (cmd &) while allowing command # This blocks background execution (cmd &) while allowing command
# chaining (cmd1 && cmd2) which is common in build commands. # chaining (cmd1 && cmd2) which is common in build commands.
_SINGLE_AMPERSAND_RE = re.compile(r"(?<!&)&(?:&&)*(?!&)") _SINGLE_AMPERSAND_RE = re.compile(r"(?<!&)&(?:&&)*(?!&)")
_GIT_HTTP_AUTHORITY_RES = (
re.compile(r"git\+https?://(?P<authority>[^/\s\"']+)", re.I),
re.compile(r"\bgit\s*=\s*[\"']https?://(?P<authority>[^/\s\"']+)", re.I),
)
_API_VERSION_PATTERN = re.compile( _API_VERSION_PATTERN = re.compile(
r"^(?P<major>\d+)" r"^(?P<major>\d+)"
r"(?:\.(?P<minor>\d+))?" r"(?:\.(?P<minor>\d+))?"
@@ -83,62 +78,6 @@ def has_disallowed_build_command_content(command: str) -> bool:
return False return False
def _has_git_http_url_userinfo(dependency: str) -> bool:
"""Check whether a Git HTTP URL contains userinfo."""
return any(
"@" in match.group("authority")
for pattern in _GIT_HTTP_AUTHORITY_RES
for match in pattern.finditer(dependency)
)
def _validate_git_http_url_userinfo(
values: Iterable[str], *, source: pathlib.Path | None = None
) -> None:
"""Reject credential-bearing Git HTTP URLs without echoing their values."""
if not any(_has_git_http_url_userinfo(value) for value in values):
return
message = (
"Git dependency URLs must not contain credentials or other URL "
"userinfo because generated Dockerfiles and image layers can retain "
"them. Use a credential-free Git URL and provide short-lived "
"credentials through your build environment's secret-backed Git "
"credential helper."
)
if source is not None:
message += f" Found in: {source}"
raise click.UsageError(message)
def _validate_git_http_url_userinfo_files(paths: Iterable[pathlib.Path]) -> None:
"""Reject credential-bearing Git HTTP URLs in dependency files."""
for path in paths:
path = path.resolve()
if not path.is_file():
continue
try:
contents = path.read_text(encoding="utf-8", errors="replace")
except OSError:
raise click.UsageError(
f"Could not inspect dependency file for embedded credentials: {path}"
) from None
_validate_git_http_url_userinfo([contents], source=path)
def _validate_local_dependency_files(config_path: pathlib.Path, config: Config) -> None:
"""Validate dependency files copied into a non-uv Python image."""
paths: list[pathlib.Path] = []
for dependency in config["dependencies"]:
if not isinstance(dependency, str) or not dependency.startswith("."):
continue
root = (config_path.parent / dependency).resolve()
paths.extend(
root / name
for name in ("requirements.txt", "pyproject.toml", "setup.py", "setup.cfg")
)
_validate_git_http_url_userinfo_files(paths)
MIN_PYTHON_VERSION = "3.11" MIN_PYTHON_VERSION = "3.11"
DEFAULT_PYTHON_VERSION = "3.11" DEFAULT_PYTHON_VERSION = "3.11"
@@ -381,9 +320,7 @@ def _get_source_kind(config: Config) -> str | None:
return kind if isinstance(kind, str) else None return kind if isinstance(kind, str) else None
def validate_config( def validate_config(config: Config) -> Config:
config: Config, *, source_path: pathlib.Path | None = None
) -> Config:
"""Validate a configuration dictionary.""" """Validate a configuration dictionary."""
graphs = config.get("graphs", {}) graphs = config.get("graphs", {})
@@ -478,15 +415,6 @@ def validate_config(
' "source": {"kind": "uv", "root": ".."}' ' "source": {"kind": "uv", "root": ".."}'
) )
_validate_git_http_url_userinfo(
(
dependency
for dependency in config["dependencies"]
if isinstance(dependency, str)
),
source=source_path,
)
source = config.get("source") source = config.get("source")
source_kind = _get_source_kind(config) source_kind = _get_source_kind(config)
if source is not None and not isinstance(source, dict): if source is not None and not isinstance(source, dict):
@@ -681,7 +609,7 @@ def validate_config_file(config_path: pathlib.Path) -> Config:
"""Load and validate a configuration file.""" """Load and validate a configuration file."""
with open(config_path) as f: with open(config_path) as f:
config = json.load(f) config = json.load(f)
validated = validate_config(config, source_path=config_path.resolve()) validated = validate_config(config)
# Enforce the package.json doesn't enforce an # Enforce the package.json doesn't enforce an
# incompatible Node.js version # incompatible Node.js version
if validated.get("node_version"): if validated.get("node_version"):
@@ -1352,7 +1280,6 @@ def python_config_to_docker(
api_version=api_version, api_version=api_version,
build_tools_to_uninstall=build_tools_to_uninstall, build_tools_to_uninstall=build_tools_to_uninstall,
) )
_validate_local_dependency_files(config_path, config)
if pip_installer == "auto": if pip_installer == "auto":
if _image_supports_uv(base_image): if _image_supports_uv(base_image):
pip_installer = "uv" pip_installer = "uv"
@@ -1563,18 +1490,7 @@ def node_config_to_docker(
) -> tuple[str, dict[str, str]]: ) -> tuple[str, dict[str, str]]:
# Calculate paths for monorepo support # Calculate paths for monorepo support
install_root = ( install_root = (
pathlib.Path(build_context).resolve() pathlib.Path(build_context).resolve() if build_context else config_path.parent
if build_context
else config_path.parent.resolve()
)
config_root = config_path.parent.resolve()
dependency_roots = (
(install_root, config_root) if install_root != config_root else (install_root,)
)
_validate_git_http_url_userinfo_files(
root / name
for root in dependency_roots
for name in ("package.json", "package-lock.json", "yarn.lock", "pnpm-lock.yaml")
) )
install_cmd = install_command or _get_node_pm_install_cmd(install_root) install_cmd = install_command or _get_node_pm_install_cmd(install_root)
if build_context: if build_context:
+1 -36
View File
@@ -142,29 +142,6 @@ def check_capabilities(runner) -> DockerCapabilities:
) )
def debugger_compose(*, port: int | None = None, base_url: str | None = None) -> dict:
if port is None:
return ""
config = {
"langgraph-debugger": {
"image": "langchain/langgraph-debugger",
"restart": "on-failure",
"depends_on": {
"langgraph-postgres": {"condition": "service_healthy"},
},
"ports": [f'"{port}:3968"'],
}
}
if base_url:
config["langgraph-debugger"]["environment"] = {
"VITE_STUDIO_LOCAL_GRAPH_URL": base_url
}
return config
# Function to convert dictionary to YAML # Function to convert dictionary to YAML
def dict_to_yaml(d: dict, *, indent: int = 0) -> str: def dict_to_yaml(d: dict, *, indent: int = 0) -> str:
"""Convert a dictionary to a YAML string.""" """Convert a dictionary to a YAML string."""
@@ -191,8 +168,6 @@ def compose_as_dict(
capabilities: DockerCapabilities, capabilities: DockerCapabilities,
*, *,
port: int, port: int,
debugger_port: int | None = None,
debugger_base_url: str | None = None,
# postgres://user:password@host:port/database?option=value # postgres://user:password@host:port/database?option=value
postgres_uri: str | None = None, postgres_uri: str | None = None,
# If you are running against an already-built image, you can pass it here # If you are running against an already-built image, you can pass it here
@@ -253,12 +228,6 @@ def compose_as_dict(
else: else:
services["langgraph-postgres"]["healthcheck"]["interval"] = "5s" services["langgraph-postgres"]["healthcheck"]["interval"] = "5s"
# Add optional debugger service if debugger_port is specified
if debugger_port:
services["langgraph-debugger"] = debugger_compose(
port=debugger_port, base_url=debugger_base_url
)["langgraph-debugger"]
# Add langgraph-api service # Add langgraph-api service
api_environment = { api_environment = {
"REDIS_URI": "redis://langgraph-redis:6379", "REDIS_URI": "redis://langgraph-redis:6379",
@@ -289,7 +258,7 @@ def compose_as_dict(
"test": "python /api/healthcheck.py", "test": "python /api/healthcheck.py",
"interval": "60s", "interval": "60s",
"start_interval": "1s", "start_interval": "1s",
"start_period": "10s", "start_period": "60s",
} }
# Final compose dictionary with volumes included if needed # Final compose dictionary with volumes included if needed
@@ -305,8 +274,6 @@ def compose(
capabilities: DockerCapabilities, capabilities: DockerCapabilities,
*, *,
port: int, port: int,
debugger_port: int | None = None,
debugger_base_url: str | None = None,
# postgres://user:password@host:port/database?option=value # postgres://user:password@host:port/database?option=value
postgres_uri: str | None = None, postgres_uri: str | None = None,
image: str | None = None, image: str | None = None,
@@ -318,8 +285,6 @@ def compose(
compose_content = compose_as_dict( compose_content = compose_as_dict(
capabilities, capabilities,
port=port, port=port,
debugger_port=debugger_port,
debugger_base_url=debugger_base_url,
postgres_uri=postgres_uri, postgres_uri=postgres_uri,
image=image, image=image,
base_image=base_image, base_image=base_image,
+1 -5
View File
@@ -650,8 +650,7 @@ class Config(TypedDict, total=False):
pip_config_file: str | None pip_config_file: str | None
"""Optional. Path to a pip config file (e.g., "/etc/pip.conf" or "pip.ini") for controlling """Optional. Path to a pip config file (e.g., "/etc/pip.conf" or "pip.ini") for controlling
package installation (custom indices, timeouts, etc.). The file is copied into the package installation (custom indices, credentials, etc.).
generated image, so it must not contain credentials or other secrets.
Only relevant if Python dependencies are installed via pip. If omitted, default pip settings are used. Only relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.
""" """
@@ -690,9 +689,6 @@ class Config(TypedDict, total=False):
- "." or "./src" if you have a local Python package - "." or "./src" if you have a local Python package
- str (aka "anthropic") for a PyPI package - str (aka "anthropic") for a PyPI package
- "git+https://github.com/org/repo.git@main" for a Git-based package - "git+https://github.com/org/repo.git@main" for a Git-based package
Git HTTP URLs must not contain userinfo such as a username or token. For private
dependencies, provide short-lived credentials through the build environment's
secret-backed Git credential helper.
Defaults to an empty list, meaning no additional packages installed beyond your base environment. Defaults to an empty list, meaning no additional packages installed beyond your base environment.
This field is not supported when `source.kind` is `uv`. This field is not supported when `source.kind` is `uv`.
-10
View File
@@ -880,7 +880,6 @@ def python_config_to_docker_uv_lock(
_get_node_pm_install_cmd, _get_node_pm_install_cmd,
_get_pip_cleanup_lines, _get_pip_cleanup_lines,
_image_supports_uv, _image_supports_uv,
_validate_git_http_url_userinfo_files,
docker_tag, docker_tag,
) )
@@ -891,20 +890,11 @@ def python_config_to_docker_uv_lock(
) )
config_root = config_path.parent.resolve() config_root = config_path.parent.resolve()
source_root = config["source"].get("root", ".")
project_root = (config_root / source_root).resolve()
_validate_git_http_url_userinfo_files(
[project_root / "pyproject.toml", project_root / "uv.lock"]
)
install_cmd = "uv pip install --system" install_cmd = "uv pip install --system"
_, global_reqs_pip_install, pip_config_file_str = _build_python_install_commands( _, global_reqs_pip_install, pip_config_file_str = _build_python_install_commands(
config, install_cmd config, install_cmd
) )
plan = _plan_uv_lock_workspace(config_path, config) plan = _plan_uv_lock_workspace(config_path, config)
_validate_git_http_url_userinfo_files(
package.pyproject_path for package in plan.install_order
)
_update_uv_lock_graph_paths(config_path, config, plan) _update_uv_lock_graph_paths(config_path, config, plan)
for section, key in [ for section, key in [
+2 -2
View File
@@ -28,7 +28,7 @@
"type": "null" "type": "null"
} }
], ],
"description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, timeouts, etc.). The file is copied into the\ngenerated image, so it must not contain credentials or other secrets.\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n" "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, credentials, etc.).\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
}, },
"_INTERNAL_docker_tag": { "_INTERNAL_docker_tag": {
"anyOf": [ "anyOf": [
@@ -270,7 +270,7 @@
"type": "null" "type": "null"
} }
], ],
"description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, timeouts, etc.). The file is copied into the\ngenerated image, so it must not contain credentials or other secrets.\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n" "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, credentials, etc.).\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
}, },
"_INTERNAL_docker_tag": { "_INTERNAL_docker_tag": {
"anyOf": [ "anyOf": [
+2 -2
View File
@@ -28,7 +28,7 @@
"type": "null" "type": "null"
} }
], ],
"description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, timeouts, etc.). The file is copied into the\ngenerated image, so it must not contain credentials or other secrets.\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n" "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, credentials, etc.).\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
}, },
"_INTERNAL_docker_tag": { "_INTERNAL_docker_tag": {
"anyOf": [ "anyOf": [
@@ -270,7 +270,7 @@
"type": "null" "type": "null"
} }
], ],
"description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, timeouts, etc.). The file is copied into the\ngenerated image, so it must not contain credentials or other secrets.\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n" "description": "Optional. Path to a pip config file (e.g., \"/etc/pip.conf\" or \"pip.ini\") for controlling\npackage installation (custom indices, credentials, etc.).\n\nOnly relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.\n"
}, },
"_INTERNAL_docker_tag": { "_INTERNAL_docker_tag": {
"anyOf": [ "anyOf": [
+80 -31
View File
@@ -8,10 +8,11 @@ from contextlib import contextmanager
from pathlib import Path from pathlib import Path
import click import click
import pytest
from click.testing import CliRunner from click.testing import CliRunner
import langgraph_cli.deploy as deploy_module import langgraph_cli.deploy as deploy_module
from langgraph_cli.cli import cli, prepare_args_and_stdin from langgraph_cli.cli import _studio_link, cli, prepare_args_and_stdin
from langgraph_cli.config import Config, _get_pip_cleanup_lines, validate_config from langgraph_cli.config import Config, _get_pip_cleanup_lines, validate_config
from langgraph_cli.docker import DEFAULT_POSTGRES_URI, DockerCapabilities, Version from langgraph_cli.docker import DEFAULT_POSTGRES_URI, DockerCapabilities, Version
from langgraph_cli.util import clean_empty_lines from langgraph_cli.util import clean_empty_lines
@@ -56,8 +57,6 @@ def test_prepare_args_and_stdin() -> None:
Config(dependencies=[".", "../../.."], graphs={"agent": "agent.py:graph"}) Config(dependencies=[".", "../../.."], graphs={"agent": "agent.py:graph"})
) )
port = 8000 port = 8000
debugger_port = 8001
debugger_graph_url = f"http://127.0.0.1:{port}"
actual_args, actual_stdin = prepare_args_and_stdin( actual_args, actual_stdin = prepare_args_and_stdin(
capabilities=DEFAULT_DOCKER_CAPABILITIES, capabilities=DEFAULT_DOCKER_CAPABILITIES,
@@ -65,8 +64,6 @@ def test_prepare_args_and_stdin() -> None:
config=config, config=config,
docker_compose=pathlib.Path("custom-docker-compose.yml"), docker_compose=pathlib.Path("custom-docker-compose.yml"),
port=port, port=port,
debugger_port=debugger_port,
debugger_base_url=debugger_graph_url,
watch=True, watch=True,
) )
@@ -110,16 +107,6 @@ services:
retries: 5 retries: 5
interval: 60s interval: 60s
start_interval: 1s start_interval: 1s
langgraph-debugger:
image: langchain/langgraph-debugger
restart: on-failure
depends_on:
langgraph-postgres:
condition: service_healthy
ports:
- "{debugger_port}:3968"
environment:
VITE_STUDIO_LOCAL_GRAPH_URL: {debugger_graph_url}
langgraph-api: langgraph-api:
ports: ports:
- "8000:8000" - "8000:8000"
@@ -135,7 +122,7 @@ services:
test: python /api/healthcheck.py test: python /api/healthcheck.py
interval: 60s interval: 60s
start_interval: 1s start_interval: 1s
start_period: 10s start_period: 60s
pull_policy: build pull_policy: build
build: build:
@@ -178,8 +165,6 @@ def test_prepare_args_and_stdin_with_image() -> None:
Config(dependencies=[".", "../../.."], graphs={"agent": "agent.py:graph"}) Config(dependencies=[".", "../../.."], graphs={"agent": "agent.py:graph"})
) )
port = 8000 port = 8000
debugger_port = 8001
debugger_graph_url = f"http://127.0.0.1:{port}"
actual_args, actual_stdin = prepare_args_and_stdin( actual_args, actual_stdin = prepare_args_and_stdin(
capabilities=DEFAULT_DOCKER_CAPABILITIES, capabilities=DEFAULT_DOCKER_CAPABILITIES,
@@ -187,8 +172,6 @@ def test_prepare_args_and_stdin_with_image() -> None:
config=config, config=config,
docker_compose=pathlib.Path("custom-docker-compose.yml"), docker_compose=pathlib.Path("custom-docker-compose.yml"),
port=port, port=port,
debugger_port=debugger_port,
debugger_base_url=debugger_graph_url,
watch=True, watch=True,
image="my-cool-image", image="my-cool-image",
) )
@@ -233,16 +216,6 @@ services:
retries: 5 retries: 5
interval: 60s interval: 60s
start_interval: 1s start_interval: 1s
langgraph-debugger:
image: langchain/langgraph-debugger
restart: on-failure
depends_on:
langgraph-postgres:
condition: service_healthy
ports:
- "{debugger_port}:3968"
environment:
VITE_STUDIO_LOCAL_GRAPH_URL: {debugger_graph_url}
langgraph-api: langgraph-api:
ports: ports:
- "8000:8000" - "8000:8000"
@@ -259,7 +232,7 @@ services:
test: python /api/healthcheck.py test: python /api/healthcheck.py
interval: 60s interval: 60s
start_interval: 1s start_interval: 1s
start_period: 10s start_period: 60s
develop: develop:
@@ -289,6 +262,82 @@ def test_version_option() -> None:
) )
def test_up_help_shows_hosted_studio_options() -> None:
result = CliRunner().invoke(cli, ["up", "--help"])
assert result.exit_code == 0, result.output
assert "--studio-url" in result.output
assert "--api-url" in result.output
assert "--debugger-port" not in result.output
assert "--debugger-base-url" not in result.output
def test_studio_link_defaults_to_hosted_studio() -> None:
assert _studio_link(
port=8123,
studio_url=None,
api_url=None,
debugger_base_url=None,
) == ("https://smith.langchain.com/studio/?baseUrl=http%3A%2F%2F127.0.0.1%3A8123")
def test_studio_link_supports_self_hosted_and_remote_urls() -> None:
assert _studio_link(
port=8123,
studio_url="https://langsmith.example.com/prefix/",
api_url="https://api.example.com/graph?tenant=a&region=eu",
debugger_base_url=None,
) == (
"https://langsmith.example.com/prefix/studio/"
"?baseUrl=https%3A%2F%2Fapi.example.com%2Fgraph%3Ftenant%3Da%26region%3Deu"
)
def test_studio_link_supports_deprecated_debugger_base_url(capsys) -> None:
assert _studio_link(
port=8123,
studio_url=None,
api_url=None,
debugger_base_url="https://api.example.com",
).endswith("?baseUrl=https%3A%2F%2Fapi.example.com")
assert "--debugger-base-url is deprecated; use --api-url" in capsys.readouterr().err
@pytest.mark.parametrize(
("studio_url", "api_url"),
[
("javascript:alert(1)", None),
("https://user:password@example.com", None),
("https://smith.langchain.com?workspace=test", None),
(None, "file:///tmp/langgraph.sock"),
(None, "https://user:password@example.com"),
],
)
def test_studio_link_rejects_unsafe_urls(
studio_url: str | None, api_url: str | None
) -> None:
with pytest.raises(click.UsageError):
_studio_link(
port=8123,
studio_url=studio_url,
api_url=api_url,
debugger_base_url=None,
)
def test_studio_link_rejects_conflicting_api_url_aliases() -> None:
with pytest.raises(
click.UsageError,
match="cannot specify different URLs",
):
_studio_link(
port=8123,
studio_url=None,
api_url="https://api.example.com",
debugger_base_url="https://other.example.com",
)
def test_top_level_help_shows_deploy_subcommands() -> None: def test_top_level_help_shows_deploy_subcommands() -> None:
runner = CliRunner() runner = CliRunner()
-237
View File
@@ -255,243 +255,6 @@ def test_validate_config():
) )
@pytest.mark.parametrize(
"dependency",
[
"git+https://user:secret-token@github.com/org/private.git@main",
"private-package @ git+http://token@github.com/org/private.git",
"git+HTTPS://user%40example.com:secret%2Ftoken@github.com/org/private.git",
"git+https://${GIT_TOKEN}@github.com/org/private.git",
],
)
def test_validate_config_rejects_git_http_url_userinfo(dependency: str):
with pytest.raises(click.UsageError) as exc_info:
validate_config(
{
"python_version": "3.11",
"dependencies": [dependency],
"graphs": {"agent": "./agent.py:graph"},
}
)
message = str(exc_info.value)
assert "must not contain credentials or other URL userinfo" in message
assert "secret-token" not in message
assert "secret%2Ftoken" not in message
def test_validate_config_file_reports_source_for_git_http_url_userinfo(
tmp_path: pathlib.Path,
):
config_path = tmp_path / "langgraph.json"
config_path.write_text(
json.dumps(
{
"python_version": "3.11",
"dependencies": ["git+https://secret-token@github.com/org/private.git"],
"graphs": {"agent": "./agent.py:graph"},
}
)
)
with pytest.raises(click.UsageError) as exc_info:
validate_config_file(config_path)
message = str(exc_info.value)
assert "secret-token" not in message
assert f"Found in: {config_path.resolve()}" in message
@pytest.mark.parametrize(
"manifest", ["package.json", "package-lock.json", "yarn.lock", "pnpm-lock.yaml"]
)
def test_config_to_docker_rejects_git_http_url_userinfo_in_node_files(
tmp_path: pathlib.Path, manifest: str
):
config_path = tmp_path / "langgraph.json"
config_path.write_text("{}\n")
(tmp_path / "agent.js").write_text("export const graph = {};\n")
(tmp_path / "package.json").write_text('{"name":"agent"}\n')
(tmp_path / manifest).write_text(
'"priv": "git+https://user:secret-token@github.com/org/private.git"\n'
)
config = validate_config(
{
"node_version": "20",
"graphs": {"agent": "./agent.js:graph"},
}
)
with pytest.raises(click.UsageError) as exc_info:
config_to_docker(
config_path,
config,
base_image="langchain/langgraphjs-api",
)
message = str(exc_info.value)
assert "must not contain credentials or other URL userinfo" in message
assert "secret-token" not in message
assert f"Found in: {(tmp_path / manifest).resolve()}" in message
def test_config_to_docker_allows_node_git_urls_without_http_userinfo(
tmp_path: pathlib.Path,
):
config_path = tmp_path / "langgraph.json"
config_path.write_text("{}\n")
(tmp_path / "agent.js").write_text("export const graph = {};\n")
(tmp_path / "package.json").write_text(
'{"dependencies":{"public":"git+https://github.com/org/public.git"}}\n'
)
config = validate_config(
{
"node_version": "20",
"graphs": {"agent": "./agent.js:graph"},
}
)
docker, _ = config_to_docker(
config_path,
config,
base_image="langchain/langgraphjs-api",
)
assert f"ADD . /deps/{tmp_path.name}" in docker
def test_config_to_docker_rejects_git_http_url_userinfo_in_node_workspace(
tmp_path: pathlib.Path,
):
config_root = tmp_path / "apps" / "agent"
config_root.mkdir(parents=True)
config_path = config_root / "langgraph.json"
config_path.write_text("{}\n")
(config_root / "agent.js").write_text("export const graph = {};\n")
(config_root / "package.json").write_text(
'{"dependencies":{"priv":"git+https://secret-token@github.com/org/private.git"}}\n'
)
(tmp_path / "package.json").write_text('{"name":"workspace"}\n')
config = validate_config(
{
"node_version": "20",
"graphs": {"agent": "./agent.js:graph"},
}
)
with pytest.raises(click.UsageError) as exc_info:
config_to_docker(
config_path,
config,
base_image="langchain/langgraphjs-api",
build_context=str(tmp_path),
)
message = str(exc_info.value)
assert "secret-token" not in message
assert f"Found in: {(config_root / 'package.json').resolve()}" in message
@pytest.mark.parametrize(
"dependency",
[
"git+https://github.com/org/public.git@main",
"private-package @ git+https://github.com/org/private.git@main",
"git+ssh://git@github.com/org/private.git@main",
],
)
def test_validate_config_allows_git_urls_without_http_userinfo(dependency: str):
config = validate_config(
{
"python_version": "3.11",
"dependencies": [dependency],
"graphs": {"agent": "./agent.py:graph"},
}
)
assert config["dependencies"] == [dependency]
def test_config_to_docker_rejects_git_http_url_userinfo_in_requirements(
tmp_path: pathlib.Path,
):
config_path = tmp_path / "langgraph.json"
config_path.write_text("{}\n")
(tmp_path / "agent.py").write_text("graph = object()\n")
(tmp_path / "requirements.txt").write_text(
"private @ git+https://secret-token@github.com/org/private.git\n"
)
config = validate_config(
{
"python_version": "3.11",
"dependencies": ["."],
"graphs": {"agent": "./agent.py:graph"},
}
)
with pytest.raises(click.UsageError) as exc_info:
config_to_docker(
config_path,
config,
base_image="langchain/langgraph-api:0.2.47",
)
message = str(exc_info.value)
assert "must not contain credentials or other URL userinfo" in message
assert "secret-token" not in message
assert f"Found in: {(tmp_path / 'requirements.txt').resolve()}" in message
@pytest.mark.parametrize("manifest", ["pyproject.toml", "uv.lock"])
def test_config_to_docker_rejects_git_http_url_userinfo_in_uv_files(
tmp_path: pathlib.Path, manifest: str
):
config_path = tmp_path / "langgraph.json"
config_path.write_text("{}\n")
(tmp_path / "src").mkdir()
(tmp_path / "src" / "agent.py").write_text("graph = object()\n")
pyproject = textwrap.dedent(
"""
[project]
name = "agent"
version = "0.1.0"
dependencies = ["private"]
[tool.uv.sources]
private = { git = "https://github.com/org/private.git" }
"""
).strip()
uv_lock = "# uv lock file\n"
if manifest == "pyproject.toml":
pyproject = pyproject.replace(
"https://github.com", "https://secret-token@github.com"
)
else:
uv_lock += (
'source = { git = "https://secret-token@github.com/org/private.git" }\n'
)
(tmp_path / "pyproject.toml").write_text(pyproject + "\n")
(tmp_path / "uv.lock").write_text(uv_lock)
config = validate_config(
{
"python_version": "3.11",
"graphs": {"agent": "./src/agent.py:graph"},
"source": {"kind": "uv"},
}
)
with pytest.raises(click.UsageError) as exc_info:
config_to_docker(
config_path,
config,
base_image="langchain/langgraph-api:0.2.47",
)
message = str(exc_info.value)
assert "must not contain credentials or other URL userinfo" in message
assert "secret-token" not in message
def test_validate_config_image_distro(): def test_validate_config_image_distro():
"""Test validation of image_distro field.""" """Test validation of image_distro field."""
# Valid image_distro values should work # Valid image_distro values should work
+4 -98
View File
@@ -16,7 +16,7 @@ DEFAULT_DOCKER_CAPABILITIES = DockerCapabilities(
) )
def test_compose_with_no_debugger_and_custom_db(): def test_compose_with_custom_db():
port = 8123 port = 8123
custom_postgres_uri = "custom_postgres_uri" custom_postgres_uri = "custom_postgres_uri"
actual_compose_str = compose( actual_compose_str = compose(
@@ -42,7 +42,7 @@ def test_compose_with_no_debugger_and_custom_db():
assert clean_empty_lines(actual_compose_str) == expected_compose_str assert clean_empty_lines(actual_compose_str) == expected_compose_str
def test_compose_with_no_debugger_and_custom_db_with_healthcheck(): def test_compose_with_custom_db_and_healthcheck():
port = 8123 port = 8123
custom_postgres_uri = "custom_postgres_uri" custom_postgres_uri = "custom_postgres_uri"
actual_compose_str = compose( actual_compose_str = compose(
@@ -71,39 +71,11 @@ def test_compose_with_no_debugger_and_custom_db_with_healthcheck():
test: python /api/healthcheck.py test: python /api/healthcheck.py
interval: 60s interval: 60s
start_interval: 1s start_interval: 1s
start_period: 10s""" start_period: 60s"""
assert clean_empty_lines(actual_compose_str) == expected_compose_str assert clean_empty_lines(actual_compose_str) == expected_compose_str
def test_compose_with_debugger_and_custom_db(): def test_compose_with_default_db():
port = 8123
custom_postgres_uri = "custom_postgres_uri"
actual_compose_str = compose(
DEFAULT_DOCKER_CAPABILITIES,
port=port,
postgres_uri=custom_postgres_uri,
)
expected_compose_str = f"""services:
langgraph-redis:
image: redis:6
healthcheck:
test: redis-cli ping
interval: 5s
timeout: 1s
retries: 5
langgraph-api:
ports:
- "{port}:8000"
depends_on:
langgraph-redis:
condition: service_healthy
environment:
REDIS_URI: redis://langgraph-redis:6379
POSTGRES_URI: {custom_postgres_uri}"""
assert clean_empty_lines(actual_compose_str) == expected_compose_str
def test_compose_with_debugger_and_default_db():
port = 8123 port = 8123
actual_compose_str = compose(DEFAULT_DOCKER_CAPABILITIES, port=port) actual_compose_str = compose(DEFAULT_DOCKER_CAPABILITIES, port=port)
expected_compose_str = f"""volumes: expected_compose_str = f"""volumes:
@@ -302,72 +274,6 @@ def test_compose_with_api_version_and_custom_postgres():
assert clean_empty_lines(actual_compose_str) == expected_compose_str assert clean_empty_lines(actual_compose_str) == expected_compose_str
def test_compose_with_api_version_and_debugger():
"""Test compose function with api_version and debugger port."""
port = 8123
debugger_port = 8001
api_version = "0.2.74"
actual_compose_str = compose(
DEFAULT_DOCKER_CAPABILITIES,
port=port,
api_version=api_version,
debugger_port=debugger_port,
)
expected_compose_str = f"""volumes:
langgraph-data:
driver: local
services:
langgraph-redis:
image: redis:6
healthcheck:
test: redis-cli ping
interval: 5s
timeout: 1s
retries: 5
langgraph-postgres:
image: pgvector/pgvector:pg16
ports:
- "5433:5432"
environment:
POSTGRES_DB: postgres
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
command:
- postgres
- -c
- shared_preload_libraries=vector
volumes:
- langgraph-data:/var/lib/postgresql/data
healthcheck:
test: pg_isready -U postgres
start_period: 10s
timeout: 1s
retries: 5
interval: 5s
langgraph-debugger:
image: langchain/langgraph-debugger
restart: on-failure
depends_on:
langgraph-postgres:
condition: service_healthy
ports:
- "{debugger_port}:3968"
langgraph-api:
ports:
- "{port}:8000"
depends_on:
langgraph-redis:
condition: service_healthy
langgraph-postgres:
condition: service_healthy
environment:
REDIS_URI: redis://langgraph-redis:6379
POSTGRES_URI: {DEFAULT_POSTGRES_URI}"""
assert clean_empty_lines(actual_compose_str) == expected_compose_str
def test_compose_distributed_mode_with_custom_db(): def test_compose_distributed_mode_with_custom_db():
"""Test compose with engine_runtime_mode='distributed' adds N_JOBS_PER_WORKER=0.""" """Test compose with engine_runtime_mode='distributed' adds N_JOBS_PER_WORKER=0."""
port = 8123 port = 8123
+35 -11
View File
@@ -119,7 +119,6 @@ from langgraph.pregel._io import (
) )
from langgraph.pregel._messages import ensure_message_ids from langgraph.pregel._messages import ensure_message_ids
from langgraph.pregel._read import PregelNode from langgraph.pregel._read import PregelNode
from langgraph.pregel._task_status import read_task_statuses
from langgraph.pregel._utils import get_new_channel_versions, is_xxh3_128_hexdigest from langgraph.pregel._utils import get_new_channel_versions, is_xxh3_128_hexdigest
from langgraph.pregel.debug import ( from langgraph.pregel.debug import (
map_debug_checkpoint, map_debug_checkpoint,
@@ -737,14 +736,17 @@ class PregelLoop:
def _reapply_writes_to_succeeded_nodes( def _reapply_writes_to_succeeded_nodes(
self, tasks: Mapping[str, PregelExecutableTask] self, tasks: Mapping[str, PregelExecutableTask]
) -> None: ) -> None:
"""Restore the output of finished tasks from checkpoint to in-memory tasks. """Restore successful channel writes from checkpoint to in-memory tasks.
Unfinished (failed or interrupted) tasks keep empty writes, so the Skips control signals (ERROR, ERROR_SOURCE_NODE, INTERRUPT, RESUME)
runner re-executes them or routes them to error handlers. so that failed/interrupted tasks remain with empty writes and will be
re-executed (or routed to error handlers) by the runner.
""" """
for tid, status in read_task_statuses(self.checkpoint_pending_writes).items(): for tid, k, v in self.checkpoint_pending_writes:
if k in (ERROR, ERROR_SOURCE_NODE, INTERRUPT, RESUME):
continue
if task := tasks.get(tid): if task := tasks.get(tid):
task.writes.extend(status.output) task.writes.append((k, v))
def _resume_error_handlers_if_applicable(self) -> None: def _resume_error_handlers_if_applicable(self) -> None:
"""On resume, schedule error handlers for tasks that failed in a prior run. """On resume, schedule error handlers for tasks that failed in a prior run.
@@ -814,13 +816,35 @@ class PregelLoop:
self.tasks[handler_task.id] = handler_task self.tasks[handler_task.id] = handler_task
def _pending_interrupts(self) -> set[str]: def _pending_interrupts(self) -> set[str]:
"""Return the ids of interrupts that are still waiting for an answer.""" """Return the set of interrupt ids that are pending without corresponding resume values."""
return { # mapping of task ids to interrupt ids
interrupt.id pending_interrupts: dict[str, str] = {}
for status in read_task_statuses(self.checkpoint_pending_writes).values()
for interrupt in status.pending_interrupts # set of resume task ids
pending_resumes: set[str] = set()
for task_id, write_type, value in self.checkpoint_pending_writes:
if write_type == INTERRUPT:
# interrupts is always a list, but there should only be one element
pending_interrupts[task_id] = value[0].id
elif write_type == RESUME:
pending_resumes.add(task_id)
resumed_interrupt_ids = {
pending_interrupts[task_id]
for task_id in pending_resumes
if task_id in pending_interrupts
} }
# Keep only interrupts whose interrupt_id is not resumed
hanging_interrupts: set[str] = {
interrupt_id
for interrupt_id in pending_interrupts.values()
if interrupt_id not in resumed_interrupt_ids
}
return hanging_interrupts
def _first( def _first(
self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None
) -> set[str] | None: ) -> set[str] | None:
+2 -4
View File
@@ -45,7 +45,6 @@ from langgraph.errors import GraphBubbleUp, GraphInterrupt
from langgraph.pregel._algo import Call from langgraph.pregel._algo import Call
from langgraph.pregel._executor import Submit from langgraph.pregel._executor import Submit
from langgraph.pregel._retry import arun_with_retry, run_with_retry from langgraph.pregel._retry import arun_with_retry, run_with_retry
from langgraph.pregel._task_status import CONTROL_WRITES
from langgraph.types import ( from langgraph.types import (
CachePolicy, CachePolicy,
PregelExecutableTask, PregelExecutableTask,
@@ -607,9 +606,8 @@ class PregelRunner:
task.config is None or TAG_HIDDEN not in task.config.get("tags", []) task.config is None or TAG_HIDDEN not in task.config.get("tags", [])
): ):
self.node_finished(task.name) self.node_finished(task.name)
if all(chan in CONTROL_WRITES for chan, _ in task.writes): if not task.writes:
# record that the task finished, even if it produced no output # add no writes marker
# (see `langgraph.pregel._task_status`)
task.writes.append((NO_WRITES, None)) task.writes.append((NO_WRITES, None))
# save task writes to checkpointer # save task writes to checkpointer
self.put_writes()(task.id, task.writes) # type: ignore[misc] self.put_writes()(task.id, task.writes) # type: ignore[misc]
@@ -1,127 +0,0 @@
"""Read the status of each task from the writes recorded for a superstep.
While a superstep is open, the checkpointer keeps a log of writes for each
task in that step. Entries are added as tasks run and are only discarded when
the whole superstep finishes and a new checkpoint is saved. When a task runs
again, for example after being resumed, its earlier entries stay in the log.
This module is the single place that turns that log into task status. Code that
needs to know whether a task finished, which interrupts it raised, which of them
are still waiting for an answer, or which output it produced must use
`read_task_statuses` instead of inspecting the writes directly.
The log uses two kinds of writes:
- Control writes describe what happened to a task: `INTERRUPT` (the task asked
a question), `RESUME` (answers the task has received), `ERROR`, and
`ERROR_SOURCE_NODE`. `INTERRUPT`, `RESUME` and `ERROR` each have a fixed slot
per task (`WRITES_IDX_MAP`), so a newer write of the same kind can replace an
older one.
- Every other write is output: channel writes, `RETURN` for functional tasks,
and the `NO_WRITES` marker.
The rules are:
1. When a task that ran finishes successfully, `PregelRunner.commit` records at
least one output write, adding `NO_WRITES` if the task produced no other
output.
2. A task that pauses at an interrupt records only control writes.
3. A task is therefore treated as finished if and only if it has an output
write.
4. Because `INTERRUPT` is stored in a fixed slot, its recorded value is the most
recent question the task asked. That question is waiting for an answer only
while the task is unfinished.
A `RESUME` write never means a task is finished: it can hold the answer to an
earlier question while the task waits on a later one.
What these rules cannot see:
- A task whose result came from the cache does not go through
`PregelRunner.commit`, so nothing is recorded for it. It reads as not
finished.
- A task that fails can record partial output writes along with its error. It
reads as finished, which is how the executor has always treated it.
- Writes recorded before rule 1 existed may describe a finished task with no
output using only control writes. Those tasks read as unfinished, which
matches how they were treated before.
"""
from __future__ import annotations
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from typing import Any
from langgraph.checkpoint.base import PendingWrite
from langgraph._internal._constants import (
ERROR,
ERROR_SOURCE_NODE,
INTERRUPT,
NULL_TASK_ID,
RESUME,
)
from langgraph.types import Interrupt
__all__ = ("CONTROL_WRITES", "TaskStatus", "read_task_statuses")
CONTROL_WRITES = frozenset((ERROR, ERROR_SOURCE_NODE, INTERRUPT, RESUME))
"""Channels that describe what happened to a task rather than what it produced."""
@dataclass(frozen=True, slots=True)
class TaskStatus:
"""The status of one task, read from the writes recorded for its superstep."""
output: tuple[tuple[str, Any], ...] = ()
"""Output writes in recorded order. Empty if the task has not finished."""
interrupts: tuple[Interrupt, ...] = ()
"""The most recent interrupts the task raised, whether or not they were answered."""
error: BaseException | None = None
"""The recorded error, if any."""
@property
def finished(self) -> bool:
"""Whether the task ran to completion."""
return bool(self.output)
@property
def pending_interrupts(self) -> tuple[Interrupt, ...]:
"""Interrupts waiting for an answer. Always empty for a finished task."""
return () if self.finished else self.interrupts
def read_task_statuses(
pending_writes: Iterable[PendingWrite],
) -> dict[str, TaskStatus]:
"""Return the status of every task that has recorded writes, keyed by task id.
Writes from `NULL_TASK_ID` are input to the superstep, not task activity, so
they are not included.
"""
output: dict[str, list[tuple[str, Any]]] = {}
interrupts: dict[str, list[Interrupt]] = {}
errors: dict[str, BaseException] = {}
for task_id, channel, value in pending_writes:
if task_id == NULL_TASK_ID:
continue
output.setdefault(task_id, [])
if channel == INTERRUPT:
interrupts.setdefault(task_id, []).extend(
value if isinstance(value, Sequence) else [value]
)
elif channel == ERROR:
errors.setdefault(task_id, value)
elif channel not in CONTROL_WRITES:
output[task_id].append((channel, value))
return {
task_id: TaskStatus(
output=tuple(task_output),
interrupts=tuple(interrupts.get(task_id, ())),
error=errors.get(task_id),
)
for task_id, task_output in output.items()
}
+33 -18
View File
@@ -26,7 +26,6 @@ from langgraph._internal._typing import MISSING
from langgraph.channels.base import BaseChannel from langgraph.channels.base import BaseChannel
from langgraph.constants import TAG_HIDDEN from langgraph.constants import TAG_HIDDEN
from langgraph.pregel._io import read_channels from langgraph.pregel._io import read_channels
from langgraph.pregel._task_status import TaskStatus, read_task_statuses
from langgraph.types import ( from langgraph.types import (
CheckpointPayload, CheckpointPayload,
PregelExecutableTask, PregelExecutableTask,
@@ -38,8 +37,6 @@ from langgraph.types import (
TASK_NAMESPACE = UUID("6ba7b831-9dad-11d1-80b4-00c04fd430c8") TASK_NAMESPACE = UUID("6ba7b831-9dad-11d1-80b4-00c04fd430c8")
_NOT_STARTED = TaskStatus()
def map_debug_tasks(tasks: Iterable[PregelExecutableTask]) -> Iterator[TaskPayload]: def map_debug_tasks(tasks: Iterable[PregelExecutableTask]) -> Iterator[TaskPayload]:
"""Produce "task" events for stream_mode=debug.""" """Produce "task" events for stream_mode=debug."""
@@ -214,21 +211,35 @@ def tasks_w_writes(
pending_writes: list[PendingWrite] | None, pending_writes: list[PendingWrite] | None,
states: dict[str, RunnableConfig | StateSnapshot] | None, states: dict[str, RunnableConfig | StateSnapshot] | None,
output_keys: str | Sequence[str], output_keys: str | Sequence[str],
*,
live: bool = False,
) -> tuple[PregelTask, ...]: ) -> tuple[PregelTask, ...]:
"""Apply writes / subgraph states to tasks to be returned in a StateSnapshot. """Apply writes / subgraph states to tasks to be returned in a StateSnapshot."""
pending_writes = pending_writes or []
With `live=True`, tasks report only the interrupts still waiting for an
answer, as of the most recent writes. Otherwise tasks report the interrupts
they raised in the step, including answered ones, as a record of the step.
"""
statuses = read_task_statuses(pending_writes or [])
out: list[PregelTask] = [] out: list[PregelTask] = []
for task in tasks: for task in tasks:
status = statuses.get(task.id, _NOT_STARTED) rtn = next(
rtn = next((val for chan, val in status.output if chan == RETURN), MISSING) (
task_writes = [(chan, val) for chan, val in status.output if chan != RETURN] val
for tid, chan, val in pending_writes
if tid == task.id and chan == RETURN
),
MISSING,
)
task_error = next(
(exc for tid, n, exc in pending_writes if tid == task.id and n == ERROR),
None,
)
task_interrupts = tuple(
v
for tid, n, vv in pending_writes
if tid == task.id and n == INTERRUPT
for v in (vv if isinstance(vv, Sequence) else [vv])
)
task_writes = [
(chan, val)
for tid, chan, val in pending_writes
if tid == task.id and chan not in (ERROR, INTERRUPT, RETURN)
]
if rtn is not MISSING: if rtn is not MISSING:
task_result = rtn task_result = rtn
@@ -250,15 +261,19 @@ def tasks_w_writes(
mapped_writes = map_task_result_writes(filtered_writes) mapped_writes = map_task_result_writes(filtered_writes)
task_result = mapped_writes if filtered_writes else {} task_result = mapped_writes if filtered_writes else {}
has_writes = rtn is not MISSING or any(
w[0] == task.id and w[1] not in (ERROR, INTERRUPT) for w in pending_writes
)
out.append( out.append(
PregelTask( PregelTask(
task.id, task.id,
task.name, task.name,
task.path, task.path,
status.error, task_error,
status.pending_interrupts if live else status.interrupts, task_interrupts,
states.get(task.id) if states else None, states.get(task.id) if states else None,
task_result if status.finished else None, task_result if has_writes else None,
) )
) )
return tuple(out) return tuple(out)
+33 -43
View File
@@ -79,6 +79,7 @@ from langgraph._internal._constants import (
CONFIG_KEY_STREAM_MESSAGES_V2, CONFIG_KEY_STREAM_MESSAGES_V2,
CONFIG_KEY_TASK_ID, CONFIG_KEY_TASK_ID,
CONFIG_KEY_THREAD_ID, CONFIG_KEY_THREAD_ID,
ERROR,
INPUT, INPUT,
INTERRUPT, INTERRUPT,
NS_END, NS_END,
@@ -148,7 +149,6 @@ from langgraph.pregel._messages import (
from langgraph.pregel._read import DEFAULT_BOUND, PregelNode from langgraph.pregel._read import DEFAULT_BOUND, PregelNode
from langgraph.pregel._retry import RetryPolicy from langgraph.pregel._retry import RetryPolicy
from langgraph.pregel._runner import PregelRunner from langgraph.pregel._runner import PregelRunner
from langgraph.pregel._task_status import read_task_statuses
from langgraph.pregel._tools import StreamToolCallHandler from langgraph.pregel._tools import StreamToolCallHandler
from langgraph.pregel._utils import ( from langgraph.pregel._utils import (
get_new_channel_versions, get_new_channel_versions,
@@ -1147,16 +1147,8 @@ class Pregel(
config: RunnableConfig, config: RunnableConfig,
saved: CheckpointTuple | None, saved: CheckpointTuple | None,
recurse: BaseCheckpointSaver | None = None, recurse: BaseCheckpointSaver | None = None,
live: bool = False, apply_pending_writes: bool = False,
) -> StateSnapshot: ) -> StateSnapshot:
"""Build a `StateSnapshot` from a saved checkpoint and its pending writes.
With `live=True` the snapshot shows current status: values include the
output of tasks that already finished, `next` lists only tasks that still
need to run, and `interrupts` lists only questions still waiting for an
answer. Otherwise the snapshot is a record of the step: values as of the
start of the step, every task in the step, and the interrupts they raised.
"""
if not saved: if not saved:
return StateSnapshot( return StateSnapshot(
values={}, values={},
@@ -1244,10 +1236,13 @@ class Pregel(
None, None,
self.trigger_to_nodes, self.trigger_to_nodes,
) )
if live and saved.pending_writes: if apply_pending_writes and saved.pending_writes:
for tid, status in read_task_statuses(saved.pending_writes).items(): for tid, k, v in saved.pending_writes:
if tid in next_tasks: if k in (ERROR, INTERRUPT):
next_tasks[tid].writes.extend(status.output) continue
if tid not in next_tasks:
continue
next_tasks[tid].writes.append((k, v))
if tasks := [t for t in next_tasks.values() if t.writes]: if tasks := [t for t in next_tasks.values() if t.writes]:
apply_writes( apply_writes(
saved.checkpoint, channels, tasks, None, self.trigger_to_nodes saved.checkpoint, channels, tasks, None, self.trigger_to_nodes
@@ -1257,7 +1252,6 @@ class Pregel(
saved.pending_writes, saved.pending_writes,
task_states, task_states,
self.stream_channels_asis, self.stream_channels_asis,
live=live,
) )
# assemble the state snapshot # assemble the state snapshot
return StateSnapshot( return StateSnapshot(
@@ -1276,16 +1270,8 @@ class Pregel(
config: RunnableConfig, config: RunnableConfig,
saved: CheckpointTuple | None, saved: CheckpointTuple | None,
recurse: BaseCheckpointSaver | None = None, recurse: BaseCheckpointSaver | None = None,
live: bool = False, apply_pending_writes: bool = False,
) -> StateSnapshot: ) -> StateSnapshot:
"""Build a `StateSnapshot` from a saved checkpoint and its pending writes.
With `live=True` the snapshot shows current status: values include the
output of tasks that already finished, `next` lists only tasks that still
need to run, and `interrupts` lists only questions still waiting for an
answer. Otherwise the snapshot is a record of the step: values as of the
start of the step, every task in the step, and the interrupts they raised.
"""
if not saved: if not saved:
return StateSnapshot( return StateSnapshot(
values={}, values={},
@@ -1373,10 +1359,13 @@ class Pregel(
None, None,
self.trigger_to_nodes, self.trigger_to_nodes,
) )
if live and saved.pending_writes: if apply_pending_writes and saved.pending_writes:
for tid, status in read_task_statuses(saved.pending_writes).items(): for tid, k, v in saved.pending_writes:
if tid in next_tasks: if k in (ERROR, INTERRUPT):
next_tasks[tid].writes.extend(status.output) continue
if tid not in next_tasks:
continue
next_tasks[tid].writes.append((k, v))
if tasks := [t for t in next_tasks.values() if t.writes]: if tasks := [t for t in next_tasks.values() if t.writes]:
apply_writes( apply_writes(
saved.checkpoint, channels, tasks, None, self.trigger_to_nodes saved.checkpoint, channels, tasks, None, self.trigger_to_nodes
@@ -1387,7 +1376,6 @@ class Pregel(
saved.pending_writes, saved.pending_writes,
task_states, task_states,
self.stream_channels_asis, self.stream_channels_asis,
live=live,
) )
# assemble the state snapshot # assemble the state snapshot
return StateSnapshot( return StateSnapshot(
@@ -1442,7 +1430,7 @@ class Pregel(
config, config,
saved, saved,
recurse=checkpointer if subgraphs else None, recurse=checkpointer if subgraphs else None,
live=CONFIG_KEY_CHECKPOINT_ID not in config[CONF], apply_pending_writes=CONFIG_KEY_CHECKPOINT_ID not in config[CONF],
) )
async def aget_state( async def aget_state(
@@ -1486,7 +1474,7 @@ class Pregel(
config, config,
saved, saved,
recurse=checkpointer if subgraphs else None, recurse=checkpointer if subgraphs else None,
live=CONFIG_KEY_CHECKPOINT_ID not in config[CONF], apply_pending_writes=CONFIG_KEY_CHECKPOINT_ID not in config[CONF],
) )
def get_state_history( def get_state_history(
@@ -1722,12 +1710,13 @@ class Pregel(
checkpointer.get_next_version, checkpointer.get_next_version,
self.trigger_to_nodes, self.trigger_to_nodes,
) )
# apply writes from tasks that already finished # apply writes from tasks that already ran
for tid, status in read_task_statuses( for tid, k, v in saved.pending_writes or []:
saved.pending_writes or [] if k in (ERROR, INTERRUPT):
).items(): continue
if tid in next_tasks: if tid not in next_tasks:
next_tasks[tid].writes.extend(status.output) continue
next_tasks[tid].writes.append((k, v))
# clear all current tasks # clear all current tasks
apply_writes( apply_writes(
checkpoint, checkpoint,
@@ -2185,12 +2174,13 @@ class Pregel(
checkpointer.get_next_version, checkpointer.get_next_version,
self.trigger_to_nodes, self.trigger_to_nodes,
) )
# apply writes from tasks that already finished # apply writes from tasks that already ran
for tid, status in read_task_statuses( for tid, k, v in saved.pending_writes or []:
saved.pending_writes or [] if k in (ERROR, INTERRUPT):
).items(): continue
if tid in next_tasks: if tid not in next_tasks:
next_tasks[tid].writes.extend(status.output) continue
next_tasks[tid].writes.append((k, v))
# clear all current tasks # clear all current tasks
apply_writes( apply_writes(
checkpoint, checkpoint,
+1 -7
View File
@@ -726,13 +726,7 @@ class StateSnapshot(NamedTuple):
tasks: tuple[PregelTask, ...] tasks: tuple[PregelTask, ...]
"""Tasks to execute in this step. If already attempted, may contain an error.""" """Tasks to execute in this step. If already attempted, may contain an error."""
interrupts: tuple[Interrupt, ...] interrupts: tuple[Interrupt, ...]
"""Interrupts that occurred in this step. """Interrupts that occurred in this step that are pending resolution."""
When reading the latest state (`get_state` without a `checkpoint_id`), this
contains only interrupts still waiting for an answer. When reading a specific
checkpoint or state history, it contains the most recent interrupt each task
raised in that step, including ones answered later in the same step.
"""
class Send: class Send:
@@ -1,534 +0,0 @@
"""State reads while some tasks of a superstep are finished and others are paused.
When parallel tasks each call `interrupt()` and only some of them are resumed,
the superstep stays open. Its recorded writes then contain the old interrupt of
each finished task next to that task's output. These tests check that state
reads, which are rebuilt from the checkpointer, report only the interrupts that
still need an answer.
"""
import operator
import sys
import uuid
from collections import Counter
from typing import Annotated, Any
import pytest
from langgraph.checkpoint.base import BaseCheckpointSaver
from typing_extensions import TypedDict
from langgraph._internal._constants import (
ERROR,
INTERRUPT,
NO_WRITES,
NULL_TASK_ID,
RESUME,
RETURN,
)
from langgraph.func import entrypoint, task
from langgraph.graph import END, START, StateGraph
from langgraph.pregel._task_status import read_task_statuses
from langgraph.types import Command, Durability, Interrupt, Send, interrupt
pytestmark = pytest.mark.anyio
NEEDS_CONTEXTVARS = pytest.mark.skipif(
sys.version_info < (3, 11),
reason="Python 3.11+ is required for async contextvars support",
)
class State(TypedDict, total=False):
log: Annotated[list[str], operator.add]
count: int
def _config() -> dict[str, Any]:
return {"configurable": {"thread_id": str(uuid.uuid4())}}
def _build_parallel(
checkpointer: BaseCheckpointSaver,
calls: Counter[str],
*,
a_questions: int = 1,
a_returns: Any = "log",
):
"""Build a graph where nodes `a` and `b` start in parallel and both ask questions.
`a` asks `a_questions` questions in a row. `a_returns` controls what `a`
returns after its last answer. The default `"log"` returns the answers in
`log`. Any other value is returned as-is.
"""
def a(state: State) -> Any:
calls["a"] += 1
answers = [interrupt(f"A{i + 1}") for i in range(a_questions)]
if a_returns == "log":
return {"log": [f"a:{answer}" for answer in answers]}
return a_returns
def b(state: State) -> State:
calls["b"] += 1
return {"log": [f"b:{interrupt('B')}"]}
builder = StateGraph(State)
builder.add_node("a", a)
builder.add_node("b", b)
builder.add_edge(START, "a")
builder.add_edge(START, "b")
builder.add_edge("a", END)
builder.add_edge("b", END)
return builder.compile(checkpointer=checkpointer)
def _interrupt_by_value(snapshot: Any, value: str) -> Interrupt:
return next(i for i in snapshot.interrupts if i.value == value)
def _task(snapshot: Any, name: str) -> Any:
return next(t for t in snapshot.tasks if t.name == name)
def _interrupt_values(interrupts: Any) -> list[str]:
return sorted(i.value for i in interrupts)
# --- Task A answered and finished, task B still paused ---
def test_finished_task_does_not_report_answered_interrupt(
sync_checkpointer: BaseCheckpointSaver, durability: Durability
) -> None:
calls: Counter[str] = Counter()
graph = _build_parallel(sync_checkpointer, calls)
config = _config()
graph.invoke({"log": []}, config, durability=durability)
snapshot = graph.get_state(config)
assert _interrupt_values(snapshot.interrupts) == ["A1", "B"]
graph.invoke(
Command(resume={_interrupt_by_value(snapshot, "A1").id: "yes"}),
config,
durability=durability,
)
snapshot = graph.get_state(config)
assert _interrupt_values(snapshot.interrupts) == ["B"]
assert snapshot.next == ("b",)
assert _task(snapshot, "a").interrupts == ()
assert _task(snapshot, "a").result == {"log": ["a:yes"]}
assert _interrupt_values(_task(snapshot, "b").interrupts) == ["B"]
assert _task(snapshot, "b").result is None
# Reading the same checkpoint by id gives the record of the step: every task
# in it, and every question asked, including the one A already answered.
record = graph.get_state(snapshot.config)
assert sorted(record.next) == ["a", "b"]
assert _interrupt_values(record.interrupts) == ["A1", "B"]
assert _interrupt_values(_task(record, "a").interrupts) == ["A1"]
assert _task(record, "a").result == {"log": ["a:yes"]}
# B can still be answered, and the graph finishes normally.
result = graph.invoke(
Command(resume={_interrupt_by_value(snapshot, "B").id: "ok"}),
config,
durability=durability,
)
assert sorted(result["log"]) == ["a:yes", "b:ok"]
assert calls == {"a": 2, "b": 3}
snapshot = graph.get_state(config)
assert snapshot.next == ()
assert snapshot.interrupts == ()
# History still shows where each question was asked.
asked = [
_interrupt_values(s.interrupts)
for s in graph.get_state_history(config)
if s.interrupts
]
if durability != "exit":
assert asked == [["A1", "B"]]
@NEEDS_CONTEXTVARS
async def test_finished_task_does_not_report_answered_interrupt_async(
async_checkpointer: BaseCheckpointSaver,
) -> None:
calls: Counter[str] = Counter()
graph = _build_parallel(async_checkpointer, calls)
config = _config()
await graph.ainvoke({"log": []}, config)
snapshot = await graph.aget_state(config)
await graph.ainvoke(
Command(resume={_interrupt_by_value(snapshot, "A1").id: "yes"}), config
)
snapshot = await graph.aget_state(config)
assert _interrupt_values(snapshot.interrupts) == ["B"]
assert snapshot.next == ("b",)
assert _task(snapshot, "a").interrupts == ()
assert _task(snapshot, "a").result == {"log": ["a:yes"]}
assert _interrupt_values(_task(snapshot, "b").interrupts) == ["B"]
record = await graph.aget_state(snapshot.config)
assert _interrupt_values(record.interrupts) == ["A1", "B"]
assert _interrupt_values(_task(record, "a").interrupts) == ["A1"]
result = await graph.ainvoke(
Command(resume={_interrupt_by_value(snapshot, "B").id: "ok"}), config
)
assert sorted(result["log"]) == ["a:yes", "b:ok"]
assert calls == {"a": 2, "b": 3}
# --- Task A answered its first question and asked a second one ---
def test_task_paused_at_second_question_stays_pending(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
calls: Counter[str] = Counter()
graph = _build_parallel(sync_checkpointer, calls, a_questions=2)
config = _config()
graph.invoke({"log": []}, config)
snapshot = graph.get_state(config)
graph.invoke(
Command(resume={_interrupt_by_value(snapshot, "A1").id: "one"}), config
)
snapshot = graph.get_state(config)
assert _interrupt_values(snapshot.interrupts) == ["A2", "B"]
# A is not finished: it has a saved answer, but no output.
assert sorted(snapshot.next) == ["a", "b"]
assert _interrupt_values(_task(snapshot, "a").interrupts) == ["A2"]
assert _task(snapshot, "a").result is None
assert _interrupt_values(_task(snapshot, "b").interrupts) == ["B"]
# Both remaining questions can be answered together.
result = graph.invoke(
Command(
resume={
_interrupt_by_value(snapshot, "A2").id: "two",
_interrupt_by_value(snapshot, "B").id: "ok",
}
),
config,
)
assert sorted(result["log"]) == ["a:one", "a:two", "b:ok"]
snapshot = graph.get_state(config)
assert snapshot.next == ()
assert snapshot.interrupts == ()
@NEEDS_CONTEXTVARS
async def test_task_paused_at_second_question_stays_pending_async(
async_checkpointer: BaseCheckpointSaver,
) -> None:
calls: Counter[str] = Counter()
graph = _build_parallel(async_checkpointer, calls, a_questions=2)
config = _config()
await graph.ainvoke({"log": []}, config)
snapshot = await graph.aget_state(config)
await graph.ainvoke(
Command(resume={_interrupt_by_value(snapshot, "A1").id: "one"}), config
)
snapshot = await graph.aget_state(config)
assert _interrupt_values(snapshot.interrupts) == ["A2", "B"]
assert sorted(snapshot.next) == ["a", "b"]
assert _interrupt_values(_task(snapshot, "a").interrupts) == ["A2"]
assert _task(snapshot, "a").result is None
def test_task_paused_at_second_question_then_other_task_finishes(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
calls: Counter[str] = Counter()
graph = _build_parallel(sync_checkpointer, calls, a_questions=2)
config = _config()
graph.invoke({"log": []}, config)
snapshot = graph.get_state(config)
graph.invoke(
Command(resume={_interrupt_by_value(snapshot, "A1").id: "one"}), config
)
snapshot = graph.get_state(config)
graph.invoke(Command(resume={_interrupt_by_value(snapshot, "B").id: "ok"}), config)
snapshot = graph.get_state(config)
assert _interrupt_values(snapshot.interrupts) == ["A2"]
assert snapshot.next == ("a",)
assert _task(snapshot, "b").interrupts == ()
assert _task(snapshot, "b").result == {"log": ["b:ok"]}
result = graph.invoke(
Command(resume={_interrupt_by_value(snapshot, "A2").id: "two"}), config
)
assert sorted(result["log"]) == ["a:one", "a:two", "b:ok"]
def test_resume_without_id_rejected_when_second_question_and_other_task_pending(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
calls: Counter[str] = Counter()
graph = _build_parallel(sync_checkpointer, calls, a_questions=2)
config = _config()
graph.invoke({"log": []}, config)
snapshot = graph.get_state(config)
graph.invoke(
Command(resume={_interrupt_by_value(snapshot, "A1").id: "one"}), config
)
# A2 and B are both waiting, so a resume value without an id is ambiguous.
with pytest.raises(RuntimeError, match="multiple pending interrupts"):
graph.invoke(Command(resume="ambiguous"), config)
def test_resume_without_id_rejected_when_subgraph_has_parallel_interrupts(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
# A subgraph node whose child graph pauses in two parallel nodes records
# both interrupts under one parent task. Both count as pending, so a resume
# value without an id is ambiguous. (Before, only the first was counted and
# the value went to whichever interrupt consumed it first.)
child_builder = StateGraph(State)
child_builder.add_node("a", lambda s: {"log": [f"a:{interrupt('A')}"]})
child_builder.add_node("b", lambda s: {"log": [f"b:{interrupt('B')}"]})
child_builder.add_edge(START, "a")
child_builder.add_edge(START, "b")
builder = StateGraph(State)
builder.add_node("child", child_builder.compile())
builder.add_edge(START, "child")
graph = builder.compile(checkpointer=sync_checkpointer)
config = _config()
graph.invoke({"log": []}, config)
snapshot = graph.get_state(config)
assert _interrupt_values(snapshot.interrupts) == ["A", "B"]
with pytest.raises(RuntimeError, match="multiple pending interrupts"):
graph.invoke(Command(resume="ambiguous"), config)
result = graph.invoke(
Command(
resume={
_interrupt_by_value(snapshot, "A").id: "x",
_interrupt_by_value(snapshot, "B").id: "y",
}
),
config,
)
assert sorted(result["log"]) == ["a:x", "b:y"]
# --- Task A finished with an empty or falsy result ---
@pytest.mark.parametrize(
"a_returns",
[None, {}, {"count": 0}, {"log": []}],
ids=["none", "empty_dict", "zero", "empty_list"],
)
def test_task_finished_with_falsy_result(
sync_checkpointer: BaseCheckpointSaver, a_returns: Any
) -> None:
calls: Counter[str] = Counter()
graph = _build_parallel(sync_checkpointer, calls, a_returns=a_returns)
config = _config()
graph.invoke({"log": []}, config)
snapshot = graph.get_state(config)
graph.invoke(
Command(resume={_interrupt_by_value(snapshot, "A1").id: "yes"}), config
)
snapshot = graph.get_state(config)
assert _interrupt_values(snapshot.interrupts) == ["B"]
assert snapshot.next == ("b",)
assert _task(snapshot, "a").interrupts == ()
graph.invoke(Command(resume={_interrupt_by_value(snapshot, "B").id: "ok"}), config)
# A already finished, so resuming B must not run A again.
assert calls == {"a": 2, "b": 3}
snapshot = graph.get_state(config)
assert snapshot.next == ()
assert snapshot.interrupts == ()
@pytest.mark.parametrize("a_returns", [None, {"count": 0}], ids=["none", "zero"])
@NEEDS_CONTEXTVARS
async def test_task_finished_with_falsy_result_async(
async_checkpointer: BaseCheckpointSaver, a_returns: Any
) -> None:
calls: Counter[str] = Counter()
graph = _build_parallel(async_checkpointer, calls, a_returns=a_returns)
config = _config()
await graph.ainvoke({"log": []}, config)
snapshot = await graph.aget_state(config)
await graph.ainvoke(
Command(resume={_interrupt_by_value(snapshot, "A1").id: "yes"}), config
)
snapshot = await graph.aget_state(config)
assert _interrupt_values(snapshot.interrupts) == ["B"]
assert snapshot.next == ("b",)
assert _task(snapshot, "a").interrupts == ()
await graph.ainvoke(
Command(resume={_interrupt_by_value(snapshot, "B").id: "ok"}), config
)
assert calls == {"a": 2, "b": 3}
# --- Subgraphs and the functional API ---
def test_parallel_subgraphs_report_only_pending_interrupts(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
class ChildState(TypedDict):
prompt: str
answers: Annotated[list[str], operator.add]
def ask(state: ChildState) -> dict[str, Any]:
return {"answers": [interrupt(state["prompt"])]}
child_builder = StateGraph(ChildState)
child_builder.add_node("ask", ask)
child_builder.add_edge(START, "ask")
child = child_builder.compile()
class ParentState(TypedDict):
answers: Annotated[list[str], operator.add]
builder = StateGraph(ParentState)
builder.add_node("child", child)
builder.add_conditional_edges(
START,
lambda _: [Send("child", {"prompt": p, "answers": []}) for p in ("a", "b")],
["child"],
)
graph = builder.compile(checkpointer=sync_checkpointer)
config = _config()
graph.invoke({"answers": []}, config)
snapshot = graph.get_state(config)
assert _interrupt_values(snapshot.interrupts) == ["a", "b"]
graph.invoke(Command(resume={_interrupt_by_value(snapshot, "a").id: "x"}), config)
snapshot = graph.get_state(config)
assert _interrupt_values(snapshot.interrupts) == ["b"]
assert snapshot.next == ("child",)
finished = next(t for t in snapshot.tasks if t.result is not None)
assert finished.interrupts == ()
assert finished.result == {"answers": ["x"]}
result = graph.invoke(
Command(resume={_interrupt_by_value(snapshot, "b").id: "y"}), config
)
assert sorted(result["answers"]) == ["x", "y"]
def test_functional_task_finished_with_none_is_not_rerun(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
calls: Counter[str] = Counter()
@task
def ask_a() -> None:
calls["a"] += 1
interrupt("A")
@task
def ask_b() -> str:
calls["b"] += 1
return interrupt("B")
@entrypoint(checkpointer=sync_checkpointer)
def workflow(_: Any) -> list[Any]:
a, b = ask_a(), ask_b()
return [a.result(), b.result()]
config = _config()
workflow.invoke(1, config)
snapshot = workflow.get_state(config)
workflow.invoke(
Command(resume={_interrupt_by_value(snapshot, "A").id: "x"}), config
)
snapshot = workflow.get_state(config)
assert _interrupt_values(snapshot.interrupts) == ["B"]
result = workflow.invoke(
Command(resume={_interrupt_by_value(snapshot, "B").id: "y"}), config
)
assert result == [None, "y"]
assert calls == {"a": 2, "b": 3}
# --- Reading task status from recorded writes ---
def test_read_task_statuses() -> None:
a1 = Interrupt(value="A1", id="a")
a2 = Interrupt(value="A2", id="a")
b = Interrupt(value="B", id="b")
error = ValueError("boom")
statuses = read_task_statuses(
[
# answered and finished: old interrupt stays recorded
("finished", INTERRUPT, (a1,)),
("finished", RESUME, ["yes"]),
("finished", "log", ["a:yes"]),
# answered once, then paused at a second question
("paused", INTERRUPT, (a2,)),
("paused", RESUME, ["one"]),
# finished with no output
("no_output", INTERRUPT, (b,)),
("no_output", RESUME, ["ok"]),
("no_output", NO_WRITES, None),
# functional task that returned None
("returned_none", RETURN, None),
# failed
("failed", ERROR, error),
# not a task
(NULL_TASK_ID, RESUME, "global"),
]
)
assert set(statuses) == {
"finished",
"paused",
"no_output",
"returned_none",
"failed",
}
assert statuses["finished"].finished
assert statuses["finished"].interrupts == (a1,)
assert statuses["finished"].pending_interrupts == ()
assert statuses["finished"].output == (("log", ["a:yes"]),)
assert not statuses["paused"].finished
assert statuses["paused"].interrupts == (a2,)
assert statuses["paused"].pending_interrupts == (a2,)
assert statuses["paused"].output == ()
assert statuses["no_output"].finished
assert statuses["no_output"].interrupts == (b,)
assert statuses["no_output"].pending_interrupts == ()
assert statuses["returned_none"].finished
assert statuses["returned_none"].output == ((RETURN, None),)
assert not statuses["failed"].finished
assert statuses["failed"].error is error