fix(cli): defer parent agent env validation

release(cli): 0.4.32
feat(cli): support agent environment defaults
2026-09-28 20:45:05 +02:00 · 2026-09-23 10:56:12 -07:00 · 2026-09-23 10:44:26 -07:00 · 2026-09-23 10:29:09 -07:00 · 2026-09-23 09:58:59 -07:00 · 2026-09-23 09:52:01 -07:00
21 changed files with 292 additions and 2464 deletions
@@ -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.
 ## Development
@@ -6,7 +6,6 @@ import re
 import shlex
 import textwrap
 from collections import Counter
 from collections.abc import Iterable
 from typing import Literal, NamedTuple
 import click
@@ -37,10 +36,6 @@ DISALLOWED_BUILD_COMMAND_CHARS = [
 # This blocks background execution (cmd &) while allowing command
 # chaining (cmd1 && cmd2) which is common in build commands.
 _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(
    r"^(?P<major>\d+)"
    r"(?:\.(?P<minor>\d+))?"
@@ -83,62 +78,6 @@ def has_disallowed_build_command_content(command: str) -> bool:
    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"
 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
-def validate_config(
+def validate_config(config: Config) -> Config:
    config: Config, *, source_path: pathlib.Path | None = None
 ) -> Config:
    """Validate a configuration dictionary."""
    graphs = config.get("graphs", {})
@@ -478,15 +415,6 @@ def validate_config(
                '  "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_kind = _get_source_kind(config)
    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."""
    with open(config_path) as 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
    # incompatible Node.js version
    if validated.get("node_version"):
@@ -1352,7 +1280,6 @@ def python_config_to_docker(
            api_version=api_version,
            build_tools_to_uninstall=build_tools_to_uninstall,
        )
    _validate_local_dependency_files(config_path, config)
    if pip_installer == "auto":
        if _image_supports_uv(base_image):
            pip_installer = "uv"
@@ -1563,18 +1490,7 @@ def node_config_to_docker(
 ) -> tuple[str, dict[str, str]]:
    # Calculate paths for monorepo support
    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)
    if build_context:
@@ -12,6 +12,7 @@ from collections.abc import Callable, Mapping, Sequence
 from contextlib import contextmanager
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from functools import partial
 from typing import Protocol, TypeVar
 import click
@@ -26,7 +27,6 @@ from langgraph_cli.dependency_tracking import find_tracked_packages
 from langgraph_cli.docker import build_docker_image, can_build_locally
 from langgraph_cli.exec import CommandRunner, Runner, subp_exec
 from langgraph_cli.host_backend import (
    MAX_PAGE_SIZE,
    ControlPlaneEndpoints,
    HostBackendClient,
    HostBackendError,
@@ -102,16 +102,16 @@ _NATIVE_AMD64_MACHINE = "x86_64"
 _PUSH_ATTEMPTS = 3
 _LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp"
 _OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {}
 _CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker"
 _LISTENER_REQUIRED_MARKER = "listener_id' is required"
-_LISTENERS_SHOWN = 10
+_HYBRID_LISTENER_GUIDANCE = (
-_LISTENER_NOT_FOUND_STATUSES = frozenset({404, 422})
+    "This workspace deploys through a listener in your own cluster, and the "
-_LISTENERS_DOCS_URL = "https://docs.langchain.com/langsmith/control-plane#listeners"
+    "control plane needs a listener ID to create a deployment. Create the "
-_NO_LISTENERS = (
+    "deployment once in the LangSmith UI, choosing the listener and namespace, "
-    "This workspace has no listeners, so --listener-id and --k8s-namespace "
+    "then re-run with --deployment-id <id>."
    "do not apply."
 )
 _CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker"
 _TERMINAL_STATUSES = frozenset(
    [
@@ -162,134 +162,6 @@ class ByAgent:
 DeploymentSelector = ById | ByName | ByAgent
@dataclass(frozen=True, slots=True)
 class Listener:
    id: str
    compute_id: str
    namespaces: tuple[str, ...]
    @classmethod
    def from_resource(cls, resource: Mapping[str, object]) -> "Listener":
        identifier = str(resource.get("id") or "")
        if not identifier:
            raise HostBackendError(
                "The control plane returned a listener without an id."
            )
        compute_config = resource.get("compute_config")
        namespaces = (
            compute_config.get("k8s_namespaces")
            if isinstance(compute_config, Mapping)
            else None
        )
        return cls(
            identifier,
            str(resource.get("compute_id", "")),
            tuple(str(namespace) for namespace in namespaces)
            if isinstance(namespaces, list)
            else (),
        )
@dataclass(frozen=True, slots=True)
 class Unplaced:
    @property
    def summary(self) -> str:
        return ""
    def source_config(self) -> dict[str, object]:
        return {}
@dataclass(frozen=True, slots=True)
 class OnListener:
    listener_id: str
    k8s_namespace: str
    @property
    def summary(self) -> str:
        return (
            f"Deploying through listener {self.listener_id} "
            f"in namespace {self.k8s_namespace}"
        )
    def source_config(self) -> dict[str, object]:
        return {
            "listener_id": self.listener_id,
            "listener_config": {"k8s_namespace": self.k8s_namespace},
        }
 Placement = Unplaced | OnListener
@dataclass(frozen=True, slots=True)
 class RequestedPlacement:
    listener_id: str | None = None
    k8s_namespace: str | None = None
    @property
    def requested(self) -> bool:
        return self.listener_id is not None or self.k8s_namespace is not None
    def ensure_not_requested(self, deployment_id: str) -> None:
        if self.requested:
            raise click.UsageError(
                "Listener and namespace are fixed when a deployment is created. "
                f"Deployment {deployment_id} already exists, so drop --listener-id "
                "and --k8s-namespace, or create a new deployment with a different "
                "--name."
            )
    def on(self, listener: Listener) -> Placement:
        return OnListener(listener.id, self._namespace(listener))
    def among(self, listeners: Sequence[Listener]) -> Placement:
        if not listeners:
            if self.requested:
                raise click.UsageError(_NO_LISTENERS)
            return Unplaced()
        if len(listeners) > 1:
            raise click.UsageError(
                "This workspace has several listeners. Choose one with "
                f"--listener-id:\n{_describe_listeners(listeners)}"
            )
        return self.on(listeners[0])
    def _namespace(self, listener: Listener) -> str:
        if not listener.namespaces:
            raise click.UsageError(
                f"Listener {listener.id} serves no namespaces. Check its configuration."
            )
        if self.k8s_namespace is None:
            if len(listener.namespaces) == 1:
                return listener.namespaces[0]
            raise click.UsageError(
                f"Listener {listener.id} serves several namespaces. Choose one with "
                f"--k8s-namespace: {', '.join(listener.namespaces)}"
            )
        if self.k8s_namespace not in listener.namespaces:
            raise click.UsageError(
                f"Listener {listener.id} does not serve namespace "
                f"'{self.k8s_namespace}'. Choose one of: "
                f"{', '.join(listener.namespaces)}"
            )
        return self.k8s_namespace
 def _describe_listeners(listeners: Sequence[Listener]) -> str:
    shown = listeners[:_LISTENERS_SHOWN]
    lines = [
        f"  {listener.id}  cluster {listener.compute_id}  "
        f"namespaces: {', '.join(listener.namespaces)}"
        for listener in shown
    ]
    if len(listeners) > len(shown):
        lines.append(f"  ... and {len(listeners) - len(shown)} more")
    if len(listeners) == MAX_PAGE_SIZE:
        lines.append(f"  (only the first {MAX_PAGE_SIZE} listeners were read)")
    return "\n".join(lines)
@dataclass(frozen=True, slots=True)
 class ExistingDeployment:
    id: str
@@ -508,16 +380,15 @@ def _source_of(resource: object) -> str | None:
 def find_deployment_by_name(
    client: HostBackendClient, name: str
 ) -> ExistingDeployment | None:
-    listed = client.list_deployments(name=name, name_contains=name, limit=MAX_PAGE_SIZE)
+    listed = client.list_deployments(name_contains=name)
-    for resource in listed:
+    resources = listed.get("resources", []) if isinstance(listed, dict) else []
-        if resource.get("name") == name and resource.get("id"):
+    for resource in resources:
        if (
            isinstance(resource, dict)
            and resource.get("name") == name
            and resource.get("id")
        ):
            return ExistingDeployment(str(resource["id"]), _source_of(resource))
    if len(listed) >= MAX_PAGE_SIZE:
        raise click.ClickException(
            "This workspace has more deployments than the CLI can search, so it "
            f"cannot tell whether '{name}' already exists. Pass --deployment-id to "
            "update an existing deployment."
        )
    return None
@@ -813,22 +684,14 @@ def _find_deployment(
        existing = _call_host_backend_with_optional_tenant(
            client,
            lambda c: c.list_deployments(
-                agent_id=selector.agent_id,
+                agent_id=selector.agent_id, agent_environment=selector.environment
                agent_environment=selector.environment,
                limit=MAX_PAGE_SIZE,
            ),
        )
        if len(existing) > 1:
            raise click.ClickException(
                "This control plane does not filter deployments by agent, so the "
                f"CLI cannot tell which one belongs to '{selector.agent_id}' in "
                f"{selector.environment}. Deploy by --name instead."
            )
        found = next(
            (
                ExistingDeployment(str(dep["id"]), _source_of(dep))
-                for dep in existing
+                for dep in existing.get("resources", [])
-                if dep.get("id") and not dep.get("is_preview")
+                if not dep.get("is_preview")
            ),
            None,
        )
@@ -896,18 +759,21 @@ def _create_deployment(
 def _get_deployment_status_url(
-    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
+    updated: object, deployment_id: str, host_url: str
 ) -> str | None:
    """Compute the LangSmith dashboard URL for a deployment, if possible."""
    tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None
    if not tenant_id:
        return None
-    return f"{endpoints.dashboard_url}/o/{tenant_id}/host/deployments/{deployment_id}"
+    base = ControlPlaneEndpoints.from_control_plane_url(host_url).dashboard_url
    return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}"
 def _emit_deployment_status_url(
-    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
+    updated: object, deployment_id: str, host_url: str
 ) -> str | None:
-    url = _get_deployment_status_url(updated, deployment_id, endpoints)
+    """Emit the deployment status URL and return it."""
    url = _get_deployment_status_url(updated, deployment_id, host_url)
    if url:
        _get_emitter().status_url(url)
    return url
@@ -925,11 +791,14 @@ def _poll_revision_status(
 ) -> tuple[str, str | None]:
    """Poll latest revision status until terminal status or timeout."""
    em = _get_emitter()
-    revisions = client.list_revisions(deployment_id, limit=1)
+    revisions_resp = client.list_revisions(deployment_id, limit=1)
-    if not revisions:
+    resources = (
        revisions_resp.get("resources", []) if isinstance(revisions_resp, dict) else []
    )
    if not resources:
        return "", None
-    revision_id = str(revisions[0]["id"])
+    revision_id = str(resources[0]["id"])
    last_status = ""
    deadline = time.time() + timeout_seconds
    start_time = time.monotonic()
@@ -1450,7 +1319,6 @@ def _run_remote_build(
@dataclass(frozen=True, slots=True)
 class DeployContext:
    client: HostBackendClient
    endpoints: ControlPlaneEndpoints
    spec: BuildSpec
    verbose: bool
    selector: DeploymentSelector
@@ -1480,66 +1348,19 @@ def _resolve_or_create(
    )
    if found is not None:
        return found.id, step
-    try:
+    created, step = _create_deployment(
-        created, step = _create_deployment(
+        ctx.client,
-            ctx.client,
+        step,
-            step,
+        name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
-            name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
+        agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
-            agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
+        source=source,
-            source=source,
+        source_config={"deployment_type": ctx.deployment_type},
-            source_config={"deployment_type": ctx.deployment_type},
+        source_revision_config={},
-            source_revision_config={},
+        secrets=ctx.secrets,
-            secrets=ctx.secrets,
+    )
        )
    except HostBackendError as err:
        if _needs_a_listener(err):
            raise ListenerRequiredError(
                "The image has to come from a registry you manage, so re-run with "
                "--push-to <registry>/<repository>."
            ) from None
        raise
    return created.id, step
 class ListenerRequiredError(click.UsageError):
    def __init__(self, remedy: str) -> None:
        super().__init__(
            "This workspace deploys through a listener in your own cluster. "
            f"{remedy}\nLearn about listeners: {_LISTENERS_DOCS_URL}"
        )
 def _needs_a_listener(err: HostBackendError) -> bool:
    return err.status_code == 400 and _LISTENER_REQUIRED_MARKER in (
        err.detail or err.message
    )
 def _requested_listener(client: HostBackendClient, listener_id: str) -> Listener:
    try:
        resource = _call_host_backend_with_optional_tenant(
            client, lambda c: c.get_listener(listener_id)
        )
    except HostBackendError as err:
        if err.status_code not in _LISTENER_NOT_FOUND_STATUSES:
            raise
        available = _available_listeners(client)
        if not available:
            raise click.UsageError(_NO_LISTENERS) from None
        raise click.UsageError(
            f"Listener {listener_id} was not found in this workspace. "
            f"Available listeners:\n{_describe_listeners(available)}"
        ) from None
    return Listener.from_resource(resource)
 def _available_listeners(client: HostBackendClient) -> tuple[Listener, ...]:
    resources = _call_host_backend_with_optional_tenant(
        client, lambda c: c.list_listeners()
    )
    return tuple(Listener.from_resource(resource) for resource in resources)
 def _ensure_customer_registry_source(existing: ExistingDeployment) -> None:
    if existing.source != _CUSTOMER_REGISTRY_SOURCE:
        raise click.UsageError(
@@ -1602,7 +1423,6 @@ class RemoteBuildSource:
 class CustomerRegistrySource:
    reference: ImageReference
    prebuilt_image: str | None
    requested_placement: RequestedPlacement
    def run(self, ctx: DeployContext) -> DeployOutcome:
        if isinstance(ctx.selector, ById):
@@ -1624,7 +1444,6 @@ class CustomerRegistrySource:
        self, ctx: DeployContext, existing: ExistingDeployment, step: int
    ) -> DeployOutcome:
        _ensure_customer_registry_source(existing)
        self.requested_placement.ensure_not_requested(existing.id)
        image_uri, step = self._publish(ctx, step)
        _log_deploy_step(step, f"Updating deployment {existing.id}")
        updated = ctx.client.update_deployment(
@@ -1638,25 +1457,7 @@ class CustomerRegistrySource:
            existing.id, _image_revision_result(updated, "Deployment updated")
        )
    def _resolve_placement(self, ctx: DeployContext) -> Placement:
        requested = self.requested_placement
        if requested.listener_id is not None:
            return requested.on(_requested_listener(ctx.client, requested.listener_id))
        if not (ctx.endpoints.is_cloud or requested.requested):
            return Unplaced()
        return requested.among(_available_listeners(ctx.client))
    def _announce(self, placement: Placement) -> None:
        if isinstance(placement, OnListener):
            _get_emitter().info(
                placement.summary,
                listener_id=placement.listener_id,
                k8s_namespace=placement.k8s_namespace,
            )
    def _create(self, ctx: DeployContext, name: str | None, step: int) -> DeployOutcome:
        placement = self._resolve_placement(ctx)
        self._announce(placement)
        image_uri, step = self._publish(ctx, step)
        try:
            created, _ = _create_deployment(
@@ -1667,19 +1468,13 @@ class CustomerRegistrySource:
                if isinstance(ctx.selector, ByAgent)
                else None,
                source=_CUSTOMER_REGISTRY_SOURCE,
-                source_config={
+                source_config={"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC},
                    "resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC,
                    **placement.source_config(),
                },
                source_revision_config={"image_uri": image_uri},
                secrets=ctx.secrets,
            )
        except HostBackendError as err:
-            if _needs_a_listener(err):
+            if err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message:
-                raise ListenerRequiredError(
+                raise click.ClickException(_HYBRID_LISTENER_GUIDANCE) from None
                    "Re-run with --listener-id and --k8s-namespace.\n"
                    f"{err.detail or err.message}"
                ) from None
            raise
        return DeployOutcome(
            created.id, _image_revision_result(created.resource, "Deployment created")
@@ -1740,31 +1535,14 @@ def _select_source(
    image_name: str | None,
    tag: str | None,
    remote_build_flag: bool | None,
    placement: RequestedPlacement,
    selector: DeploymentSelector,
 ) -> DeploymentSource:
    if push_to is None and placement.requested:
        raise click.UsageError(
            "--listener-id and --k8s-namespace only apply when creating a "
            "deployment with --push-to."
        )
    if placement.requested and isinstance(selector, ById):
        raise click.UsageError(
            "Listener and namespace are fixed when a deployment is created, so "
            "they cannot be set for an existing --deployment-id. Drop them, or "
            "create a new deployment with --name."
        )
    if push_to is not None:
        if remote_build_flag is True:
            raise click.UsageError("--push-to cannot be combined with --remote.")
        reference = _push_reference(push_to, tag)
        if image is None:
            _require_local_docker()
-        return CustomerRegistrySource(
+        return CustomerRegistrySource(reference, prebuilt_image=image)
            reference=reference,
            prebuilt_image=image,
            requested_placement=placement,
        )
    if image and remote_build_flag is True:
        raise click.UsageError("--image cannot be combined with --remote builds.")
    use_remote_build, local_build_error = _resolve_build_mode(
@@ -1870,7 +1648,9 @@ def _call_host_backend_with_optional_tenant(
                prompted_for_tenant = True
                continue
            if err.status_code == 403 and "not enabled" in err.message.lower():
-                smith_base = client.endpoints.dashboard_url
+                smith_base = ControlPlaneEndpoints.from_control_plane_url(
                    client.base_url
                ).dashboard_url
                raise HostBackendError(
                    "LangSmith Deployment is not enabled for this organization. "
                    f"Enable it at {smith_base}/host/deployments"
@@ -1917,7 +1697,8 @@ OPT_AGENT_ID = click.option(
    help="Logical agent ID (requires agent mode enabled for the tenant).",
 )
-OPT_AGENT_ENVIRONMENT = click.option(
+OPT_AGENT_ENVIRONMENT = partial(
    click.option,
    "--agent-environment",
    "environment",
    envvar="LANGSMITH_AGENT_ENVIRONMENT",
@@ -2025,7 +1806,9 @@ def _deploy_base_options(
            OPT_HOST_API_KEY,
            OPT_HOST_DEPLOYMENT_NAME,
            OPT_AGENT_ID,
-            OPT_AGENT_ENVIRONMENT,
+            OPT_AGENT_ENVIRONMENT()
            if include_docker_args
            else OPT_AGENT_ENVIRONMENT(type=str),
            click.option(
                "--deployment-id",
                help=(
@@ -2075,21 +1858,6 @@ def _deploy_base_options(
                    "Give the tag here or with --tag (default: latest)."
                ),
            ),
            click.option(
                "--listener-id",
                help=(
                    "Listener that will run the deployment, for workspaces that "
                    "deploy through a listener in your own cluster. Only used when "
                    "creating a deployment with --push-to."
                ),
            ),
            click.option(
                "--k8s-namespace",
                help=(
                    "Kubernetes namespace the listener deploys into. Only used when "
                    "creating a deployment with --push-to."
                ),
            ),
            click.option(
                "--config",
                "-c",
@@ -2172,6 +1940,12 @@ def deploy(ctx: click.Context, **_: object):
    # 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
    environment_param = next(
        param for param in _deploy_cmd.params if param.name == "environment"
    )
    ctx.params["environment"] = environment_param.type_cast_value(
        ctx, ctx.params["environment"]
    )
    if (
        ctx.params.get("agent_id") is not None
        or ctx.params.get("environment") is not None
@@ -2200,8 +1974,6 @@ def _deploy_cmd(
    image_name: str | None,
    image: str | None,
    push_to: str | None,
    listener_id: str | None,
    k8s_namespace: str | None,
    tag: str | None,
    base_image: str | None,
    install_command: str | None,
@@ -2269,15 +2041,12 @@ def _deploy_cmd(
    secrets = _secrets_from_env(_env_without_deployment_name(env_vars))
    selector = ByAgent(**agent) if agent else deployment_selector(deployment_id, name)
    source = _select_source(
        push_to=push_to,
        image=image,
        image_name=image_name,
        tag=tag,
        remote_build_flag=remote_build_flag,
        placement=RequestedPlacement(listener_id, k8s_namespace),
        selector=selector,
    )
    client = _create_host_backend_client(host_url, api_key, env_vars=env_vars)
@@ -2290,7 +2059,6 @@ def _deploy_cmd(
    outcome = source.run(
        DeployContext(
            client=client,
            endpoints=client.endpoints,
            spec=BuildSpec(
                config=config,
                config_json=config_json,
@@ -2302,7 +2070,9 @@ def _deploy_cmd(
                build_command=build_command,
            ),
            verbose=verbose,
-            selector=selector,
+            selector=ByAgent(**agent)
            if agent
            else deployment_selector(deployment_id, name),
            deployment_type=deployment_type,
            secrets=secrets,
            tracked_packages=tracked_packages,
@@ -2311,7 +2081,7 @@ def _deploy_cmd(
    dep_status_url = _emit_deployment_status_url(
        outcome.build_result.updated,
        outcome.deployment_id,
-        client.endpoints,
+        client.base_url,
    )
    if no_wait:
@@ -2371,7 +2141,7 @@ def _deploy_cmd(
@OPT_HOST_API_KEY
@OPT_HOST_URL
@OPT_AGENT_ID
-@OPT_AGENT_ENVIRONMENT
+@OPT_AGENT_ENVIRONMENT()
@click.option(
    "--name-contains",
    default="",
@@ -2398,10 +2168,16 @@ def deploy_list(
    if environment is not None:
        filters["agent_environment"] = environment
    client = _create_host_backend_client(host_url, api_key)
-    deployments = _call_host_backend_with_optional_tenant(
+    response = _call_host_backend_with_optional_tenant(
        client,
        lambda c: c.list_deployments(name_contains=name_contains, **filters),
    )
    resources = response.get("resources") if isinstance(response, dict) else None
    deployments = (
        [item for item in resources if isinstance(item, dict)]
        if isinstance(resources, list)
        else []
    )
    if not deployments:
        click.echo("No deployments found.")
        return
@@ -2441,10 +2217,16 @@ def deploy_revisions_list(
    api_key: str | None, host_url: str | None, limit: int, deployment_id: str
 ) -> None:
    client = _create_host_backend_client(host_url, api_key)
-    revisions = _call_host_backend_with_optional_tenant(
+    response = _call_host_backend_with_optional_tenant(
        client,
        lambda c: c.list_revisions(deployment_id, limit=limit),
    )
    resources = response.get("resources") if isinstance(response, dict) else None
    revisions = (
        [item for item in resources if isinstance(item, dict)]
        if isinstance(resources, list)
        else []
    )
    if not revisions:
        click.echo(f"No revisions found for deployment {deployment_id}.")
        return
@@ -2594,12 +2376,17 @@ def deploy_logs(
        dep_id = found.id
    if log_type == "build" and not revision_id:
-        revisions = client.list_revisions(dep_id, limit=1)
+        revisions_resp = client.list_revisions(dep_id, limit=1)
-        if not revisions:
+        resources = (
            revisions_resp.get("resources", [])
            if isinstance(revisions_resp, dict)
            else []
        )
        if not resources:
            raise click.ClickException(
                "No revisions found for this deployment. Cannot fetch build logs."
            )
-        revision_id = str(revisions[0]["id"])
+        revision_id = str(resources[0]["id"])
        click.secho(f"Using latest revision: {revision_id}", fg="cyan")
    payload: dict = {"limit": limit, "order": "desc"}
@@ -18,7 +18,6 @@ CLOUD_DASHBOARD_HOST = "smith.langchain.com"
 CONTROL_PLANE_PATH = "/api-host"
 LANGSMITH_API_PATHS = ("/api/v1", "/api")
 LOCAL_HOSTNAMES = ("localhost", "127.0.0.1")
 MAX_PAGE_SIZE = 100
 SourceName = Literal["internal_docker", "internal_source", "external_docker"]
@@ -37,13 +36,6 @@ class ControlPlaneEndpoints:
            return cls.from_langsmith_endpoint(langsmith_endpoint)
        return cls(CLOUD_CONTROL_PLANE_URL, CLOUD_DASHBOARD_URL)
    @property
    def is_cloud(self) -> bool:
        hostname = urlparse(self.control_plane_url).hostname or ""
        return hostname == CLOUD_CONTROL_PLANE_HOST or hostname.endswith(
            f".{CLOUD_CONTROL_PLANE_HOST}"
        )
    @classmethod
    def from_control_plane_url(cls, url: str) -> ControlPlaneEndpoints:
        control_plane_url = url.rstrip("/")
@@ -91,36 +83,12 @@ def _without_api_path(path: str) -> str:
    return path
 def _resources(payload: object) -> list[dict[str, Any]]:
    if not isinstance(payload, dict):
        return []
    resources = payload.get("resources")
    if not isinstance(resources, list):
        return []
    return [item for item in resources if isinstance(item, dict)]
 class HostBackendError(click.ClickException):
    """Raised when the host backend returns an error response."""
-    def __init__(
+    def __init__(self, message: str, status_code: int | None = None):
        self,
        message: str,
        status_code: int | None = None,
        detail: str | None = None,
    ):
        super().__init__(message)
        self.status_code = status_code
        self.detail = detail
 def _error_detail(response: httpx.Response) -> str | None:
    try:
        body = response.json()
    except ValueError:
        return None
    detail = body.get("detail") if isinstance(body, dict) else None
    return detail if isinstance(detail, str) else None
 class HostBackendClient:
@@ -142,8 +110,7 @@ class HostBackendClient:
        }
        if tenant_id:
            headers["X-Tenant-ID"] = tenant_id
-        self._endpoints = ControlPlaneEndpoints.from_control_plane_url(base_url)
+        self._base_url = base_url.rstrip("/")
        self._base_url = self._endpoints.control_plane_url
        self._client = httpx.Client(
            base_url=self._base_url,
            headers=headers,
@@ -155,10 +122,6 @@ class HostBackendClient:
    def base_url(self) -> str:
        return self._base_url
    @property
    def endpoints(self) -> ControlPlaneEndpoints:
        return self._endpoints
    def set_tenant(self, tenant_id: str) -> None:
        self._client.headers["X-Tenant-ID"] = tenant_id
@@ -173,12 +136,10 @@ class HostBackendClient:
            resp = self._client.request(method, path, json=payload, params=params)
            resp.raise_for_status()
        except httpx.HTTPStatusError as err:
-            detail = _error_detail(err.response)
+            detail = err.response.text or str(err.response.status_code)
            reason = detail or err.response.text or str(err.response.status_code)
            raise HostBackendError(
-                f"{method} {path} failed with status {err.response.status_code}: {reason}",
+                f"{method} {path} failed with status {err.response.status_code}: {detail}",
                status_code=err.response.status_code,
                detail=detail,
            ) from None
        except httpx.TransportError as err:
            raise HostBackendError(str(err)) from None
@@ -217,29 +178,20 @@ class HostBackendClient:
    def list_deployments(
        self,
        name_contains: str = "",
        *,
        name: str | None = None,
        name_contains: str | None = None,
        limit: int | None = None,
        agent_id: str | None = None,
        agent_environment: str | None = None,
-    ) -> list[dict[str, Any]]:
+    ) -> dict[str, Any]:
-        given = (
+        params = {"name_contains": name_contains}
-            ("name", name),
+        if agent_id is not None:
-            ("name_contains", name_contains),
+            params["agent_id"] = agent_id
-            ("limit", limit),
+        if agent_environment is not None:
-            ("agent_id", agent_id),
+            params["agent_environment"] = agent_environment
-            ("agent_environment", agent_environment),
+        return self._request(
-        )
+            "GET",
-        params = {key: value for key, value in given if value is not None}
+            "/v2/deployments",
-        return _resources(self._request("GET", "/v2/deployments", params=params))
+            params=params,
    def get_listener(self, listener_id: str) -> dict[str, Any]:
        return self._request("GET", f"/v2/listeners/{listener_id}")
    def list_listeners(self) -> list[dict[str, Any]]:
        return _resources(
            self._request("GET", "/v2/listeners", params={"limit": MAX_PAGE_SIZE})
        )
    def get_deployment(self, deployment_id: str) -> dict[str, Any]:
@@ -314,15 +266,10 @@ class HostBackendClient:
            payload["secrets"] = secrets
        return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)
-    def list_revisions(
+    def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]:
-        self, deployment_id: str, limit: int = 1
+        return self._request(
-    ) -> list[dict[str, Any]]:
+            "GET",
-        return _resources(
+            f"/v2/deployments/{deployment_id}/revisions?limit={limit}",
            self._request(
                "GET",
                f"/v2/deployments/{deployment_id}/revisions",
                params={"limit": limit},
            )
        )
    def get_revision(self, deployment_id: str, revision_id: str) -> dict[str, Any]:
@@ -650,8 +650,7 @@ class Config(TypedDict, total=False):
    pip_config_file: str | None
    """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.
    """
@@ -690,9 +689,6 @@ class Config(TypedDict, total=False):
      - "." or "./src" if you have a local Python package
      - str (aka "anthropic") for a PyPI 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.
    This field is not supported when `source.kind` is `uv`.
@@ -880,7 +880,6 @@ def python_config_to_docker_uv_lock(
        _get_node_pm_install_cmd,
        _get_pip_cleanup_lines,
        _image_supports_uv,
        _validate_git_http_url_userinfo_files,
        docker_tag,
    )
@@ -891,20 +890,11 @@ def python_config_to_docker_uv_lock(
        )
    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"
    _, global_reqs_pip_install, pip_config_file_str = _build_python_install_commands(
        config, install_cmd
    )
    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)
    for section, key in [
@@ -28,7 +28,7 @@
                  "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": {
              "anyOf": [
@@ -270,7 +270,7 @@
                  "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": {
              "anyOf": [
@@ -28,7 +28,7 @@
                  "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": {
              "anyOf": [
@@ -270,7 +270,7 @@
                  "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": {
              "anyOf": [
@@ -382,18 +382,20 @@ def test_deploy_list_command(monkeypatch) -> None:
        def list_deployments(self, name_contains: str = ""):
            captured["name_contains"] = name_contains
-            return [
+            return {
-                {
+                "resources": [
-                    "id": "dep-123",
+                    {
-                    "name": "alpha",
+                        "id": "dep-123",
-                    "source_config": {"custom_url": "https://alpha.example.com"},
+                        "name": "alpha",
-                },
+                        "source_config": {"custom_url": "https://alpha.example.com"},
-                {
+                    },
-                    "id": "dep-456",
+                    {
-                    "name": "beta",
+                        "id": "dep-456",
-                    "source_config": {"custom_url": "https://beta.example.com"},
+                        "name": "beta",
-                },
+                        "source_config": {"custom_url": "https://beta.example.com"},
-            ]
+                    },
                ]
            }
    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
@@ -433,7 +435,7 @@ def test_deploy_list_command_no_results(monkeypatch) -> None:
            pass
        def list_deployments(self, name_contains: str = ""):
-            return []
+            return {"resources": []}
    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
@@ -466,18 +468,20 @@ def test_deploy_revisions_list_command(monkeypatch) -> None:
        def list_revisions(self, deployment_id: str, limit: int = 1):
            captured["deployment_id"] = deployment_id
            captured["limit"] = str(limit)
-            return [
+            return {
-                {
+                "resources": [
-                    "id": "rev-123",
+                    {
-                    "status": "CREATING",
+                        "id": "rev-123",
-                    "created_at": "2023-11-07T05:31:56Z",
+                        "status": "CREATING",
-                },
+                        "created_at": "2023-11-07T05:31:56Z",
-                {
+                    },
-                    "id": "rev-456",
+                    {
-                    "status": "DEPLOYED",
+                        "id": "rev-456",
-                    "created_at": "2023-11-08T10:00:00Z",
+                        "status": "DEPLOYED",
-                },
+                        "created_at": "2023-11-08T10:00:00Z",
-            ]
+                    },
                ]
            }
    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
@@ -518,7 +522,7 @@ def test_deploy_revisions_list_command_no_results(monkeypatch) -> None:
            pass
        def list_revisions(self, deployment_id: str, limit: int = 1):
-            return []
+            return {"resources": []}
    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
@@ -551,7 +555,7 @@ def test_deploy_revisions_list_command_with_explicit_limit(monkeypatch) -> None:
        def list_revisions(self, deployment_id: str, limit: int = 1):
            captured["deployment_id"] = deployment_id
            captured["limit"] = str(limit)
-            return []
+            return {"resources": []}
    monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
@@ -1,6 +1,5 @@
 import asyncio
 import json
 import uuid
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass, field
@@ -18,7 +17,6 @@ from langgraph_cli.host_backend import HostBackendClient
 from langgraph_cli.image_reference import ImageReference
 CONTROL_PLANE_URL = "https://control-plane.example.com"
 CLOUD_CONTROL_PLANE_URL = "https://api.host.langchain.com"
 REGISTRY_URL = "https://registry.example.com/team"
 PUSH_TOKEN = "push-token"
 PUSHED_IMAGE = "registry.example.com/team/my-app:latest"
@@ -26,25 +24,10 @@ PUSHED_DIGEST = "registry.example.com/team/my-app@sha256:abc123"
 PUSH_REPOSITORY = "registry.example.com/team/agent"
 EXTERNAL_IMAGE = f"{PUSH_REPOSITORY}:latest"
 EXTERNAL_DIGEST = f"{PUSH_REPOSITORY}@sha256:abc123"
-LISTENER_ID = "11111111-1111-4111-8111-111111111111"
+LISTENER_REQUIRED = (
-OTHER_LISTENER_ID = "22222222-2222-4222-8222-222222222222"
+    "Source configuration error: 'source_config.listener_id' is required for "
-PAGE_TWO_LISTENER_ID = "33333333-3333-4333-8333-333333333333"
+    "workspace with available listener IDs: ['listener-1']"
-UNKNOWN_LISTENER_ID = "99999999-9999-4999-8999-999999999999"
+)
 LISTENER = {
    "id": LISTENER_ID,
    "compute_id": "prod-cluster",
    "compute_config": {"k8s_namespaces": ["agents"]},
 }
 OTHER_LISTENER = {
    "id": OTHER_LISTENER_ID,
    "compute_id": "other-cluster",
    "compute_config": {"k8s_namespaces": ["agents"]},
 }
 TWO_NAMESPACE_LISTENER = {
    "id": LISTENER_ID,
    "compute_id": "prod-cluster",
    "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
 }
 CREATED_ID = "dep-created"
 TRACKED_PACKAGES = ["langgraph:1.0.0"]
 SIGNED_UPLOAD_URL = "https://storage.example.com/signed"
@@ -55,12 +38,7 @@ DIGESTS_FORMAT = "{{json .RepoDigests}}"
 NOT_A_CLI_DEPLOYMENT = (
    "push token is only available for 'internal_docker' source deployments"
 )
 LISTENER_REQUIRED = (
    "Source configuration error: 'source_config.listener_id' is required "
    f"for workspace with available listener IDs: ['{LISTENER_ID}']"
 )
 LIST_DEPLOYMENTS = "GET /v2/deployments"
 LIST_LISTENERS = "GET /v2/listeners"
 CREATE_DEPLOYMENT = "POST /v2/deployments"
@@ -80,22 +58,12 @@ def _get(deployment_id: str) -> str:
    return f"GET /v2/deployments/{deployment_id}"
 def _looks_like_a_uuid(value: str) -> bool:
    try:
        uuid.UUID(value)
    except ValueError:
        return False
    return True
@dataclass
 class ControlPlaneDouble:
    timeline: list[str]
    existing_deployments: list[dict] = field(default_factory=list)
    push_token_status: int = 200
    create_error: str | None = None
    listeners: list[dict] = field(default_factory=list)
    listeners_by_id: dict[str, dict] = field(default_factory=dict)
    bodies: dict[str, dict] = field(default_factory=dict)
    def handle(self, request: httpx.Request) -> httpx.Response:
@@ -103,45 +71,14 @@ class ControlPlaneDouble:
        self.timeline.append(route)
        if request.content:
            self.bodies[route] = json.loads(request.content)
-        return self._respond(request)
+        return self._respond(request.method, request.url.path)
-    def _respond(self, request: httpx.Request) -> httpx.Response:
+    def _respond(self, method: str, path: str) -> httpx.Response:
        method, path = request.method, request.url.path
        if (method, path) == ("GET", "/v2/listeners"):
            return httpx.Response(200, json={"resources": self.listeners})
        if method == "GET" and path.startswith("/v2/listeners/"):
            listener_id = path.rsplit("/", 1)[-1]
            if not _looks_like_a_uuid(listener_id):
                return httpx.Response(
                    422,
                    json={
                        "detail": [
                            {"type": "uuid_parsing", "loc": ["path", "listener_id"]}
                        ]
                    },
                )
            known = {listener["id"]: listener for listener in self.listeners}
            known.update(self.listeners_by_id)
            if listener_id not in known:
                return httpx.Response(
                    404, json={"detail": f"Listener ID {listener_id} not found."}
                )
            return httpx.Response(200, json=known[listener_id])
        if (method, path) == ("GET", "/v2/deployments"):
-            name = request.url.params.get("name")
+            return httpx.Response(200, json={"resources": self.existing_deployments})
            return httpx.Response(
                200,
                json={
                    "resources": [
                        deployment
                        for deployment in self.existing_deployments
                        if name is None or deployment.get("name") == name
                    ]
                },
            )
        if (method, path) == ("POST", "/v2/deployments"):
            if self.create_error is not None:
-                return httpx.Response(400, json={"detail": self.create_error})
+                return httpx.Response(400, text=self.create_error)
            return httpx.Response(201, json={"id": CREATED_ID, "tenant_id": "tenant-1"})
        if path.endswith("/push-token"):
            if self.push_token_status != 200:
@@ -262,7 +199,7 @@ class DeployProject:
    timeline: list[str]
    uploads: list[tuple[str, str, int]]
-    def run(self, *args: str, host_url: str = CONTROL_PLANE_URL) -> Result:
+    def run(self, *args: str) -> Result:
        return CliRunner().invoke(
            cli,
            [
@@ -270,7 +207,7 @@ class DeployProject:
                "--api-key",
                "test-key",
                "--host-url",
-                host_url,
+                CONTROL_PLANE_URL,
                "--name",
                "my-app",
                "--no-input",
@@ -674,6 +611,18 @@ def test_push_to_rejects_a_non_external_deployment_before_any_docker_work(
    assert deploy_project.docker.verbs() == []
 def test_push_to_explains_the_listener_requirement_of_hybrid_workspaces(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.create_error = LISTENER_REQUIRED
    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
    assert result.exit_code != 0
    assert "listener" in result.output
    assert "--deployment-id" in result.output
 def test_push_to_with_deployment_id_fetches_the_deployment_once(
    deploy_project: DeployProject,
 ) -> None:
@@ -703,414 +652,3 @@ def test_invalid_tag_fails_before_any_control_plane_call(
    assert result.exit_code != 0
    assert "Image tag may only contain" in result.output
    assert deploy_project.timeline == []
 def test_push_to_places_a_new_deployment_on_the_only_listener(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER]
    result = deploy_project.run(
        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
    )
    assert result.exit_code == 0, result.output
    assert deploy_project.timeline == [
        LIST_DEPLOYMENTS,
        LIST_LISTENERS,
        "docker build",
        "docker push",
        "docker inspect-digest",
        CREATE_DEPLOYMENT,
    ]
    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
        "resource_spec": {},
        "listener_id": LISTENER_ID,
        "listener_config": {"k8s_namespace": "agents"},
    }
    assert f"Deploying through listener {LISTENER_ID} in namespace agents" in (
        result.output
    )
 def test_push_to_places_a_new_deployment_on_the_chosen_listener(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER, OTHER_LISTENER]
    result = deploy_project.run(
        "--push-to",
        PUSH_REPOSITORY,
        "--listener-id",
        OTHER_LISTENER_ID,
        "--k8s-namespace",
        "agents",
        host_url=CLOUD_CONTROL_PLANE_URL,
    )
    assert result.exit_code == 0, result.output
    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
        "resource_spec": {},
        "listener_id": OTHER_LISTENER_ID,
        "listener_config": {"k8s_namespace": "agents"},
    }
@pytest.mark.parametrize(
    ("listeners", "args", "message"),
    [
        pytest.param(
            [LISTENER, OTHER_LISTENER], (), "--listener-id", id="two_listeners"
        ),
        pytest.param(
            [TWO_NAMESPACE_LISTENER], (), "--k8s-namespace", id="two_namespaces"
        ),
        pytest.param(
            [LISTENER],
            ("--k8s-namespace", "nope"),
            "does not serve namespace",
            id="unknown_namespace",
        ),
    ],
 )
 def test_push_to_refuses_an_unresolved_placement_before_any_docker_work(
    deploy_project: DeployProject, listeners, args, message
 ) -> None:
    deploy_project.control_plane.listeners = listeners
    result = deploy_project.run(
        "--push-to", PUSH_REPOSITORY, *args, host_url=CLOUD_CONTROL_PLANE_URL
    )
    assert result.exit_code != 0
    assert message in result.output
    assert deploy_project.docker.verbs() == []
    assert CREATE_DEPLOYMENT not in deploy_project.timeline
 def test_self_hosted_control_plane_keeps_its_default_placement(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER]
    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
    assert result.exit_code == 0, result.output
    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
        "resource_spec": {}
    }
 def test_self_hosted_control_plane_places_when_asked(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER]
    result = deploy_project.run(
        "--push-to", PUSH_REPOSITORY, "--listener-id", LISTENER_ID
    )
    assert result.exit_code == 0, result.output
    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
        "resource_spec": {},
        "listener_id": LISTENER_ID,
        "listener_config": {"k8s_namespace": "agents"},
    }
 def test_updating_a_deployment_never_looks_up_listeners(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER]
    deploy_project.control_plane.existing_deployments = [
        {"id": "dep-ext", "name": "my-app", "source": "external_docker"}
    ]
    result = deploy_project.run(
        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
    )
    assert result.exit_code == 0, result.output
    assert LIST_LISTENERS not in deploy_project.timeline
 def test_listener_flags_are_refused_for_a_deployment_id_without_any_call(
    deploy_project: DeployProject,
 ) -> None:
    result = deploy_project.run(
        "--push-to",
        PUSH_REPOSITORY,
        "--deployment-id",
        "dep-ext",
        "--k8s-namespace",
        "agents",
        host_url=CLOUD_CONTROL_PLANE_URL,
    )
    assert result.exit_code != 0
    assert "fixed when a deployment is created" in result.output
    assert deploy_project.timeline == []
 def test_listener_flags_are_refused_on_an_existing_deployment(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER]
    deploy_project.control_plane.existing_deployments = [
        {"id": "dep-ext", "name": "my-app", "source": "external_docker"}
    ]
    result = deploy_project.run(
        "--push-to",
        PUSH_REPOSITORY,
        "--listener-id",
        LISTENER_ID,
        host_url=CLOUD_CONTROL_PLANE_URL,
    )
    assert result.exit_code != 0
    assert "fixed when a deployment is created" in result.output
    assert deploy_project.docker.verbs() == []
 def test_a_deployment_without_a_listener_announces_nothing(
    deploy_project: DeployProject,
 ) -> None:
    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
    assert result.exit_code == 0, result.output
    assert "listener" not in result.output
 def test_a_self_hosted_create_without_flags_never_looks_up_listeners(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER]
    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
    assert result.exit_code == 0, result.output
    assert LIST_LISTENERS not in deploy_project.timeline
 def test_a_control_plane_that_demands_a_listener_names_the_flags(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.create_error = LISTENER_REQUIRED
    result = deploy_project.run("--push-to", PUSH_REPOSITORY)
    assert result.exit_code != 0
    assert "--listener-id" in result.output
    assert "--k8s-namespace" in result.output
    assert LISTENER_ID in result.output
    assert "{" not in result.output
    assert "POST /v2/deployments failed" not in result.output
 def test_listener_flags_without_push_to_make_no_call_at_all(
    deploy_project: DeployProject,
 ) -> None:
    result = deploy_project.run("--listener-id", LISTENER_ID)
    assert result.exit_code != 0
    assert "--push-to" in result.output
    assert deploy_project.timeline == []
 def test_a_truncated_listener_page_says_so(deploy_project: DeployProject) -> None:
    deploy_project.control_plane.listeners = [
        {
            "id": str(uuid.UUID(int=index)),
            "compute_id": "cluster",
            "compute_config": {"k8s_namespaces": ["agents"]},
        }
        for index in range(100)
    ]
    result = deploy_project.run(
        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
    )
    assert result.exit_code != 0
    assert "first 100" in result.output
 def test_a_managed_build_in_a_listener_workspace_points_at_push_to(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.create_error = LISTENER_REQUIRED
    result = deploy_project.run("--no-remote")
    assert result.exit_code != 0
    assert "--push-to" in result.output
    assert deploy_project.docker.verbs() == []
@pytest.mark.parametrize(
    "args",
    [
        pytest.param(("--no-remote",), id="managed_build"),
        pytest.param(("--push-to", PUSH_REPOSITORY), id="push_to"),
    ],
 )
 def test_a_listener_requirement_links_the_listener_docs(
    deploy_project: DeployProject, args: tuple[str, ...]
 ) -> None:
    deploy_project.control_plane.create_error = LISTENER_REQUIRED
    result = deploy_project.run(*args)
    assert result.exit_code != 0
    assert "https://docs.langchain.com/langsmith/control-plane#listeners" in (
        result.output
    )
 def test_a_managed_control_plane_without_listeners_creates_as_before(
    deploy_project: DeployProject,
 ) -> None:
    result = deploy_project.run(
        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
    )
    assert result.exit_code == 0, result.output
    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
        "resource_spec": {}
    }
    assert deploy_project.timeline.count(LIST_LISTENERS) == 1
 def test_a_listener_without_an_id_is_reported_rather_than_ignored(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [
        {"compute_id": "broken", "compute_config": {"k8s_namespaces": ["agents"]}},
        LISTENER,
    ]
    result = deploy_project.run(
        "--push-to", PUSH_REPOSITORY, host_url=CLOUD_CONTROL_PLANE_URL
    )
    assert result.exit_code != 0
    assert "without an id" in result.output
    assert deploy_project.docker.verbs() == []
 def _listener_route(listener_id: str) -> str:
    return f"GET /v2/listeners/{listener_id}"
 def test_an_explicit_listener_is_fetched_by_id_not_searched(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER, OTHER_LISTENER]
    result = deploy_project.run(
        "--push-to",
        PUSH_REPOSITORY,
        "--listener-id",
        OTHER_LISTENER_ID,
        host_url=CLOUD_CONTROL_PLANE_URL,
    )
    assert result.exit_code == 0, result.output
    assert _listener_route(OTHER_LISTENER_ID) in deploy_project.timeline
    assert LIST_LISTENERS not in deploy_project.timeline
    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
        "resource_spec": {},
        "listener_id": OTHER_LISTENER_ID,
        "listener_config": {"k8s_namespace": "agents"},
    }
 def test_an_explicit_listener_beyond_the_first_page_still_works(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [
        {
            "id": str(uuid.UUID(int=index)),
            "compute_id": "cluster",
            "compute_config": {"k8s_namespaces": ["agents"]},
        }
        for index in range(100)
    ]
    deploy_project.control_plane.listeners_by_id = {
        PAGE_TWO_LISTENER_ID: {
            "id": PAGE_TWO_LISTENER_ID,
            "compute_id": "far-cluster",
            "compute_config": {"k8s_namespaces": ["agents"]},
        }
    }
    result = deploy_project.run(
        "--push-to",
        PUSH_REPOSITORY,
        "--listener-id",
        PAGE_TWO_LISTENER_ID,
        host_url=CLOUD_CONTROL_PLANE_URL,
    )
    assert result.exit_code == 0, result.output
    assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source_config"] == {
        "resource_spec": {},
        "listener_id": PAGE_TWO_LISTENER_ID,
        "listener_config": {"k8s_namespace": "agents"},
    }
 def test_an_unknown_listener_names_the_ones_that_exist(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER]
    result = deploy_project.run(
        "--push-to",
        PUSH_REPOSITORY,
        "--listener-id",
        UNKNOWN_LISTENER_ID,
        host_url=CLOUD_CONTROL_PLANE_URL,
    )
    assert result.exit_code != 0
    assert "was not found" in result.output
    assert LISTENER_ID in result.output
    assert "prod-cluster" in result.output
    assert deploy_project.docker.verbs() == []
 def test_an_explicit_listener_in_a_workspace_without_any_is_refused(
    deploy_project: DeployProject,
 ) -> None:
    result = deploy_project.run(
        "--push-to",
        PUSH_REPOSITORY,
        "--listener-id",
        LISTENER_ID,
        host_url=CLOUD_CONTROL_PLANE_URL,
    )
    assert result.exit_code != 0
    assert "no listeners" in result.output
    assert deploy_project.docker.verbs() == []
 def test_a_listener_id_that_is_not_an_identifier_still_names_the_real_ones(
    deploy_project: DeployProject,
 ) -> None:
    deploy_project.control_plane.listeners = [LISTENER]
    result = deploy_project.run(
        "--push-to",
        PUSH_REPOSITORY,
        "--listener-id",
        "not-a-listener",
        host_url=CLOUD_CONTROL_PLANE_URL,
    )
    assert result.exit_code != 0
    assert "was not found" in result.output
    assert LISTENER_ID in result.output
    assert "uuid_parsing" not in result.output
@@ -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():
    """Test validation of image_distro field."""
    # Valid image_distro values should work
@@ -72,9 +72,9 @@ def test_agent_create(deployment_api, tmp_path, monkeypatch):
    result = CliRunner().invoke(cli, AGENT_ARGS)
    assert result.exit_code == 0, result.output
    assert dict(requests[0].url.params) == {
        "name_contains": "",
        "agent_id": "customer-support",
        "agent_environment": "staging",
        "limit": "100",
    }
    payload = json.loads(requests[1].content)
    assert payload["agent"] == {
@@ -103,17 +103,3 @@ def test_agent_rejects_explicit_name(deployment_api, monkeypatch):
    assert result.exit_code == 2
    assert "cannot be combined" in result.output
    assert not requests
 def test_agent_lookup_refuses_a_control_plane_that_ignores_the_filter(deployment_api):
    state, requests, _ = deployment_api
    state["resources"] = [
        {"id": "someone-elses", "is_preview": False},
        {"id": "another", "is_preview": False},
    ]
    result = CliRunner().invoke(cli, AGENT_ARGS)
    assert result.exit_code != 0
    assert "does not filter deployments by agent" in result.output
    assert len(requests) == 1
@@ -13,17 +13,10 @@ import pytest
 import langgraph_cli.deploy as deploy_mod
 from langgraph_cli.deploy import (
    ById,
    ByName,
    CustomerRegistrySource,
    DockerBuildCommand,
    ExistingDeployment,
    Listener,
    ManagedRegistrySource,
    OnListener,
    RemoteBuildSource,
    RequestedPlacement,
    Unplaced,
    _call_host_backend_with_optional_tenant,
    _create_host_backend_client,
    _docker_config_for_token,
@@ -34,7 +27,6 @@ from langgraph_cli.deploy import (
    _resolve_pushed_image_digest,
    _select_source,
    _validate_prebuilt_image,
    find_deployment_by_name,
    normalize_image_tag,
    normalize_name,
 )
@@ -288,13 +280,11 @@ class TestCallHostBackendWithOptionalTenant:
        return c
    def test_success_passes_through(self):
-        client = self._make_client(
+        client = self._make_client(lambda req: httpx.Response(200, json={"ok": True}))
            lambda req: httpx.Response(200, json={"resources": [{"id": "dep-1"}]})
        )
        result = _call_host_backend_with_optional_tenant(
            client, lambda c: c.list_deployments()
        )
-        assert result == [{"id": "dep-1"}]
+        assert result == {"ok": True}
    def test_403_not_enabled_gives_actionable_error(self):
        detail = (
@@ -617,8 +607,6 @@ class TestSelectSource:
        "image_name": None,
        "tag": None,
        "remote_build_flag": None,
        "placement": RequestedPlacement(),
        "selector": ByName("my-app"),
    }
    REPOSITORY = "registry.example.com/app"
@@ -629,9 +617,7 @@ class TestSelectSource:
                {"push_to": REPOSITORY},
                True,
                CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "latest"),
+                    ImageReference(REPOSITORY, "latest"), prebuilt_image=None
                    prebuilt_image=None,
                    requested_placement=RequestedPlacement(),
                ),
                id="push_to_selects_the_external_source_with_the_default_tag",
            ),
@@ -639,9 +625,7 @@ class TestSelectSource:
                {"push_to": f"{REPOSITORY}:v2"},
                True,
                CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "v2"),
+                    ImageReference(REPOSITORY, "v2"), prebuilt_image=None
                    prebuilt_image=None,
                    requested_placement=RequestedPlacement(),
                ),
                id="push_to_keeps_a_tag_given_in_the_reference",
            ),
@@ -649,9 +633,7 @@ class TestSelectSource:
                {"push_to": REPOSITORY, "tag": "v3"},
                True,
                CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "v3"),
+                    ImageReference(REPOSITORY, "v3"), prebuilt_image=None
                    prebuilt_image=None,
                    requested_placement=RequestedPlacement(),
                ),
                id="tag_flag_composes_with_push_to",
            ),
@@ -659,25 +641,10 @@ class TestSelectSource:
                {"push_to": REPOSITORY, "image": "app:dev"},
                False,
                CustomerRegistrySource(
-                    reference=ImageReference(REPOSITORY, "latest"),
+                    ImageReference(REPOSITORY, "latest"), prebuilt_image="app:dev"
                    prebuilt_image="app:dev",
                    requested_placement=RequestedPlacement(),
                ),
                id="prebuilt_image_is_retagged_for_push_to_without_docker_checks",
            ),
            pytest.param(
                {
                    "push_to": REPOSITORY,
                    "placement": RequestedPlacement("listener-1", "agents"),
                },
                True,
                CustomerRegistrySource(
                    reference=ImageReference(REPOSITORY, "latest"),
                    prebuilt_image=None,
                    requested_placement=RequestedPlacement("listener-1", "agents"),
                ),
                id="push_to_carries_the_requested_placement",
            ),
            pytest.param(
                {"remote_build_flag": True},
                True,
@@ -753,16 +720,6 @@ class TestSelectSource:
                "--image cannot be combined with --remote builds.",
                id="image_with_remote",
            ),
            pytest.param(
                {"placement": RequestedPlacement(listener_id="listener-1")},
                "only apply when creating a deployment with --push-to",
                id="listener_without_push_to",
            ),
            pytest.param(
                {"placement": RequestedPlacement(k8s_namespace="agents")},
                "only apply when creating a deployment with --push-to",
                id="namespace_without_push_to",
            ),
        ],
    )
    def test_conflicting_flags_are_rejected(self, monkeypatch, flags, message):
@@ -933,289 +890,3 @@ class TestResolvePushedImageDigest:
        frame_locals = captured["coro"].cr_frame.f_locals
        assert "--config" not in frame_locals["args"]
        captured["coro"].close()
 class TestListener:
    @pytest.mark.parametrize(
        ("resource", "expected"),
        [
            pytest.param(
                {
                    "id": "listener-1",
                    "compute_id": "prod-cluster",
                    "compute_config": {"k8s_namespaces": ["agents", "agents-staging"]},
                },
                Listener("listener-1", "prod-cluster", ("agents", "agents-staging")),
                id="reads_id_cluster_and_namespaces",
            ),
            pytest.param(
                {"id": "listener-1", "compute_id": "c", "compute_config": {}},
                Listener("listener-1", "c", ()),
                id="missing_namespaces",
            ),
            pytest.param(
                {"id": "listener-1", "compute_id": "c", "compute_config": None},
                Listener("listener-1", "c", ()),
                id="null_compute_config",
            ),
            pytest.param(
                {"id": "listener-1"},
                Listener("listener-1", "", ()),
                id="only_an_id",
            ),
        ],
    )
    def test_from_resource_reads_the_control_plane_shape(self, resource, expected):
        assert Listener.from_resource(resource) == expected
 ONE_NAMESPACE = Listener("listener-1", "prod-cluster", ("agents",))
 TWO_NAMESPACES = Listener("listener-2", "multi-cluster", ("agents", "agents-staging"))
 NO_NAMESPACE = Listener("listener-3", "broken-cluster", ())
 class TestRequestedPlacement:
    @pytest.mark.parametrize(
        ("request_", "listeners", "expected"),
        [
            pytest.param(
                RequestedPlacement(), (), Unplaced(), id="no_listeners_no_request"
            ),
            pytest.param(
                RequestedPlacement(),
                (ONE_NAMESPACE,),
                OnListener("listener-1", "agents"),
                id="uses_the_only_possible_answer",
            ),
            pytest.param(
                RequestedPlacement(k8s_namespace="agents-staging"),
                (TWO_NAMESPACES,),
                OnListener("listener-2", "agents-staging"),
                id="namespace_alone_picks_the_only_listener",
            ),
        ],
    )
    def test_resolves_to_a_placement(self, request_, listeners, expected):
        assert request_.among(listeners) == expected
    @pytest.mark.parametrize(
        ("request_", "listeners", "message"),
        [
            pytest.param(
                RequestedPlacement(listener_id="listener-1"),
                (),
                "no listeners",
                id="workspace_has_no_listeners",
            ),
            pytest.param(
                RequestedPlacement(),
                (ONE_NAMESPACE, TWO_NAMESPACES),
                "--listener-id",
                id="several_listeners_need_a_choice",
            ),
            pytest.param(
                RequestedPlacement(k8s_namespace="agents"),
                (ONE_NAMESPACE, TWO_NAMESPACES),
                "--listener-id",
                id="namespace_alone_is_ambiguous_with_several_listeners",
            ),
            pytest.param(
                RequestedPlacement(k8s_namespace="agents"),
                (),
                "no listeners",
                id="namespace_without_any_listener",
            ),
            pytest.param(
                RequestedPlacement(),
                (TWO_NAMESPACES,),
                "--k8s-namespace",
                id="several_namespaces_need_a_choice",
            ),
        ],
    )
    def test_refuses_and_names_the_choices(self, request_, listeners, message):
        with pytest.raises(click.UsageError, match=message):
            request_.among(listeners)
    def test_the_error_lists_every_listener_with_its_cluster_and_namespaces(self):
        with pytest.raises(click.UsageError) as error:
            RequestedPlacement().among((ONE_NAMESPACE, TWO_NAMESPACES))
        assert "listener-1" in error.value.message
        assert "prod-cluster" in error.value.message
        assert "agents-staging" in error.value.message
    @pytest.mark.parametrize(
        ("placement", "expected"),
        [
            pytest.param(Unplaced(), {}, id="unplaced_adds_nothing"),
            pytest.param(
                OnListener("listener-1", "agents"),
                {
                    "listener_id": "listener-1",
                    "listener_config": {"k8s_namespace": "agents"},
                },
                id="placed_carries_listener_and_namespace",
            ),
        ],
    )
    def test_source_config_matches_the_control_plane_shape(self, placement, expected):
        assert placement.source_config() == expected
 def test_finding_a_deployment_by_name_narrows_the_search_for_every_server_version():
    seen: dict = {}
    def handler(req: httpx.Request) -> httpx.Response:
        seen["params"] = dict(req.url.params)
        return httpx.Response(
            200,
            json={"resources": [{"id": "dep-1", "name": "agent", "source": "github"}]},
        )
    client = HostBackendClient(
        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
    )
    found = find_deployment_by_name(client, "agent")
    assert seen["params"] == {
        "name": "agent",
        "name_contains": "agent",
        "limit": "100",
    }
    assert found == ExistingDeployment("dep-1", "github")
 def test_a_server_that_ignores_the_exact_name_filter_never_matches_another_deployment():
    client = HostBackendClient(
        "https://api.example.com",
        "key",
        transport=httpx.MockTransport(
            lambda req: httpx.Response(
                200,
                json={
                    "resources": [
                        {
                            "id": "dep-other",
                            "name": "another-teams-agent",
                            "source": "external_docker",
                        }
                    ]
                },
            )
        ),
    )
    assert find_deployment_by_name(client, "brand-new-agent") is None
 def test_a_full_page_without_a_match_refuses_to_claim_the_name_is_free():
    page = [
        {"id": f"dep-{index}", "name": f"other-agent-{index}"} for index in range(100)
    ]
    client = HostBackendClient(
        "https://api.example.com",
        "key",
        transport=httpx.MockTransport(
            lambda req: httpx.Response(200, json={"resources": page})
        ),
    )
    with pytest.raises(click.ClickException, match="--deployment-id"):
        find_deployment_by_name(client, "brand-new-agent")
 def test_a_partial_page_without_a_match_means_the_name_is_free():
    client = HostBackendClient(
        "https://api.example.com",
        "key",
        transport=httpx.MockTransport(
            lambda req: httpx.Response(
                200, json={"resources": [{"id": "dep-1", "name": "other"}]}
            )
        ),
    )
    assert find_deployment_by_name(client, "brand-new-agent") is None
@pytest.mark.parametrize(
    "resource",
    [
        pytest.param({"compute_id": "c"}, id="no_id"),
        pytest.param({"id": ""}, id="empty_id"),
    ],
 )
 def test_a_listener_without_an_id_is_refused(resource):
    with pytest.raises(HostBackendError, match="without an id"):
        Listener.from_resource(resource)
 def test_a_deployment_id_with_listener_flags_is_refused_without_probing_docker(
    monkeypatch,
 ):
    def explode() -> tuple[bool, str | None]:
        raise AssertionError("docker must not be probed for an argv-only conflict")
    monkeypatch.setattr(deploy_mod, "can_build_locally", explode)
    with pytest.raises(click.UsageError, match="--deployment-id"):
        _select_source(
            push_to="registry.example.com/app",
            image=None,
            image_name=None,
            tag=None,
            remote_build_flag=None,
            placement=RequestedPlacement(listener_id="listener-1"),
            selector=ById("dep-1"),
        )
 class TestPlacementOnAKnownListener:
    @pytest.mark.parametrize(
        ("request_", "listener", "expected"),
        [
            pytest.param(
                RequestedPlacement(listener_id="listener-1"),
                ONE_NAMESPACE,
                OnListener("listener-1", "agents"),
                id="the_only_namespace_is_used",
            ),
            pytest.param(
                RequestedPlacement(listener_id="listener-2", k8s_namespace="agents"),
                TWO_NAMESPACES,
                OnListener("listener-2", "agents"),
                id="the_chosen_namespace_is_used",
            ),
        ],
    )
    def test_places_on_the_listener(self, request_, listener, expected):
        assert request_.on(listener) == expected
    @pytest.mark.parametrize(
        ("request_", "listener", "message"),
        [
            pytest.param(
                RequestedPlacement(listener_id="listener-2"),
                TWO_NAMESPACES,
                "--k8s-namespace",
                id="several_namespaces_need_a_choice",
            ),
            pytest.param(
                RequestedPlacement(listener_id="listener-2", k8s_namespace="nope"),
                TWO_NAMESPACES,
                "does not serve namespace",
                id="unknown_namespace",
            ),
            pytest.param(
                RequestedPlacement(listener_id="listener-3"),
                NO_NAMESPACE,
                "serves no namespaces",
                id="listener_without_namespaces",
            ),
        ],
    )
    def test_refuses_and_names_the_namespaces(self, request_, listener, message):
        with pytest.raises(click.UsageError, match=message):
            request_.on(listener)
@@ -79,6 +79,19 @@ def test_request_transport_error_raises():
        c._request("GET", "/test")
 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", transport=httpx.MockTransport(handler)
    )
    result = c.list_deployments("my app")
    assert result == {"ok": True}
 def _capturing_client(captured: dict) -> HostBackendClient:
    def handler(req: httpx.Request) -> httpx.Response:
        captured["body"] = req.read()
@@ -408,7 +421,7 @@ def test_injected_transport_receives_requests_under_the_prefixed_base_url():
        transport=httpx.MockTransport(handler),
    )
-    assert c.list_revisions("dep-1", limit=2) == []
+    assert c.list_revisions("dep-1", limit=2) == {"ok": True}
    assert seen == {
        "url": "https://smith.example.com/api-host/v2/deployments/dep-1/revisions?limit=2",
        "api_key": "key",
@@ -533,144 +546,3 @@ def test_control_plane_endpoints_resolve(host_url, langsmith_endpoint, expected)
    endpoints = ControlPlaneEndpoints.resolve(host_url, langsmith_endpoint)
    assert (endpoints.control_plane_url, endpoints.dashboard_url) == expected
@pytest.mark.parametrize(
    ("payload", "expected"),
    [
        pytest.param(
            {"resources": [{"id": "a"}, {"id": "b"}]},
            [{"id": "a"}, {"id": "b"}],
            id="list_returns_the_resources",
        ),
        pytest.param({"resources": []}, [], id="empty_list"),
        pytest.param({}, [], id="missing_key"),
        pytest.param({"resources": None}, [], id="null_resources"),
        pytest.param(
            {"resources": ["nope", {"id": "a"}]}, [{"id": "a"}], id="skips_non_objects"
        ),
        pytest.param([], [], id="unexpected_envelope"),
    ],
 )
 def test_list_endpoints_return_resource_objects(payload, expected):
    def handler(req: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=payload)
    c = HostBackendClient(
        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
    )
    assert c.list_deployments() == expected
 def test_list_listeners_asks_for_a_full_page():
    seen: dict = {}
    def handler(req: httpx.Request) -> httpx.Response:
        seen["url"] = str(req.url)
        return httpx.Response(200, json={"resources": [{"id": "listener-1"}]})
    c = HostBackendClient(
        "https://api.example.com", "key", transport=httpx.MockTransport(handler)
    )
    assert c.list_listeners() == [{"id": "listener-1"}]
    assert seen["url"] == "https://api.example.com/v2/listeners?limit=100"
@pytest.mark.parametrize(
    ("control_plane_url", "expected"),
    [
        pytest.param("https://api.host.langchain.com", True, id="cloud"),
        pytest.param("https://eu.api.host.langchain.com", True, id="cloud_region"),
        pytest.param("https://dev.api.host.langchain.com", True, id="cloud_dev"),
        pytest.param("https://smith.example.com/api-host", False, id="self_hosted"),
        pytest.param(
            "https://corp.example.com/langsmith/api-host",
            False,
            id="self_hosted_prefix",
        ),
        pytest.param("http://localhost:8080/api-host", False, id="local"),
        pytest.param(
            "https://evil-api.host.langchain.com", False, id="lookalike_needs_a_dot"
        ),
    ],
 )
 def test_is_cloud_recognises_the_managed_control_plane(control_plane_url, expected):
    endpoints = ControlPlaneEndpoints.from_control_plane_url(control_plane_url)
    assert endpoints.is_cloud is expected
@pytest.mark.parametrize(
    ("call", "expected_params"),
    [
        pytest.param(
            lambda c: c.list_deployments(name="agent"),
            {"name": "agent"},
            id="exact_name_filters_server_side",
        ),
        pytest.param(
            lambda c: c.list_deployments(name_contains="age"),
            {"name_contains": "age"},
            id="substring_search_keeps_its_own_parameter",
        ),
        pytest.param(
            lambda c: c.list_deployments(),
            {},
            id="no_filter_sends_no_parameters",
        ),
        pytest.param(
            lambda c: c.list_deployments(
                name="agent", name_contains="agent", limit=100
            ),
            {"name": "agent", "name_contains": "agent", "limit": "100"},
            id="both_filters_travel_together_for_older_servers",
        ),
    ],
 )
 def test_list_deployments_sends_one_name_filter(call, expected_params):
    seen: dict = {}
    def handler(req: httpx.Request) -> httpx.Response:
        seen.update(dict(req.url.params))
        return httpx.Response(200, json={"resources": []})
    call(
        HostBackendClient(
            "https://api.example.com", "key", transport=httpx.MockTransport(handler)
        )
    )
    assert seen == expected_params
@pytest.mark.parametrize(
    ("body", "expected"),
    [
        pytest.param(
            {"detail": "Source configuration error: bad listener"},
            "Source configuration error: bad listener",
            id="fastapi_detail_is_unwrapped",
        ),
        pytest.param(
            {"detail": {"loc": ["body"], "msg": "nope"}},
            None,
            id="a_structured_detail_is_left_alone",
        ),
        pytest.param({"other": "shape"}, None, id="an_unknown_shape_is_left_alone"),
    ],
 )
 def test_error_detail_is_readable(body, expected):
    c = HostBackendClient(
        "https://api.example.com",
        "key",
        transport=httpx.MockTransport(lambda req: httpx.Response(400, json=body)),
    )
    with pytest.raises(HostBackendError) as error:
        c.get_deployment("dep-1")
    assert error.value.detail == expected
    if expected is not None:
        assert error.value.message.endswith(expected)
@@ -119,7 +119,6 @@ from langgraph.pregel._io import (
 )
 from langgraph.pregel._messages import ensure_message_ids
 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.debug import (
    map_debug_checkpoint,
@@ -737,14 +736,17 @@ class PregelLoop:
    def _reapply_writes_to_succeeded_nodes(
        self, tasks: Mapping[str, PregelExecutableTask]
    ) -> 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):
-                task.writes.extend(status.output)
+                task.writes.append((k, v))
    def _resume_error_handlers_if_applicable(self) -> None:
        """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
    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(
        self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None
    ) -> set[str] | None:
@@ -45,7 +45,6 @@ from langgraph.errors import GraphBubbleUp, GraphInterrupt
 from langgraph.pregel._algo import Call
 from langgraph.pregel._executor import Submit
 from langgraph.pregel._retry import arun_with_retry, run_with_retry
 from langgraph.pregel._task_status import CONTROL_WRITES
 from langgraph.types import (
    CachePolicy,
    PregelExecutableTask,
@@ -607,9 +606,8 @@ class PregelRunner:
                task.config is None or TAG_HIDDEN not in task.config.get("tags", [])
            ):
                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))
            # save task writes to checkpointer
            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()
    }
@@ -26,7 +26,6 @@ from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel
 from langgraph.constants import TAG_HIDDEN
 from langgraph.pregel._io import read_channels
 from langgraph.pregel._task_status import TaskStatus, read_task_statuses
 from langgraph.types import (
    CheckpointPayload,
    PregelExecutableTask,
@@ -38,8 +37,6 @@ from langgraph.types import (
 TASK_NAMESPACE = UUID("6ba7b831-9dad-11d1-80b4-00c04fd430c8")
 _NOT_STARTED = TaskStatus()
 def map_debug_tasks(tasks: Iterable[PregelExecutableTask]) -> Iterator[TaskPayload]:
    """Produce "task" events for stream_mode=debug."""
@@ -214,21 +211,35 @@ def tasks_w_writes(
    pending_writes: list[PendingWrite] | None,
    states: dict[str, RunnableConfig | StateSnapshot] | None,
    output_keys: str | Sequence[str],
    *,
    live: bool = False,
 ) -> 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] = []
    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:
            task_result = rtn
@@ -250,15 +261,19 @@ def tasks_w_writes(
            mapped_writes = map_task_result_writes(filtered_writes)
            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(
            PregelTask(
                task.id,
                task.name,
                task.path,
-                status.error,
+                task_error,
-                status.pending_interrupts if live else status.interrupts,
+                task_interrupts,
                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)
@@ -79,6 +79,7 @@ from langgraph._internal._constants import (
    CONFIG_KEY_STREAM_MESSAGES_V2,
    CONFIG_KEY_TASK_ID,
    CONFIG_KEY_THREAD_ID,
    ERROR,
    INPUT,
    INTERRUPT,
    NS_END,
@@ -148,7 +149,6 @@ from langgraph.pregel._messages import (
 from langgraph.pregel._read import DEFAULT_BOUND, PregelNode
 from langgraph.pregel._retry import RetryPolicy
 from langgraph.pregel._runner import PregelRunner
 from langgraph.pregel._task_status import read_task_statuses
 from langgraph.pregel._tools import StreamToolCallHandler
 from langgraph.pregel._utils import (
    get_new_channel_versions,
@@ -1147,16 +1147,8 @@ class Pregel(
        config: RunnableConfig,
        saved: CheckpointTuple | None,
        recurse: BaseCheckpointSaver | None = None,
-        live: bool = False,
+        apply_pending_writes: bool = False,
    ) -> 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:
            return StateSnapshot(
                values={},
@@ -1244,10 +1236,13 @@ class Pregel(
                None,
                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]:
                apply_writes(
                    saved.checkpoint, channels, tasks, None, self.trigger_to_nodes
@@ -1257,7 +1252,6 @@ class Pregel(
            saved.pending_writes,
            task_states,
            self.stream_channels_asis,
            live=live,
        )
        # assemble the state snapshot
        return StateSnapshot(
@@ -1276,16 +1270,8 @@ class Pregel(
        config: RunnableConfig,
        saved: CheckpointTuple | None,
        recurse: BaseCheckpointSaver | None = None,
-        live: bool = False,
+        apply_pending_writes: bool = False,
    ) -> 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:
            return StateSnapshot(
                values={},
@@ -1373,10 +1359,13 @@ class Pregel(
                None,
                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]:
                apply_writes(
                    saved.checkpoint, channels, tasks, None, self.trigger_to_nodes
@@ -1387,7 +1376,6 @@ class Pregel(
            saved.pending_writes,
            task_states,
            self.stream_channels_asis,
            live=live,
        )
        # assemble the state snapshot
        return StateSnapshot(
@@ -1442,7 +1430,7 @@ class Pregel(
            config,
            saved,
            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(
@@ -1486,7 +1474,7 @@ class Pregel(
            config,
            saved,
            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(
@@ -1722,12 +1710,13 @@ class Pregel(
                            checkpointer.get_next_version,
                            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
                    apply_writes(
                        checkpoint,
@@ -2185,12 +2174,13 @@ class Pregel(
                            checkpointer.get_next_version,
                            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
                    apply_writes(
                        checkpoint,
@@ -726,13 +726,7 @@ class StateSnapshot(NamedTuple):
    tasks: tuple[PregelTask, ...]
    """Tasks to execute in this step. If already attempted, may contain an error."""
    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:
@@ -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
Author	SHA1	Message	Date
syachamaneni-lc	b0c97cfcc9	fix(cli): defer parent agent env validation	2026-09-23 10:56:12 -07:00
syachamaneni-lc	03e543cbb6	release(cli): 0.4.32	2026-09-23 10:44:26 -07:00
syachamaneni-lc	d14e3cab57	feat(cli): support agent environment defaults	2026-09-23 10:29:09 -07:00
syachamaneni-lc	a79c8740f9	note	2026-09-23 09:58:59 -07:00
syachamaneni-lc	1dbd1dc4a0	test(cli): remove beta notice tests	2026-09-23 09:52:01 -07:00
syachamaneni-lc	f085820dd3	fix(cli): label agent flags as private beta	2026-09-23 09:49:51 -07:00