feat(cli): add native TypeScript terminal UI

Rebuild the terminal client on OpenTUI while keeping the Python gateway as the single agent, session, tool, and memory runtime. Preserve a classic prompt fallback and publish version-matched native sidecars for supported platforms.

Co-authored-by: Bingxi Zhao <150592536+pancacake@users.noreply.github.com>
Co-authored-by: chengyongru <2755839590@qq.com>

nanobot/cli/agent.py

@@ -47,7 +47,7 @@ console = Console()
 
 
 def agent(
-    message: str = typer.Option(None, "--message", "-m", help="Message to send to the agent"),
+    message: str | None = typer.Option(None, "--message", "-m", help="Message to send to the agent"),
     session_id: str = typer.Option("cli:direct", "--session", "-s", help="Session ID"),
     workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
     config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
@@ -61,6 +61,12 @@ def agent(
         "--logs/--no-logs",
         help="Show nanobot runtime logs during chat",
     ),
+    classic: bool = typer.Option(
+        False,
+        "--classic",
+        "--no-tui",
+        help="Use the classic Python prompt instead of the native terminal UI",
+    ),
 ):
     """Interact with the agent directly."""
     from nanobot.bus.queue import MessageBus
@@ -69,6 +75,33 @@ def agent(
     from nanobot.providers.image_generation import image_gen_provider_configs
 
     runtime_config = _load_runtime_config(config, workspace)
+    native_tui = (
+        message is None
+        and not classic
+        and markdown
+        and not logs
+        and sys.stdin.isatty()
+        and sys.stdout.isatty()
+    )
+    if native_tui:
+        from nanobot.cli.tui_launcher import TuiUnavailableError, launch_tui
+        from nanobot.config.loader import get_config_path
+
+        try:
+            exit_code = launch_tui(
+                runtime_config,
+                config_path=get_config_path().resolve(strict=False),
+                workspace_override=workspace,
+                session_id=session_id,
+            )
+        except TuiUnavailableError as exc:
+            console.print(f"[yellow]Native TUI unavailable: {exc}[/yellow]")
+            console.print("[dim]Falling back to the classic prompt.[/dim]")
+        else:
+            if exit_code:
+                raise typer.Exit(exit_code)
+            return
+
     try:
         provider = make_provider(runtime_config)
     except ValueError as exc:

nanobot/cli/tui_launcher.py

@@ -0,0 +1,277 @@
+"""Launch the TypeScript terminal client against the local gateway."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import platform
+import shutil
+import subprocess
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, cast
+
+from nanobot import __version__
+from nanobot.cli.runtime_config import _model_display
+from nanobot.cli.webui_support import (
+    _gateway_health_ready,
+    _webui_browser_url,
+    _webui_endpoint_reachable,
+    webui_bootstrap_secret,
+)
+from nanobot.config.paths import get_data_dir
+from nanobot.config.schema import Config
+
+
+class TuiUnavailableError(RuntimeError):
+    """Raised when the native TypeScript TUI cannot run on this installation."""
+
+
+@dataclass(frozen=True)
+class _GatewayLease:
+    runtime: Any
+    owned: bool
+    base_url: str
+
+    def close(self) -> None:
+        if self.owned:
+            self.runtime.stop(timeout_s=20)
+
+
+def launch_tui(
+    config: Config,
+    *,
+    config_path: Path,
+    workspace_override: str | None,
+    session_id: str,
+) -> int:
+    """Run the native TUI, owning a local gateway only when one is not running."""
+    command = _resolve_tui_command()
+    lease = _ensure_gateway(
+        config,
+        config_path=config_path,
+        workspace_override=workspace_override,
+    )
+    try:
+        bootstrap = _fetch_bootstrap(
+            lease.base_url,
+            secret=webui_bootstrap_secret(config),
+        )
+        env = os.environ.copy()
+        env.update(
+            {
+                "NANOBOT_TUI_WS_URL": _authenticated_ws_url(bootstrap),
+                "NANOBOT_TUI_API_URL": lease.base_url,
+                "NANOBOT_TUI_API_TOKEN": str(bootstrap.get("api_token") or ""),
+                "NANOBOT_TUI_MODEL": _model_display(config)[0],
+                "NANOBOT_TUI_WORKSPACE": str(config.workspace_path),
+                "NANOBOT_TUI_VERSION": __version__,
+                "NANOBOT_TUI_ACCESS": (
+                    "workspace access" if config.tools.restrict_to_workspace else "full access"
+                ),
+            }
+        )
+        chat_id = _websocket_chat_id(session_id)
+        if chat_id:
+            env["NANOBOT_TUI_CHAT_ID"] = chat_id
+        else:
+            env.pop("NANOBOT_TUI_CHAT_ID", None)
+        try:
+            return subprocess.run(command, env=env, check=False).returncode
+        except OSError as exc:
+            raise TuiUnavailableError(f"could not start the native TUI: {exc}") from exc
+    finally:
+        lease.close()
+
+
+def _resolve_tui_command() -> list[str]:
+    override = os.environ.get("NANOBOT_TUI_BIN", "").strip()
+    if override:
+        executable = Path(override).expanduser().resolve(strict=False)
+        if not executable.is_file():
+            raise TuiUnavailableError(f"NANOBOT_TUI_BIN does not exist: {executable}")
+        return [str(executable)]
+
+    suffix = ".exe" if os.name == "nt" else ""
+    system = {"Windows": "win32", "Darwin": "darwin", "Linux": "linux"}.get(
+        platform.system(),
+        platform.system().lower(),
+    )
+    machine = {"x86_64": "x64", "AMD64": "x64", "aarch64": "arm64"}.get(
+        platform.machine(),
+        platform.machine().lower(),
+    )
+    asset = f"nanobot-tui-{system}-{machine}{suffix}"
+    packaged = Path(__file__).resolve().parents[1] / "tui" / "bin" / asset
+    if packaged.is_file():
+        return [str(packaged)]
+
+    source_dir = Path(__file__).resolve().parents[2] / "tui"
+    bun = shutil.which("bun")
+    if bun and (source_dir / "package.json").is_file():
+        if not (source_dir / "node_modules" / "@opentui" / "core").is_dir():
+            raise TuiUnavailableError(
+                f"TUI dependencies are missing; run `bun install --cwd {source_dir}`"
+            )
+        return [bun, str(source_dir / "src" / "index.ts")]
+
+    downloaded = _download_release_tui(asset)
+    if downloaded is not None:
+        return [str(downloaded)]
+
+    raise TuiUnavailableError(
+        "this build does not include the native TUI; install Bun for a source checkout "
+        "or use `nanobot agent --classic`"
+    )
+
+
+def _download_release_tui(asset: str) -> Path | None:
+    """Install the version-matched release sidecar into nanobot's data directory."""
+    if os.environ.get("NANOBOT_TUI_NO_DOWNLOAD") == "1":
+        return None
+    version = __version__.strip()
+    if not version or version.endswith((".dev0", "+dev")):
+        return None
+
+    target_dir = get_data_dir() / "bin" / "tui" / version
+    target = target_dir / asset
+    if target.is_file():
+        return target
+
+    base = f"https://github.com/HKUDS/nanobot/releases/download/v{version}"
+    try:
+        checksum = _read_release_asset(f"{base}/{asset}.sha256", max_bytes=1024).decode()
+        expected = checksum.split()[0].lower()
+        if len(expected) != 64:
+            return None
+        binary = _read_release_asset(f"{base}/{asset}", max_bytes=150 * 1024 * 1024)
+    except (OSError, TimeoutError, urllib.error.URLError, urllib.error.HTTPError):
+        return None
+    if hashlib.sha256(binary).hexdigest() != expected:
+        raise TuiUnavailableError("downloaded TUI binary failed checksum verification")
+
+    temporary = target.with_suffix(f"{target.suffix}.tmp-{os.getpid()}")
+    try:
+        target_dir.mkdir(parents=True, exist_ok=True)
+        temporary.write_bytes(binary)
+        if os.name != "nt":
+            temporary.chmod(0o755)
+        temporary.replace(target)
+    except OSError:
+        temporary.unlink(missing_ok=True)
+        return None
+    return target
+
+
+def _read_release_asset(url: str, *, max_bytes: int) -> bytes:
+    request = urllib.request.Request(url, headers={"User-Agent": f"nanobot/{__version__}"})
+    with urllib.request.urlopen(request, timeout=5) as response:
+        content_length = response.headers.get("Content-Length")
+        if content_length and int(content_length) > max_bytes:
+            raise OSError("release asset exceeds size limit")
+        body = response.read(max_bytes + 1)
+    if len(body) > max_bytes:
+        raise OSError("release asset exceeds size limit")
+    return body
+
+
+def _ensure_gateway(
+    config: Config,
+    *,
+    config_path: Path,
+    workspace_override: str | None,
+) -> _GatewayLease:
+    from nanobot.gateway import GatewayRuntime, GatewayRuntimePaths, GatewayStartOptions
+
+    base_url = _webui_browser_url(config).split("/#/", 1)[0].rstrip("/")
+    if _webui_endpoint_reachable(base_url):
+        return _GatewayLease(runtime=None, owned=False, base_url=base_url)
+
+    workspace = (
+        str(Path(workspace_override).expanduser().resolve(strict=False))
+        if workspace_override
+        else None
+    )
+    runtime = GatewayRuntime(
+        paths=GatewayRuntimePaths.for_instance(
+            data_dir=config_path.parent,
+            workspace=workspace,
+            config_path=str(config_path),
+        )
+    )
+    result = runtime.start_background(
+        GatewayStartOptions(
+            port=config.gateway.port,
+            workspace=workspace,
+            config_path=str(config_path),
+        )
+    )
+    owned = result.ok
+    if not result.ok and result.message != "gateway_already_running":
+        raise TuiUnavailableError(
+            f"could not start the local gateway ({result.message}); logs: {result.status.log_path}"
+        )
+
+    deadline = time.monotonic() + 20
+    while time.monotonic() < deadline:
+        if _webui_endpoint_reachable(base_url):
+            return _GatewayLease(runtime=runtime, owned=owned, base_url=base_url)
+        if not runtime.status().running and not _gateway_health_ready(
+            config.gateway.host,
+            config.gateway.port,
+        ):
+            break
+        time.sleep(0.1)
+
+    if owned:
+        runtime.stop(timeout_s=5)
+    raise TuiUnavailableError(
+        f"local gateway did not become ready; logs: {result.status.log_path}"
+    )
+
+
+def _fetch_bootstrap(base_url: str, *, secret: str) -> dict[str, Any]:
+    headers = {"X-Nanobot-Auth": secret} if secret else {}
+    request = urllib.request.Request(f"{base_url}/webui/bootstrap", headers=headers)
+    try:
+        with urllib.request.urlopen(request, timeout=5) as response:
+            raw_payload: Any = json.loads(response.read().decode("utf-8"))
+    except (OSError, TimeoutError, urllib.error.URLError, json.JSONDecodeError) as exc:
+        raise TuiUnavailableError(
+            f"could not authenticate with the local gateway: {exc}"
+        ) from exc
+    if not isinstance(raw_payload, dict):
+        raise TuiUnavailableError("gateway bootstrap response is missing ws_path")
+    payload = cast(dict[str, Any], raw_payload)
+    if not payload.get("ws_path"):
+        raise TuiUnavailableError("gateway bootstrap response is missing ws_path")
+    return payload
+
+
+def _authenticated_ws_url(bootstrap: dict[str, Any]) -> str:
+    raw_url = str(bootstrap.get("ws_url") or "").strip()
+    if not raw_url:
+        raise TuiUnavailableError("gateway bootstrap response is missing ws_url")
+    parsed = urllib.parse.urlsplit(raw_url)
+    query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
+    token = str(bootstrap.get("token") or "").strip()
+    if token:
+        query.append(("token", token))
+    query.append(("client_id", f"tui-{os.getpid()}"))
+    return urllib.parse.urlunsplit(
+        (parsed.scheme, parsed.netloc, parsed.path, urllib.parse.urlencode(query), parsed.fragment)
+    )
+
+
+def _websocket_chat_id(session_id: str) -> str | None:
+    """Map the CLI selector to the WebSocket namespace used by the native TUI."""
+    if session_id.startswith("websocket:"):
+        return session_id.split(":", 1)[1] or None
+    if session_id == "cli:direct":
+        return "tui-direct"
+    return session_id.split(":", 1)[-1] or None

nanobot/cli/webui_support.py

@@ -49,6 +49,7 @@ __all__ = [
     "_validate_gateway_startup",
     "_warn_webui_bind_scope",
     "_webui_browser_url",
+    "webui_bootstrap_secret",
     "_webui_build_mode_for_interactive",
     "_webui_channel_enabled",
     "_webui_display_url",
@@ -224,7 +225,8 @@ def _gateway_health_bind_note(host: str) -> str:
     return "" if is_loopback_host(host) else f" [dim](listening on {host})[/dim]"
 
 
-def _webui_bootstrap_secret(config: Config) -> str:
+def webui_bootstrap_secret(config: Config) -> str:
+    """Return the shared local bootstrap credential for WebUI protocol clients."""
     ws_cfg = _webui_config_dict(config)
     return str(ws_cfg.get("tokenIssueSecret") or ws_cfg.get("token") or "").strip()
 
@@ -236,7 +238,7 @@ def _webui_browser_url(config: Config) -> str:
     host = _host_for_local_browser(str(ws_cfg.get("host") or "127.0.0.1"))
     port = int(ws_cfg.get("port") or 8765)
     base_url = f"http://{host}:{port}"
-    secret = _webui_bootstrap_secret(config)
+    secret = webui_bootstrap_secret(config)
     if not secret:
         return base_url
     return f"{base_url}/#/?bootstrapSecret={quote(secret, safe='')}"