refactor(cli): isolate local agent runtime

README.md

@@ -209,7 +209,7 @@ Use `nanobot gateway --background` for the same direct entry point without keepi
 nanobot agent
 ```
 
-This opens the native terminal client with the same configured model, workspace, tools, streaming protocol, and session engine as the WebUI. Use `/sessions` to switch saved conversations, `/new-chat` to preserve this conversation and start another one, `/branch` to fork from a completed reply, `/context` to inspect the compacted summary and raw message suffix available to the agent, or `/diff` to review the latest turn's file changes. Type `@` to mention an installed app, configured MCP server, or saved session. While nanobot is working, `Enter` steers the current turn, `Tab` queues a visible follow-up for the next turn, and `Option+Up` on macOS (`Alt+Up` on Windows/Linux) returns the latest queued message for editing. Press `Shift+Enter` to add a newline; `Ctrl+J` is the universal fallback for terminals that cannot distinguish modified Enter keys. Use `PageUp` at the top to load earlier transcript pages. The next launch returns to your last session unless `--session` selects another WebSocket session; use `--classic` to resume a session from another channel. The existing nanobot `/new` command keeps its original behavior: it resets the current chat. `nanobot agent` and `nanobot webui` share one on-demand local gateway: either command can start it, each launcher releases only its own client, and the last interactive launcher to exit stops it. Use `nanobot gateway --background` when the gateway must stay alive with no local clients. Type `exit` or press `Ctrl+C` when you are done. Use `nanobot agent --classic` for the legacy Python prompt.
+This opens the native terminal client with the same configured model, workspace, tools, streaming protocol, and session engine as the WebUI. Use `/sessions` to switch saved conversations, `/new-chat` to preserve this conversation and start another one, `/branch` to fork from a completed reply, `/context` to inspect the compacted summary and raw message suffix available to the agent, or `/diff` to review the latest turn's file changes. Type `@` to mention an installed app, configured MCP server, or saved session. While nanobot is working, `Enter` steers the current turn, `Tab` queues a visible follow-up for the next turn, and `Option+Up` on macOS (`Alt+Up` on Windows/Linux) returns the latest queued message for editing. Press `Shift+Enter` to add a newline; `Ctrl+J` is the universal fallback for terminals that cannot distinguish modified Enter keys. Use `PageUp` at the top to load earlier transcript pages. The next launch returns to your last session unless `--session` selects another WebSocket session; use `--classic` to resume a session from another channel. The existing nanobot `/new` command keeps its original behavior: it resets the current chat. `nanobot agent` and `nanobot webui` share one on-demand local gateway: either command can start it, each launcher releases only its own client, and the last interactive launcher to exit stops it. Use `nanobot gateway --background` when the gateway must stay alive with no local clients. Type `exit` or press `Ctrl+C` when you are done. Use `nanobot agent --classic` only when you need the compatibility Python prompt.
 
 For one request and an immediate exit, use:
 

docs/cli-reference.md

@@ -91,7 +91,7 @@ follow the printed WebUI **Settings → Models** or `nanobot onboard --wizard` r
 | `nanobot agent --session <id>` | Use a WebSocket session key; add `--classic` for another channel |
 | `nanobot agent --workspace <path>` | Override workspace |
 | `nanobot agent --config <path>` | Use a specific config file |
-| `nanobot agent --classic` | Use the classic Python prompt instead of the native terminal UI |
+| `nanobot agent --classic` | Use the compatibility Python prompt instead of the native terminal UI |
 | `nanobot agent --theme auto\|dark\|light` | Auto-detect the terminal appearance or force a TUI palette |
 | `nanobot agent --no-markdown` | Use the classic prompt and print plain text instead of Markdown |
 | `nanobot agent --logs` | Use the classic prompt and show runtime logs while chatting |
@@ -127,7 +127,7 @@ The default `--theme auto` mode probes the terminal's real foreground and backgr
 
 `Enter` sends the current message. While a turn is active, `Enter` steers it immediately, `Tab` queues a visible follow-up for the next turn, and `Option+Up` on macOS (`Alt+Up` on Windows/Linux) returns the latest queued message to the composer. Press `Shift+Enter` to add a newline; `Ctrl+J` is the universal fallback when a terminal cannot distinguish modified Enter keys. `Alt+Enter` and `Ctrl+Enter` are also accepted when distinguishable. Use `Up`/`Down` at the composer edge to recall prompts from the current saved session. Large pastes appear as a compact placeholder in the composer but are sent unchanged. Type `/` to discover nanobot commands and terminal navigation in one palette, or type `@` to complete installed apps, configured MCP servers, and saved sessions. Use the arrow keys to choose an item and `Tab` to complete it. `/sessions` opens a searchable conversation picker, `/new-chat` preserves the current conversation and starts another one, and `/branch` forks from a completed reply. `/diff` opens a read-only unified diff for the newest turn; use `Left`/`Right` to switch edits and `Esc` to close it. The core `/new` command retains its cross-channel behavior and resets the current chat. `Ctrl+C` copies a selection, stops a running turn, clears a non-empty composer, or exits when idle. Use `PageUp`/`PageDown` to scroll, `Ctrl+Home`/`Ctrl+End` to jump to the transcript edges, and `Ctrl+O` to expand or collapse long tool traces. When you leave the bottom, the TUI shows a scrollbar and a `Ctrl+End` hint until you return. The footer reports provider token/cache usage when available. Selections copy through OSC 52 when the terminal supports it. The transcript reflows when the terminal is resized, and exiting restores the previous screen.
 
-Packaged releases fetch a version-matched, checksummed terminal archive for macOS (Apple Silicon and Intel), Linux (x64 and ARM64), or Windows x64 on first use. The cache keeps the executable together with its licenses, third-party notices, source offer, relinking instructions, and corresponding TUI source. Windows ARM64 currently falls back to the classic prompt because the Bun runtime disables the FFI required by OpenTUI on that platform. Set `NANOBOT_TUI_NO_DOWNLOAD=1` or pass `--classic` to keep the Python-only path. A local source install requires Bun and runs its own `tui/` source while the original checkout remains available; it never silently falls back to a release binary.
+Packaged releases fetch a version-matched, checksummed terminal archive for macOS (Apple Silicon and Intel), Linux (x64 and ARM64), or Windows x64 on first use. The cache keeps the executable together with its licenses, third-party notices, source offer, relinking instructions, and corresponding TUI source. Windows ARM64 must currently use `--classic` because the Bun runtime disables the FFI required by OpenTUI on that platform. Set `NANOBOT_TUI_NO_DOWNLOAD=1` or pass `--classic` to keep the Python-only path. A local source install requires Bun and runs its own `tui/` source while the original checkout remains available; it never silently falls back to a release binary.
 
 Non-interactive input/output, `--logs`, and `--no-markdown` automatically retain the classic prompt so existing scripts and diagnostic workflows do not acquire terminal control sequences or silently ignore their options.
 

nanobot/cli/agent.py

@@ -1,47 +1,11 @@
-"""Direct and interactive agent CLI command."""
+"""Agent CLI command."""
 
-import asyncio
-import signal
 import sys
-from collections.abc import Awaitable, Callable
-from types import FrameType
-from typing import Any
 
 import typer
 from rich.console import Console
 
-from nanobot import __logo__
+from nanobot.cli.runtime_config import _load_runtime_config
-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,
-    _migrate_cron_store,
-    _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()
 
@@ -64,34 +28,28 @@ def agent(
     classic: bool = typer.Option(
         False,
         "--classic",
-        "--no-tui",
+        help="Use the compatibility Python prompt instead of the terminal UI",
-        help="Use the classic Python prompt instead of the native terminal UI",
     ),
     theme: str = typer.Option(
         "auto",
         "--theme",
-        help="Terminal UI appearance: auto, dark, or light",
+        help="Native terminal UI appearance: auto, dark, or light",
     ),
-):
+) -> None:
     """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"}:
         raise typer.BadParameter("must be auto, dark, or light", param_hint="--theme")
-    native_tui = message is None and not classic
+
-    if native_tui:
+    if message is None and not classic:
         from nanobot.cli.tui_launcher import TuiSessionError, TuiUnavailableError, launch_tui
         from nanobot.config.loader import get_config_path
 
         if not sys.stdin.isatty() or not sys.stdout.isatty():
             raise typer.BadParameter(
                 "the native TUI requires an interactive terminal; use --message for "
-                "one-shot input or --classic for the legacy prompt",
+                "one-shot input or --classic for the compatibility prompt",
                 param_hint="terminal",
             )
         if not markdown:
@@ -110,308 +68,20 @@ def agent(
             raise typer.BadParameter(str(exc), param_hint="--session") from exc
         except TuiUnavailableError as exc:
             console.print(f"[red]Native TUI unavailable: {exc}[/red]")
-            console.print("[dim]Use `nanobot agent --classic` only if you want the old prompt.[/dim]")
+            console.print(
+                "[dim]Use `nanobot agent --classic` only if you want the compatibility prompt.[/dim]"
+            )
             raise typer.Exit(1) from exc
-        else:
         if exit_code:
             raise typer.Exit(exit_code)
         return
 
-    session_id = session_id or "cli:direct"
+    from nanobot.cli.agent_runtime import run_local_agent
 
-    try:
+    run_local_agent(
-        provider = make_provider(runtime_config)
-    except ValueError as exc:
-        _print_agent_start_error(exc)
-        raise typer.Exit(1) from exc
-
-    sync_workspace_templates(runtime_config.workspace_path)
-
-    bus = MessageBus()
-
-    # Preserve existing single-workspace installs, but keep custom workspaces clean.
-    if is_default_workspace(runtime_config.workspace_path):
-        _migrate_cron_store(runtime_config)
-
-    # Create cron service with workspace-scoped store
-    cron_store_path = runtime_config.workspace_path / "cron" / "jobs.json"
-    cron = CronService(cron_store_path)
-    tools = ToolRegistry()
-    mcp_provider = MCPProvider.from_config(runtime_config, tools)
-
-    _set_nanobot_logs(logs)
-
-    try:
-        agent_loop = AgentLoop.from_config(
         runtime_config,
-            bus,
+        message=message,
-            provider=provider,
+        session_id=session_id or "cli:direct",
-            cron_service=cron,
+        markdown=markdown,
-            image_generation_provider_configs=image_gen_provider_configs(runtime_config),
+        logs=logs,
-            hook_factories=[create_file_edit_activity_hook],
-            tool_registry=tools,
     )
-    except ValueError as exc:
-        _print_agent_start_error(exc)
-        raise typer.Exit(1) from exc
-    restart_notice = consume_restart_notice_from_env()
-    if restart_notice and should_show_cli_restart_notice(restart_notice, session_id):
-        cli_terminal._print_agent_response(
-            format_restart_completed_message(restart_notice.started_at_raw),
-            render_markdown=False,
-        )
-
-    async def _close_runtime() -> None:
-        try:
-            await agent_loop.aclose()
-        finally:
-            await mcp_provider.aclose()
-
-    # Shared reference for progress callbacks
-    _thinking: ThinkingSpinner | None = None
-
-    def _make_progress(
-        renderer: StreamRenderer | None = None,
-    ) -> Callable[..., Awaitable[None]]:
-        reasoning_buffer = cli_terminal._ReasoningBuffer()
-
-        async def _cli_progress(
-            content: str,
-            *,
-            tool_hint: bool = False,
-            reasoning: bool = False,
-            **_kwargs: Any,
-        ) -> None:
-            ch = agent_loop.channels_config
-
-            if _kwargs.get("reasoning_end"):
-                if ch and not ch.show_reasoning:
-                    reasoning_buffer.clear()
-                else:
-                    cli_terminal._flush_cli_reasoning(reasoning_buffer, _thinking, renderer)
-                return
-
-            if reasoning:
-                if ch and not ch.show_reasoning:
-                    reasoning_buffer.clear()
-                    return
-                text = reasoning_buffer.add(content)
-                if text:
-                    cli_terminal._print_cli_reasoning(text, _thinking, renderer)
-                return
-            if ch and tool_hint and not ch.send_tool_hints:
-                return
-            if ch and not tool_hint and not ch.send_progress:
-                return
-            cli_terminal._print_cli_progress_line(content, _thinking, renderer)
-
-        return _cli_progress
-
-    if message is not None:
-        # Single message mode — direct call, no bus needed
-        async def run_once() -> None:
-            try:
-                await mcp_provider.connect()
-                renderer = StreamRenderer(
-                    render_markdown=markdown,
-                    bot_name=runtime_config.agents.defaults.bot_name,
-                    bot_icon=runtime_config.agents.defaults.bot_icon,
-                )
-                response = await agent_loop.process_direct(
-                    message,
-                    session_id,
-                    on_progress=_make_progress(renderer),
-                    on_stream=renderer.on_delta,
-                    on_stream_end=renderer.on_end,
-                )
-                if not renderer.streamed:
-                    await renderer.close()
-                    print_kwargs: dict[str, Any] = {}
-                    if renderer.header_printed:
-                        print_kwargs["show_header"] = False
-                    cli_terminal._print_agent_response(
-                        response.content if response else "",
-                        render_markdown=markdown,
-                        metadata=response.metadata if response else None,
-                        **print_kwargs,
-                    )
-            finally:
-                await _close_runtime()
-
-        asyncio.run(run_once())
-    else:
-        # Interactive mode — route through bus like other channels
-        from nanobot.bus.events import InboundMessage
-
-        cli_terminal._init_prompt_session()
-        _model, _preset_tag = _model_display(runtime_config)
-        _icon = runtime_config.agents.defaults.bot_icon or __logo__
-        console.print(
-            f"{_icon} Interactive mode [bold blue]({_model})[/bold blue]{_preset_tag} "
-            "— type [bold]exit[/bold] or [bold]Ctrl+C[/bold] to quit\n"
-        )
-
-        if ":" in session_id:
-            cli_channel, cli_chat_id = session_id.split(":", 1)
-        else:
-            cli_channel, cli_chat_id = "cli", session_id
-
-        def _handle_signal(signum: int, _frame: FrameType | None) -> None:
-            sig_name = signal.Signals(signum).name
-            cli_terminal._restore_terminal()
-            console.print(f"\nReceived {sig_name}, goodbye!")
-            sys.exit(0)
-
-        signal.signal(signal.SIGINT, _handle_signal)
-        signal.signal(signal.SIGTERM, _handle_signal)
-        # SIGHUP is not available on Windows
-        if hasattr(signal, "SIGHUP"):
-            signal.signal(signal.SIGHUP, _handle_signal)
-        # Ignore SIGPIPE to prevent silent process termination when writing to closed pipes
-        # SIGPIPE is not available on Windows
-        if hasattr(signal, "SIGPIPE"):
-            signal.signal(signal.SIGPIPE, signal.SIG_IGN)
-
-        async def run_interactive() -> None:
-            await mcp_provider.connect()
-            bus_task = asyncio.create_task(agent_loop.run())
-            turn_done = asyncio.Event()
-            turn_done.set()
-            turn_response: list[Any] = []
-            renderer: StreamRenderer | None = None
-            reasoning_buffer = cli_terminal._ReasoningBuffer()
-
-            async def _consume_outbound() -> None:
-                while True:
-                    try:
-                        msg = await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
-                        event = outbound_event_from_message(msg)
-
-                        if isinstance(event, StreamDeltaEvent):
-                            if renderer:
-                                await renderer.on_delta(msg.content)
-                            continue
-                        if isinstance(event, StreamEndEvent):
-                            if renderer:
-                                await renderer.on_end(
-                                    resuming=event.resuming,
-                                )
-                            continue
-                        if isinstance(event, StreamedResponseEvent):
-                            if msg.content and renderer and not renderer.streamed:
-                                await renderer.close()
-                                print_kwargs: dict[str, Any] = {}
-                                if renderer.header_printed:
-                                    print_kwargs["show_header"] = False
-                                cli_terminal._print_agent_response(
-                                    msg.content,
-                                    render_markdown=markdown,
-                                    metadata=msg.metadata,
-                                    **print_kwargs,
-                                )
-                            turn_done.set()
-                            continue
-
-                        if await cli_terminal._maybe_print_interactive_progress(
-                            msg,
-                            None,
-                            agent_loop.channels_config,
-                            renderer,
-                            reasoning_buffer,
-                        ):
-                            continue
-
-                        if not turn_done.is_set():
-                            if msg.content:
-                                turn_response.append(msg)
-                            turn_done.set()
-                        elif msg.content:
-                            await cli_terminal._print_interactive_response(
-                                msg.content,
-                                render_markdown=markdown,
-                                metadata=msg.metadata,
-                            )
-
-                    except asyncio.TimeoutError:
-                        continue
-                    except asyncio.CancelledError:
-                        break
-
-            outbound_task = asyncio.create_task(_consume_outbound())
-
-            try:
-                while True:
-                    try:
-                        cli_terminal._flush_pending_tty_input()
-                        # Stop spinner before user input to avoid prompt_toolkit conflicts
-                        if renderer:
-                            renderer.stop_for_input()
-                        user_input = _sanitize_surrogates(
-                            await cli_terminal._read_interactive_input_async()
-                        )
-                        command = user_input.strip()
-                        if not command:
-                            continue
-
-                        if cli_terminal._is_exit_command(command):
-                            cli_terminal._restore_terminal()
-                            console.print("\nGoodbye!")
-                            break
-
-                        turn_done.clear()
-                        turn_response.clear()
-                        reasoning_buffer.clear()
-                        renderer = StreamRenderer(
-                            render_markdown=markdown,
-                            bot_name=runtime_config.agents.defaults.bot_name,
-                            bot_icon=runtime_config.agents.defaults.bot_icon,
-                        )
-
-                        await bus.publish_inbound(
-                            InboundMessage(
-                                channel=cli_channel,
-                                sender_id="user",
-                                chat_id=cli_chat_id,
-                                content=user_input,
-                                metadata={"_wants_stream": True},
-                            )
-                        )
-
-                        await turn_done.wait()
-
-                        if turn_response:
-                            response_msg = turn_response[0]
-                            content = response_msg.content
-                            meta = response_msg.metadata
-                            if content and not isinstance(
-                                response_msg.event,
-                                StreamedResponseEvent,
-                            ):
-                                if renderer:
-                                    await renderer.close()
-                                print_kwargs: dict[str, Any] = {}
-                                if renderer and renderer.header_printed:
-                                    print_kwargs["show_header"] = False
-                                cli_terminal._print_agent_response(
-                                    content,
-                                    render_markdown=markdown,
-                                    metadata=meta,
-                                    **print_kwargs,
-                                )
-                        elif renderer and not renderer.streamed:
-                            await renderer.close()
-                    except KeyboardInterrupt:
-                        cli_terminal._restore_terminal()
-                        console.print("\nGoodbye!")
-                        break
-                    except EOFError:
-                        cli_terminal._restore_terminal()
-                        console.print("\nGoodbye!")
-                        break
-            finally:
-                agent_loop.stop()
-                outbound_task.cancel()
-                await asyncio.gather(bus_task, outbound_task, return_exceptions=True)
-                await _close_runtime()
-
-        asyncio.run(run_interactive())

nanobot/cli/agent_runtime.py

@@ -0,0 +1,308 @@
+"""Python runtime for one-shot agent calls and the compatibility prompt."""
+
+import asyncio
+import signal
+import sys
+from types import FrameType
+from typing import Any
+
+import typer
+
+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.events import InboundMessage, OutboundMessage
+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.log_control import _set_nanobot_logs
+from nanobot.cli.runtime_config import (
+    _migrate_cron_store,
+    _model_display,
+    _print_agent_start_error,
+)
+from nanobot.cli.stream import StreamRenderer
+from nanobot.config.paths import is_default_workspace
+from nanobot.config.schema import Config
+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, sync_workspace_templates
+from nanobot.utils.restart import (
+    consume_restart_notice_from_env,
+    format_restart_completed_message,
+    should_show_cli_restart_notice,
+)
+
+
+def run_local_agent(
+    config: Config,
+    *,
+    message: str | None,
+    session_id: str,
+    markdown: bool,
+    logs: bool,
+) -> None:
+    """Run without the gateway: once for a message, otherwise as the classic prompt."""
+    runtime = _LocalAgent(config, logs=logs, session_id=session_id)
+    if message is not None:
+        asyncio.run(runtime.run_once(message, session_id=session_id, markdown=markdown))
+    else:
+        runtime.run_classic(session_id=session_id, markdown=markdown)
+
+
+class _LocalAgent:
+    def __init__(self, config: Config, *, logs: bool, session_id: str) -> None:
+        self.config = config
+        try:
+            provider = make_provider(config)
+        except ValueError as exc:
+            _print_agent_start_error(exc)
+            raise typer.Exit(1) from exc
+
+        sync_workspace_templates(config.workspace_path)
+        if is_default_workspace(config.workspace_path):
+            _migrate_cron_store(config)
+
+        self.bus = MessageBus()
+        tools = ToolRegistry()
+        self.mcp = MCPProvider.from_config(config, tools)
+        _set_nanobot_logs(logs)
+        try:
+            self.loop = AgentLoop.from_config(
+                config,
+                self.bus,
+                provider=provider,
+                cron_service=CronService(config.workspace_path / "cron" / "jobs.json"),
+                image_generation_provider_configs=image_gen_provider_configs(config),
+                hook_factories=[create_file_edit_activity_hook],
+                tool_registry=tools,
+            )
+        except ValueError as exc:
+            _print_agent_start_error(exc)
+            raise typer.Exit(1) from exc
+
+        notice = consume_restart_notice_from_env()
+        if notice and should_show_cli_restart_notice(notice, session_id):
+            cli_terminal._print_agent_response(
+                format_restart_completed_message(notice.started_at_raw),
+                render_markdown=False,
+            )
+
+    async def close(self) -> None:
+        try:
+            await self.loop.aclose()
+        finally:
+            await self.mcp.aclose()
+
+    def renderer(self, markdown: bool) -> StreamRenderer:
+        return StreamRenderer(
+            render_markdown=markdown,
+            bot_name=self.config.agents.defaults.bot_name,
+            bot_icon=self.config.agents.defaults.bot_icon,
+        )
+
+    async def run_once(self, message: str, *, session_id: str, markdown: bool) -> None:
+        try:
+            await self.mcp.connect()
+            renderer = self.renderer(markdown)
+            reasoning_buffer = cli_terminal._ReasoningBuffer()
+
+            async def report(
+                content: str,
+                *,
+                tool_hint: bool = False,
+                reasoning: bool = False,
+                **kwargs: Any,
+            ) -> None:
+                channel_config = self.loop.channels_config
+                if kwargs.get("reasoning_end"):
+                    if channel_config and not channel_config.show_reasoning:
+                        reasoning_buffer.clear()
+                    else:
+                        cli_terminal._flush_cli_reasoning(reasoning_buffer, None, renderer)
+                    return
+                if reasoning:
+                    if channel_config and not channel_config.show_reasoning:
+                        reasoning_buffer.clear()
+                        return
+                    text = reasoning_buffer.add(content)
+                    if text:
+                        cli_terminal._print_cli_reasoning(text, None, renderer)
+                    return
+                if channel_config and tool_hint and not channel_config.send_tool_hints:
+                    return
+                if channel_config and not tool_hint and not channel_config.send_progress:
+                    return
+                cli_terminal._print_cli_progress_line(content, None, renderer)
+
+            response = await self.loop.process_direct(
+                message,
+                session_id,
+                on_progress=report,
+                on_stream=renderer.on_delta,
+                on_stream_end=renderer.on_end,
+            )
+            if renderer.streamed:
+                return
+            await renderer.close()
+            cli_terminal._print_agent_response(
+                response.content if response else "",
+                render_markdown=markdown,
+                metadata=response.metadata if response else None,
+                **({"show_header": False} if renderer.header_printed else {}),
+            )
+        finally:
+            await self.close()
+
+    def run_classic(self, *, session_id: str, markdown: bool) -> None:
+        cli_terminal._init_prompt_session()
+        model, preset_tag = _model_display(self.config)
+        icon = self.config.agents.defaults.bot_icon or __logo__
+        cli_terminal.console.print(
+            f"{icon} Interactive mode [bold blue]({model})[/bold blue]{preset_tag} "
+            "— type [bold]exit[/bold] or [bold]Ctrl+C[/bold] to quit\n"
+        )
+        channel, chat_id = (
+            session_id.split(":", 1) if ":" in session_id else ("cli", session_id)
+        )
+        self._install_signal_handlers()
+        asyncio.run(self._run_classic_loop(channel, chat_id, markdown=markdown))
+
+    @staticmethod
+    def _install_signal_handlers() -> None:
+        def exit_on_signal(signum: int, _frame: FrameType | None) -> None:
+            cli_terminal._restore_terminal()
+            cli_terminal.console.print(f"\nReceived {signal.Signals(signum).name}, goodbye!")
+            sys.exit(0)
+
+        signal.signal(signal.SIGINT, exit_on_signal)
+        signal.signal(signal.SIGTERM, exit_on_signal)
+        if hasattr(signal, "SIGHUP"):
+            signal.signal(signal.SIGHUP, exit_on_signal)
+        if hasattr(signal, "SIGPIPE"):
+            signal.signal(signal.SIGPIPE, signal.SIG_IGN)
+
+    async def _run_classic_loop(self, channel: str, chat_id: str, *, markdown: bool) -> None:
+        await self.mcp.connect()
+        bus_task = asyncio.create_task(self.loop.run())
+        turn_done = asyncio.Event()
+        turn_done.set()
+        turn_response: list[OutboundMessage] = []
+        renderer: StreamRenderer | None = None
+        reasoning_buffer = cli_terminal._ReasoningBuffer()
+
+        async def consume_outbound() -> None:
+            while True:
+                try:
+                    msg = await asyncio.wait_for(self.bus.consume_outbound(), timeout=1.0)
+                    event = outbound_event_from_message(msg)
+                    if isinstance(event, StreamDeltaEvent):
+                        if renderer:
+                            await renderer.on_delta(msg.content)
+                        continue
+                    if isinstance(event, StreamEndEvent):
+                        if renderer:
+                            await renderer.on_end(resuming=event.resuming)
+                        continue
+                    if isinstance(event, StreamedResponseEvent):
+                        if msg.content and renderer and not renderer.streamed:
+                            await renderer.close()
+                            cli_terminal._print_agent_response(
+                                msg.content,
+                                render_markdown=markdown,
+                                metadata=msg.metadata,
+                                **({"show_header": False} if renderer.header_printed else {}),
+                            )
+                        turn_done.set()
+                        continue
+                    if await cli_terminal._maybe_print_interactive_progress(
+                        msg,
+                        None,
+                        self.loop.channels_config,
+                        renderer,
+                        reasoning_buffer,
+                    ):
+                        continue
+                    if not turn_done.is_set():
+                        if msg.content:
+                            turn_response.append(msg)
+                        turn_done.set()
+                    elif msg.content:
+                        await cli_terminal._print_interactive_response(
+                            msg.content,
+                            render_markdown=markdown,
+                            metadata=msg.metadata,
+                        )
+                except asyncio.TimeoutError:
+                    continue
+                except asyncio.CancelledError:
+                    break
+
+        outbound_task = asyncio.create_task(consume_outbound())
+        try:
+            while True:
+                try:
+                    cli_terminal._flush_pending_tty_input()
+                    if renderer:
+                        renderer.stop_for_input()
+                    user_input = sanitize_surrogates(
+                        await cli_terminal._read_interactive_input_async()
+                    )
+                    command = user_input.strip()
+                    if not command:
+                        continue
+                    if cli_terminal._is_exit_command(command):
+                        cli_terminal._restore_terminal()
+                        cli_terminal.console.print("\nGoodbye!")
+                        break
+
+                    turn_done.clear()
+                    turn_response.clear()
+                    reasoning_buffer.clear()
+                    renderer = self.renderer(markdown)
+                    await self.bus.publish_inbound(
+                        InboundMessage(
+                            channel=channel,
+                            sender_id="user",
+                            chat_id=chat_id,
+                            content=user_input,
+                            metadata={"_wants_stream": True},
+                        )
+                    )
+                    await turn_done.wait()
+                    if turn_response:
+                        response = turn_response[0]
+                        if response.content and not isinstance(
+                            response.event, StreamedResponseEvent
+                        ):
+                            if renderer:
+                                await renderer.close()
+                            cli_terminal._print_agent_response(
+                                response.content,
+                                render_markdown=markdown,
+                                metadata=response.metadata,
+                                **(
+                                    {"show_header": False}
+                                    if renderer and renderer.header_printed
+                                    else {}
+                                ),
+                            )
+                    elif renderer and not renderer.streamed:
+                        await renderer.close()
+                except (KeyboardInterrupt, EOFError):
+                    cli_terminal._restore_terminal()
+                    cli_terminal.console.print("\nGoodbye!")
+                    break
+        finally:
+            self.loop.stop()
+            outbound_task.cancel()
+            await asyncio.gather(bus_task, outbound_task, return_exceptions=True)
+            await self.close()

tests/cli/test_agent_interactive.py

@@ -103,20 +103,24 @@ def test_interactive_agent_routes_a_complete_user_turn(
     read_input = AsyncMock(side_effect=["hello nanobot", "exit"])
     print_response = MagicMock()
     monkeypatch.setattr("nanobot.cli.agent._load_runtime_config", lambda *_args: config)
-    monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda *_args: None)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.sync_workspace_templates", lambda *_args: None)
-    monkeypatch.setattr("nanobot.cli.agent.is_default_workspace", lambda *_args: False)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.is_default_workspace", lambda *_args: False)
-    monkeypatch.setattr("nanobot.cli.agent._set_nanobot_logs", lambda *_args: None)
+    monkeypatch.setattr("nanobot.cli.agent_runtime._set_nanobot_logs", lambda *_args: None)
-    monkeypatch.setattr("nanobot.cli.agent._model_display", lambda *_args: ("test-model", ""))
-    monkeypatch.setattr("nanobot.cli.agent.consume_restart_notice_from_env", lambda: None)
-    monkeypatch.setattr("nanobot.cli.agent.AgentLoop", _AgentLoop)
-    monkeypatch.setattr("nanobot.cli.agent.StreamRenderer", _Renderer)
-    monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda *_args: object())
     monkeypatch.setattr(
-        "nanobot.providers.image_generation.image_gen_provider_configs",
+        "nanobot.cli.agent_runtime._model_display", lambda *_args: ("test-model", "")
+    )
+    monkeypatch.setattr(
+        "nanobot.cli.agent_runtime.consume_restart_notice_from_env", lambda: None
+    )
+    monkeypatch.setattr("nanobot.cli.agent_runtime.AgentLoop", _AgentLoop)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.StreamRenderer", _Renderer)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.make_provider", lambda *_args: object())
+    monkeypatch.setattr(
+        "nanobot.cli.agent_runtime.image_gen_provider_configs",
         lambda *_args: [],
     )
-    monkeypatch.setattr("nanobot.cron.service.CronService", lambda *_args: object())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.CronService", lambda *_args: object())
-    monkeypatch.setattr("nanobot.cli.agent.signal.signal", lambda *_args: None)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.signal.signal", lambda *_args: None)
     monkeypatch.setattr("nanobot.cli.terminal._init_prompt_session", lambda: None)
     monkeypatch.setattr("nanobot.cli.terminal._flush_pending_tty_input", lambda: None)
     monkeypatch.setattr("nanobot.cli.terminal._restore_terminal", lambda: None)

tests/cli/test_commands.py

@@ -1532,12 +1532,12 @@ def mock_agent_runtime(tmp_path):
 
     with patch("nanobot.config.loader.load_config", return_value=config) as mock_load_config, \
          patch("nanobot.config.loader.resolve_config_env_vars", side_effect=lambda c: c), \
-         patch("nanobot.cli.agent.sync_workspace_templates") as mock_sync_templates, \
+         patch("nanobot.cli.agent_runtime.sync_workspace_templates") as mock_sync_templates, \
-         patch("nanobot.providers.factory.make_provider", return_value=_fake_provider()), \
+         patch("nanobot.cli.agent_runtime.make_provider", return_value=_fake_provider()), \
          patch("nanobot.cli.terminal._print_agent_response") as mock_print_response, \
-         patch("nanobot.bus.queue.MessageBus"), \
+         patch("nanobot.cli.agent_runtime.MessageBus"), \
-         patch("nanobot.cron.service.CronService"), \
+         patch("nanobot.cli.agent_runtime.CronService"), \
-         patch("nanobot.cli.agent.AgentLoop.from_config") as mock_from_config:
+         patch("nanobot.cli.agent_runtime.AgentLoop.from_config") as mock_from_config:
         agent_loop = MagicMock()
         agent_loop.channels_config = None
         agent_loop.process_direct = AsyncMock(
@@ -1566,6 +1566,8 @@ def test_agent_help_shows_workspace_and_config_options():
     assert "--config" in stripped_output
     assert "-c" in stripped_output
     assert "--theme" in stripped_output
+    assert "--classic" in stripped_output
+    assert "--no-tui" not in stripped_output
 
 
 def test_agent_rejects_unknown_tui_theme(mock_agent_runtime):
@@ -1615,10 +1617,10 @@ def test_agent_config_sets_active_path(monkeypatch, tmp_path: Path) -> None:
         lambda path: seen.__setitem__("config_path", path),
     )
     monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
-    monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda _path: None)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.sync_workspace_templates", lambda _path: None)
-    monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.make_provider", lambda _config: _fake_provider())
-    monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.MessageBus", lambda: object())
-    monkeypatch.setattr("nanobot.cron.service.CronService", lambda _store: object())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.CronService", lambda _store: object())
 
     class _FakeAgentLoop:
         @classmethod
@@ -1633,7 +1635,7 @@ def test_agent_config_sets_active_path(monkeypatch, tmp_path: Path) -> None:
         async def aclose(self) -> None:
             return None
 
-    monkeypatch.setattr("nanobot.cli.agent.AgentLoop", _FakeAgentLoop)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.AgentLoop", _FakeAgentLoop)
     monkeypatch.setattr("nanobot.cli.terminal._print_agent_response", lambda *_args, **_kwargs: None)
 
     result = runner.invoke(app, ["agent", "-m", "hello", "-c", str(config_file)])
@@ -1653,9 +1655,9 @@ def test_agent_uses_workspace_directory_for_cron_store(monkeypatch, tmp_path: Pa
 
     monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
     monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
-    monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda _path: None)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.sync_workspace_templates", lambda _path: None)
-    monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.make_provider", lambda _config: _fake_provider())
-    monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.MessageBus", lambda: object())
 
     class _FakeCron:
         def __init__(self, store_path: Path) -> None:
@@ -1674,8 +1676,8 @@ def test_agent_uses_workspace_directory_for_cron_store(monkeypatch, tmp_path: Pa
         async def aclose(self) -> None:
             return None
 
-    monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.CronService", _FakeCron)
-    monkeypatch.setattr("nanobot.cli.agent.AgentLoop", _FakeAgentLoop)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.AgentLoop", _FakeAgentLoop)
     monkeypatch.setattr("nanobot.cli.terminal._print_agent_response", lambda *_args, **_kwargs: None)
 
     result = runner.invoke(app, ["agent", "-m", "hello", "-c", str(config_file)])
@@ -1702,9 +1704,9 @@ def test_agent_workspace_override_does_not_migrate_legacy_cron(
 
     monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
     monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
-    monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda _path: None)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.sync_workspace_templates", lambda _path: None)
-    monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.make_provider", lambda _config: _fake_provider())
-    monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.MessageBus", lambda: object())
     monkeypatch.setattr("nanobot.config.paths.get_cron_dir", lambda: legacy_dir)
 
     class _FakeCron:
@@ -1724,8 +1726,8 @@ def test_agent_workspace_override_does_not_migrate_legacy_cron(
         async def aclose(self) -> None:
             return None
 
-    monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.CronService", _FakeCron)
-    monkeypatch.setattr("nanobot.cli.agent.AgentLoop", _FakeAgentLoop)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.AgentLoop", _FakeAgentLoop)
     monkeypatch.setattr("nanobot.cli.terminal._print_agent_response", lambda *_args, **_kwargs: None)
 
     result = runner.invoke(
@@ -1758,9 +1760,9 @@ def test_agent_custom_config_workspace_does_not_migrate_legacy_cron(
 
     monkeypatch.setattr("nanobot.config.loader.set_config_path", lambda _path: None)
     monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
-    monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda _path: None)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.sync_workspace_templates", lambda _path: None)
-    monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.make_provider", lambda _config: _fake_provider())
-    monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
+    monkeypatch.setattr("nanobot.cli.agent_runtime.MessageBus", lambda: object())
     monkeypatch.setattr("nanobot.config.paths.get_cron_dir", lambda: legacy_dir)
 
     class _FakeCron:
@@ -1780,8 +1782,8 @@ def test_agent_custom_config_workspace_does_not_migrate_legacy_cron(
         async def aclose(self) -> None:
             return None
 
-    monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.CronService", _FakeCron)
-    monkeypatch.setattr("nanobot.cli.agent.AgentLoop", _FakeAgentLoop)
+    monkeypatch.setattr("nanobot.cli.agent_runtime.AgentLoop", _FakeAgentLoop)
     monkeypatch.setattr(
         "nanobot.cli.terminal._print_agent_response", lambda *_args, **_kwargs: None
     )

tests/cli/test_tui_launcher.py

@@ -285,7 +285,7 @@ def test_interactive_agent_does_not_silently_fall_back(
     assert exc_info.value.exit_code == 1
     assert output == [
         "[red]Native TUI unavailable: missing sidecar[/red]",
-        "[dim]Use `nanobot agent --classic` only if you want the old prompt.[/dim]",
+        "[dim]Use `nanobot agent --classic` only if you want the compatibility prompt.[/dim]",
     ]
 
 

tui/README.md

@@ -9,7 +9,7 @@ bun run --cwd tui test
 bun run --cwd tui build
 ```
 
-`nanobot agent` launches this client, leases the shared local gateway or starts it on demand, and passes an authenticated local endpoint through environment variables. Other terminals and the WebUI keep that gateway alive; the final interactive launcher to exit releases the on-demand process. Only `nanobot gateway --background` makes it persistent without clients. Source checkouts automatically align dependencies with `bun.lock` before launch; released installs use a version-matched, checksum-verified archive that keeps the executable together with its licenses, notices, corresponding application source, source offer, and relinking instructions. Startup fails explicitly if the native client is unavailable. The legacy Python prompt is only selected with `nanobot agent --classic`.
+`nanobot agent` launches this client, leases the shared local gateway or starts it on demand, and passes an authenticated local endpoint through environment variables. Other terminals and the WebUI keep that gateway alive; the final interactive launcher to exit releases the on-demand process. Only `nanobot gateway --background` makes it persistent without clients. Source checkouts automatically align dependencies with `bun.lock` before launch; released installs use a version-matched, checksum-verified archive that keeps the executable together with its licenses, notices, corresponding application source, source offer, and relinking instructions. Startup fails explicitly if the native client is unavailable. The compatibility Python prompt is selected only with `nanobot agent --classic`.
 
 Standalone terminals use OpenTUI's retained full-screen layout: the transcript reflows with the terminal while the composer stays fixed at the bottom. Mouse and keyboard scrolling operate inside the transcript, and leaving the TUI restores the previous terminal screen.