fix(lint): remove redundant cast flagged by mypy

chore: apply ruff format/lint fixes
fix(message): full add_messages parity for _messages_delta_reducer
2026-10-16 21:25:35 +02:00 · 2026-05-06 15:08:37 -04:00 · 2026-05-06 15:03:13 -04:00 · 2026-05-06 14:59:12 -04:00 · 2026-05-06 14:42:36 -04:00 · 2026-05-05 21:58:10 -07:00
7 changed files with 747 additions and 133 deletions
@@ -115,6 +115,172 @@ class BuildResult:
    show_build_logs_on_failure: bool = False


+# ---------------------------------------------------------------------------
+# Structured output emitter
+# ---------------------------------------------------------------------------
+
+_emitter: "_Emitter | None" = None
+_no_input: bool = False
+
+
+class _Emitter:
+    """Dual-mode output: JSON-lines (``--json``) or human-readable click text."""
+
+    def __init__(self, json_mode: bool) -> None:
+        self._json = json_mode
+
+    @property
+    def json_mode(self) -> bool:
+        return self._json
+
+    # -- Structured event helpers ------------------------------------------
+
+    def step(self, step: int, message: str, **extra: object) -> None:
+        if self._json:
+            self._write({"event": "step", "step": step, "message": message, **extra})
+        else:
+            click.secho(f"{step}. {message}", fg="cyan")
+
+    def info(self, message: str, **extra: object) -> None:
+        if self._json:
+            self._write({"event": "info", "message": message, **extra})
+        else:
+            click.secho(f"   {message}", fg="green")
+
+    def warn(self, message: str, **extra: object) -> None:
+        """Warning nested under a step. Text mode indents; JSON mode strips leading whitespace."""
+        if self._json:
+            self._write({"event": "warn", "message": message.lstrip(), **extra})
+        else:
+            click.secho(f"   {message}", fg="yellow")
+
+    def note(self, message: str, **extra: object) -> None:
+        """Top-level banner (pre-step). Text mode does not indent."""
+        if self._json:
+            self._write({"event": "note", "message": message, **extra})
+        else:
+            click.secho(message, fg="yellow")
+
+    def error(self, message: str, **extra: object) -> None:
+        if self._json:
+            self._write({"event": "error", "message": message, **extra})
+        else:
+            click.secho(f"   {message}", fg="red")
+
+    def status_change(
+        self,
+        status: str,
+        elapsed_seconds: float,
+        finished: bool = False,
+    ) -> None:
+        mins, secs = divmod(int(elapsed_seconds), 60)
+        elapsed_str = f"{mins}m {secs:02d}s" if mins else f"{secs}s"
+        if self._json:
+            self._write(
+                {
+                    "event": "status_change",
+                    "status": status,
+                    "elapsed_seconds": round(elapsed_seconds, 1),
+                    "message": f"{status}... ({elapsed_str})",
+                }
+            )
+        else:
+            click.echo(f"   {status}... ({elapsed_str})")
+
+    def log(self, message: str) -> None:
+        if self._json:
+            self._write({"event": "log", "message": message})
+        else:
+            click.echo(f"   | {message}")
+
+    def status_url(self, url: str) -> None:
+        if self._json:
+            self._write({"event": "status_url", "url": url})
+        else:
+            click.secho(f"   View status: {url}", fg="cyan")
+
+    def result(
+        self,
+        status: str,
+        *,
+        deployment_id: str,
+        url: str | None = None,
+        status_url: str | None = None,
+        fallback_status_message: str | None = None,
+    ) -> None:
+        if self._json:
+            if status == "succeeded":
+                message = "Deployment successful!"
+            elif status == "failed":
+                message = "Deployment failed"
+            else:
+                message = "Timed out waiting for deployment."
+            payload: dict = {
+                "event": "result",
+                "status": status,
+                "deployment_id": deployment_id,
+                "message": message,
+            }
+            if url:
+                payload["url"] = url
+            if status_url:
+                payload["status_url"] = status_url
+            self._write(payload)
+        else:
+            if status == "succeeded":
+                click.secho("   Deployment successful!", fg="green")
+                if url:
+                    click.secho(f"   URL: {url}", fg="green")
+                if status_url:
+                    click.secho(f"   View status: {status_url}", fg="green")
+            elif status == "failed":
+                click.secho("   Deployment failed", fg="red")
+                if status_url:
+                    click.secho(f"   View status: {status_url}", fg="red")
+            elif status == "timed_out":
+                click.secho("   Timed out waiting for deployment.", fg="yellow")
+                if status_url:
+                    click.secho(f"   Check status at: {status_url}", fg="yellow")
+                elif fallback_status_message:
+                    click.secho(f"   {fallback_status_message}", fg="yellow")
+
+    def heartbeat(self, status: str, elapsed_seconds: float) -> None:
+        if self._json:
+            mins, secs = divmod(int(elapsed_seconds), 60)
+            elapsed_str = f"{mins}m {secs:02d}s" if mins else f"{secs}s"
+            self._write(
+                {
+                    "event": "heartbeat",
+                    "status": status,
+                    "elapsed_seconds": round(elapsed_seconds, 1),
+                    "message": f"{status}... ({elapsed_str})",
+                }
+            )
+
+    def upload_progress(self, size_mb: float, pct: int) -> None:
+        if self._json:
+            self._write(
+                {
+                    "event": "upload_progress",
+                    "size_mb": round(size_mb, 1),
+                    "pct": pct,
+                }
+            )
+        else:
+            click.echo(f"\r   Uploading ({size_mb:.1f} MB)... {pct}%", nl=False)
+
+    def _write(self, obj: dict) -> None:
+        import sys as _sys
+
+        _sys.stdout.write(json_mod.dumps(obj, default=str) + "\n")
+        _sys.stdout.flush()
+
+
+def _get_emitter() -> _Emitter:
+    """Return the module-level emitter (falls back to text mode)."""
+    return _emitter or _Emitter(json_mode=False)
+
+
 # ---------------------------------------------------------------------------
 # Validators
 # ---------------------------------------------------------------------------
@@ -172,15 +338,16 @@ def find_deployment_id_by_name(
 # ---------------------------------------------------------------------------


-def normalize_image_name(value: str | None) -> str:
-    """Sanitize a deployment/directory name into a valid Docker repository name.
+def normalize_name(value: str | None) -> str:
+    """Sanitize a deployment/directory name into a valid deployment name.

-    Docker repository names must be lowercase and may only contain
-    [a-z0-9._-].  Invalid characters are replaced with hyphens.
+    LangSmith Deployment names only allow lowercase
+    alphanumeric characters and hyphens ([a-z0-9-]).
+    Invalid characters are replaced with hyphens.
    """
    if not value:
        return "app"
-    slug = re.sub(r"[^a-z0-9._-]+", "-", value.lower()).strip("-.")
+    slug = re.sub(r"[^a-z0-9-]+", "-", value.lower()).strip("-")
    return slug or "app"


@@ -307,9 +474,8 @@ def _resolve_env_path(
    if isinstance(env_field, str):
        env_path = (config_path.parent / env_field).resolve()
        if not env_path.exists():
-            click.secho(
-                f"Warning: env file '{env_field}' specified in langgraph.json not found.",
-                fg="yellow",
+            _get_emitter().note(
+                f"Warning: env file '{env_field}' specified in langgraph.json not found."
            )
            return None
        return env_path
@@ -343,7 +509,7 @@ def _secrets_from_env(
    secrets: list[dict[str, str]] = []
    for name, value in env_vars.items():
        if name in RESERVED_ENV_VARS:
-            click.secho(f"   Skipping reserved env var: {name}", fg="yellow")
+            _get_emitter().note(f"Skipping reserved env var: {name}")
            continue
        if not value:
            continue
@@ -386,8 +552,8 @@ def _resolve_build_mode(
 # ---------------------------------------------------------------------------


-def _log_deploy_step(step: int, message: str) -> None:
-    click.secho(f"{step}. {message}", fg="cyan")
+def _log_deploy_step(step: int, message: str, **extra: object) -> None:
+    _get_emitter().step(step, message, **extra)


 def _resolve_deployment(
@@ -411,12 +577,13 @@ def _resolve_deployment(
    found_id = _call_host_backend_with_optional_tenant(
        client, lambda c: find_deployment_id_by_name(c, name)
    )
+    em = _get_emitter()
    if found_id:
        deployment_id = str(found_id)
-        click.secho(f"   Found existing deployment (ID: {deployment_id})", fg="green")
+        em.info(f"Found existing deployment (ID: {deployment_id})")
    else:
        needs_creation = True
-        click.secho(not_found_message, fg="yellow")
+        em.warn(not_found_message)
    return deployment_id, needs_creation, step + 1


@@ -444,7 +611,7 @@ def _create_deployment(
        raise HostBackendError(
            "POST /v2/deployments succeeded but response missing a valid 'id'"
        )
-    click.secho(f"   Deployment ID: {created_id}", fg="green")
+    _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id)
    return created_id, step + 1


@@ -458,21 +625,36 @@ def _smith_dashboard_base_url(host_url: str | None) -> str:
    hostname = parsed.hostname or ""
    if hostname in ("localhost", "127.0.0.1"):
        return host_url.rstrip("/")
-    if hostname.startswith("eu."):
-        return "https://eu.smith.langchain.com"
+
+    api_host_suffix = "api.host.langchain.com"
+    if hostname == api_host_suffix:
+        return "https://smith.langchain.com"
+    if hostname.endswith(f".{api_host_suffix}"):
+        prefix = hostname[: -(len(api_host_suffix) + 1)]
+        return f"https://{prefix}.smith.langchain.com"
+
    return "https://smith.langchain.com"


-def _print_deployment_status_url(
+def _get_deployment_status_url(
    updated: object, deployment_id: str, host_url: str | None = None
-) -> None:
-    """Print the deployment status URL when tenant metadata is available."""
+) -> 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
+        return None
    base = _smith_dashboard_base_url(host_url)
-    status_url = f"{base}/o/{tenant_id}/host/deployments/{deployment_id}"
-    click.secho(f"   View status: {status_url}", fg="cyan")
+    return f"{base}/o/{tenant_id}/host/deployments/{deployment_id}"
+
+
+def _emit_deployment_status_url(
+    updated: object, deployment_id: str, host_url: str | None = None
+) -> str | None:
+    """Emit the deployment status URL and return it."""
+    url = _get_deployment_status_url(updated, deployment_id, host_url)
+    if url:
+        _get_emitter().status_url(url)
+    return url


 def _poll_revision_status(
@@ -486,6 +668,7 @@ def _poll_revision_status(
    on_interrupt: Callable[[str], None] | None = None,
 ) -> 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 []
@@ -497,7 +680,11 @@ def _poll_revision_status(
    last_status = ""
    deadline = time.time() + timeout_seconds
    start_time = time.monotonic()
-    with Progress(message=progress_message, elapsed=True) as set_progress:
+    last_heartbeat = start_time
+    json_mode = em.json_mode
+    with Progress(
+        message=progress_message, elapsed=True, json_mode=json_mode
+    ) as set_progress:
        while time.time() < deadline:
            try:
                rev = client.get_revision(deployment_id, revision_id)
@@ -514,14 +701,15 @@ def _poll_revision_status(
            if status != last_status:
                set_progress("")
                if last_status:
-                    elapsed = time.monotonic() - start_time
-                    mins, secs = divmod(int(elapsed), 60)
-                    elapsed_str = f"{mins}m {secs:02d}s" if mins else f"{secs}s"
-                    click.echo(f"   {last_status}... ({elapsed_str})")
+                    em.status_change(last_status, time.monotonic() - start_time)
                last_status = status
                if status in _TERMINAL_STATUSES:
                    break
                set_progress(f"{status}...")
+                last_heartbeat = time.monotonic()
+            elif json_mode and time.monotonic() - last_heartbeat > 10:
+                em.heartbeat(last_status, time.monotonic() - start_time)
+                last_heartbeat = time.monotonic()

            if on_poll is not None:
                on_poll(status, revision_id, set_progress)
@@ -538,8 +726,10 @@ def _print_deployment_result(
    last_status: str,
    *,
    dashboard_label: str,
+    status_url: str | None = None,
 ) -> None:
    """Print final deployment status and raise on failure."""
+    em = _get_emitter()
    dep_info = client.get_deployment(deployment_id)
    custom_url = None
    if isinstance(dep_info, dict):
@@ -548,24 +738,28 @@ def _print_deployment_result(
            custom_url = sc.get("custom_url")

    if last_status == "DEPLOYED":
-        click.secho("   Deployment successful!", fg="green")
-        if custom_url:
-            click.secho(f"   URL: {custom_url}", fg="green")
+        em.result(
+            "succeeded",
+            deployment_id=deployment_id,
+            url=custom_url,
+            status_url=status_url,
+        )
    elif last_status in ("BUILD_FAILED", "DEPLOY_FAILED", "CREATE_FAILED"):
-        click.secho(f"   Deployment failed: {last_status}", fg="red")
+        em.result(
+            "failed",
+            deployment_id=deployment_id,
+            status_url=status_url,
+        )
        raise click.exceptions.Exit(1)
    else:
-        click.secho(
-            f"   Timed out waiting for deployment (last status: {last_status}).",
-            fg="yellow",
+        em.result(
+            "timed_out",
+            deployment_id=deployment_id,
+            status_url=status_url,
+            fallback_status_message=(
+                f"Check status in the LangSmith {dashboard_label}."
+            ),
        )
-        if custom_url:
-            click.secho(f"   Check status at: {custom_url}", fg="yellow")
-        else:
-            click.secho(
-                f"   Check status in the LangSmith {dashboard_label}.",
-                fg="yellow",
-            )


 # ---------------------------------------------------------------------------
@@ -594,14 +788,16 @@ def _docker_config_for_token(registry_host: str, token: str):
 # ---------------------------------------------------------------------------

 _UPLOAD_TIMEOUT_SECONDS = 300
+_BYTES_PER_MIB = 1_048_576


 class _ProgressReader:
-    """File-like wrapper that displays upload progress via click."""
+    """File-like wrapper that reports upload progress via the emitter."""

-    def __init__(self, fobj, file_size: int):
+    def __init__(self, fobj, file_size: int, emitter: "_Emitter"):
        self._fobj = fobj
        self._file_size = file_size
+        self._emitter = emitter
        self._uploaded = 0

    def read(self, size=-1):
@@ -611,10 +807,7 @@ class _ProgressReader:
            pct = (
                int(self._uploaded * 100 / self._file_size) if self._file_size else 100
            )
-            click.echo(
-                f"\r   Uploading ({self._file_size / 1_048_576:.1f} MB)... {pct}%",
-                nl=False,
-            )
+            self._emitter.upload_progress(self._file_size / _BYTES_PER_MIB, pct)
        return data

    def __len__(self):
@@ -626,10 +819,13 @@ def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None:
    import urllib.error
    import urllib.request

+    em = _get_emitter()
+
    with open(file_path, "rb") as f:
+        reader = _ProgressReader(f, file_size, em)
        req = urllib.request.Request(
            signed_url,
-            data=_ProgressReader(f, file_size),
+            data=reader,
            method="PUT",
            headers={
                "Content-Type": "application/gzip",
@@ -644,7 +840,8 @@ def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None:
            raise click.ClickException(
                f"Upload failed with status {err.code}: {detail}"
            ) from None
-    click.echo()
+    if not em.json_mode:
+        click.echo()


 # ---------------------------------------------------------------------------
@@ -752,7 +949,7 @@ def _run_local_build(
        if "://" in normalized_registry:
            normalized_registry = normalized_registry.split("//", 1)[1]
        repo_seed = image_name or name or config.parent.name
-        repo_name = normalize_image_name(repo_seed)
+        repo_name = normalize_name(repo_seed)
        tag_value = normalize_image_tag(tag)
        remote_image = f"{normalized_registry}/{repo_name}:{tag_value}"

@@ -811,9 +1008,8 @@ def _run_local_build(
                    break
                except click.exceptions.Exit:
                    if attempt < max_push_retries - 1:
-                        click.secho(
-                            f"   Push failed, retrying (attempt {attempt + 2} of {max_push_retries})...",
-                            fg="yellow",
+                        _get_emitter().warn(
+                            f"   Push failed, retrying (attempt {attempt + 2} of {max_push_retries})..."
                        )
                    else:
                        raise
@@ -847,9 +1043,10 @@ def _run_remote_build(
    """Upload source tarball and trigger a remote build."""
    from langgraph_cli.archive import create_archive

+    em = _get_emitter()
    _log_deploy_step(step, "Creating source archive")
    with create_archive(config, config_json) as (archive_path, file_size, config_rel):
-        click.secho(f"   Archive created ({file_size / 1_048_576:.1f} MB)", fg="green")
+        em.info(f"Archive created ({file_size / _BYTES_PER_MIB:.1f} MB)")
        step += 1

        _log_deploy_step(step, "Requesting upload URL")
@@ -897,12 +1094,12 @@ def _run_remote_build(
                if has_output:
                    set_progress("")
                    if not logs_header_printed:
-                        click.echo(f"   {status} (build logs):")
+                        em.info(f"{status} (build logs):")
                        logs_header_printed = True
                for entry in entries:
                    msg = entry.get("message", "")
                    if msg:
-                        click.echo(f"   | {msg}")
+                        em.log(msg)
                log_offset = logs_resp.get("next_offset") or log_offset
                if has_output:
                    set_progress(f"{status}...")
@@ -910,11 +1107,10 @@ def _run_remote_build(
            pass

    def _handle_interrupt(revision_id: str) -> None:
-        click.secho(
-            f"\n   Interrupted. Deployment ID: {deployment_id}, Revision ID: {revision_id}",
-            fg="yellow",
+        em.warn(
+            f"\nInterrupted. Deployment ID: {deployment_id}, Revision ID: {revision_id}"
        )
-        click.secho("   The build will continue remotely.", fg="yellow")
+        em.warn("The build will continue remotely.")

    return BuildResult(
        updated=updated if isinstance(updated, dict) else {},
@@ -952,12 +1148,20 @@ def _create_host_backend_client(
                resolved_api_key = val
                break
    if not resolved_api_key:
+        if _no_input:
+            raise click.ClickException(
+                "No LangSmith API key found. Set LANGSMITH_API_KEY in the "
+                "environment or .env file."
+            )
        click.secho(
            "No LangSmith API key found. Create one at Settings > API Keys in LangSmith.",
            fg="yellow",
        )
        resolved_api_key = click.prompt("Enter LangSmith API key", hide_input=True)
-    return HostBackendClient(host_url, resolved_api_key)
+    tenant_id = env_vars.get("LANGSMITH_TENANT_ID") or os.environ.get(
+        "LANGSMITH_TENANT_ID"
+    )
+    return HostBackendClient(host_url, resolved_api_key, tenant_id=tenant_id)


 def _call_host_backend_with_optional_tenant(
@@ -982,6 +1186,12 @@ def _call_host_backend_with_optional_tenant(
                and err.status_code == 403
                and "requires workspace specification" in err.message
            ):
+                if _no_input:
+                    raise click.ClickException(
+                        "API key is org-scoped and requires a workspace ID. "
+                        "Set LANGSMITH_TENANT_ID in your .env file or "
+                        "use a workspace-scoped API key."
+                    ) from None
                click.secho(
                    "Your API key is org-scoped and requires a workspace ID.",
                    fg="yellow",
@@ -1189,6 +1399,19 @@ def _deploy_base_options(
                    "if Docker is not available locally."
                ),
            ),
+            click.option(
+                "--json",
+                "json_output",
+                is_flag=True,
+                default=False,
+                help="Emit structured JSON-lines to stdout instead of human-readable text.",
+            ),
+            click.option(
+                "--no-input",
+                is_flag=True,
+                default=False,
+                help="Never prompt for input; fail with an error if a required value is missing.",
+            ),
        ]
        if include_docker_args:
            # Only attach build args to the default command; on the group they
@@ -1256,36 +1479,50 @@ def _deploy_cmd(
    no_wait: bool,
    remote_build_flag: bool | None,
    docker_build_args: Sequence[str],
+    json_output: bool,
+    no_input: bool,
 ):
-    click.secho(
-        "Note: 'langgraph deploy' is in beta. Expect frequent updates and improvements.",
-        fg="yellow",
+    global _emitter, _no_input
+    _emitter = _Emitter(json_mode=json_output)
+    _no_input = no_input
+    em = _emitter
+
+    em.note(
+        "Note: 'langgraph deploy' is in beta. Expect frequent updates and improvements."
    )
-    click.echo()
+    if not json_output:
+        click.echo()

    # -- 1. Preflight --
    validate_deploy_commands(install_command, build_command)
    config_json = langgraph_cli.config.validate_config_file(config)
-    warn_non_wolfi_distro(config_json)
+    warn_non_wolfi_distro(config_json, emit=em.note)

    env_vars = _parse_env_from_config(config_json, config)

    if not deployment_id and not name:
        name = env_vars.get(_DEPLOYMENT_NAME_ENV)
    if not deployment_id and not name:
-        default_name = normalize_image_name(pathlib.Path.cwd().name)
-        name = click.prompt("Deployment name", default=default_name)
-        env_path = _resolve_env_path(config_json, config)
-        if env_path is not None:
-            set_key(str(env_path), _DEPLOYMENT_NAME_ENV, name)
-            click.echo(f"Saved deployment name to {env_path}")
+        default_name = normalize_name(pathlib.Path.cwd().name)
+        if no_input:
+            name = default_name
+        else:
+            name = click.prompt("Deployment name", default=default_name)
+    if name and not deployment_id:
+        name = normalize_name(name)
+        if not no_input:
+            env_path = _resolve_env_path(config_json, config)
+            if env_path is not None:
+                set_key(str(env_path), _DEPLOYMENT_NAME_ENV, name)
+                em.info(f"Saved deployment name to {env_path}")

    secrets = _secrets_from_env(_env_without_deployment_name(env_vars))

    use_remote_build, local_build_error = _resolve_build_mode(remote_build_flag)
    if use_remote_build and remote_build_flag is None and local_build_error:
-        click.secho(f"{local_build_error}\nUsing remote build instead.", fg="yellow")
-        click.echo()
+        em.note(f"{local_build_error}\nUsing remote build instead.")
+        if not json_output:
+            click.echo()

    # -- 2. Resolve / create deployment --
    client = _create_host_backend_client(host_url, api_key, env_vars=env_vars)
@@ -1297,9 +1534,9 @@ def _deploy_cmd(
        deployment_id,
        name,
        not_found_message=(
-            "   No deployment found. Will create."
+            "No deployment found. Will create."
            if use_remote_build
-            else "   No deployment found. Will create after build."
+            else "No deployment found. Will create after build."
        ),
    )

@@ -1350,10 +1587,14 @@ def _deploy_cmd(
        )

    # -- 4. Shared wait + result --
-    _print_deployment_status_url(build_result.updated, deployment_id, host_url)
+    dep_status_url = _emit_deployment_status_url(
+        build_result.updated,
+        deployment_id,
+        host_url,
+    )

    if no_wait:
-        click.secho(f"   {build_result.no_result_message}", fg="green")
+        em.info(build_result.no_result_message)
        return

    last_status, revision_id = _poll_revision_status(
@@ -1366,7 +1607,7 @@ def _deploy_cmd(
        on_interrupt=build_result.on_interrupt,
    )
    if not last_status:
-        click.secho(f"   {build_result.no_result_message}", fg="green")
+        em.info(build_result.no_result_message)
        return

    if (
@@ -1375,7 +1616,7 @@ def _deploy_cmd(
        and not verbose
        and revision_id is not None
    ):
-        click.secho("   Last build log lines:", fg="red")
+        em.error("Last build log lines:")
        try:
            logs_resp = client.get_build_logs(
                deployment_id,
@@ -1387,19 +1628,17 @@ def _deploy_cmd(
                for entry in entries:
                    msg = entry.get("message", "")
                    if msg:
-                        click.echo(f"   | {msg}")
+                        em.log(msg)
        except Exception:
-            click.secho("   (failed to fetch build logs)", fg="red")
-        click.secho(
-            "   Re-run with --verbose to see full build output.",
-            fg="yellow",
-        )
+            em.error("(failed to fetch build logs)")
+        em.warn("Re-run with --verbose to see full build output.")

    _print_deployment_result(
        client,
        deployment_id,
        last_status,
        dashboard_label="Deployment dashboard",
+        status_url=dep_status_url,
    )


@@ -12,10 +12,11 @@ class Progress:
        while True:
            yield from "|/-\\"

-    def __init__(self, *, message="", elapsed: bool = False):
+    def __init__(self, *, message="", elapsed: bool = False, json_mode: bool = False):
        self.message = message
        self._base_message = message
        self._show_elapsed = elapsed
+        self._json_mode = json_mode
        # use this to make sure we don't kill thread when we set msg to ""
        self._stop = threading.Event()
        # signalled when the spinner has no text on screen
@@ -69,6 +70,9 @@ class Progress:
            self._line_clear.set()

    def __enter__(self) -> Callable[[str], None]:
+        if self._json_mode:
+            return lambda message: None
+
        if sys.stdout.isatty():
            self.thread = threading.Thread(target=self.spinner_task)
            self.thread.start()
@@ -90,6 +94,8 @@ class Progress:
            return set_message

    def __exit__(self, exception, value, tb):
+        if self._json_mode:
+            return
        if sys.stdout.isatty():
            self.message = ""
            self._stop.set()
@@ -1,5 +1,7 @@
 """General-purpose utilities shared across the LangGraph CLI."""

+from collections.abc import Callable
+
 import click


@@ -7,21 +9,42 @@ def clean_empty_lines(input_str: str):
    return "\n".join(filter(None, input_str.splitlines()))


-def warn_non_wolfi_distro(config_json: dict) -> None:
-    """Show warning if image_distro is not set to 'wolfi'."""
+def warn_non_wolfi_distro(
+    config_json: dict,
+    *,
+    emit: Callable[[str], None] | None = None,
+) -> None:
+    """Show warning if image_distro is not set to 'wolfi'.
+
+    When ``emit`` is provided, each warning line is sent through it (used by
+    callers that need JSON-aware output). Otherwise falls back to colored
+    ``click.secho`` output.
+    """
    image_distro = config_json.get("image_distro", "debian")  # Default is debian
-    if image_distro != "wolfi":
-        click.secho(
-            "⚠️  Security Recommendation: Consider switching to Wolfi Linux for enhanced security.",
-            fg="yellow",
-            bold=True,
+    if image_distro == "wolfi":
+        return
+    if emit is not None:
+        emit(
+            "⚠️  Security Recommendation: Consider switching to Wolfi Linux for enhanced security."
        )
-        click.secho(
-            "   Wolfi is a security-oriented, minimal Linux distribution designed for containers.",
-            fg="yellow",
+        emit(
+            "   Wolfi is a security-oriented, minimal Linux distribution designed for containers."
        )
-        click.secho(
-            '   To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.',
-            fg="yellow",
+        emit(
+            '   To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.'
        )
-        click.secho("")  # Empty line for better readability
+        return
+    click.secho(
+        "⚠️  Security Recommendation: Consider switching to Wolfi Linux for enhanced security.",
+        fg="yellow",
+        bold=True,
+    )
+    click.secho(
+        "   Wolfi is a security-oriented, minimal Linux distribution designed for containers.",
+        fg="yellow",
+    )
+    click.secho(
+        '   To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.',
+        fg="yellow",
+    )
+    click.secho("")  # Empty line for better readability
@@ -1,6 +1,8 @@
 import base64
+import io
 import json
 import os
+import sys

 import click
 import httpx
@@ -8,12 +10,15 @@ import pytest

 from langgraph_cli.deploy import (
    _call_host_backend_with_optional_tenant,
+    _create_host_backend_client,
    _docker_config_for_token,
+    _Emitter,
    _env_without_deployment_name,
    _parse_env_from_config,
    _resolve_env_path,
-    normalize_image_name,
+    _smith_dashboard_base_url,
    normalize_image_tag,
+    normalize_name,
 )
 from langgraph_cli.host_backend import HostBackendClient, HostBackendError

@@ -40,30 +45,33 @@ class TestDockerConfigForToken:
            assert "gcr.io" in data["auths"]


-class TestNormalizeImageName:
+class TestNormalizeName:
    def test_simple_name(self):
-        assert normalize_image_name("myapp") == "myapp"
+        assert normalize_name("myapp") == "myapp"

    def test_uppercase_lowered(self):
-        assert normalize_image_name("MyApp") == "myapp"
+        assert normalize_name("MyApp") == "myapp"

    def test_special_chars_replaced(self):
-        assert normalize_image_name("my app!@#v2") == "my-app-v2"
+        assert normalize_name("my app!@#v2") == "my-app-v2"

-    def test_dots_and_hyphens_kept(self):
-        assert normalize_image_name("my-app.v2") == "my-app.v2"
+    def test_dots_replaced_with_hyphens(self):
+        assert normalize_name("my-app.v2") == "my-app-v2"
+
+    def test_underscores_replaced_with_hyphens(self):
+        assert normalize_name("simple_graph_name") == "simple-graph-name"

    def test_leading_trailing_stripped(self):
-        assert normalize_image_name("--my-app..") == "my-app"
+        assert normalize_name("--my-app..") == "my-app"

    def test_empty_string_returns_app(self):
-        assert normalize_image_name("") == "app"
+        assert normalize_name("") == "app"

    def test_none_returns_app(self):
-        assert normalize_image_name(None) == "app"
+        assert normalize_name(None) == "app"

    def test_all_invalid_chars_returns_app(self):
-        assert normalize_image_name("!!!") == "app"
+        assert normalize_name("!!!") == "app"


 class TestNormalizeImageTag:
@@ -271,3 +279,256 @@ class TestCallHostBackendWithOptionalTenant:
            _call_host_backend_with_optional_tenant(
                client, lambda c: c.list_deployments()
            )
+
+    def test_workspace_prompt_blocked_by_no_input(self, monkeypatch):
+        """With _no_input=True, 403 requiring workspace should raise ClickException."""
+        import langgraph_cli.deploy as deploy_mod
+
+        monkeypatch.setattr(deploy_mod, "_no_input", True)
+
+        requires_workspace = '{"detail":"requires workspace specification"}'
+        client = self._make_client(
+            lambda req: httpx.Response(403, text=requires_workspace)
+        )
+        with pytest.raises(click.ClickException, match="workspace"):
+            _call_host_backend_with_optional_tenant(
+                client, lambda c: c.list_deployments()
+            )
+
+
+# ---------------------------------------------------------------------------
+# _Emitter JSON mode
+# ---------------------------------------------------------------------------
+
+
+class TestEmitterJsonMode:
+    """Verify that _Emitter in json_mode writes valid JSON-lines to stdout."""
+
+    def _capture(self, fn):
+        """Run fn with stdout captured and return parsed JSON objects."""
+        buf = io.StringIO()
+        old = sys.stdout
+        sys.stdout = buf
+        try:
+            fn()
+        finally:
+            sys.stdout = old
+        lines = [line for line in buf.getvalue().splitlines() if line.strip()]
+        return [json.loads(line) for line in lines]
+
+    def test_step_event(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.step(1, "Building image"))
+        assert len(events) == 1
+        assert events[0]["event"] == "step"
+        assert events[0]["step"] == 1
+        assert events[0]["message"] == "Building image"
+
+    def test_info_event(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.info("All good"))
+        assert events[0]["event"] == "info"
+        assert events[0]["message"] == "All good"
+
+    def test_warn_event(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.warn("Careful"))
+        assert events[0]["event"] == "warn"
+
+    def test_error_event(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.error("Boom"))
+        assert events[0]["event"] == "error"
+        assert events[0]["message"] == "Boom"
+
+    def test_status_change_event(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.status_change("building", 12.345))
+        assert events[0]["event"] == "status_change"
+        assert events[0]["status"] == "building"
+        assert events[0]["elapsed_seconds"] == 12.3
+        assert events[0]["message"] == "building... (12s)"
+
+    def test_status_change_event_with_minutes(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.status_change("deploying", 95.0))
+        assert events[0]["message"] == "deploying... (1m 35s)"
+
+    def test_log_event(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.log("some output"))
+        assert events[0] == {"event": "log", "message": "some output"}
+
+    def test_status_url_event(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(
+            lambda: em.status_url("https://smith.langchain.com/deploy/123")
+        )
+        assert events[0]["event"] == "status_url"
+        assert events[0]["url"] == "https://smith.langchain.com/deploy/123"
+
+    def test_result_event_full(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(
+            lambda: em.result(
+                "succeeded",
+                deployment_id="dep-1",
+                url="https://app.example.com",
+                status_url="https://smith.langchain.com/deploy/dep-1",
+            )
+        )
+        assert events[0]["event"] == "result"
+        assert events[0]["status"] == "succeeded"
+        assert events[0]["deployment_id"] == "dep-1"
+        assert events[0]["message"] == "Deployment successful!"
+        assert events[0]["url"] == "https://app.example.com"
+        assert events[0]["status_url"] == "https://smith.langchain.com/deploy/dep-1"
+
+    def test_result_event_minimal(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.result("failed", deployment_id="dep-2"))
+        assert events[0]["event"] == "result"
+        assert events[0]["status"] == "failed"
+        assert events[0]["message"] == "Deployment failed"
+        assert "url" not in events[0]
+        assert "status_url" not in events[0]
+
+    def test_heartbeat_event(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.heartbeat("building", 30.789))
+        assert events[0]["event"] == "heartbeat"
+        assert events[0]["elapsed_seconds"] == 30.8
+        assert events[0]["message"] == "building... (30s)"
+
+    def test_heartbeat_silent_in_text_mode(self, capsys):
+        em = _Emitter(json_mode=False)
+        em.heartbeat("building", 10.0)
+        captured = capsys.readouterr()
+        assert captured.out == ""
+
+    def test_upload_progress_event(self):
+        em = _Emitter(json_mode=True)
+        events = self._capture(lambda: em.upload_progress(5.678, 42))
+        assert events[0]["event"] == "upload_progress"
+        assert events[0]["size_mb"] == 5.7
+        assert events[0]["pct"] == 42
+
+
+# ---------------------------------------------------------------------------
+# _Emitter text mode (non-json)
+# ---------------------------------------------------------------------------
+
+
+class TestEmitterTextMode:
+    """Verify that _Emitter in text mode uses click.echo/click.secho."""
+
+    def test_step_writes_text(self, capsys):
+        em = _Emitter(json_mode=False)
+        em.step(1, "Hello")
+        captured = capsys.readouterr()
+        assert "1. Hello" in captured.out
+
+    def test_log_writes_text(self, capsys):
+        em = _Emitter(json_mode=False)
+        em.log("my line")
+        captured = capsys.readouterr()
+        assert "my line" in captured.out
+
+    def test_result_succeeded_text(self, capsys):
+        em = _Emitter(json_mode=False)
+        em.result("succeeded", deployment_id="d1", url="https://app.test")
+        captured = capsys.readouterr()
+        lines = [line.strip() for line in captured.out.splitlines() if line.strip()]
+        assert "Deployment successful!" in lines
+        assert "URL: https://app.test" in lines
+
+
+# ---------------------------------------------------------------------------
+# --no-input guard on _create_host_backend_client
+# ---------------------------------------------------------------------------
+
+
+class TestCreateHostBackendClientNoInput:
+    def test_raises_when_no_api_key_and_no_input(self, monkeypatch, tmp_path):
+        import langgraph_cli.deploy as deploy_mod
+
+        monkeypatch.setattr(deploy_mod, "_no_input", True)
+        monkeypatch.delenv("LANGSMITH_API_KEY", raising=False)
+        monkeypatch.delenv("LANGCHAIN_API_KEY", raising=False)
+        monkeypatch.delenv("LANGGRAPH_HOST_API_KEY", raising=False)
+
+        with pytest.raises(click.ClickException, match="API key"):
+            _create_host_backend_client(
+                host_url="https://api.example.com",
+                api_key=None,
+                env_vars={},
+            )
+
+    def test_succeeds_with_api_key_in_env(self, monkeypatch, tmp_path):
+        import langgraph_cli.deploy as deploy_mod
+
+        monkeypatch.setattr(deploy_mod, "_no_input", True)
+        monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test")
+
+        client = _create_host_backend_client(
+            host_url="https://api.example.com",
+            api_key=None,
+            env_vars={},
+        )
+        assert client is not None
+
+
+class TestSmithDashboardBaseUrl:
+    def test_none_returns_default(self):
+        assert _smith_dashboard_base_url(None) == "https://smith.langchain.com"
+
+    def test_empty_returns_default(self):
+        assert _smith_dashboard_base_url("") == "https://smith.langchain.com"
+
+    def test_prod_host_url(self):
+        assert (
+            _smith_dashboard_base_url("https://api.host.langchain.com")
+            == "https://smith.langchain.com"
+        )
+
+    def test_dev_host_url(self):
+        assert (
+            _smith_dashboard_base_url("https://dev.api.host.langchain.com")
+            == "https://dev.smith.langchain.com"
+        )
+
+    def test_eu_host_url(self):
+        assert (
+            _smith_dashboard_base_url("https://eu.api.host.langchain.com")
+            == "https://eu.smith.langchain.com"
+        )
+
+    def test_staging_host_url(self):
+        assert (
+            _smith_dashboard_base_url("https://staging.api.host.langchain.com")
+            == "https://staging.smith.langchain.com"
+        )
+
+    def test_localhost(self):
+        assert (
+            _smith_dashboard_base_url("http://localhost:8080")
+            == "http://localhost:8080"
+        )
+
+    def test_localhost_trailing_slash(self):
+        assert (
+            _smith_dashboard_base_url("http://localhost:8080/")
+            == "http://localhost:8080"
+        )
+
+    def test_127_0_0_1(self):
+        assert (
+            _smith_dashboard_base_url("http://127.0.0.1:3000")
+            == "http://127.0.0.1:3000"
+        )
+
+    def test_unknown_domain_returns_default(self):
+        assert (
+            _smith_dashboard_base_url("https://custom.example.com")
+            == "https://smith.langchain.com"
+        )
@@ -249,17 +249,15 @@ def _messages_delta_reducer(
 ) -> list[AnyMessage]:
    """**Experimental.** Batch reducer for use with `DeltaChannel`.

-    Processes all writes in one pass — dedup by ID, `RemoveMessage`
-    tombstoning — without calling `add_messages`.
+    Provides full `add_messages` parity: dedup by ID, `RemoveMessage`
+    tombstoning, `REMOVE_ALL_MESSAGES` reset, `BaseMessageChunk` coercion,
+    and UUID assignment for ID-less messages — all in a single batched pass.

    This reducer is batching-invariant, as required by `DeltaChannel`:
    `reducer(reducer(state, xs), ys) == reducer(state, xs + ys)`.

    Raw dict / string / tuple inputs are coerced to typed `BaseMessage`
-    objects so that HTTP-driven graphs work without a separate coercion
-    step. This is not full `add_messages` parity — `REMOVE_ALL_MESSAGES`,
-    unknown-id `RemoveMessage` errors, missing-id UUID assignment, and
-    `BaseMessageChunk` conversion are not handled here.
+    objects so that HTTP-driven graphs work without a separate coercion step.

    Example::

@@ -280,24 +278,51 @@ def _messages_delta_reducer(
            flat.extend(w)
        else:
            flat.append(w)
-    # Steady state: the reducer's own output is already typed, so skip
-    # `convert_to_messages` on state when the first element is a BaseMessage.
+    # Steady state: the reducer's own output is already typed BaseMessages
+    # (never chunks), so skip convert_to_messages on the fast path.
    # Only raw input (initial dicts, deserialized blobs) hits the slow path.
    if state and isinstance(state[0], BaseMessage):
        state_msgs = state
    else:
-        state_msgs = cast("list[AnyMessage]", convert_to_messages(state))
-    msgs = cast("list[AnyMessage]", convert_to_messages(flat))
+        state_msgs = cast(
+            "list[AnyMessage]",
+            [
+                message_chunk_to_message(cast(BaseMessageChunk, m))
+                for m in convert_to_messages(state)
+            ],
+        )
+    # Coerce chunks to full messages — streaming nodes can emit BaseMessageChunk.
+    msgs = cast(
+        "list[AnyMessage]",
+        [
+            message_chunk_to_message(cast(BaseMessageChunk, m))
+            for m in convert_to_messages(flat)
+        ],
+    )

-    index: dict[str, int] = {
-        m.id: i for i, m in enumerate(state_msgs) if m.id is not None
-    }
+    # REMOVE_ALL_MESSAGES resets everything; find the last sentinel and
+    # discard all state plus all writes before it.
+    remove_all_idx = None
+    for idx, m in enumerate(msgs):
+        if isinstance(m, RemoveMessage) and m.id == REMOVE_ALL_MESSAGES:
+            remove_all_idx = idx
+    if remove_all_idx is not None:
+        state_msgs = []
+        msgs = msgs[remove_all_idx + 1 :]
+
+    # Build index and assign missing IDs in one pass (parity with add_messages
+    # so that eviction and RemoveMessage tombstoning work on ID-less messages).
+    index: dict[str, int] = {}
+    for i, m in enumerate(state_msgs):
+        if m.id is None:
+            m.id = str(uuid.uuid4())
+        index[m.id] = i
    result: list[AnyMessage | None] = list(state_msgs)
    for msg in msgs:
+        if msg.id is None:
+            msg.id = str(uuid.uuid4())
        mid = msg.id
-        if mid is None:
-            result.append(msg)
-        elif isinstance(msg, RemoveMessage):
+        if isinstance(msg, RemoveMessage):
            if mid in index:
                result[index[mid]] = None
                del index[mid]
@@ -3,7 +3,12 @@ from collections.abc import Sequence
 from typing import Annotated

 import pytest
-from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
+from langchain_core.messages import (
+    AIMessage,
+    AIMessageChunk,
+    HumanMessage,
+    RemoveMessage,
+)
 from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import NotRequired, TypedDict
@@ -16,7 +21,7 @@ from langgraph.channels.topic import Topic
 from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.errors import EmptyChannelError, InvalidUpdateError
 from langgraph.graph import START, StateGraph
-from langgraph.graph.message import _messages_delta_reducer
+from langgraph.graph.message import REMOVE_ALL_MESSAGES, _messages_delta_reducer
 from langgraph.graph.state import _get_channel
 from langgraph.types import Overwrite

@@ -284,6 +289,61 @@ def test_messages_delta_reducer_tuple_write_is_one_message() -> None:
    assert result[0].content == "hi"


+def test_messages_delta_reducer_assigns_uuid_to_id_less_messages() -> None:
+    """Messages without IDs get UUIDs assigned, matching add_messages behavior.
+
+    Without UUID assignment, RemoveMessage tombstoning fails on messages that
+    were created without explicit IDs.
+    """
+    m1 = HumanMessage(content="hi")
+    m2 = AIMessage(content="hello")
+    assert m1.id is None
+    assert m2.id is None
+
+    result = _messages_delta_reducer([], [[m1, m2]])
+    assert len(result) == 2
+    assert result[0].id is not None
+    assert result[1].id is not None
+
+    # RemoveMessage tombstoning must work on the now-assigned IDs.
+    result2 = _messages_delta_reducer(result, [RemoveMessage(id=result[1].id)])
+    assert len(result2) == 1
+    assert result2[0].content == "hi"
+
+
+def test_messages_delta_reducer_remove_all_messages() -> None:
+    """REMOVE_ALL_MESSAGES sentinel clears all state and preceding writes."""
+    state = [HumanMessage(content="old", id="h1"), AIMessage(content="prior", id="a1")]
+
+    # Sentinel mid-batch: everything before it (including state) is discarded.
+    result = _messages_delta_reducer(
+        state,
+        [
+            [
+                RemoveMessage(id=REMOVE_ALL_MESSAGES),
+                HumanMessage(content="fresh", id="h2"),
+            ]
+        ],
+    )
+    assert len(result) == 1
+    assert result[0].content == "fresh"
+
+    # Batching-invariant: split across two calls must equal one combined call.
+    step1 = _messages_delta_reducer(state, [[RemoveMessage(id=REMOVE_ALL_MESSAGES)]])
+    step2 = _messages_delta_reducer(step1, [[HumanMessage(content="fresh", id="h2")]])
+    assert step2 == result
+
+
+def test_messages_delta_reducer_coerces_message_chunks() -> None:
+    """BaseMessageChunk writes are coerced to full messages."""
+    chunk = AIMessageChunk(content="hello", id="a1")
+    result = _messages_delta_reducer([], [[chunk]])
+    assert len(result) == 1
+    assert not isinstance(result[0], AIMessageChunk)
+    assert result[0].content == "hello"
+    assert result[0].id == "a1"
+
+
 def test_delta_channel_checkpoint_returns_missing() -> None:
    """checkpoint() always returns MISSING regardless of state.

@@ -1246,7 +1246,7 @@ wheels = [

 [[package]]
 name = "jupyter-server"
-version = "2.17.0"
+version = "2.18.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
    { name = "anyio" },
@@ -1269,9 +1269,9 @@ dependencies = [
    { name = "traitlets" },
    { name = "websocket-client" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/5b/ac/e040ec363d7b6b1f11304cc9f209dac4517ece5d5e01821366b924a64a50/jupyter_server-2.17.0.tar.gz", hash = "sha256:c38ea898566964c888b4772ae1ed58eca84592e88251d2cfc4d171f81f7e99d5", size = 731949, upload-time = "2025-08-21T14:42:54.042Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/f1/ec/9302cec1ccacdd33c1b1312ac31681c8975cae56c626d783ab49edf9c681/jupyter_server-2.18.0.tar.gz", hash = "sha256:568b27bce4320a53c3eebf1bdcbee9acf48a8ab7f66ec83d900ca9909d4fb770", size = 751152, upload-time = "2026-05-04T13:39:29.685Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/80/a24767e6ca280f5a49525d987bf3e4d7552bf67c8be07e8ccf20271f8568/jupyter_server-2.17.0-py3-none-any.whl", hash = "sha256:e8cb9c7db4251f51ed307e329b81b72ccf2056ff82d50524debde1ee1870e13f", size = 388221, upload-time = "2025-08-21T14:42:52.034Z" },
+    { url = "https://files.pythonhosted.org/packages/cf/f9/050312d92072ddb9ce14c11171804c07435790c98d4350935a780d9e10c2/jupyter_server-2.18.0-py3-none-any.whl", hash = "sha256:69a5397a039d689da81a45955f9b23e95ee167f6d8a8d64372fb616f2aac650a", size = 391687, upload-time = "2026-05-04T13:39:27.549Z" },
 ]

 [[package]]