perf(tui): reduce cold-start latency

nanobot/__main__.py

@@ -2,7 +2,7 @@
 Entry point for running nanobot as a module: python -m nanobot
 """
 
-from nanobot.cli.commands import app
+from nanobot.cli.entry import main
 
 if __name__ == "__main__":
-    app()
+    main()

nanobot/cli/agent.py

@@ -1,6 +1,7 @@
 """Direct and interactive agent CLI command."""
 
 import asyncio
+import importlib
 import signal
 import sys
 from collections.abc import Awaitable, Callable
@@ -11,17 +12,6 @@ import typer
 from rich.console import Console
 
 from nanobot import __logo__
-from nanobot.agent.hooks import create_file_edit_activity_hook
-from nanobot.agent.loop import AgentLoop
-from nanobot.agent.tools.mcp import MCPProvider
-from nanobot.agent.tools.registry import ToolRegistry
-from nanobot.bus.outbound_events import (
-    StreamDeltaEvent,
-    StreamedResponseEvent,
-    StreamEndEvent,
-    outbound_event_from_message,
-)
-from nanobot.cli import terminal as cli_terminal
 from nanobot.cli.log_control import _set_nanobot_logs
 from nanobot.cli.runtime_config import (
     _load_runtime_config,
@@ -29,22 +19,37 @@ from nanobot.cli.runtime_config import (
     _model_display,
     _print_agent_start_error,
 )
-from nanobot.cli.stream import StreamRenderer, ThinkingSpinner
-from nanobot.config.paths import is_default_workspace
-from nanobot.utils.helpers import (
-    sanitize_surrogates as _sanitize_surrogates,
-)
-from nanobot.utils.helpers import (
-    sync_workspace_templates,
-)
-from nanobot.utils.restart import (
-    consume_restart_notice_from_env,
-    format_restart_completed_message,
-    should_show_cli_restart_notice,
-)
 
 console = Console()
 
+_CLASSIC_DEPENDENCIES = {
+    "AgentLoop": ("nanobot.agent.loop", "AgentLoop"),
+    "StreamRenderer": ("nanobot.cli.stream", "StreamRenderer"),
+    "consume_restart_notice_from_env": (
+        "nanobot.utils.restart",
+        "consume_restart_notice_from_env",
+    ),
+    "is_default_workspace": ("nanobot.config.paths", "is_default_workspace"),
+    "sync_workspace_templates": ("nanobot.utils.helpers", "sync_workspace_templates"),
+}
+
+
+def __getattr__(name: str) -> Any:
+    """Preserve patchable classic-agent symbols without loading them for the TUI."""
+    dependency = _CLASSIC_DEPENDENCIES.get(name)
+    if dependency is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    module_name, attribute = dependency
+    value = getattr(importlib.import_module(module_name), attribute)
+    globals()[name] = value
+    return value
+
+
+def _classic_dependency(name: str) -> Any:
+    if name in globals():
+        return globals()[name]
+    return __getattr__(name)
+
 
 def agent(
     message: str | None = typer.Option(None, "--message", "-m", help="Message to send to the agent"),
@@ -74,11 +79,6 @@ def agent(
     ),
 ):
     """Chat in the terminal or send one message non-interactively."""
-    from nanobot.bus.queue import MessageBus
-    from nanobot.cron.service import CronService
-    from nanobot.providers.factory import make_provider
-    from nanobot.providers.image_generation import image_gen_provider_configs
-
     runtime_config = _load_runtime_config(config, workspace)
     theme = theme.strip().lower()
     if theme not in {"auto", "dark", "light"}:
@@ -117,6 +117,33 @@ def agent(
                 raise typer.Exit(exit_code)
             return
 
+    from nanobot.agent.hooks import create_file_edit_activity_hook
+    from nanobot.agent.tools.mcp import MCPProvider
+    from nanobot.agent.tools.registry import ToolRegistry
+    from nanobot.bus.outbound_events import (
+        StreamDeltaEvent,
+        StreamedResponseEvent,
+        StreamEndEvent,
+        outbound_event_from_message,
+    )
+    from nanobot.bus.queue import MessageBus
+    from nanobot.cli import terminal as cli_terminal
+    from nanobot.cli.stream import ThinkingSpinner
+    from nanobot.cron.service import CronService
+    from nanobot.providers.factory import make_provider
+    from nanobot.providers.image_generation import image_gen_provider_configs
+    from nanobot.utils.helpers import sanitize_surrogates as _sanitize_surrogates
+    from nanobot.utils.restart import (
+        format_restart_completed_message,
+        should_show_cli_restart_notice,
+    )
+
+    agent_loop_class = _classic_dependency("AgentLoop")
+    stream_renderer_class = _classic_dependency("StreamRenderer")
+    consume_restart_notice_from_env = _classic_dependency("consume_restart_notice_from_env")
+    is_default_workspace = _classic_dependency("is_default_workspace")
+    sync_workspace_templates = _classic_dependency("sync_workspace_templates")
+
     session_id = session_id or "cli:direct"
 
     try:
@@ -142,7 +169,7 @@ def agent(
     _set_nanobot_logs(logs)
 
     try:
-        agent_loop = AgentLoop.from_config(
+        agent_loop = agent_loop_class.from_config(
             runtime_config,
             bus,
             provider=provider,
@@ -171,7 +198,7 @@ def agent(
     _thinking: ThinkingSpinner | None = None
 
     def _make_progress(
-        renderer: StreamRenderer | None = None,
+        renderer: Any | None = None,
     ) -> Callable[..., Awaitable[None]]:
         reasoning_buffer = cli_terminal._ReasoningBuffer()
 
@@ -212,7 +239,7 @@ def agent(
         async def run_once() -> None:
             try:
                 await mcp_provider.connect()
-                renderer = StreamRenderer(
+                renderer = stream_renderer_class(
                     render_markdown=markdown,
                     bot_name=runtime_config.agents.defaults.bot_name,
                     bot_icon=runtime_config.agents.defaults.bot_icon,
@@ -278,7 +305,7 @@ def agent(
             turn_done = asyncio.Event()
             turn_done.set()
             turn_response: list[Any] = []
-            renderer: StreamRenderer | None = None
+            renderer: Any | None = None
             reasoning_buffer = cli_terminal._ReasoningBuffer()
 
             async def _consume_outbound() -> None:
@@ -361,7 +388,7 @@ def agent(
                         turn_done.clear()
                         turn_response.clear()
                         reasoning_buffer.clear()
-                        renderer = StreamRenderer(
+                        renderer = stream_renderer_class(
                             render_markdown=markdown,
                             bot_name=runtime_config.agents.defaults.bot_name,
                             bot_icon=runtime_config.agents.defaults.bot_icon,

nanobot/cli/entry.py

@@ -0,0 +1,51 @@
+"""Low-overhead console entrypoint for the native terminal client."""
+
+from __future__ import annotations
+
+import os
+import sys
+from contextlib import suppress
+
+
+def _native_tui_candidate(args: list[str]) -> bool:
+    """Return whether ``agent`` can start without the classic agent stack."""
+    if not args or args[0] != "agent":
+        return False
+    for argument in args[1:]:
+        if argument in {"--classic", "--no-tui", "-m", "--message"}:
+            return False
+        if argument.startswith("--message=") or (
+            argument.startswith("-m") and not argument.startswith("--")
+        ):
+            return False
+    return True
+
+
+def _configure_windows_console() -> None:
+    if sys.platform != "win32" or sys.stdout.encoding == "utf-8":
+        return
+    os.environ["PYTHONIOENCODING"] = "utf-8"
+    with suppress(Exception):
+        for stream in (sys.stdout, sys.stderr):
+            reconfigure = getattr(stream, "reconfigure", None)
+            if callable(reconfigure):
+                reconfigure(encoding="utf-8", errors="replace")
+
+
+def main() -> None:
+    """Dispatch native TUI startup without importing the complete CLI graph."""
+    _configure_windows_console()
+    if _native_tui_candidate(sys.argv[1:]):
+        import typer
+
+        from nanobot.cli.agent import agent
+
+        fast_app = typer.Typer(add_completion=False)
+        fast_app.command()(agent)
+        command = typer.main.get_command(fast_app)
+        command.main(args=sys.argv[2:], prog_name="nanobot agent")
+        return
+
+    from nanobot.cli.commands import app
+
+    app()

nanobot/cli/tui_launcher.py

@@ -11,7 +11,6 @@ import shutil
 import subprocess
 import time
 import urllib.error
-import urllib.parse
 import urllib.request
 import zipfile
 from dataclasses import dataclass
@@ -22,9 +21,8 @@ from nanobot import __version__
 from nanobot.cli.runtime_config import _model_display
 from nanobot.cli.webui_support import (
     _gateway_health_ready,
-    _webui_browser_url,
+    _host_for_local_browser,
     _webui_endpoint_reachable,
-    webui_bootstrap_secret,
 )
 from nanobot.config.paths import get_data_dir
 from nanobot.config.schema import Config
@@ -82,22 +80,17 @@ def launch_tui(
     state_path = config_path.parent / "tui" / "state.json"
     chat_id = _initial_tui_chat_id(session_id, state_path)
     command = _resolve_tui_command()
-    gateway = _ensure_gateway(
-        config,
-        config_path=config_path,
-        workspace_override=workspace_override,
-    )
+    base_url, bootstrap_secret = _tui_gateway_connection(config)
+    gateway: _GatewayHandle | None = None
+    process: subprocess.Popen[Any] | None = None
     try:
-        bootstrap = _fetch_bootstrap(
-            gateway.base_url,
-            secret=webui_bootstrap_secret(config),
-        )
         env = os.environ.copy()
+        env.pop("NANOBOT_TUI_WS_URL", None)
+        env.pop("NANOBOT_TUI_API_TOKEN", None)
         env.update(
             {
-                "NANOBOT_TUI_WS_URL": _authenticated_ws_url(bootstrap),
-                "NANOBOT_TUI_API_URL": gateway.base_url,
-                "NANOBOT_TUI_API_TOKEN": str(bootstrap.get("api_token") or ""),
+                "NANOBOT_TUI_BOOTSTRAP_URL": f"{base_url}/webui/bootstrap",
+                "NANOBOT_TUI_API_URL": base_url,
                 "NANOBOT_TUI_MODEL": _model_display(config)[0],
                 "NANOBOT_TUI_MODEL_PRESET": config.agents.defaults.model_preset or "default",
                 "NANOBOT_TUI_WORKSPACE": str(config.workspace_path),
@@ -108,18 +101,42 @@ def launch_tui(
                 "NANOBOT_TUI_THEME": theme,
             }
         )
+        if bootstrap_secret:
+            env["NANOBOT_TUI_BOOTSTRAP_SECRET"] = bootstrap_secret
+        else:
+            env.pop("NANOBOT_TUI_BOOTSTRAP_SECRET", None)
         env["NANOBOT_TUI_STATE_PATH"] = str(state_path)
         if chat_id:
             env["NANOBOT_TUI_CHAT_ID"] = chat_id
         else:
             env.pop("NANOBOT_TUI_CHAT_ID", None)
-        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
+        try:
+            process = subprocess.Popen(command, env=env)
+        except OSError as exc:
+            raise TuiUnavailableError(f"could not start the native TUI: {exc}") from exc
+        gateway = _ensure_gateway(
+            config,
+            config_path=config_path,
+            workspace_override=workspace_override,
+            wait_until_ready=False,
+        )
+        return process.wait()
+    except BaseException:
+        if process is not None and process.poll() is None:
+            process.terminate()
+            try:
+                process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+        raise
     finally:
-        lease = getattr(gateway, "lease", None)
+        lease = getattr(gateway, "lease", None) if gateway is not None else None
         if lease is not None:
-            lease.release()
+            # Returning to the shell must not wait for process termination. The
+            # gateway's client monitor observes the released last lease and owns
+            # the orderly on-demand shutdown.
+            lease.release(wait_for_stop=False)
 
 
 def _resolve_tui_command() -> list[str]:
@@ -364,6 +381,7 @@ def _ensure_gateway(
     *,
     config_path: Path,
     workspace_override: str | None,
+    wait_until_ready: bool = True,
 ) -> _GatewayHandle:
     from nanobot.gateway import (
         GatewayClientLease,
@@ -371,7 +389,7 @@ def _ensure_gateway(
         GatewayRuntime,
     )
 
-    base_url = _webui_browser_url(config).split("/#/", 1)[0].rstrip("/")
+    base_url, _bootstrap_secret = _tui_gateway_connection(config)
     instance = GatewayInstance.resolve(
         config_path=config_path,
         workspace=workspace_override,
@@ -388,7 +406,7 @@ def _ensure_gateway(
                     "the matching gateway instance is running on a different port; "
                     "restart it or use `nanobot agent --classic`"
                 )
-            if endpoint_reachable:
+            if endpoint_reachable or not wait_until_ready:
                 return _GatewayHandle(base_url=base_url, lease=lease)
         elif endpoint_reachable:
             raise TuiUnavailableError(
@@ -405,6 +423,9 @@ def _ensure_gateway(
                 f"logs: {result.status.log_path}"
             )
 
+        if not wait_until_ready:
+            return _GatewayHandle(base_url=base_url, lease=lease)
+
         deadline = time.monotonic() + 20
         while time.monotonic() < deadline:
             if _webui_endpoint_reachable(base_url):
@@ -427,37 +448,22 @@ def _ensure_gateway(
         raise
 
 
-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)
+def _tui_gateway_connection(config: Config) -> tuple[str, str]:
+    """Read the small bootstrap subset without importing the WebSocket runtime."""
+    raw: object = getattr(config.channels, "websocket", None)
+    settings = cast(dict[str, Any], raw) if isinstance(raw, dict) else {}
+    host = _host_for_local_browser(str(settings.get("host") or "127.0.0.1"))
     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)
-    )
+        port = int(settings.get("port") or 8765)
+    except (TypeError, ValueError):
+        port = 8765
+    secret = str(
+        settings.get("tokenIssueSecret")
+        or settings.get("token_issue_secret")
+        or settings.get("token")
+        or ""
+    ).strip()
+    return f"http://{host}:{port}", secret
 
 
 def _websocket_chat_id(session_id: str) -> str | None:

nanobot/gateway/runtime.py

@@ -466,8 +466,8 @@ class GatewayClientLease:
             self._write_state(state)
             return True
 
-    def release(self, *, timeout_s: int = 20) -> bool:
-        """Release this client and stop an ephemeral gateway when it was the last."""
+    def release(self, *, timeout_s: int = 20, wait_for_stop: bool = True) -> bool:
+        """Release this client, optionally leaving last-client shutdown to the monitor."""
         if not self._acquired:
             return False
         while True:
@@ -482,7 +482,7 @@ class GatewayClientLease:
                     self._acquired = False
                     should_stop = not clients and bool(state.get("auto_stop"))
                     self._write_or_clear(state)
-                if not should_stop:
+                if not should_stop or not wait_for_stop:
                     return False
                 result = self.runtime._stop(timeout_s=timeout_s)
                 stopped = result.ok or result.message in {