chore: merge main into dev package branch

Keep the upstream agent CLI and deployment validation while preserving
the dev release version and regional LangSmith dashboard URLs.

Validation: format, lint, and type checks pass; 487 CLI tests pass.
Two Docker-related failures also reproduce on unchanged origin/main.

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.31"
+__version__ = "0.4.31.dev0"

libs/cli/langgraph_cli/deploy.py

@@ -26,6 +26,7 @@ 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,
@@ -101,15 +102,15 @@ _NATIVE_AMD64_MACHINE = "x86_64"
 _PUSH_ATTEMPTS = 3
 _LOCAL_BUILD_TAG_PREFIX = "langgraph-deploy-tmp"
 _OPERATOR_DEFAULT_RESOURCE_SPEC: Mapping[str, object] = {}
-_LISTENER_REQUIRED_MARKER = "listener_id' is required"
-_HYBRID_LISTENER_GUIDANCE = (
-    "This workspace deploys through a listener in your own cluster, and the "
-    "control plane needs a listener ID to create a deployment. Create the "
-    "deployment once in the LangSmith UI, choosing the listener and namespace, "
-    "then re-run with --deployment-id <id>."
-)
-
 _CUSTOMER_REGISTRY_SOURCE: SourceName = "external_docker"
+_LISTENER_REQUIRED_MARKER = "listener_id' is required"
+_LISTENERS_SHOWN = 10
+_LISTENER_NOT_FOUND_STATUSES = frozenset({404, 422})
+_LISTENERS_DOCS_URL = "https://docs.langchain.com/langsmith/control-plane#listeners"
+_NO_LISTENERS = (
+    "This workspace has no listeners, so --listener-id and --k8s-namespace "
+    "do not apply."
+)
 
 
 _TERMINAL_STATUSES = frozenset(
@@ -161,6 +162,134 @@ 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
@@ -379,15 +508,16 @@ def _source_of(resource: object) -> str | None:
 def find_deployment_by_name(
     client: HostBackendClient, name: str
 ) -> ExistingDeployment | None:
-    listed = client.list_deployments(name_contains=name)
-    resources = listed.get("resources", []) if isinstance(listed, dict) else []
-    for resource in resources:
-        if (
-            isinstance(resource, dict)
-            and resource.get("name") == name
-            and resource.get("id")
-        ):
+    listed = client.list_deployments(name=name, name_contains=name, limit=MAX_PAGE_SIZE)
+    for resource in listed:
+        if 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
 
 
@@ -683,14 +813,22 @@ def _find_deployment(
         existing = _call_host_backend_with_optional_tenant(
             client,
             lambda c: c.list_deployments(
-                agent_id=selector.agent_id, agent_environment=selector.environment
+                agent_id=selector.agent_id,
+                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.get("resources", [])
-                if not dep.get("is_preview")
+                for dep in existing
+                if dep.get("id") and not dep.get("is_preview")
             ),
             None,
         )
@@ -758,21 +896,18 @@ def _create_deployment(
 
 
 def _get_deployment_status_url(
-    updated: object, deployment_id: str, host_url: str
+    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
 ) -> 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
-    base = ControlPlaneEndpoints.from_control_plane_url(host_url).dashboard_url
-    return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}"
+    return f"{endpoints.dashboard_url}/o/{tenant_id}/host/deployments/{deployment_id}"
 
 
 def _emit_deployment_status_url(
-    updated: object, deployment_id: str, host_url: str
+    updated: object, deployment_id: str, endpoints: ControlPlaneEndpoints
 ) -> str | None:
-    """Emit the deployment status URL and return it."""
-    url = _get_deployment_status_url(updated, deployment_id, host_url)
+    url = _get_deployment_status_url(updated, deployment_id, endpoints)
     if url:
         _get_emitter().status_url(url)
     return url
@@ -790,14 +925,11 @@ def _poll_revision_status(
 ) -> tuple[str, str | None]:
     """Poll latest revision status until terminal status or timeout."""
     em = _get_emitter()
-    revisions_resp = client.list_revisions(deployment_id, limit=1)
-    resources = (
-        revisions_resp.get("resources", []) if isinstance(revisions_resp, dict) else []
-    )
-    if not resources:
+    revisions = client.list_revisions(deployment_id, limit=1)
+    if not revisions:
         return "", None
 
-    revision_id = str(resources[0]["id"])
+    revision_id = str(revisions[0]["id"])
     last_status = ""
     deadline = time.time() + timeout_seconds
     start_time = time.monotonic()
@@ -1318,6 +1450,7 @@ def _run_remote_build(
 @dataclass(frozen=True, slots=True)
 class DeployContext:
     client: HostBackendClient
+    endpoints: ControlPlaneEndpoints
     spec: BuildSpec
     verbose: bool
     selector: DeploymentSelector
@@ -1347,19 +1480,66 @@ def _resolve_or_create(
     )
     if found is not None:
         return found.id, step
-    created, step = _create_deployment(
-        ctx.client,
-        step,
-        name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
-        agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
-        source=source,
-        source_config={"deployment_type": ctx.deployment_type},
-        source_revision_config={},
-        secrets=ctx.secrets,
-    )
+    try:
+        created, step = _create_deployment(
+            ctx.client,
+            step,
+            name=ctx.selector.name if isinstance(ctx.selector, ByName) else None,
+            agent=asdict(ctx.selector) if isinstance(ctx.selector, ByAgent) else None,
+            source=source,
+            source_config={"deployment_type": ctx.deployment_type},
+            source_revision_config={},
+            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(
@@ -1422,6 +1602,7 @@ class RemoteBuildSource:
 class CustomerRegistrySource:
     reference: ImageReference
     prebuilt_image: str | None
+    requested_placement: RequestedPlacement
 
     def run(self, ctx: DeployContext) -> DeployOutcome:
         if isinstance(ctx.selector, ById):
@@ -1443,6 +1624,7 @@ 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(
@@ -1456,7 +1638,25 @@ 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(
@@ -1467,13 +1667,19 @@ class CustomerRegistrySource:
                 if isinstance(ctx.selector, ByAgent)
                 else None,
                 source=_CUSTOMER_REGISTRY_SOURCE,
-                source_config={"resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC},
+                source_config={
+                    "resource_spec": _OPERATOR_DEFAULT_RESOURCE_SPEC,
+                    **placement.source_config(),
+                },
                 source_revision_config={"image_uri": image_uri},
                 secrets=ctx.secrets,
             )
         except HostBackendError as err:
-            if err.status_code == 400 and _LISTENER_REQUIRED_MARKER in err.message:
-                raise click.ClickException(_HYBRID_LISTENER_GUIDANCE) from None
+            if _needs_a_listener(err):
+                raise ListenerRequiredError(
+                    "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")
@@ -1534,14 +1740,31 @@ 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(reference, prebuilt_image=image)
+        return CustomerRegistrySource(
+            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(
@@ -1647,9 +1870,7 @@ 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 = ControlPlaneEndpoints.from_control_plane_url(
-                    client.base_url
-                ).dashboard_url
+                smith_base = client.endpoints.dashboard_url
                 raise HostBackendError(
                     "LangSmith Deployment is not enabled for this organization. "
                     f"Enable it at {smith_base}/host/deployments"
@@ -1690,11 +1911,17 @@ OPT_HOST_URL = click.option(
 )
 
 OPT_AGENT_ID = click.option(
-    "--agent-id", help="Logical agent ID (requires agent mode enabled for the tenant)."
+    "--agent-id",
+    envvar="LANGSMITH_AGENT_ID",
+    show_envvar=True,
+    help="Logical agent ID (requires agent mode enabled for the tenant).",
 )
 
 OPT_AGENT_ENVIRONMENT = click.option(
-    "--environment",
+    "--agent-environment",
+    "environment",
+    envvar="LANGSMITH_AGENT_ENVIRONMENT",
+    show_envvar=True,
     type=click.Choice(["development", "staging", "production"]),
     help="Agent environment (requires agent mode enabled for the tenant).",
 )
@@ -1848,6 +2075,21 @@ 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",
@@ -1958,6 +2200,8 @@ 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,
@@ -1982,13 +2226,14 @@ def _deploy_cmd(
     validate_deploy_commands(install_command, build_command)
     agent = None
     if agent_id is not None or environment is not None:
+        em.note("Note: --agent-id and --agent-environment flags are in private beta")
         if not agent_id or not agent_id.strip() or not environment:
             raise click.UsageError(
-                "--agent-id and --environment are required together."
+                "--agent-id and --agent-environment are required together."
             )
         if name is not None or deployment_id is not None:
             raise click.UsageError(
-                "--agent-id and --environment cannot be combined with --name or --deployment-id."
+                "--agent-id and --agent-environment cannot be combined with --name or --deployment-id."
             )
         agent = {"agent_id": agent_id, "environment": environment}
     if not config.exists():
@@ -2024,12 +2269,15 @@ 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)
@@ -2042,6 +2290,7 @@ def _deploy_cmd(
     outcome = source.run(
         DeployContext(
             client=client,
+            endpoints=client.endpoints,
             spec=BuildSpec(
                 config=config,
                 config_json=config_json,
@@ -2053,9 +2302,7 @@ def _deploy_cmd(
                 build_command=build_command,
             ),
             verbose=verbose,
-            selector=ByAgent(**agent)
-            if agent
-            else deployment_selector(deployment_id, name),
+            selector=selector,
             deployment_type=deployment_type,
             secrets=secrets,
             tracked_packages=tracked_packages,
@@ -2064,7 +2311,7 @@ def _deploy_cmd(
     dep_status_url = _emit_deployment_status_url(
         outcome.build_result.updated,
         outcome.deployment_id,
-        client.base_url,
+        client.endpoints,
     )
 
     if no_wait:
@@ -2138,6 +2385,11 @@ def deploy_list(
     agent_id: str | None,
     environment: str | None,
 ) -> None:
+    if agent_id is not None or environment is not None:
+        click.secho(
+            "Note: --agent-id and --agent-environment flags are in private beta",
+            fg="yellow",
+        )
     if agent_id is not None and not agent_id.strip():
         raise click.UsageError("--agent-id must not be empty.")
     filters = {}
@@ -2146,16 +2398,10 @@ def deploy_list(
     if environment is not None:
         filters["agent_environment"] = environment
     client = _create_host_backend_client(host_url, api_key)
-    response = _call_host_backend_with_optional_tenant(
+    deployments = _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
@@ -2195,16 +2441,10 @@ 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)
-    response = _call_host_backend_with_optional_tenant(
+    revisions = _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
@@ -2354,17 +2594,12 @@ def deploy_logs(
         dep_id = found.id
 
     if log_type == "build" and not revision_id:
-        revisions_resp = client.list_revisions(dep_id, limit=1)
-        resources = (
-            revisions_resp.get("resources", [])
-            if isinstance(revisions_resp, dict)
-            else []
-        )
-        if not resources:
+        revisions = client.list_revisions(dep_id, limit=1)
+        if not revisions:
             raise click.ClickException(
                 "No revisions found for this deployment. Cannot fetch build logs."
             )
-        revision_id = str(resources[0]["id"])
+        revision_id = str(revisions[0]["id"])
         click.secho(f"Using latest revision: {revision_id}", fg="cyan")
 
     payload: dict = {"limit": limit, "order": "desc"}

libs/cli/langgraph_cli/host_backend.py

@@ -18,6 +18,7 @@ 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"]
 
 
@@ -36,6 +37,13 @@ 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("/")
@@ -70,9 +78,10 @@ def _cloud_control_plane_host_for(langsmith_api_host: str) -> str:
 
 
 def _cloud_dashboard_for(control_plane_host: str) -> str:
-    if control_plane_host.endswith(f".{CLOUD_CONTROL_PLANE_HOST}"):
-        region = control_plane_host[: -len(CLOUD_CONTROL_PLANE_HOST) - 1]
-        return f"https://{region}.{CLOUD_DASHBOARD_HOST}"
+    for api_host in (CLOUD_CONTROL_PLANE_HOST, CLOUD_API_HOST):
+        if control_plane_host.endswith(f".{api_host}"):
+            region = control_plane_host[: -len(api_host) - 1]
+            return f"https://{region}.{CLOUD_DASHBOARD_HOST}"
     return CLOUD_DASHBOARD_URL
 
 
@@ -83,12 +92,36 @@ 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__(self, message: str, status_code: int | None = None):
+    def __init__(
+        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:
@@ -110,7 +143,8 @@ class HostBackendClient:
         }
         if tenant_id:
             headers["X-Tenant-ID"] = tenant_id
-        self._base_url = base_url.rstrip("/")
+        self._endpoints = ControlPlaneEndpoints.from_control_plane_url(base_url)
+        self._base_url = self._endpoints.control_plane_url
         self._client = httpx.Client(
             base_url=self._base_url,
             headers=headers,
@@ -122,6 +156,10 @@ 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
 
@@ -136,10 +174,12 @@ class HostBackendClient:
             resp = self._client.request(method, path, json=payload, params=params)
             resp.raise_for_status()
         except httpx.HTTPStatusError as err:
-            detail = err.response.text or str(err.response.status_code)
+            detail = _error_detail(err.response)
+            reason = detail or err.response.text or str(err.response.status_code)
             raise HostBackendError(
-                f"{method} {path} failed with status {err.response.status_code}: {detail}",
+                f"{method} {path} failed with status {err.response.status_code}: {reason}",
                 status_code=err.response.status_code,
+                detail=detail,
             ) from None
         except httpx.TransportError as err:
             raise HostBackendError(str(err)) from None
@@ -178,20 +218,29 @@ 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,
-    ) -> dict[str, Any]:
-        params = {"name_contains": name_contains}
-        if agent_id is not None:
-            params["agent_id"] = agent_id
-        if agent_environment is not None:
-            params["agent_environment"] = agent_environment
-        return self._request(
-            "GET",
-            "/v2/deployments",
-            params=params,
+    ) -> list[dict[str, Any]]:
+        given = (
+            ("name", name),
+            ("name_contains", name_contains),
+            ("limit", limit),
+            ("agent_id", agent_id),
+            ("agent_environment", agent_environment),
+        )
+        params = {key: value for key, value in given if value is not None}
+        return _resources(self._request("GET", "/v2/deployments", 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]:
@@ -266,10 +315,15 @@ class HostBackendClient:
             payload["secrets"] = secrets
         return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)
 
-    def list_revisions(self, deployment_id: str, limit: int = 1) -> dict[str, Any]:
-        return self._request(
-            "GET",
-            f"/v2/deployments/{deployment_id}/revisions?limit={limit}",
+    def list_revisions(
+        self, deployment_id: str, limit: int = 1
+    ) -> list[dict[str, Any]]:
+        return _resources(
+            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]:

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

@@ -382,20 +382,18 @@ def test_deploy_list_command(monkeypatch) -> None:
 
         def list_deployments(self, name_contains: str = ""):
             captured["name_contains"] = name_contains
-            return {
-                "resources": [
-                    {
-                        "id": "dep-123",
-                        "name": "alpha",
-                        "source_config": {"custom_url": "https://alpha.example.com"},
-                    },
-                    {
-                        "id": "dep-456",
-                        "name": "beta",
-                        "source_config": {"custom_url": "https://beta.example.com"},
-                    },
-                ]
-            }
+            return [
+                {
+                    "id": "dep-123",
+                    "name": "alpha",
+                    "source_config": {"custom_url": "https://alpha.example.com"},
+                },
+                {
+                    "id": "dep-456",
+                    "name": "beta",
+                    "source_config": {"custom_url": "https://beta.example.com"},
+                },
+            ]
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -435,7 +433,7 @@ def test_deploy_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_deployments(self, name_contains: str = ""):
-            return {"resources": []}
+            return []
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -468,20 +466,18 @@ 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 {
-                "resources": [
-                    {
-                        "id": "rev-123",
-                        "status": "CREATING",
-                        "created_at": "2023-11-07T05:31:56Z",
-                    },
-                    {
-                        "id": "rev-456",
-                        "status": "DEPLOYED",
-                        "created_at": "2023-11-08T10:00:00Z",
-                    },
-                ]
-            }
+            return [
+                {
+                    "id": "rev-123",
+                    "status": "CREATING",
+                    "created_at": "2023-11-07T05:31:56Z",
+                },
+                {
+                    "id": "rev-456",
+                    "status": "DEPLOYED",
+                    "created_at": "2023-11-08T10:00:00Z",
+                },
+            ]
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -522,7 +518,7 @@ def test_deploy_revisions_list_command_no_results(monkeypatch) -> None:
             pass
 
         def list_revisions(self, deployment_id: str, limit: int = 1):
-            return {"resources": []}
+            return []
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 
@@ -555,7 +551,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 {"resources": []}
+            return []
 
     monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient)
 

libs/cli/tests/unit_tests/cli/test_deploy_command.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import uuid
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass, field
@@ -17,6 +18,7 @@ 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"
@@ -24,10 +26,25 @@ 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_REQUIRED = (
-    "Source configuration error: 'source_config.listener_id' is required for "
-    "workspace with available listener IDs: ['listener-1']"
-)
+LISTENER_ID = "11111111-1111-4111-8111-111111111111"
+OTHER_LISTENER_ID = "22222222-2222-4222-8222-222222222222"
+PAGE_TWO_LISTENER_ID = "33333333-3333-4333-8333-333333333333"
+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"
@@ -38,7 +55,12 @@ 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"
 
 
@@ -58,12 +80,22 @@ 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:
@@ -71,14 +103,45 @@ class ControlPlaneDouble:
         self.timeline.append(route)
         if request.content:
             self.bodies[route] = json.loads(request.content)
-        return self._respond(request.method, request.url.path)
+        return self._respond(request)
 
-    def _respond(self, method: str, path: str) -> httpx.Response:
+    def _respond(self, request: httpx.Request) -> 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"):
-            return httpx.Response(200, json={"resources": self.existing_deployments})
+            name = request.url.params.get("name")
+            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, text=self.create_error)
+                return httpx.Response(400, json={"detail": 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:
@@ -199,7 +262,7 @@ class DeployProject:
     timeline: list[str]
     uploads: list[tuple[str, str, int]]
 
-    def run(self, *args: str) -> Result:
+    def run(self, *args: str, host_url: str = CONTROL_PLANE_URL) -> Result:
         return CliRunner().invoke(
             cli,
             [
@@ -207,7 +270,7 @@ class DeployProject:
                 "--api-key",
                 "test-key",
                 "--host-url",
-                CONTROL_PLANE_URL,
+                host_url,
                 "--name",
                 "my-app",
                 "--no-input",
@@ -611,18 +674,6 @@ 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:
@@ -652,3 +703,414 @@ 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

libs/cli/tests/unit_tests/test_deploy_agents.py

@@ -58,7 +58,7 @@ AGENT_ARGS = [
     "deploy",
     "--agent-id",
     "customer-support",
-    "--environment",
+    "--agent-environment",
     "staging",
     "--remote",
     "--no-wait",
@@ -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,3 +103,17 @@ 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

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -13,10 +13,17 @@ 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,
@@ -27,6 +34,7 @@ from langgraph_cli.deploy import (
     _resolve_pushed_image_digest,
     _select_source,
     _validate_prebuilt_image,
+    find_deployment_by_name,
     normalize_image_tag,
     normalize_name,
 )
@@ -280,11 +288,13 @@ class TestCallHostBackendWithOptionalTenant:
         return c
 
     def test_success_passes_through(self):
-        client = self._make_client(lambda req: httpx.Response(200, json={"ok": True}))
+        client = self._make_client(
+            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 == {"ok": True}
+        assert result == [{"id": "dep-1"}]
 
     def test_403_not_enabled_gives_actionable_error(self):
         detail = (
@@ -607,6 +617,8 @@ class TestSelectSource:
         "image_name": None,
         "tag": None,
         "remote_build_flag": None,
+        "placement": RequestedPlacement(),
+        "selector": ByName("my-app"),
     }
     REPOSITORY = "registry.example.com/app"
 
@@ -617,7 +629,9 @@ class TestSelectSource:
                 {"push_to": REPOSITORY},
                 True,
                 CustomerRegistrySource(
-                    ImageReference(REPOSITORY, "latest"), prebuilt_image=None
+                    reference=ImageReference(REPOSITORY, "latest"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement(),
                 ),
                 id="push_to_selects_the_external_source_with_the_default_tag",
             ),
@@ -625,7 +639,9 @@ class TestSelectSource:
                 {"push_to": f"{REPOSITORY}:v2"},
                 True,
                 CustomerRegistrySource(
-                    ImageReference(REPOSITORY, "v2"), prebuilt_image=None
+                    reference=ImageReference(REPOSITORY, "v2"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement(),
                 ),
                 id="push_to_keeps_a_tag_given_in_the_reference",
             ),
@@ -633,7 +649,9 @@ class TestSelectSource:
                 {"push_to": REPOSITORY, "tag": "v3"},
                 True,
                 CustomerRegistrySource(
-                    ImageReference(REPOSITORY, "v3"), prebuilt_image=None
+                    reference=ImageReference(REPOSITORY, "v3"),
+                    prebuilt_image=None,
+                    requested_placement=RequestedPlacement(),
                 ),
                 id="tag_flag_composes_with_push_to",
             ),
@@ -641,10 +659,25 @@ class TestSelectSource:
                 {"push_to": REPOSITORY, "image": "app:dev"},
                 False,
                 CustomerRegistrySource(
-                    ImageReference(REPOSITORY, "latest"), prebuilt_image="app:dev"
+                    reference=ImageReference(REPOSITORY, "latest"),
+                    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,
@@ -720,6 +753,16 @@ 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):
@@ -890,3 +933,289 @@ 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)

libs/cli/tests/unit_tests/test_host_backend.py

@@ -79,19 +79,6 @@ 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()
@@ -421,7 +408,7 @@ def test_injected_transport_receives_requests_under_the_prefixed_base_url():
         transport=httpx.MockTransport(handler),
     )
 
-    assert c.list_revisions("dep-1", limit=2) == {"ok": True}
+    assert c.list_revisions("dep-1", limit=2) == []
     assert seen == {
         "url": "https://smith.example.com/api-host/v2/deployments/dep-1/revisions?limit=2",
         "api_key": "key",
@@ -540,9 +527,156 @@ CLOUD = ("https://api.host.langchain.com", "https://smith.langchain.com")
             ("https://eu.api.host.langchain.com", "https://eu.smith.langchain.com"),
             id="regional_control_plane_maps_to_regional_dashboard",
         ),
+        pytest.param(
+            "https://dev.api.smith.langchain.com/",
+            None,
+            ("https://dev.api.smith.langchain.com", "https://dev.smith.langchain.com"),
+            id="dev_smith_api_maps_to_dev_dashboard",
+        ),
     ],
 )
 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)

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -80,14 +80,12 @@ def get_updated_channels_from_tasks(
 
 def get_delta_channels_from_all_channels(
     channels: Mapping[str, BaseChannel],
-    *,
-    include_unavailable: bool = False,
 ) -> set[str]:
-    """DeltaChannels to snapshot on the first update_state of a fresh thread or fork."""
+    """DeltaChannels to snapshot on the first update_state of a fresh thread."""
     return {
         k
         for k, ch in channels.items()
-        if isinstance(ch, DeltaChannel) and (include_unavailable or ch.is_available())
+        if isinstance(ch, DeltaChannel) and ch.is_available()
     }
 
 
@@ -124,22 +122,15 @@ def create_checkpoint_plan_for_update_state_api(
     parents: dict[str, Any],
     saved_metadata: Mapping[str, Any] | None,
     is_fresh_thread: bool,
-    is_fork: bool,
 ) -> tuple[set[str], dict[str, Any]]:
-    """Return ``(channels_to_snapshot, metadata)`` for an update_state head.
-
-    A fork snapshots everything, like a fresh thread: its base also holds the
-    writes of the branch it abandons, so the ancestor walk must stop here.
-    """
+    """Return ``(channels_to_snapshot, metadata)`` for an update_state head."""
     metadata: dict[str, Any] = {
         "source": "update",
         "step": step,
         "parents": parents,
     }
-    if is_fresh_thread or is_fork:
-        return get_delta_channels_from_all_channels(
-            channels, include_unavailable=is_fork
-        ), metadata
+    if is_fresh_thread:
+        return get_delta_channels_from_all_channels(channels), metadata
 
     new_counters = create_metadata_for_update_state_api(
         channels,
@@ -155,34 +146,6 @@ def create_checkpoint_plan_for_update_state_api(
     return channels_to_snapshot, metadata
 
 
-def create_fork_checkpoint(
-    checkpoint: Checkpoint,
-    channels: Mapping[str, BaseChannel],
-    step: int,
-    *,
-    is_fork: bool,
-    get_next_version: GetNextVersion,
-) -> Checkpoint:
-    """``create_checkpoint`` for the update_state paths that skip the plan.
-
-    The fork has to be sealed by its first checkpoint: any later superstep
-    has already rebuilt its delta channels through the shared base. These
-    paths never write the delta channel, so its version must be bumped here
-    or ``put`` drops the blob; derive ``new_versions`` from the result.
-    """
-    if not is_fork:
-        return create_checkpoint(checkpoint, channels, step)
-    return create_checkpoint(
-        checkpoint,
-        channels,
-        step,
-        get_next_version=get_next_version,
-        channels_to_snapshot=get_delta_channels_from_all_channels(
-            channels, include_unavailable=True
-        ),
-    )
-
-
 def create_checkpoint(
     checkpoint: Checkpoint,
     channels: Mapping[str, BaseChannel] | None,
@@ -211,23 +174,14 @@ def create_checkpoint(
         values = {}
         channel_versions = dict(checkpoint["channel_versions"])
         for k in channels:
-            ch = channels[k]
             if k not in channel_versions:
-                # A forced snapshot of a never-written channel still has to
-                # land to stop the ancestor walk, and `put` only stores blobs
-                # for versioned channels.
-                if k in channels_to_snapshot and get_next_version is not None:
-                    channel_versions[k] = get_next_version(None, None)
-                    values[k] = _DeltaSnapshot(
-                        ch.get() if ch.is_available() else ch.typ()
-                    )
                 continue
+            ch = channels[k]
             if k in channels_to_snapshot:
                 # Callers force a full snapshot blob here: exit mode when a
-                # delta channel reaches its snapshot cadence, update_state on
-                # a fresh thread (no ancestor to replay writes from), and a
-                # fork. The manual version-bump below only applies to the
-                # exit-mode case.
+                # delta channel reaches its snapshot cadence, and update_state
+                # on a fresh thread (no ancestor to replay writes from). The
+                # manual version-bump below only applies to the exit-mode case.
                 #
                 # In exit mode, the snapshot decision is deferred to exit
                 # time (intermediate steps have do_checkpoint=False). The

libs/langgraph/langgraph/pregel/_loop.py

@@ -222,16 +222,10 @@ class PregelLoop:
     # under the saver's `ORDER BY task_id, idx` sorting.
     _exit_delta_writes: list[tuple[int, str, str, Any]] | None = None
 
-    # Delta channels that must snapshot at the next checkpoint, whatever their
-    # cadence counters say:
-    # * an Overwrite arrived since the last checkpoint, so sparse replay has to
-    #   start from the post-overwrite value;
-    # * this run forked off an explicitly addressed checkpoint. That base also
-    #   holds the writes of the branch the fork abandons, and nothing records
-    #   which child consumed which, so the ancestor walk must stop inside the
-    #   fork. Any addressed checkpoint counts, because telling a real fork
-    #   apart would mean trusting the base's `pending_writes` to be complete.
-    _delta_channels_forced_snapshot: set[str]
+    # Delta channels that saw an Overwrite since the last checkpoint. These
+    # channels must snapshot after live update applies overwrite semantics so
+    # sparse replay starts from the same post-overwrite value.
+    _delta_channels_with_overwrite: set[str]
 
     # The checkpoint_config that points at the parent loaded at `__enter__`
     # (or the synthetic-empty checkpoint, on first run). We capture it
@@ -375,13 +369,6 @@ class PregelLoop:
             if self.config[CONF].get(CONFIG_KEY_CHECKPOINT_NS)
             else ()
         )
-        # Value, not key presence like `is_replaying`: subgraph task configs
-        # always carry an explicit `None` checkpoint_id.
-        self._delta_channels_forced_snapshot = (
-            {k for k, spec in specs.items() if isinstance(spec, DeltaChannel)}
-            if self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
-            else set()
-        )
         self.prev_checkpoint_config = None
         runtime = self.config[CONF].get(CONFIG_KEY_RUNTIME)
         self.control = runtime.control if isinstance(runtime, Runtime) else None
@@ -696,7 +683,7 @@ class PregelLoop:
     def after_tick(self) -> None:
         # finish superstep
         writes = [w for t in self.tasks.values() for w in t.writes]
-        self._delta_channels_forced_snapshot.update(
+        self._delta_channels_with_overwrite.update(
             ch
             for ch, v in writes
             if isinstance(self.specs.get(ch), DeltaChannel) and _get_overwrite(v)[0]
@@ -1004,7 +991,7 @@ class PregelLoop:
                 manager=None,
                 updated_channels=updated_channels,
             )
-            self._delta_channels_forced_snapshot.update(
+            self._delta_channels_with_overwrite.update(
                 c
                 for c, v in input_writes
                 if isinstance(self.specs.get(c), DeltaChannel) and _get_overwrite(v)[0]
@@ -1149,7 +1136,7 @@ class PregelLoop:
         # create new checkpoint
         channels_to_snapshot = (
             delta_channels_to_snapshot(self.channels, new_counters)
-            | self._delta_channels_forced_snapshot
+            | self._delta_channels_with_overwrite
             if do_checkpoint
             else set()
         )
@@ -1167,7 +1154,7 @@ class PregelLoop:
         for k in channels_to_snapshot:
             new_counters[k] = (0, 0)
         if do_checkpoint:
-            self._delta_channels_forced_snapshot.difference_update(channels_to_snapshot)
+            self._delta_channels_with_overwrite.difference_update(channels_to_snapshot)
         non_zero = {k: v for k, v in new_counters.items() if v != (0, 0)}
         if non_zero:
             self.checkpoint_metadata["counters_since_delta_snapshot"] = non_zero
@@ -1252,7 +1239,7 @@ class PregelLoop:
         )
         channels_to_snapshot = (
             delta_channels_to_snapshot(self.channels, counters)
-            | self._delta_channels_forced_snapshot
+            | self._delta_channels_with_overwrite
         )
 
         pending = [
@@ -1697,6 +1684,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         )
         self._delta_write_futs = []
         self._error_handler_write_futs = []
+        self._delta_channels_with_overwrite = set()
         self._exit_delta_writes = (
             [] if self.durability == "exit" and self.checkpointer is not None else None
         )
@@ -1954,6 +1942,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         )
         self._delta_write_futs = []
         self._error_handler_write_futs = []
+        self._delta_channels_with_overwrite = set()
         self._exit_delta_writes = (
             [] if self.durability == "exit" and self.checkpointer is not None else None
         )

libs/langgraph/langgraph/pregel/main.py

@@ -108,7 +108,6 @@ from langgraph.callbacks import (
     get_sync_graph_callback_manager_for_config,
 )
 from langgraph.channels.base import BaseChannel
-from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.topic import Topic
 from langgraph.config import get_config
 from langgraph.constants import END
@@ -134,7 +133,6 @@ from langgraph.pregel._checkpoint import (
     copy_checkpoint,
     create_checkpoint,
     create_checkpoint_plan_for_update_state_api,
-    create_fork_checkpoint,
     empty_checkpoint,
     get_updated_channels_from_tasks,
 )
@@ -1639,21 +1637,8 @@ class Pregel(
             else:
                 raise ValueError(f"Subgraph {recast} not found")
 
-        # Read once from the caller's config: every later superstep receives
-        # the config of the checkpoint just written, which always names one.
-        # Cleared by the first checkpoint that carries the snapshots, which
-        # `__copy__` does not write.
-        fork_pending: set[str] = (
-            {k for k, v in self.channels.items() if isinstance(v, DeltaChannel)}
-            if config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
-            else set()
-        )
-
         def perform_superstep(
-            input_config: RunnableConfig,
-            updates: Sequence[StateUpdate],
-            *,
-            is_fork: bool,
+            input_config: RunnableConfig, updates: Sequence[StateUpdate]
         ) -> RunnableConfig:
             # get last checkpoint
             config = ensure_config(self.config, input_config)
@@ -1741,17 +1726,9 @@ class Pregel(
                         self.trigger_to_nodes,
                     )
                 # save checkpoint
-                next_checkpoint = create_fork_checkpoint(
-                    checkpoint,
-                    channels,
-                    step,
-                    is_fork=is_fork,
-                    get_next_version=checkpointer.get_next_version,
-                )
-                fork_pending.difference_update(next_checkpoint["channel_values"])
                 next_config = checkpointer.put(
                     checkpoint_config,
-                    next_checkpoint,
+                    create_checkpoint(checkpoint, channels, step),
                     {
                         "source": "update",
                         "step": step + 1,
@@ -1759,7 +1736,7 @@ class Pregel(
                     },
                     get_new_channel_versions(
                         checkpoint_previous_versions,
-                        next_checkpoint["channel_versions"],
+                        checkpoint["channel_versions"],
                     ),
                 )
                 return patch_checkpoint_map(
@@ -1788,17 +1765,9 @@ class Pregel(
                         if saved and saved.metadata.get("step") is not None
                         else -1
                     )
-                    next_checkpoint = create_fork_checkpoint(
-                        checkpoint,
-                        channels,
-                        next_step,
-                        is_fork=is_fork,
-                        get_next_version=checkpointer.get_next_version,
-                    )
-                    fork_pending.difference_update(next_checkpoint["channel_values"])
                     next_config = checkpointer.put(
                         checkpoint_config,
-                        next_checkpoint,
+                        create_checkpoint(checkpoint, channels, next_step),
                         {
                             "source": "input",
                             "step": next_step,
@@ -1808,7 +1777,7 @@ class Pregel(
                         },
                         get_new_channel_versions(
                             checkpoint_previous_versions,
-                            next_checkpoint["channel_versions"],
+                            checkpoint["channel_versions"],
                         ),
                     )
 
@@ -1904,7 +1873,6 @@ class Pregel(
                     return perform_superstep(
                         patch_checkpoint_map(next_config, saved.metadata),
                         [item for lst in user_group_by.values() for item in lst],
-                        is_fork=is_fork,
                     )
 
                 return patch_checkpoint_map(next_config, saved.metadata)
@@ -2052,7 +2020,6 @@ class Pregel(
                     parents=saved.metadata.get("parents", {}) if saved else {},
                     saved_metadata=saved.metadata if saved else None,
                     is_fresh_thread=saved is None,
-                    is_fork=is_fork,
                 )
             )
             checkpoint = create_checkpoint(
@@ -2065,8 +2032,6 @@ class Pregel(
                 else None,
                 channels_to_snapshot=channels_to_snapshot,
             )
-            if is_fork:
-                fork_pending.difference_update(checkpoint["channel_values"])
             next_config = checkpointer.put(
                 checkpoint_config,
                 checkpoint,
@@ -2085,9 +2050,7 @@ class Pregel(
             config, {CONFIG_KEY_THREAD_ID: str(config[CONF][CONFIG_KEY_THREAD_ID])}
         )
         for superstep in supersteps:
-            current_config = perform_superstep(
-                current_config, superstep, is_fork=bool(fork_pending)
-            )
+            current_config = perform_superstep(current_config, superstep)
         return current_config
 
     async def abulk_update_state(
@@ -2140,21 +2103,8 @@ class Pregel(
             else:
                 raise ValueError(f"Subgraph {recast} not found")
 
-        # Read once from the caller's config: every later superstep receives
-        # the config of the checkpoint just written, which always names one.
-        # Cleared by the first checkpoint that carries the snapshots, which
-        # `__copy__` does not write.
-        fork_pending: set[str] = (
-            {k for k, v in self.channels.items() if isinstance(v, DeltaChannel)}
-            if config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
-            else set()
-        )
-
         async def aperform_superstep(
-            input_config: RunnableConfig,
-            updates: Sequence[StateUpdate],
-            *,
-            is_fork: bool,
+            input_config: RunnableConfig, updates: Sequence[StateUpdate]
         ) -> RunnableConfig:
             # get last checkpoint
             config = ensure_config(self.config, input_config)
@@ -2240,25 +2190,16 @@ class Pregel(
                         self.trigger_to_nodes,
                     )
                 # save checkpoint
-                next_checkpoint = create_fork_checkpoint(
-                    checkpoint,
-                    channels,
-                    step,
-                    is_fork=is_fork,
-                    get_next_version=checkpointer.get_next_version,
-                )
-                fork_pending.difference_update(next_checkpoint["channel_values"])
                 next_config = await checkpointer.aput(
                     checkpoint_config,
-                    next_checkpoint,
+                    create_checkpoint(checkpoint, channels, step),
                     {
                         "source": "update",
                         "step": step + 1,
                         "parents": saved.metadata.get("parents", {}) if saved else {},
                     },
                     get_new_channel_versions(
-                        checkpoint_previous_versions,
-                        next_checkpoint["channel_versions"],
+                        checkpoint_previous_versions, checkpoint["channel_versions"]
                     ),
                 )
                 return patch_checkpoint_map(
@@ -2287,17 +2228,9 @@ class Pregel(
                         if saved and saved.metadata.get("step") is not None
                         else -1
                     )
-                    next_checkpoint = create_fork_checkpoint(
-                        checkpoint,
-                        channels,
-                        next_step,
-                        is_fork=is_fork,
-                        get_next_version=checkpointer.get_next_version,
-                    )
-                    fork_pending.difference_update(next_checkpoint["channel_values"])
                     next_config = await checkpointer.aput(
                         checkpoint_config,
-                        next_checkpoint,
+                        create_checkpoint(checkpoint, channels, next_step),
                         {
                             "source": "input",
                             "step": next_step,
@@ -2307,7 +2240,7 @@ class Pregel(
                         },
                         get_new_channel_versions(
                             checkpoint_previous_versions,
-                            next_checkpoint["channel_versions"],
+                            checkpoint["channel_versions"],
                         ),
                     )
 
@@ -2402,7 +2335,6 @@ class Pregel(
                     return await aperform_superstep(
                         patch_checkpoint_map(next_config, saved.metadata),
                         [item for lst in user_group_by.values() for item in lst],
-                        is_fork=is_fork,
                     )
 
                 return patch_checkpoint_map(
@@ -2548,7 +2480,6 @@ class Pregel(
                     parents=saved.metadata.get("parents", {}) if saved else {},
                     saved_metadata=saved.metadata if saved else None,
                     is_fresh_thread=saved is None,
-                    is_fork=is_fork,
                 )
             )
             checkpoint = create_checkpoint(
@@ -2561,8 +2492,6 @@ class Pregel(
                 else None,
                 channels_to_snapshot=channels_to_snapshot,
             )
-            if is_fork:
-                fork_pending.difference_update(checkpoint["channel_values"])
             next_config = await checkpointer.aput(
                 checkpoint_config,
                 checkpoint,
@@ -2580,9 +2509,7 @@ class Pregel(
             config, {CONFIG_KEY_THREAD_ID: str(config[CONF][CONFIG_KEY_THREAD_ID])}
         )
         for superstep in supersteps:
-            current_config = await aperform_superstep(
-                current_config, superstep, is_fork=bool(fork_pending)
-            )
+            current_config = await aperform_superstep(current_config, superstep)
         return current_config
 
     def update_state(

libs/langgraph/tests/memory_assert.py

@@ -85,13 +85,11 @@ class MemorySaverAssertImmutable(InMemorySaver):
                 )
                 == saved
             ), config["configurable"]["checkpoint_ns"]
-        next_config = super().put(config, checkpoint, metadata, new_versions)
-        # Read back, not the object handed in: a DeltaChannel a step did not
-        # write is refilled on read from the blob its inherited version points at.
         self.storage_for_copies[thread_id][checkpoint_ns][checkpoint["id"]] = (
-            self.serde.dumps_typed(super().get(next_config))
+            self.serde.dumps_typed(checkpoint)
         )
-        return next_config
+        # call super to write checkpoint
+        return super().put(config, checkpoint, metadata, new_versions)
 
 
 class MemorySaverNoPending(InMemorySaver):

libs/langgraph/tests/test_delta_channel_fork.py

@@ -1,346 +0,0 @@
-"""Forking a thread must not replay the abandoned branch into the fork.
-
-Every graph carries a ``DeltaChannel`` and a plain reducer channel fed the same
-values; the plain channel needs no replay, so it is the oracle.
-"""
-
-from collections.abc import Sequence
-from operator import add
-from typing import Annotated, Any
-
-import pytest
-from langchain_core.runnables import RunnableConfig
-from langgraph.checkpoint.base import BaseCheckpointSaver
-from langgraph.checkpoint.serde.types import _DeltaSnapshot
-from typing_extensions import TypedDict
-
-from langgraph._internal._constants import INPUT
-from langgraph.channels.delta import DeltaChannel
-from langgraph.graph import END, StateGraph
-from langgraph.types import Durability, StateSnapshot, StateUpdate
-
-pytestmark = pytest.mark.anyio
-
-
-def _append(current: list | None, writes: Sequence[Any]) -> list:
-    out = list(current or [])
-    for write in writes:
-        out.extend(write if isinstance(write, list) else [write])
-    return out
-
-
-class _State(TypedDict):
-    log: Annotated[list, DeltaChannel(_append, snapshot_frequency=1000)]
-    plain: Annotated[list, add]
-    other: Annotated[list, add]
-
-
-def _build(checkpointer: BaseCheckpointSaver, tag: str) -> Any:
-    def node(state: _State) -> dict:
-        return {"log": [f"{tag}-out"], "plain": [f"{tag}-out"]}
-
-    builder = StateGraph(_State)
-    builder.add_node("n", node)
-    builder.set_entry_point("n")
-    builder.set_finish_point("n")
-    return builder.compile(checkpointer=checkpointer)
-
-
-def _build_without_delta_writes(checkpointer: BaseCheckpointSaver, tag: str) -> Any:
-    def node(state: _State) -> dict:
-        return {"other": [f"{tag}-other"]}
-
-    builder = StateGraph(_State)
-    builder.add_node("n", node)
-    builder.set_entry_point("n")
-    builder.set_finish_point("n")
-    return builder.compile(checkpointer=checkpointer)
-
-
-def _thread(thread_id: str) -> RunnableConfig:
-    return {"configurable": {"thread_id": thread_id}}
-
-
-def _at(config: RunnableConfig, snapshot: StateSnapshot) -> RunnableConfig:
-    return {
-        "configurable": {
-            **config["configurable"],
-            "checkpoint_ns": "",
-            "checkpoint_id": snapshot.config["configurable"]["checkpoint_id"],
-        }
-    }
-
-
-def _input(marker: str) -> dict:
-    return {"log": [marker], "plain": [marker]}
-
-
-def _snapshotted_checkpoints(
-    checkpointer: BaseCheckpointSaver, config: RunnableConfig
-) -> list[str]:
-    return [
-        tuple_.config["configurable"]["checkpoint_id"]
-        for tuple_ in checkpointer.list(config)
-        if isinstance(tuple_.checkpoint["channel_values"].get("log"), _DeltaSnapshot)
-    ]
-
-
-def _assert_fork_is_clean(state: StateSnapshot, abandoned: str) -> None:
-    assert state.values["log"] == state.values["plain"], (
-        f"delta channel diverged from the plain channel: "
-        f"{state.values['log']} != {state.values['plain']}"
-    )
-    assert abandoned not in state.values["log"], (
-        f"{abandoned!r} belongs to the branch the fork replaced, "
-        f"but was replayed into {state.values['log']}"
-    )
-
-
-def test_fork_by_invoke(
-    sync_checkpointer: BaseCheckpointSaver, durability: Durability
-) -> None:
-    config = _thread("t")
-    _build(sync_checkpointer, "first").invoke(
-        _input("in-1"), config, durability=durability
-    )
-    graph = _build(sync_checkpointer, "second")
-    graph.invoke(_input("in-2"), config, durability=durability)
-    abandoned_head = graph.get_state(config)
-
-    base = next(
-        snapshot
-        for snapshot in graph.get_state_history(config)
-        if "in-2" not in snapshot.values["log"]
-    )
-    _build(sync_checkpointer, "third").invoke(
-        _input("in-3"), _at(config, base), durability=durability
-    )
-
-    state = graph.get_state(config)
-    _assert_fork_is_clean(state, "in-2")
-    assert state.values["log"] == [*base.values["log"], "in-3", "third-out"]
-
-    abandoned = graph.get_state(abandoned_head.config).values
-    assert abandoned["log"] == abandoned["plain"] == abandoned_head.values["log"]
-
-
-async def test_afork_by_invoke(
-    async_checkpointer: BaseCheckpointSaver, durability: Durability
-) -> None:
-    config = _thread("t")
-    await _build(async_checkpointer, "first").ainvoke(
-        _input("in-1"), config, durability=durability
-    )
-    graph = _build(async_checkpointer, "second")
-    await graph.ainvoke(_input("in-2"), config, durability=durability)
-    abandoned_head = await graph.aget_state(config)
-
-    base = await anext(
-        snapshot
-        async for snapshot in graph.aget_state_history(config)
-        if "in-2" not in snapshot.values["log"]
-    )
-    await _build(async_checkpointer, "third").ainvoke(
-        _input("in-3"), _at(config, base), durability=durability
-    )
-
-    state = await graph.aget_state(config)
-    _assert_fork_is_clean(state, "in-2")
-    assert state.values["log"] == [*base.values["log"], "in-3", "third-out"]
-
-    abandoned = (await graph.aget_state(abandoned_head.config)).values
-    assert abandoned["log"] == abandoned["plain"] == abandoned_head.values["log"]
-
-
-def test_fork_off_checkpoint_before_first_input(
-    sync_checkpointer: BaseCheckpointSaver, durability: Durability
-) -> None:
-    config = _thread("t")
-    graph = _build(sync_checkpointer, "first")
-    graph.invoke(_input("in-1"), config, durability=durability)
-
-    root = list(graph.get_state_history(config))[-1]
-    assert root.values["log"] == []
-
-    _build(sync_checkpointer, "third").invoke(
-        _input("in-9"), _at(config, root), durability=durability
-    )
-
-    state = graph.get_state(config)
-    _assert_fork_is_clean(state, "in-1")
-    assert state.values["log"] == ["in-9", "third-out"]
-
-
-async def test_afork_off_checkpoint_before_first_input(
-    async_checkpointer: BaseCheckpointSaver, durability: Durability
-) -> None:
-    config = _thread("t")
-    graph = _build(async_checkpointer, "first")
-    await graph.ainvoke(_input("in-1"), config, durability=durability)
-
-    root = [snapshot async for snapshot in graph.aget_state_history(config)][-1]
-    assert root.values["log"] == []
-
-    await _build(async_checkpointer, "third").ainvoke(
-        _input("in-9"), _at(config, root), durability=durability
-    )
-
-    state = await graph.aget_state(config)
-    _assert_fork_is_clean(state, "in-1")
-    assert state.values["log"] == ["in-9", "third-out"]
-
-
-def test_fork_by_update_state(sync_checkpointer: BaseCheckpointSaver) -> None:
-    config = _thread("t")
-    _build(sync_checkpointer, "first").invoke(_input("in-1"), config)
-    graph = _build(sync_checkpointer, "second")
-    graph.invoke(_input("in-2"), config)
-
-    base = next(
-        snapshot
-        for snapshot in graph.get_state_history(config)
-        if "in-2" not in snapshot.values["log"]
-    )
-    forked = graph.update_state(_at(config, base), _input("patched"))
-
-    state = graph.get_state(forked)
-    _assert_fork_is_clean(state, "in-2")
-    assert state.values["log"] == [*base.values["log"], "patched"]
-
-
-async def test_afork_by_update_state(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    config = _thread("t")
-    await _build(async_checkpointer, "first").ainvoke(_input("in-1"), config)
-    graph = _build(async_checkpointer, "second")
-    await graph.ainvoke(_input("in-2"), config)
-
-    base = await anext(
-        snapshot
-        async for snapshot in graph.aget_state_history(config)
-        if "in-2" not in snapshot.values["log"]
-    )
-    forked = await graph.aupdate_state(_at(config, base), _input("patched"))
-
-    state = await graph.aget_state(forked)
-    _assert_fork_is_clean(state, "in-2")
-    assert state.values["log"] == [*base.values["log"], "patched"]
-
-
-def test_unaddressed_run_keeps_snapshot_cadence(
-    sync_checkpointer: BaseCheckpointSaver, durability: Durability
-) -> None:
-    config = _thread("t")
-    graph = _build(sync_checkpointer, "first")
-    graph.invoke(_input("in-1"), config, durability=durability)
-    graph.invoke(_input("in-2"), config, durability=durability)
-
-    assert not _snapshotted_checkpoints(sync_checkpointer, config)
-
-
-def test_fork_before_first_value_when_fork_never_writes_the_channel(
-    sync_checkpointer: BaseCheckpointSaver, durability: Durability
-) -> None:
-    config = _thread("t")
-    graph = _build(sync_checkpointer, "first")
-    graph.invoke(_input("in-1"), config, durability=durability)
-
-    root = list(graph.get_state_history(config))[-1]
-    assert root.values["log"] == []
-
-    _build_without_delta_writes(sync_checkpointer, "third").invoke(
-        {"other": ["in-9"]}, _at(config, root), durability=durability
-    )
-
-    state = graph.get_state(config)
-    _assert_fork_is_clean(state, "in-1")
-    assert state.values["log"] == []
-
-
-async def test_afork_before_first_value_when_fork_never_writes_the_channel(
-    async_checkpointer: BaseCheckpointSaver, durability: Durability
-) -> None:
-    config = _thread("t")
-    graph = _build(async_checkpointer, "first")
-    await graph.ainvoke(_input("in-1"), config, durability=durability)
-
-    root = [snapshot async for snapshot in graph.aget_state_history(config)][-1]
-    assert root.values["log"] == []
-
-    await _build_without_delta_writes(async_checkpointer, "third").ainvoke(
-        {"other": ["in-9"]}, _at(config, root), durability=durability
-    )
-
-    state = await graph.aget_state(config)
-    _assert_fork_is_clean(state, "in-1")
-    assert state.values["log"] == []
-
-
-def test_fork_before_first_value_by_bulk_update(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    config = _thread("t")
-    graph = _build(sync_checkpointer, "first")
-    graph.invoke(_input("in-1"), config)
-
-    root = list(graph.get_state_history(config))[-1]
-    assert root.values["log"] == []
-
-    forked = graph.bulk_update_state(
-        _at(config, root),
-        [
-            [StateUpdate({"other": ["s1"]}, "n")],
-            [StateUpdate(_input("s2"), "n")],
-        ],
-    )
-
-    state = graph.get_state(forked)
-    _assert_fork_is_clean(state, "in-1")
-    assert state.values["log"] == ["s2"]
-
-
-@pytest.mark.parametrize("first_as_node", [INPUT, END, "__copy__"])
-def test_fork_by_bulk_update_whose_first_superstep_skips_the_plan(
-    sync_checkpointer: BaseCheckpointSaver, first_as_node: str
-) -> None:
-    config = _thread("t")
-    _build(sync_checkpointer, "first").invoke(_input("in-1"), config)
-    graph = _build(sync_checkpointer, "second")
-    graph.invoke(_input("in-2"), config)
-
-    base = next(
-        snapshot
-        for snapshot in graph.get_state_history(config)
-        if "in-2" not in snapshot.values["log"]
-    )
-    first = (
-        StateUpdate(_input("first-step"), first_as_node)
-        if first_as_node == INPUT
-        else StateUpdate(None, first_as_node)
-    )
-    forked = graph.bulk_update_state(
-        _at(config, base),
-        [[first], [StateUpdate(_input("second-step"), "n")]],
-    )
-
-    state = graph.get_state(forked)
-    assert state.values["log"] == state.values["plain"], (
-        f"delta channel diverged from the plain channel: "
-        f"{state.values['log']} != {state.values['plain']}"
-    )
-
-
-def test_unaddressed_bulk_update_keeps_snapshot_cadence(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    config = _thread("t")
-    graph = _build(sync_checkpointer, "first")
-    graph.invoke(_input("in-1"), config)
-
-    graph.bulk_update_state(
-        config,
-        [[StateUpdate(_input(f"u{i}"), "n")] for i in range(4)],
-    )
-
-    assert not _snapshotted_checkpoints(sync_checkpointer, config)