mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-04 08:28:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(cli): split commands into focused modules (#5175)

Browse Source
This commit is contained in:
chengyongru 2026-07-30 15:01:35 +08:00 committed by GitHub
parent ad6900e56c
commit e2563e2e74
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
24 changed files with 3190 additions and 2781 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
nanobot/agent/loop.py
View File

@ -1103,7 +1103,7 @@ class AgentLoop:
return
self._next_idle_compact_check_at = now + self._idle_compact_check_interval_s
self.auto_compact.check_expired(
self._schedule_background,
self.schedule_background,
self.runtime_for_session,
active_session_keys=self._pending_queues.keys(),
)
@ -1336,7 +1336,7 @@ class AgentLoop:
if errors:
raise BaseExceptionGroup("failed to close agent resources", errors)
def _schedule_background(self, coro: Coroutine[Any, Any, Any]) -> None:
def schedule_background(self, coro: Coroutine[Any, Any, Any]) -> None:
"""Schedule a coroutine as a tracked background task (drained on shutdown)."""
task = asyncio.create_task(coro)
self._background_tasks.add(task)
@ -1752,7 +1752,7 @@ class AgentLoop:
session.enforce_file_cap(
on_archive=partial(self.context.memory.raw_archive, session_key=ctx.session_key)
)
self._schedule_background(
self.schedule_background(
self.consolidator.maybe_consolidate_by_tokens(
session,
runtime=runtime,

352
nanobot/cli/agent.py Normal file
View File

@ -0,0 +1,352 @@
"""Direct and interactive 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.agent.hooks import create_file_edit_activity_hook
from nanobot.agent.loop import AgentLoop
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()
def agent(
message: str = 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"),
markdown: bool = typer.Option(
True,
"--markdown/--no-markdown",
help="Render assistant output as Markdown",
),
logs: bool = typer.Option(
False,
"--logs/--no-logs",
help="Show nanobot runtime logs during chat",
),
):
"""Interact with the agent directly."""
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)
try:
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)
_set_nanobot_logs(logs)
try:
agent_loop = AgentLoop.from_config(
runtime_config,
bus,
provider=provider,
cron_service=cron,
image_generation_provider_configs=image_gen_provider_configs(runtime_config),
hook_factories=[create_file_edit_activity_hook],
)
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,
)
# 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:
# Single message mode — direct call, no bus needed
async def run_once() -> None:
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,
)
await agent_loop.close_mcp()
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:
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 agent_loop.close_mcp()
asyncio.run(run_interactive())

2628
nanobot/cli/commands.py
View File

File diff suppressed because it is too large Load Diff

828
nanobot/cli/gateway_runtime.py Normal file
View File

@ -0,0 +1,828 @@
"""Foreground gateway runtime and lifecycle helpers."""
import asyncio
import signal
from collections.abc import Awaitable, Callable, Coroutine, Iterable
from contextlib import suppress
from pathlib import Path
from typing import Any, cast
import typer
from loguru import logger
from rich.console import Console
from nanobot import __logo__, __version__
from nanobot.agent.hooks import create_file_edit_activity_hook
from nanobot.agent.loop import AgentLoop
from nanobot.cli import terminal as cli_terminal
from nanobot.cli.runtime_config import _migrate_cron_store
from nanobot.cli.webui_support import (
_gateway_health_bind_note,
_gateway_health_url,
_host_for_local_browser,
_prepare_webui_bundle_for_gateway,
_print_foreground_port_conflict,
_tcp_endpoint_reachable,
_webui_browser_url,
_webui_channel_enabled,
_webui_endpoint_reachable,
)
from nanobot.config.paths import is_default_workspace
from nanobot.config.schema import Config
from nanobot.security.network import is_loopback_host
from nanobot.session.keys import UNIFIED_SESSION_KEY, last_channel_from_metadata
from nanobot.utils.evaluator import evaluate_response, resolve_evaluator_prompt
from nanobot.utils.helpers import sync_workspace_templates
from nanobot.webui.build import BuildMode
from nanobot.webui.sidebar_state import read_webui_sidebar_state
__all__ = ["_run_gateway"]
console = Console()
def _signal_name(signum: int) -> str:
with suppress(ValueError):
return signal.Signals(signum).name
return f"signal {signum}"
def _install_gateway_shutdown_handlers(
loop: asyncio.AbstractEventLoop,
shutdown_event: asyncio.Event,
tasks: list[asyncio.Task[Any]],
print_status: Callable[[str], None],
) -> Callable[[], None]:
"""Install foreground gateway signal handlers and return a restore callback."""
loop_signals: list[int] = []
previous_handlers: list[tuple[int, Any]] = []
shutdown_requested = False
def request_shutdown(signum: int) -> None:
nonlocal shutdown_requested
sig_name = _signal_name(signum)
if shutdown_requested:
logger.warning("Forcing gateway shutdown after repeated {}", sig_name)
for task in tasks:
if not task.done():
task.cancel()
return
shutdown_requested = True
logger.info("Gateway shutdown requested by {}", sig_name)
print_status("\nShutting down... Press Ctrl+C again to force.")
shutdown_event.set()
for signum in (signal.SIGINT, signal.SIGTERM):
try:
loop.add_signal_handler(signum, request_shutdown, signum)
except (NotImplementedError, RuntimeError, ValueError):
try:
previous = signal.getsignal(signum)
signal.signal(signum, lambda sig, _frame: request_shutdown(sig))
except (RuntimeError, ValueError):
logger.debug("Could not install gateway handler for {}", _signal_name(signum))
continue
previous_handlers.append((signum, previous))
else:
loop_signals.append(signum)
def restore() -> None:
for signum in loop_signals:
with suppress(NotImplementedError, RuntimeError, ValueError):
loop.remove_signal_handler(signum)
for signum, handler in previous_handlers:
with suppress(RuntimeError, ValueError):
signal.signal(signum, handler)
return restore
def _advance_dream_cursor_if_behind(memory: Any) -> None:
latest = memory.get_latest_cursor()
if memory.get_last_dream_cursor() < latest:
memory.set_last_dream_cursor(latest)
def _commit_dream_changes(memory: Any) -> str | None:
"""Commit durable Dream edits, without entering the commit path for a no-op run."""
if not memory.git.is_initialized():
return None
diff_body = memory.dream_content_diff()
if not diff_body:
return None
message = memory.build_dream_commit_message(
"dream: periodic memory consolidation",
diff_body,
)
return memory.git.auto_commit(message)
_HEARTBEAT_PREAMBLE = (
"[Your response will be delivered directly to the user's messaging app. "
"Output ONLY the final user-facing message. Never reference internal "
"files (HEARTBEAT.md, AWARENESS.md, etc.), your instructions, or your "
"decision process. If nothing needs reporting, respond with just "
"'All clear.' and nothing else.]\n\n"
)
def _heartbeat_has_active_tasks(content: str) -> bool:
"""True if HEARTBEAT.md has task lines, ignoring headers, blanks and comments."""
in_comment = False
in_active_section: bool = False
for line in content.splitlines():
stripped = line.strip()
if in_comment:
if "-->" in stripped:
in_comment = False
continue
if not stripped or stripped.startswith("#"):
if stripped.startswith("##") and not stripped.startswith("###"):
heading = stripped.lstrip("#").strip().lower()
in_active_section = heading.startswith("active tasks")
continue
if stripped.startswith("<!--"):
if "-->" not in stripped[4:]:
in_comment = True
continue
if in_active_section is False:
continue
return True
return False
def _pick_heartbeat_target_from_sessions(
*,
enabled_channels: Iterable[str],
sessions: Iterable[dict[str, Any]],
archived_keys: Iterable[str],
unified_session_metadata: dict[str, Any] | None = None,
) -> tuple[str, str]:
enabled = set(enabled_channels)
archived = set(archived_keys)
for item in sessions:
key = item.get("key") or ""
if key in archived:
continue
if key == UNIFIED_SESSION_KEY:
route = last_channel_from_metadata(unified_session_metadata)
if route is not None:
channel, chat_id = route
if channel not in {"cli", "system"} and channel in enabled:
return channel, chat_id
continue
if ":" not in key:
continue
channel, chat_id = key.split(":", 1)
if channel in {"cli", "system"}:
continue
if channel in enabled and chat_id:
return channel, chat_id
return "cli", "direct"
_GATEWAY_HEALTH_MAX_CONNECTIONS = 64
_GATEWAY_HEALTH_READ_TIMEOUT_SECONDS = 2.0
def _print_gateway_health_endpoint(host: str, port: int) -> None:
"""Print a usable health URL and make non-loopback binds explicit."""
console.print(
f"[green]✓[/green] Health endpoint: {_gateway_health_url(host, port)}"
f"{_gateway_health_bind_note(host)}"
)
if is_loopback_host(host):
return
console.print(
"[yellow]Warning: the unauthenticated health endpoint is listening beyond loopback "
"and may be reachable from other devices. "
f"Keep port {port} private or protect it with a firewall or reverse proxy.[/yellow]"
)
def _run_gateway(
config: Config,
*,
port: int | None = None,
open_browser_url: str | None = None,
webui_static_dist: bool = True,
webui_bundle_mode: BuildMode = "warn",
webui_runtime_surface: str = "browser",
webui_runtime_capabilities: dict[str, Any] | None = None,
health_server_enabled: bool = True,
unconfigured_provider_error: str | None = None,
) -> None:
"""Shared gateway runtime; ``open_browser_url`` opens a tab once channels are up."""
from nanobot.agent.model_presets import load_model_preset_catalog
from nanobot.agent.tools.message import MessageTool
from nanobot.agent.turn_delivery import TurnDeliveryFactory
from nanobot.bus.queue import MessageBus
from nanobot.bus.runtime_events import RuntimeEventBus
from nanobot.channels.manager import ChannelManager
from nanobot.config.watcher import watch_config_file
from nanobot.cron.bound_runner import run_bound_cron_job
from nanobot.cron.service import CronJobSkippedError, CronService
from nanobot.cron.session_turns import is_bound_cron_job
from nanobot.cron.types import CronJob
from nanobot.providers.factory import (
ProviderSnapshot,
build_provider_snapshot,
build_unconfigured_provider_snapshot,
load_provider_snapshot,
)
from nanobot.providers.fallback_provider import FallbackProvider
from nanobot.providers.image_generation import image_gen_provider_configs
from nanobot.session.manager import SessionManager
from nanobot.session.webui_turns import (
WebuiTurnCoordinator,
WebuiTurnRoutePolicy,
build_webui_fallback_model_observer,
)
from nanobot.triggers.local_runner import run_local_trigger_queue
from nanobot.triggers.local_store import LocalTriggerStore
from nanobot.webui.token_usage import TokenUsageHook
port = port if port is not None else config.gateway.port
webui_url = _webui_browser_url(config)
gateway_host_for_browser = _host_for_local_browser(config.gateway.host)
if health_server_enabled and _tcp_endpoint_reachable(gateway_host_for_browser, port):
_print_foreground_port_conflict(
webui_url=webui_url,
gateway_host=config.gateway.host,
gateway_port=port,
)
raise typer.Exit(1)
if _webui_channel_enabled(config) and _webui_endpoint_reachable(webui_url):
_print_foreground_port_conflict(
webui_url=webui_url,
gateway_host=config.gateway.host,
gateway_port=port,
)
raise typer.Exit(1)
console.print(f"{__logo__} Starting nanobot gateway version {__version__} on port {port}...")
_prepare_webui_bundle_for_gateway(
config,
mode=webui_bundle_mode,
webui_static_dist=webui_static_dist,
)
sync_workspace_templates(config.workspace_path)
bus = MessageBus()
runtime_events = RuntimeEventBus()
fallback_model_observer = build_webui_fallback_model_observer(bus)
def _observe_fallback_models(snapshot: ProviderSnapshot) -> ProviderSnapshot:
if isinstance(snapshot.provider, FallbackProvider):
snapshot.provider.set_fallback_model_observer(fallback_model_observer)
return snapshot
def _load_gateway_provider_snapshot(
*args: Any,
**kwargs: Any,
) -> ProviderSnapshot:
try:
return _observe_fallback_models(load_provider_snapshot(*args, **kwargs))
except ValueError as exc:
if unconfigured_provider_error is None:
raise
return build_unconfigured_provider_snapshot(config, str(exc))
if unconfigured_provider_error is not None:
provider_snapshot = build_unconfigured_provider_snapshot(
config,
unconfigured_provider_error,
)
else:
try:
provider_snapshot = _observe_fallback_models(build_provider_snapshot(config))
except ValueError as exc:
console.print(f"[red]Error: {exc}[/red]")
raise typer.Exit(1) from exc
session_manager = SessionManager(config.workspace_path)
# Self-heal the gateway state file with the current PID after any restart.
from nanobot.config.loader import get_config_path
from nanobot.gateway.runtime import GatewayRuntime, GatewayRuntimePaths
config_path = str(get_config_path().resolve(strict=False))
GatewayRuntime.refresh_state_pid(
paths=GatewayRuntimePaths.for_instance(
workspace=str(config.workspace_path)
if not is_default_workspace(config.workspace_path)
else None,
config_path=config_path,
)
)
# Preserve existing single-workspace installs, but keep custom workspaces clean.
if is_default_workspace(config.workspace_path):
_migrate_cron_store(config)
# Create cron service with workspace-scoped store
cron_store_path = config.workspace_path / "cron" / "jobs.json"
cron = CronService(cron_store_path)
trigger_store = LocalTriggerStore(config.workspace_path)
turn_delivery_factory = TurnDeliveryFactory(
bus,
runtime_events,
route_policy=WebuiTurnRoutePolicy(session_manager),
)
# Create agent with cron service
agent = AgentLoop.from_config(
config, bus,
provider=provider_snapshot.provider,
model=provider_snapshot.model,
context_window_tokens=provider_snapshot.context_window_tokens,
cron_service=cron,
session_manager=session_manager,
image_generation_provider_configs=image_gen_provider_configs(config),
provider_snapshot_loader=_load_gateway_provider_snapshot,
preset_catalog_loader=load_model_preset_catalog,
runtime_events=runtime_events,
turn_delivery_factory=turn_delivery_factory,
provider_signature=provider_snapshot.signature,
hooks=[TokenUsageHook(timezone_name=config.agents.defaults.timezone)],
local_trigger_store=trigger_store,
hook_factories=[create_file_edit_activity_hook],
)
def _schedule_webui_background(awaitable: Awaitable[None]) -> None:
agent.schedule_background(cast(Coroutine[Any, Any, None], awaitable))
webui_turn_coordinator = WebuiTurnCoordinator(
bus=bus,
sessions=session_manager,
schedule_background=_schedule_webui_background,
)
webui_turn_coordinator.subscribe(runtime_events)
from nanobot.bus.events import OutboundMessage
from nanobot.session.keys import session_key_for_channel
def _channel_session_key(channel: str, chat_id: str) -> str:
return session_key_for_channel(
channel,
chat_id,
unified_session=config.agents.defaults.unified_session,
)
async def _deliver_to_channel(
msg: OutboundMessage, *, record: bool = False, session_key: str | None = None,
) -> None:
"""Publish a user-visible message and mirror it into that channel's session."""
metadata = dict(msg.metadata or {})
record = record or bool(metadata.pop("_record_channel_delivery", False))
if metadata != (msg.metadata or {}):
msg = OutboundMessage(
channel=msg.channel,
chat_id=msg.chat_id,
content=msg.content,
reply_to=msg.reply_to,
media=msg.media,
metadata=metadata,
buttons=msg.buttons,
)
if (
record
and msg.channel != "cli"
and msg.content.strip()
and hasattr(session_manager, "get_or_create")
and hasattr(session_manager, "save")
):
key = session_key or _channel_session_key(msg.channel, msg.chat_id)
session = session_manager.get_or_create(key)
extra: dict[str, Any] = {"_channel_delivery": True}
if msg.media:
extra["media"] = list(msg.media)
session.add_message("assistant", msg.content, **extra)
session_manager.save(session)
await bus.publish_outbound(msg)
message_tool = agent.tools.get("message")
if isinstance(message_tool, MessageTool):
message_tool.set_send_callback(_deliver_to_channel)
# Set cron callback (needs agent)
async def on_cron_job(job: CronJob) -> str | None:
"""Execute a cron job through the agent."""
async def _silent(*_args: Any, **_kwargs: Any) -> None:
pass
# Dream is an internal job — run directly, not through the agent loop.
if job.name == "dream":
from nanobot.agent.memory import DreamRunProgress, MemoryStore
dream_session_key = MemoryStore.dream_session_key
prune_dream_sessions = MemoryStore.prune_dream_sessions
store = agent.context.memory
progress = DreamRunProgress()
resp = None
diff_body = ""
try:
result = store.build_dream_prompt()
if result is None:
logger.info("Dream: nothing to process")
return None
prompt, last_cursor = result
key = dream_session_key()
dream_runtime = agent.dream_runtime()
resp = await agent.process_direct(
prompt,
session_key=key,
ephemeral=True,
tools=store.build_dream_tools(),
on_progress=progress,
runtime=dream_runtime,
)
# The real file delta grounds the audit record; clean completion
# decides whether this history batch has finished processing.
diff_body = store.dream_content_diff()
completed = MemoryStore.dream_run_completed(
resp,
had_tool_errors=progress.had_tool_errors,
)
if completed:
store.set_last_dream_cursor(last_cursor)
if diff_body:
logger.info(
"Dream cron job completed, cursor advanced to {}",
last_cursor,
)
else:
logger.info(
"Dream cron job completed with no memory changes; "
"cursor advanced to {}",
last_cursor,
)
else:
logger.warning(
"Dream cron job did not complete; cursor remains at {}",
store.get_last_dream_cursor(),
)
except Exception:
logger.exception("Dream cron job failed")
finally:
from nanobot.webui.token_usage import record_response_token_usage
record_response_token_usage(
resp,
source="dream",
timezone_name=config.agents.defaults.timezone,
)
sha = _commit_dream_changes(store)
if sha:
logger.info("Dream commit: {}", sha)
store.compact_history()
prune_dream_sessions(agent.sessions.sessions_dir)
return None
# Heartbeat is a system job that checks HEARTBEAT.md for active tasks.
if job.name == "heartbeat":
heartbeat_file = config.workspace_path / "HEARTBEAT.md"
try:
content = heartbeat_file.read_text(encoding="utf-8")
except OSError:
logger.debug("Heartbeat: HEARTBEAT.md missing")
return None
if not _heartbeat_has_active_tasks(content):
logger.debug("Heartbeat: HEARTBEAT.md has no active tasks")
return None
channel, chat_id = _pick_heartbeat_target()
if channel == "cli":
return None
prompt = (
_HEARTBEAT_PREAMBLE
+ f"You are executing periodic heartbeat tasks. Read the active tasks below, perform each one, and report what you did:\n\n{content}"
)
# Internal check: funnel all output through the post-run gate so the
# turn can't deliver directly via the message tool and skip it.
suppress_token = None
if isinstance(message_tool, MessageTool):
suppress_token = message_tool.set_suppress_delivery(True)
try:
resp = await agent.process_direct(
prompt,
session_key="heartbeat",
channel=channel,
chat_id=chat_id,
on_progress=_silent,
)
finally:
if isinstance(message_tool, MessageTool) and suppress_token is not None:
message_tool.reset_suppress_delivery(suppress_token)
# Keep a small tail of heartbeat history so the loop stays bounded.
session = agent.sessions.get_or_create("heartbeat")
session.retain_recent_legal_suffix(hb_cfg.keep_recent_messages)
agent.sessions.save(session)
if not resp or not resp.content:
return
response = resp.content
evaluator_prompt = resolve_evaluator_prompt(config.workspace_path)
# Fail closed: stay silent on evaluator failure instead of notifying.
should_notify = await evaluate_response(
response=response,
task_context=prompt,
provider=agent.provider,
model=agent.model,
evaluator_prompt=evaluator_prompt,
default_notify=False,
)
if should_notify:
logger.info("Heartbeat: completed, delivering response")
await _deliver_to_channel(
OutboundMessage(channel=channel, chat_id=chat_id, content=response),
record=True,
)
else:
logger.info("Heartbeat: silenced by post-run evaluation")
return response
if is_bound_cron_job(job):
return await run_bound_cron_job(job, agent=agent, cron=cron)
reason = "unbound agent cron job must be recreated from a chat session"
logger.warning(
"Cron: skipped unbound agent job '{}' ({}): {}",
job.name,
job.id,
reason,
)
raise CronJobSkippedError(reason)
cron.on_job = on_cron_job
def _webui_runtime_model_name() -> str | None:
return agent.model.strip() or None
def _webui_skill_state_action(disabled_skills: set[str]) -> None:
config.agents.defaults.disabled_skills = sorted(disabled_skills)
agent.context.skills.disabled_skills = set(disabled_skills)
agent.subagents.disabled_skills = set(disabled_skills)
# Create channel manager (forwards SessionManager so the WebSocket channel
# can serve the embedded webui's REST surface).
channels = ChannelManager(
config,
bus,
session_manager=session_manager,
cron_service=cron,
local_trigger_store=trigger_store,
webui_runtime_model_name=_webui_runtime_model_name,
webui_cron_pending_job_ids=agent.pending_cron_job_ids_for_session,
webui_local_trigger_pending_ids=agent.pending_local_trigger_ids_for_session,
webui_static_dist=webui_static_dist,
webui_runtime_surface=webui_runtime_surface,
webui_runtime_capabilities=webui_runtime_capabilities,
webui_skill_state_action=_webui_skill_state_action,
)
def _pick_heartbeat_target() -> tuple[str, str]:
"""Pick a routable channel/chat target for heartbeat-triggered messages."""
sidebar_state = read_webui_sidebar_state()
unified_metadata = None
if config.agents.defaults.unified_session:
record = session_manager.read_session_metadata(UNIFIED_SESSION_KEY)
if isinstance(record, dict) and isinstance(record.get("metadata"), dict):
unified_metadata = record["metadata"]
return _pick_heartbeat_target_from_sessions(
enabled_channels=channels.enabled_channels,
sessions=session_manager.list_sessions(),
archived_keys=sidebar_state.get("archived_keys", []),
unified_session_metadata=unified_metadata,
)
if channels.enabled_channels:
console.print(f"[green]✓[/green] Channels enabled: {', '.join(channels.enabled_channels)}")
else:
console.print("[yellow]Warning: No channels enabled[/yellow]")
cron_status = cron.status()
cron_job_count = cast(int, cron_status["jobs"])
if cron_job_count > 0:
console.print(f"[green]✓[/green] Cron: {cron_job_count} scheduled jobs")
hb_cfg = config.gateway.heartbeat
if hb_cfg.enabled:
console.print(f"[green]✓[/green] Heartbeat: every {hb_cfg.interval_s}s")
else:
console.print("[yellow]✗[/yellow] Heartbeat: disabled")
async def _health_server(host: str, health_port: int) -> None:
"""Lightweight HTTP health endpoint on the gateway port."""
import json as _json
connection_slots = asyncio.Semaphore(_GATEWAY_HEALTH_MAX_CONNECTIONS)
async def handle(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
) -> None:
if connection_slots.locked():
writer.close()
return
async with connection_slots:
try:
data = await asyncio.wait_for(
reader.read(4096),
timeout=_GATEWAY_HEALTH_READ_TIMEOUT_SECONDS,
)
request_line = data.split(b"\r\n", 1)[0].decode(
"utf-8", errors="replace",
)
method, path = "", ""
parts = request_line.split(" ")
if len(parts) >= 2:
method, path = parts[0], parts[1]
if method == "GET" and path == "/health":
body = _json.dumps({"status": "ok"})
status = "200 OK"
content_type = "application/json"
else:
body = "Not Found"
status = "404 Not Found"
content_type = "text/plain"
resp = (
f"HTTP/1.0 {status}\r\n"
f"Content-Type: {content_type}\r\n"
f"Content-Length: {len(body)}\r\n"
"Connection: close\r\n"
f"\r\n{body}"
)
writer.write(resp.encode())
await writer.drain()
except (asyncio.TimeoutError, ConnectionError):
pass
finally:
writer.close()
server = await asyncio.start_server(handle, host, health_port)
_print_gateway_health_endpoint(host, health_port)
async with server:
await server.serve_forever()
# Register Dream system job (idempotent on restart)
from nanobot.cron.types import CronJob, CronPayload, CronSchedule
dream_cfg = config.agents.defaults.dream
if dream_cfg.enabled:
cron.register_system_job(CronJob(
id="dream",
name="dream",
schedule=dream_cfg.build_schedule(config.agents.defaults.timezone),
payload=CronPayload(kind="system_event"),
))
console.print(f"[green]✓[/green] Dream: {dream_cfg.describe_schedule()}")
else:
console.print("[yellow]○[/yellow] Dream: disabled")
_advance_dream_cursor_if_behind(agent.context.memory)
# Register Heartbeat system job (idempotent on restart)
if hb_cfg.enabled:
cron.register_system_job(CronJob(
id="heartbeat",
name="heartbeat",
schedule=CronSchedule(
kind="every",
every_ms=hb_cfg.interval_s * 1000,
tz=config.agents.defaults.timezone,
),
payload=CronPayload(kind="system_event"),
))
async def _open_browser_when_ready() -> None:
"""Wait for the gateway to bind, then point the user's browser at the webui."""
if not open_browser_url:
return
import webbrowser
from urllib.parse import urlparse
parsed = urlparse(open_browser_url)
target_host = parsed.hostname or config.gateway.host or "127.0.0.1"
target_port = parsed.port or port
# Channels start asynchronously; a short poll lets us avoid racing the bind.
for _ in range(40): # ~4s max
try:
_reader, writer = await asyncio.open_connection(
target_host,
target_port,
)
writer.close()
with suppress(Exception):
await writer.wait_closed()
break
except OSError:
await asyncio.sleep(0.1)
try:
webbrowser.open(open_browser_url)
console.print(f"[green]✓[/green] Opened browser at {open_browser_url}")
except Exception as e:
console.print(f"[yellow]Could not open browser ({e}); visit {open_browser_url}[/yellow]")
async def run() -> None:
tasks: list[asyncio.Task[Any]] = []
shutdown_task: asyncio.Task[Any] | None = None
runtime_tasks: asyncio.Future[list[Any]] | None = None
runtime_tasks_drained = False
shutdown_event = asyncio.Event()
cli_terminal._ensure_interactive_tty_mode()
restore_shutdown_handlers = _install_gateway_shutdown_handlers(
asyncio.get_running_loop(),
shutdown_event,
tasks,
console.print,
)
try:
await cron.start()
# Re-read once on first admission to close the watcher subscription window.
agent.runtime_resolver.invalidate()
tasks = [
asyncio.create_task(
watch_config_file(
Path(config_path),
lambda: agent.invalidate_runtime_config(),
),
name="nanobot-config-watcher",
),
asyncio.create_task(agent.run(), name="nanobot-agent-loop"),
asyncio.create_task(channels.start_all(), name="nanobot-channels"),
asyncio.create_task(
run_local_trigger_queue(
store=trigger_store,
submit_turn=agent.submit_local_trigger_turn,
is_channel_enabled=lambda name: channels.get_channel(name) is not None,
),
name="nanobot-local-triggers",
),
]
if health_server_enabled:
tasks.append(asyncio.create_task(
_health_server(config.gateway.host, port),
name="nanobot-health-server",
))
if open_browser_url:
tasks.append(asyncio.create_task(
_open_browser_when_ready(),
name="nanobot-open-browser",
))
runtime_tasks = asyncio.gather(*tasks)
shutdown_task = asyncio.create_task(
shutdown_event.wait(),
name="nanobot-gateway-shutdown",
)
done, _pending = await asyncio.wait(
{runtime_tasks, shutdown_task},
return_when=asyncio.FIRST_COMPLETED,
)
if runtime_tasks in done:
runtime_tasks_drained = True
await runtime_tasks
else:
runtime_tasks.cancel()
except KeyboardInterrupt:
console.print("\nShutting down...")
except Exception:
import traceback
console.print("\n[red]Error: Gateway crashed unexpectedly[/red]")
console.print(traceback.format_exc())
finally:
try:
if shutdown_task and not shutdown_task.done():
shutdown_task.cancel()
with suppress(asyncio.CancelledError):
await shutdown_task
cron.stop()
agent.stop()
# Some SDKs swallow task cancellation while attempting to reconnect.
# Close channel transports before waiting for their runners to exit.
await channels.stop_all()
for task in tasks:
if not task.done():
task.cancel()
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
if runtime_tasks is not None and not runtime_tasks_drained:
with suppress(asyncio.CancelledError, Exception):
await runtime_tasks
# Flush all cached sessions to durable storage before exit.
# This prevents data loss on filesystems with write-back
# caching (rclone VFS, NFS, FUSE mounts, etc.).
flushed = agent.sessions.flush_all()
if flushed:
logger.info("Shutdown: flushed {} session(s) to disk", flushed)
finally:
restore_shutdown_handlers()
asyncio.run(run())

12
nanobot/cli/log_control.py Normal file
View File

@ -0,0 +1,12 @@
"""Runtime log visibility controls shared by CLI commands."""
from loguru import logger
__all__ = ["_set_nanobot_logs"]
def _set_nanobot_logs(enabled: bool) -> None:
if enabled:
logger.enable("nanobot")
else:
logger.disable("nanobot")

372
nanobot/cli/provider.py Normal file
View File

@ -0,0 +1,372 @@
"""Typer commands for OAuth provider authentication."""
from __future__ import annotations
from collections.abc import Callable
from contextlib import suppress
from importlib import import_module
from pathlib import Path
from typing import TYPE_CHECKING, Protocol, cast
import typer
from rich.console import Console
from nanobot import __logo__
if TYPE_CHECKING:
from nanobot.providers.registry import ProviderSpec
console = Console()
provider_app = typer.Typer(help="Manage providers")
_PROVIDER_DISPLAY: dict[str, str] = {
"openai_codex": "OpenAI Codex",
"xai_grok": "xAI Grok",
"github_copilot": "GitHub Copilot",
}
_OAUTH_PROVIDER_DEFAULT_MODELS: dict[str, str] = {
"openai_codex": "openai-codex/gpt-5.6-sol",
"xai_grok": "xai-grok/grok-4.5",
"github_copilot": "github-copilot/gpt-5.4-mini",
}
class _OAuthToken(Protocol):
access: str | None
account_id: str | None
class _GetOAuthToken(Protocol):
def __call__(self, *, proxy: str | None = None) -> _OAuthToken | None: ...
class _LoginOAuthInteractive(Protocol):
def __call__(
self,
*,
print_fn: Callable[[str], None],
prompt_fn: Callable[[str], str],
proxy: str | None = None,
) -> _OAuthToken | None: ...
class _OAuthProviderConfig(Protocol):
token_filename: str
class _TokenStorage(Protocol):
def get_token_path(self) -> Path: ...
class _FileTokenStorageFactory(Protocol):
def __call__(self, *, token_filename: str) -> _TokenStorage: ...
def _required_module_attribute(module_name: str, attribute: str) -> object:
"""Load an optional dependency attribute with import-compatible errors."""
module = import_module(module_name)
try:
return getattr(module, attribute)
except AttributeError as exc:
raise ImportError(f"{module_name}.{attribute} is unavailable") from exc
def _load_openai_oauth_client() -> tuple[_GetOAuthToken, _LoginOAuthInteractive]:
"""Load the optional untyped OAuth client behind a typed boundary."""
return (
cast(_GetOAuthToken, _required_module_attribute("oauth_cli_kit", "get_token")),
cast(
_LoginOAuthInteractive,
_required_module_attribute("oauth_cli_kit", "login_oauth_interactive"),
),
)
def _load_openai_oauth_storage() -> tuple[_OAuthProviderConfig, _FileTokenStorageFactory]:
"""Load the optional untyped OAuth storage API behind a typed boundary."""
return (
cast(
_OAuthProviderConfig,
_required_module_attribute(
"oauth_cli_kit.providers",
"OPENAI_CODEX_PROVIDER",
),
),
cast(
_FileTokenStorageFactory,
_required_module_attribute("oauth_cli_kit.storage", "FileTokenStorage"),
),
)
def _resolve_oauth_provider(provider: str) -> ProviderSpec:
"""Resolve and validate an OAuth provider configuration."""
from nanobot.providers.registry import PROVIDERS
key = provider.replace("-", "_")
spec = next((s for s in PROVIDERS if s.name == key and s.is_oauth), None)
if not spec:
names = ", ".join(s.name.replace("_", "-") for s in PROVIDERS if s.is_oauth)
console.print(f"[red]Unknown OAuth provider: {provider}[/red] Supported: {names}")
raise typer.Exit(1)
return spec
def _set_oauth_provider_as_main(
provider_name: str,
*,
model: str | None = None,
config_path: str | None = None,
) -> None:
"""Persist an OAuth provider as the active agent provider."""
from nanobot.config.loader import get_config_path, load_config, save_config, set_config_path
resolved_config_path = Path(config_path).expanduser().resolve() if config_path else None
if resolved_config_path is not None and get_config_path() != resolved_config_path:
set_config_path(resolved_config_path)
console.print(f"[dim]Using config: {resolved_config_path}[/dim]")
config = load_config(resolved_config_path)
selected_model = (model or "").strip() or _OAUTH_PROVIDER_DEFAULT_MODELS[provider_name]
config.agents.defaults.model_preset = None
config.agents.defaults.provider = provider_name
config.agents.defaults.model = selected_model
if provider_name == "xai_grok" and selected_model == "xai-grok/grok-4.5":
config.agents.defaults.context_window_tokens = 500_000
save_config(config, resolved_config_path)
saved_path = resolved_config_path or get_config_path()
console.print(
f"[green]✓ Set {provider_name.replace('_', '-')} as the main provider[/green] "
f"[dim]{selected_model}[/dim]"
)
console.print(f"[dim]Saved: {saved_path}[/dim]")
@provider_app.command("login")
def provider_login(
provider: str = typer.Argument(
...,
help="OAuth provider (e.g. 'openai-codex', 'xai-grok', 'github-copilot')",
),
set_main: bool = typer.Option(
False,
"--set-main",
"--main",
help="Set this OAuth provider as the active agent provider after login",
),
model: str | None = typer.Option(
None,
"--model",
"-m",
help="Model to use when setting this provider as the active provider",
),
config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
):
"""Authenticate with an OAuth provider."""
spec = _resolve_oauth_provider(provider)
handler = _LOGIN_HANDLERS.get(spec.name)
if not handler:
console.print(f"[red]Login not implemented for {spec.label}[/red]")
raise typer.Exit(1)
if config:
from nanobot.config.loader import set_config_path
resolved_config_path = Path(config).expanduser().resolve()
set_config_path(resolved_config_path)
console.print(f"[dim]Using config: {resolved_config_path}[/dim]")
console.print(f"{__logo__} OAuth Login - {spec.label}\n")
handler()
if set_main or model:
_set_oauth_provider_as_main(spec.name, model=model, config_path=config)
@provider_app.command("logout")
def provider_logout(
provider: str = typer.Argument(
...,
help="OAuth provider (e.g. 'openai-codex', 'xai-grok', 'github-copilot')",
),
config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
):
"""Log out from an OAuth provider."""
spec = _resolve_oauth_provider(provider)
handler = _LOGOUT_HANDLERS.get(spec.name)
if not handler:
console.print(f"[red]Logout not implemented for {spec.label}[/red]")
raise typer.Exit(1)
if config:
from nanobot.config.loader import set_config_path
resolved_config_path = Path(config).expanduser().resolve()
set_config_path(resolved_config_path)
console.print(f"[dim]Using config: {resolved_config_path}[/dim]")
console.print(f"{__logo__} OAuth Logout - {spec.label}\n")
handler()
def _login_openai_codex() -> None:
try:
from nanobot.config.loader import load_config, resolve_config_env_vars
get_token, login_oauth_interactive = _load_openai_oauth_client()
proxy = None
try:
proxy = resolve_config_env_vars(load_config()).providers.openai_codex.proxy or None
except ValueError as e:
console.print(f"[red]{e}[/red]")
raise typer.Exit(1) from e
token = None
with suppress(Exception):
token = get_token(proxy=proxy)
if not (token and token.access):
console.print("[cyan]Starting interactive OAuth login...[/cyan]\n")
token = login_oauth_interactive(
print_fn=lambda s: console.print(s),
prompt_fn=lambda s: typer.prompt(s),
proxy=proxy,
)
if not (token and token.access):
console.print("[red]✗ Authentication failed[/red]")
raise typer.Exit(1)
console.print(
f"[green]✓ Authenticated with OpenAI Codex[/green] [dim]{token.account_id}[/dim]"
)
except ImportError:
console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
raise typer.Exit(1)
def _logout_openai_codex() -> None:
"""Clear local OAuth credentials for OpenAI Codex."""
try:
provider_config, storage_factory = _load_openai_oauth_storage()
except ImportError:
console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
raise typer.Exit(1)
storage = storage_factory(token_filename=provider_config.token_filename)
_delete_oauth_files(storage.get_token_path(), _PROVIDER_DISPLAY["openai_codex"])
def _login_xai_grok() -> None:
"""Authenticate with xAI using the Grok subscription OAuth contract."""
from nanobot.config.loader import load_config, resolve_config_env_vars
from nanobot.providers.xai_oauth import get_xai_oauth_token, login_xai_oauth
try:
proxy = resolve_config_env_vars(load_config()).providers.xai_grok.proxy or None
except ValueError as exc:
console.print(f"[red]{exc}[/red]")
raise typer.Exit(1) from exc
token = None
with suppress(Exception):
token = get_xai_oauth_token(proxy=proxy)
if not (token and token.access):
console.print(
"[cyan]Starting xAI browser sign-in for your X Premium / Grok subscription...[/cyan]\n"
)
try:
token = login_xai_oauth(
print_fn=lambda message: console.print(message),
prompt_fn=lambda prompt: typer.prompt(prompt),
proxy=proxy,
)
except Exception as exc:
console.print(f"[red]Authentication error: {exc}[/red]")
raise typer.Exit(1) from exc
account = token.account_id or "xAI account"
console.print(f"[green]✓ Authenticated with xAI[/green] [dim]{account}[/dim]")
console.print(
"[dim]Hosted X Search is enabled automatically when the selected model supports it.[/dim]"
)
def _logout_xai_grok() -> None:
"""Clear local xAI OAuth credentials for this nanobot instance."""
from nanobot.providers.xai_oauth import get_xai_oauth_storage_path, logout_xai_oauth
token_path = get_xai_oauth_storage_path()
provider_label = _PROVIDER_DISPLAY["xai_grok"]
if logout_xai_oauth():
console.print(f"[green]✓ Logged out from {provider_label}[/green]")
console.print(f"[dim]Removed: {token_path}[/dim]")
else:
console.print(f"[yellow]! No local OAuth credentials found for {provider_label}[/yellow]")
def _logout_github_copilot() -> None:
"""Clear local OAuth credentials for GitHub Copilot."""
try:
from nanobot.providers.github_copilot_provider import get_storage
except ImportError:
console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
raise typer.Exit(1)
storage = get_storage()
_delete_oauth_files(storage.get_token_path(), _PROVIDER_DISPLAY["github_copilot"])
def _delete_oauth_files(token_path: Path, provider_label: str) -> None:
"""Delete OAuth token and lock files, reporting the result."""
removed_paths: list[Path] = []
skipped: list[tuple[Path, OSError]] = []
for path in (token_path, token_path.with_suffix(".lock")):
try:
path.unlink()
except FileNotFoundError:
continue
except OSError as exc:
skipped.append((path, exc))
continue
removed_paths.append(path)
if not removed_paths and not skipped:
console.print(f"[yellow]! No local OAuth credentials found for {provider_label}[/yellow]")
return
if removed_paths:
console.print(f"[green]✓ Logged out from {provider_label}[/green]")
for path in removed_paths:
console.print(f"[dim]Removed: {path}[/dim]")
for path, exc in skipped:
console.print(f"[yellow]! Could not remove {path}: {exc}[/yellow]")
def _login_github_copilot() -> None:
try:
from nanobot.providers.github_copilot_provider import login_github_copilot
console.print("[cyan]Starting GitHub Copilot device flow...[/cyan]\n")
token = login_github_copilot(
print_fn=lambda s: console.print(s),
prompt_fn=lambda s: typer.prompt(s),
)
account = token.account_id or "GitHub"
console.print(
f"[green]✓ Authenticated with GitHub Copilot[/green] [dim]{account}[/dim]"
)
except Exception as e:
console.print(f"[red]Authentication error: {e}[/red]")
raise typer.Exit(1)
_LOGIN_HANDLERS: dict[str, Callable[[], None]] = {
"openai_codex": _login_openai_codex,
"xai_grok": _login_xai_grok,
"github_copilot": _login_github_copilot,
}
_LOGOUT_HANDLERS: dict[str, Callable[[], None]] = {
"openai_codex": _logout_openai_codex,
"xai_grok": _logout_xai_grok,
"github_copilot": _logout_github_copilot,
}

185
nanobot/cli/runtime_config.py Normal file
View File

@ -0,0 +1,185 @@
"""Configuration loading and diagnostics shared by CLI commands."""
from pathlib import Path
import typer
from pydantic import ValidationError
from rich.console import Console
from rich.markup import escape
from rich.text import Text
from nanobot.config.schema import Config
__all__ = [
"_load_config_for_cli",
"_load_inspection_config",
"_load_runtime_config",
"_migrate_cron_store",
"_model_display",
"_print_agent_start_error",
"_print_config_error",
"_print_model_setup_steps",
"_print_runtime_config_validation_error",
"_provider_setup_error",
]
console = Console()
def _model_display(config: Config) -> tuple[str, str]:
"""Return (resolved_model_name, preset_tag) for display strings."""
resolved = config.resolve_preset()
name = config.agents.defaults.model_preset
tag = f" (preset: {name})" if name else ""
return resolved.model, tag
def _print_config_error(error: Exception) -> None:
"""Render a configuration failure without exposing traceback internals."""
from nanobot.config.errors import ConfigLoadError
console.print(Text(str(error), style="red"))
if isinstance(error, ConfigLoadError):
command = _status_command(error.path)
console.print(f"[dim]Check again after editing: {escape(command)}[/dim]")
def _print_runtime_config_validation_error(
error: ValidationError,
*,
config_path: Path,
summary: str,
path_prefix: tuple[str | int, ...],
retry_command: str,
) -> None:
"""Render a runtime-owned Pydantic config error without exposing input values."""
from nanobot.config.errors import ConfigIssue, ConfigLoadError, validation_issues
issues = tuple(
ConfigIssue(
path=(*path_prefix, *issue.path),
message=issue.message,
)
for issue in validation_issues(error)
)
diagnostic = ConfigLoadError(
config_path,
kind="invalid_schema",
summary=summary,
issues=issues,
)
console.print(Text(str(diagnostic), style="red"))
console.print(f"[dim]Fix the listed setting, then retry: {escape(retry_command)}[/dim]")
def _status_command(config_path: Path) -> str:
return f'nanobot status --config "{config_path}"'
def _print_model_setup_steps(config_path: Path) -> None:
"""Show the shortest setup routes shared by Status and Agent startup."""
config_arg = f'--config "{config_path}"'
console.print(
f" WebUI: run [cyan]nanobot webui {escape(config_arg)}[/cyan], "
"then open Settings → Models"
)
console.print(f" CLI: run [cyan]nanobot onboard --wizard {escape(config_arg)}[/cyan]")
console.print(f" Check: [cyan]{escape(_status_command(config_path))}[/cyan]")
def _print_agent_start_error(error: ValueError) -> None:
from nanobot.config.loader import get_config_path
console.print(Text(f"Agent cannot start: {error}", style="red"))
console.print("Complete provider/model setup:")
_print_model_setup_steps(get_config_path())
def _load_config_for_cli(
config_path: Path | None = None,
*,
resolve_env: bool = False,
) -> Config:
"""Load CLI configuration and turn expected failures into a clean exit."""
from nanobot.config.errors import ConfigLoadError
from nanobot.config.loader import load_config, resolve_config_env_vars
try:
loaded = load_config(config_path)
if resolve_env:
loaded = resolve_config_env_vars(loaded)
return loaded
except ConfigLoadError as exc:
_print_config_error(exc)
raise typer.Exit(1) from exc
def _load_runtime_config(config: str | None = None, workspace: str | None = None) -> Config:
"""Load config and optionally override the active workspace."""
from nanobot.config.loader import set_config_path
config_path = None
if config:
config_path = Path(config).expanduser().resolve()
if not config_path.exists():
console.print(f"[red]Error: Config file not found: {config_path}[/red]")
raise typer.Exit(1)
set_config_path(config_path)
console.print(f"[dim]Using config: {config_path}[/dim]")
loaded = _load_config_for_cli(config_path, resolve_env=True)
if workspace:
loaded.agents.defaults.workspace = workspace
return loaded
def _load_inspection_config(
config: str | None = None,
workspace: str | None = None,
) -> tuple[Path, Config]:
"""Load config for diagnostic commands without resolving secret env refs."""
from nanobot.config.errors import ConfigLoadError
from nanobot.config.loader import get_config_path, load_config, set_config_path
config_path = None
if config:
config_path = Path(config).expanduser().resolve(strict=False)
set_config_path(config_path)
console.print(f"[dim]Using config: {config_path}[/dim]")
display_path = config_path or get_config_path()
try:
loaded = load_config(config_path)
except ConfigLoadError as exc:
_print_config_error(exc)
raise typer.Exit(1) from exc
except ValueError as exc:
console.print(f"[red]Error: {exc}[/red]")
raise typer.Exit(1) from exc
if workspace:
loaded.agents.defaults.workspace = workspace
return display_path, loaded
def _migrate_cron_store(config: "Config") -> None:
"""One-time migration: move legacy global cron store into the workspace."""
from nanobot.config.paths import get_cron_dir
legacy_path = get_cron_dir() / "jobs.json"
new_path = config.workspace_path / "cron" / "jobs.json"
if legacy_path.is_file() and not new_path.exists():
new_path.parent.mkdir(parents=True, exist_ok=True)
import shutil
shutil.move(str(legacy_path), str(new_path))
def _provider_setup_error(config: Config) -> str | None:
"""Return a local provider/model configuration error, or None."""
from nanobot.providers.factory import validate_provider_setup
try:
validate_provider_setup(config)
except ValueError as exc:
return str(exc)
return None

428
nanobot/cli/terminal.py Normal file
View File

@ -0,0 +1,428 @@
"""Terminal input and rendering helpers for the interactive CLI."""
from __future__ import annotations
import os
import select
import sys
from collections.abc import Callable
from contextlib import nullcontext, suppress
from typing import Any, Literal, cast
from loguru import logger
from prompt_toolkit import PromptSession, print_formatted_text
from prompt_toolkit.application import run_in_terminal
from prompt_toolkit.formatted_text import ANSI, HTML
from prompt_toolkit.history import FileHistory
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.key_binding.key_processor import KeyPressEvent
from prompt_toolkit.keys import Keys
from prompt_toolkit.patch_stdout import patch_stdout
from rich.console import Console
from rich.markdown import Markdown
from rich.text import Text
from nanobot import __logo__
from nanobot.bus.outbound_events import (
ProgressEvent,
RetryWaitEvent,
outbound_event_from_message,
)
from nanobot.cli.stream import StreamRenderer, ThinkingSpinner
from nanobot.utils.helpers import sanitize_surrogates as _sanitize_surrogates
__all__ = [
"_ReasoningBuffer",
"_ensure_interactive_tty_mode",
"_flush_cli_reasoning",
"_flush_pending_tty_input",
"_init_prompt_session",
"_is_exit_command",
"_maybe_print_interactive_progress",
"_print_agent_response",
"_print_cli_progress_line",
"_print_cli_reasoning",
"_print_interactive_response",
"_read_interactive_input_async",
"_restore_terminal",
]
console = Console()
EXIT_COMMANDS = {"exit", "quit", "/exit", "/quit", ":q"}
_REASONING_SENTENCE_ENDINGS = (".", "!", "?", "。", "！", "？")
_REASONING_FLUSH_CHARS = 60
_prompt_session: PromptSession[str] | None = None
_saved_term_attrs: list[Any] | None = None
def _ensure_interactive_tty_mode() -> None:
"""Restore interactive line input after a raw-mode TTY leak."""
try:
fd = sys.stdin.fileno()
if not os.isatty(fd):
return
except Exception:
return
with suppress(Exception):
import termios
attrs = termios.tcgetattr(fd)
required_lflag = termios.ISIG | termios.ICANON | termios.ECHO
blocked_input_flags = getattr(termios, "IGNCR", 0) | getattr(termios, "INLCR", 0)
if (
(attrs[3] & required_lflag) == required_lflag
and attrs[0] & termios.ICRNL
and not attrs[0] & blocked_input_flags
):
return
attrs[0] = (attrs[0] | termios.ICRNL) & ~blocked_input_flags
attrs[3] |= required_lflag
termios.tcsetattr(fd, termios.TCSANOW, attrs)
termios.tcflush(fd, termios.TCIFLUSH)
logger.debug("Restored foreground gateway TTY mode")
class SafeFileHistory(FileHistory):
"""FileHistory subclass that sanitizes surrogate characters on write.
On Windows, special Unicode input (emoji, mixed-script) can produce
surrogate characters that crash prompt_toolkit's file write.
See issue #2846.
"""
def store_string(self, string: str) -> None:
super().store_string(_sanitize_surrogates(string))
def _flush_pending_tty_input() -> None:
"""Drop unread keypresses typed while the model was generating output."""
try:
fd = sys.stdin.fileno()
if not os.isatty(fd):
return
except Exception:
return
with suppress(Exception):
import termios
termios.tcflush(fd, termios.TCIFLUSH)
return
with suppress(Exception):
while True:
ready, _, _ = select.select([fd], [], [], 0)
if not ready:
break
if not os.read(fd, 4096):
break
def _restore_terminal() -> None:
"""Restore terminal to its original state (echo, line buffering, etc.)."""
if _saved_term_attrs is None:
return
with suppress(Exception):
import termios
termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, _saved_term_attrs)
def _build_cli_key_bindings() -> KeyBindings:
"""Key bindings for the interactive prompt.
Behaviour:
* Enter -> submit the current input (keeps the familiar
single-line Enter-to-send feel even though the buffer
is multiline-capable).
* Alt+Enter -> insert a newline for multi-line input.
* Shift+Enter -> insert a newline on terminals that emit the CSI-u
(kitty / fixterms) keyboard-protocol encoding for it.
"""
# prompt_toolkit does not recognize CSI-u, so register its Shift+Enter
# sequence as a best-effort addition without overriding existing mappings.
with suppress(Exception):
from prompt_toolkit.input import ansi_escape_sequences as _aes
_aes.ANSI_SEQUENCES.setdefault("\x1b[13;2u", Keys.ControlF3)
kb = KeyBindings()
@kb.add("enter")
def _(event: KeyPressEvent) -> None:
event.current_buffer.validate_and_handle()
@kb.add("escape", "enter") # Alt+Enter / Meta+Enter (ESC + CR, "\x1b\r")
def _(event: KeyPressEvent) -> None:
event.current_buffer.insert_text("\n")
# LF-as-Enter terminals send Alt+Enter as ESC + LF rather than ESC + CR.
@kb.add("escape", Keys.ControlJ) # Alt+Enter on LF-as-Enter terminals
def _(event: KeyPressEvent) -> None:
event.current_buffer.insert_text("\n")
@kb.add(Keys.ControlF3) # Shift+Enter on CSI-u capable terminals
def _(event: KeyPressEvent) -> None:
event.current_buffer.insert_text("\n")
return kb
def _init_prompt_session() -> None:
"""Create the prompt_toolkit session with persistent file history."""
global _prompt_session, _saved_term_attrs
# Save terminal state so we can restore it on exit
with suppress(Exception):
import termios
_saved_term_attrs = termios.tcgetattr(sys.stdin.fileno())
from nanobot.config.paths import get_cli_history_path
history_file = get_cli_history_path()
history_file.parent.mkdir(parents=True, exist_ok=True)
_prompt_session = PromptSession(
history=SafeFileHistory(str(history_file)),
enable_open_in_editor=False,
# Multiline-capable buffer; Enter still submits via the custom key
# bindings, while Alt+Enter adds a newline.
multiline=True,
key_bindings=_build_cli_key_bindings(),
)
def _make_console() -> Console:
return Console(file=sys.stdout)
def _render_interactive_ansi(render_fn: Callable[[Console], None]) -> str:
"""Render Rich output to ANSI so prompt_toolkit can print it safely."""
ansi_console = Console(
force_terminal=sys.stdout.isatty(),
color_system=cast(
Literal["auto", "standard", "256", "truecolor", "windows"],
console.color_system or "standard",
),
width=console.width,
)
with ansi_console.capture() as capture:
render_fn(ansi_console)
return capture.get()
def _print_agent_response(
response: str,
render_markdown: bool,
metadata: dict[str, Any] | None = None,
show_header: bool = True,
) -> None:
"""Render assistant response with consistent terminal styling."""
console = _make_console()
content = response or ""
body = _response_renderable(content, render_markdown, metadata)
if show_header:
console.print()
console.print(f"[cyan]{__logo__} nanobot[/cyan]")
console.print(body)
console.print()
def _response_renderable(
content: str, render_markdown: bool, metadata: dict[str, Any] | None = None
) -> Text | Markdown:
"""Render plain-text command output without markdown collapsing newlines."""
if not render_markdown:
return Text(content)
if (metadata or {}).get("render_as") == "text":
return Text(content)
return Markdown(content)
async def _print_interactive_line(text: str) -> None:
"""Print async interactive updates with prompt_toolkit-safe Rich styling."""
def _write() -> None:
ansi = _render_interactive_ansi(lambda c: c.print(f" [dim]↳ {text}[/dim]"))
print_formatted_text(ANSI(ansi), end="")
await run_in_terminal(_write)
async def _print_interactive_response(
response: str,
render_markdown: bool,
metadata: dict[str, Any] | None = None,
) -> None:
"""Print async interactive replies with prompt_toolkit-safe Rich styling."""
def _write() -> None:
content = response or ""
def _render(target: Console) -> None:
target.print()
target.print(f"[cyan]{__logo__} nanobot[/cyan]")
target.print(_response_renderable(content, render_markdown, metadata))
target.print()
ansi = _render_interactive_ansi(_render)
print_formatted_text(ANSI(ansi), end="")
await run_in_terminal(_write)
def _print_cli_progress_line(
text: str,
thinking: ThinkingSpinner | None,
renderer: StreamRenderer | None = None,
) -> None:
"""Print a CLI progress line, pausing the spinner if needed."""
if not text.strip():
return
target = renderer.console if renderer else console
pause = renderer.pause_spinner() if renderer else (thinking.pause() if thinking else nullcontext())
with pause:
if renderer:
renderer.ensure_header()
target.print(f" [dim]↳ {text}[/dim]")
class _ReasoningBuffer:
def __init__(self) -> None:
self._text = ""
def add(self, text: str) -> str | None:
if not text:
return None
self._text += text
if self._should_flush(text):
return self.flush()
return None
def flush(self) -> str | None:
text = self._text.strip()
self._text = ""
return text or None
def clear(self) -> None:
self._text = ""
def _should_flush(self, text: str) -> bool:
stripped = text.rstrip()
return (
"\n" in text
or stripped.endswith(_REASONING_SENTENCE_ENDINGS)
or len(self._text) >= _REASONING_FLUSH_CHARS
)
def _print_cli_reasoning(
text: str,
thinking: ThinkingSpinner | None,
renderer: StreamRenderer | None = None,
) -> None:
"""Print reasoning/thinking content in a distinct style."""
if not text.strip():
return
target = renderer.console if renderer else console
pause = renderer.pause_spinner() if renderer else (thinking.pause() if thinking else nullcontext())
with pause:
if renderer:
renderer.ensure_header()
target.print(f"[dim italic]✻ {text}[/dim italic]")
def _flush_cli_reasoning(
reasoning_buffer: _ReasoningBuffer,
thinking: ThinkingSpinner | None,
renderer: StreamRenderer | None = None,
) -> None:
text = reasoning_buffer.flush()
if text:
_print_cli_reasoning(text, thinking, renderer)
async def _print_interactive_progress_line(
text: str,
thinking: ThinkingSpinner | None,
renderer: StreamRenderer | None = None,
) -> None:
"""Print an interactive progress line, pausing the spinner if needed."""
if not text.strip():
return
if renderer:
with renderer.pause_spinner():
renderer.ensure_header()
renderer.console.print(f" [dim]↳ {text}[/dim]")
else:
with thinking.pause() if thinking else nullcontext():
await _print_interactive_line(text)
async def _maybe_print_interactive_progress(
msg: Any,
thinking: ThinkingSpinner | None,
channels_config: Any,
renderer: StreamRenderer | None = None,
reasoning_buffer: _ReasoningBuffer | None = None,
) -> bool:
event = outbound_event_from_message(msg)
if isinstance(event, RetryWaitEvent):
await _print_interactive_progress_line(msg.content, thinking, renderer)
return True
if not isinstance(event, ProgressEvent):
return False
reasoning_buffer = reasoning_buffer or _ReasoningBuffer()
if event.reasoning_end:
if channels_config and not channels_config.show_reasoning:
reasoning_buffer.clear()
else:
_flush_cli_reasoning(reasoning_buffer, thinking, renderer)
return True
is_tool_hint = event.tool_hint
is_reasoning = event.reasoning or event.reasoning_delta
if is_reasoning:
if channels_config and not channels_config.show_reasoning:
reasoning_buffer.clear()
return True
text = reasoning_buffer.add(msg.content)
if text:
_print_cli_reasoning(text, thinking, renderer)
return True
if channels_config and is_tool_hint and not channels_config.send_tool_hints:
return True
if channels_config and not is_tool_hint and not channels_config.send_progress:
return True
await _print_interactive_progress_line(msg.content, thinking, renderer)
return True
def _is_exit_command(command: str) -> bool:
"""Return True when input should end interactive chat."""
return command.lower() in EXIT_COMMANDS
async def _read_interactive_input_async() -> str:
"""Read user input using prompt_toolkit (handles paste, history, display).
prompt_toolkit natively handles:
- Multiline paste (bracketed paste mode)
- History navigation (up/down arrows)
- Clean display (no ghost characters or artifacts)
"""
if _prompt_session is None:
raise RuntimeError("Call _init_prompt_session() first")
try:
with patch_stdout():
return await _prompt_session.prompt_async(
HTML("<b fg='ansiblue'>You:</b> "),
)
except EOFError as exc:
raise KeyboardInterrupt from exc

261
nanobot/cli/webui.py Normal file
View File

@ -0,0 +1,261 @@
"""WebUI CLI command."""
from pathlib import Path
import typer
from pydantic import ValidationError
from rich.console import Console
from nanobot.cli import terminal as cli_terminal
from nanobot.cli.gateway_runtime import _run_gateway
from nanobot.cli.runtime_config import (
_load_runtime_config,
_print_config_error,
_print_runtime_config_validation_error,
_provider_setup_error,
)
from nanobot.cli.webui_support import (
_attach_to_background_gateway,
_confirm_webui_action,
_ensure_local_webui_channel,
_gateway_health_bind_note,
_gateway_health_ready,
_gateway_health_url,
_gateway_instance_command,
_host_for_local_browser,
_load_webui_setup_config,
_open_webui_browser,
_prepare_webui_bundle_for_gateway,
_print_foreground_port_conflict,
_print_webui_foreground_lifecycle,
_resolve_webui_config_path,
_run_quick_start_for_webui,
_tcp_endpoint_reachable,
_warn_webui_bind_scope,
_webui_browser_url,
_webui_build_mode_for_interactive,
_webui_display_url,
_webui_endpoint_reachable,
)
from nanobot.config.paths import get_workspace_path
from nanobot.utils.helpers import sync_workspace_templates
console = Console()
def webui(
port: int | None = typer.Option(None, "--port", "-p", help="WebUI port"),
gateway_port: int | None = typer.Option(
None,
"--gateway-port",
help="Gateway health port",
),
workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
background: bool = typer.Option(
False,
"--background",
help="Keep the gateway running after this command exits",
),
no_open: bool = typer.Option(False, "--no-open", help="Do not open a browser"),
yes: bool = typer.Option(
False,
"--yes",
"-y",
help="Apply safe local WebUI defaults without prompting",
),
) -> None:
"""Prepare the local WebUI, start the gateway, and open the browser workbench."""
from nanobot.config.loader import resolve_config_env_vars, save_config
from nanobot.gateway import GatewayRuntime, GatewayRuntimePaths, GatewayStartOptions
cli_terminal._ensure_interactive_tty_mode()
config_path = _resolve_webui_config_path(config)
created_config = not config_path.exists()
if created_config:
console.print(f"[yellow]No config found at {config_path}.[/yellow]")
_confirm_webui_action("Create a nanobot config and workspace now?", yes=yes)
setup_config = _load_webui_setup_config(config_path)
if workspace:
setup_config.agents.defaults.workspace = workspace
try:
resolved_setup_config = resolve_config_env_vars(
setup_config.model_copy(deep=True),
config_path=config_path,
)
except ValueError as exc:
_print_config_error(exc)
raise typer.Exit(1) from exc
provider_error = _provider_setup_error(resolved_setup_config)
settings_setup_error = provider_error if provider_error and created_config else None
if settings_setup_error:
console.print(f"[yellow]Model setup is incomplete: {provider_error}[/yellow]")
console.print("Configure a provider and model in WebUI Settings → Models.")
if background:
console.print(
"[red]First-time WebUI setup must run in the foreground. "
"Run `nanobot webui` without --background.[/red]"
)
raise typer.Exit(1)
elif provider_error:
console.print(f"[dim]Provider check: {provider_error}[/dim]")
setup_config = _run_quick_start_for_webui(
setup_config,
yes=yes,
config_path=config_path,
)
if workspace:
setup_config.agents.defaults.workspace = workspace
try:
changed_webui, generated_bootstrap_secret = _ensure_local_webui_channel(
setup_config,
port=port,
yes=yes,
)
_warn_webui_bind_scope(setup_config)
webui_url = _webui_browser_url(setup_config)
except ValidationError as exc:
retry_command = f'nanobot webui --config "{config_path}"'
_print_runtime_config_validation_error(
exc,
config_path=config_path,
summary="WebUI configuration is invalid.",
path_prefix=("channels", "websocket"),
retry_command=retry_command,
)
raise typer.Exit(1) from exc
except ValueError as exc:
console.print(f"[red]Error: invalid WebUI channel config: {exc}[/red]")
raise typer.Exit(1) from exc
if created_config or provider_error or changed_webui or workspace:
save_config(setup_config, config_path)
console.print(f"[green]✓[/green] Saved config: {config_path}")
workspace_path = get_workspace_path(setup_config.workspace_path)
workspace_path.mkdir(parents=True, exist_ok=True)
sync_workspace_templates(workspace_path)
runtime_config = _load_runtime_config(str(config_path), workspace)
effective_gateway_port = gateway_port if gateway_port is not None else runtime_config.gateway.port
console.print()
console.print(f"WebUI: [cyan]{_webui_display_url(webui_url)}[/cyan]")
gateway_health_url = _gateway_health_url(
runtime_config.gateway.host,
effective_gateway_port,
)
console.print(
f"Gateway health: [cyan]{gateway_health_url}[/cyan]"
f"{_gateway_health_bind_note(runtime_config.gateway.host)}"
)
if no_open:
console.print("[dim]Browser opening disabled by --no-open.[/dim]")
if generated_bootstrap_secret:
console.print(
"[yellow]A WebUI bootstrap secret was generated and saved in this config.[/yellow]"
)
console.print(
"[dim]Open the WebUI and enter channels.websocket.tokenIssueSecret from "
f"{config_path}, or rerun without --no-open to open the authenticated URL.[/dim]"
)
webui_bundle_mode = _webui_build_mode_for_interactive(yes=yes)
config_arg = str(config_path)
workspace_arg = str(Path(workspace).expanduser().resolve(strict=False)) if workspace else None
runtime = GatewayRuntime(
paths=GatewayRuntimePaths.for_instance(
data_dir=config_path.parent,
workspace=workspace_arg,
config_path=config_arg,
)
)
start_options = GatewayStartOptions(
port=effective_gateway_port,
workspace=workspace_arg,
config_path=config_arg,
)
if background:
_prepare_webui_bundle_for_gateway(runtime_config, mode=webui_bundle_mode)
result = runtime.start_background(start_options)
restarted = False
restart_attempted = False
if not result.ok and result.message == "gateway_already_running" and changed_webui:
restart_attempted = True
console.print("[yellow]WebUI config changed; restarting the background gateway.[/yellow]")
result = runtime.restart(start_options, timeout_s=20)
restarted = result.ok
if not result.ok and (restart_attempted or result.message != "gateway_already_running"):
action = "restarted" if restart_attempted else "started"
console.print(f"[yellow]Gateway was not {action}: {result.message}[/yellow]")
console.print(f"Logs: {result.status.log_path}")
raise typer.Exit(1)
if restarted:
console.print("[green]Gateway restarted in the background.[/green]")
elif result.ok:
console.print("[green]Gateway started in the background.[/green]")
else:
console.print("[yellow]Gateway is already running in the background.[/yellow]")
console.print(
"Manage this instance: "
f"[cyan]{_gateway_instance_command('status', config_path=config_path, workspace=workspace)}[/cyan]"
)
console.print(
"View logs: "
f"[cyan]{_gateway_instance_command('logs', config_path=config_path, workspace=workspace)}[/cyan]"
)
console.print("[dim]Closing the browser does not stop channels or automations.[/dim]")
console.print(
"Stop nanobot: "
f"[cyan]{_gateway_instance_command('stop', config_path=config_path, workspace=workspace)}[/cyan]"
)
if not no_open:
_open_webui_browser(webui_url)
return
gateway_ready = _gateway_health_ready(runtime_config.gateway.host, effective_gateway_port)
webui_ready = _webui_endpoint_reachable(webui_url)
if gateway_ready and webui_ready:
console.print("[yellow]Gateway is already running; attaching to the existing WebUI.[/yellow]")
console.print(
"Restart the gateway if you need it to pick up local source changes: "
f"[cyan]{_gateway_instance_command('restart', config_path=config_path, workspace=workspace)}[/cyan]"
)
if not no_open:
_open_webui_browser(webui_url, wait=False)
if runtime.status().running:
_attach_to_background_gateway(runtime)
else:
console.print(
"[yellow]This gateway is controlled by another foreground command. "
"Stop it from that terminal.[/yellow]"
)
return
gateway_port_taken = gateway_ready or _tcp_endpoint_reachable(
_host_for_local_browser(runtime_config.gateway.host),
effective_gateway_port,
)
webui_port_taken = webui_ready
if gateway_port_taken or webui_port_taken:
_print_foreground_port_conflict(
webui_url=webui_url,
gateway_host=runtime_config.gateway.host,
gateway_port=effective_gateway_port,
)
raise typer.Exit(1)
_print_webui_foreground_lifecycle(attached=False)
_run_gateway(
runtime_config,
port=effective_gateway_port,
open_browser_url=None if no_open else webui_url,
webui_bundle_mode=webui_bundle_mode,
unconfigured_provider_error=settings_setup_error,
)

498
nanobot/cli/webui_support.py Normal file
View File

@ -0,0 +1,498 @@
"""Shared WebUI setup, URL, health, and browser helpers."""
import sys
import time
from pathlib import Path
from typing import TYPE_CHECKING, Any
import typer
from pydantic import ValidationError
from rich.console import Console
from rich.markup import escape
from rich.text import Text
from nanobot.cli.runtime_config import (
_load_config_for_cli,
_print_model_setup_steps,
_print_runtime_config_validation_error,
_provider_setup_error,
)
from nanobot.config.schema import Config
from nanobot.security.network import is_loopback_host
from nanobot.webui.build import (
BuildMode,
WebUIBuildError,
ensure_webui_bundle,
)
if TYPE_CHECKING:
from nanobot.gateway.runtime import GatewayRuntime
__all__ = [
"_attach_to_background_gateway",
"_confirm_webui_action",
"_ensure_local_webui_channel",
"_gateway_health_bind_note",
"_gateway_health_ready",
"_gateway_health_url",
"_gateway_instance_command",
"_host_for_local_browser",
"_load_webui_setup_config",
"_open_webui_browser",
"_prepare_webui_bundle_for_gateway",
"_print_foreground_port_conflict",
"_print_webui_foreground_lifecycle",
"_resolve_webui_config_path",
"_run_quick_start_for_webui",
"_tcp_endpoint_reachable",
"_validate_gateway_startup",
"_warn_webui_bind_scope",
"_webui_browser_url",
"_webui_build_mode_for_interactive",
"_webui_channel_enabled",
"_webui_display_url",
"_webui_endpoint_reachable",
]
console = Console()
def _confirm_webui_action(message: str, *, yes: bool) -> None:
"""Confirm a WebUI first-run mutation or fail clearly in non-interactive shells."""
if yes:
return
if not _cli_can_prompt():
console.print(
"[red]Error: WebUI setup needs confirmation. Re-run with --yes or use "
"`nanobot onboard --wizard`.[/red]"
)
raise typer.Exit(1)
if not typer.confirm(message, default=True):
console.print("[yellow]WebUI setup cancelled.[/yellow]")
raise typer.Exit(1)
def _cli_can_prompt() -> bool:
try:
return sys.stdin.isatty()
except Exception:
return False
def _webui_build_mode_for_interactive(*, yes: bool = False) -> BuildMode:
if yes:
return "auto"
return "prompt" if _cli_can_prompt() else "warn"
def _resolve_webui_config_path(config: str | None) -> Path:
"""Resolve the config path used by ``nanobot webui`` and bind loader state."""
from nanobot.config.loader import get_config_path, set_config_path
if not config:
return get_config_path()
config_path = Path(config).expanduser().resolve(strict=False)
set_config_path(config_path)
console.print(f"[dim]Using config: {config_path}[/dim]")
return config_path
def _load_webui_setup_config(config_path: Path) -> Config:
"""Load config for first-run mutation without resolving env-var placeholders."""
return _load_config_for_cli(config_path)
def _webui_config_dict(config: Config) -> dict[str, Any]:
"""Return the current WebSocket config as a mutable alias-key dictionary."""
from nanobot.channels.websocket.runtime import WebSocketConfig
current: Any = getattr(config.channels, "websocket", None) or {}
model = WebSocketConfig.model_validate(current)
return model.model_dump(by_alias=True, exclude_none=True)
def _webui_channel_enabled(config: Config) -> bool:
from nanobot.channels.websocket.runtime import WebSocketConfig
current: Any = getattr(config.channels, "websocket", None) or {}
return bool(WebSocketConfig.model_validate(current).enabled)
def _validate_gateway_startup(config: Config) -> str | None:
"""Validate gateway startup and return a provider error recoverable through WebUI."""
from nanobot.config.loader import get_config_path
config_path = get_config_path()
try:
webui_config = _webui_config_dict(config)
except ValidationError as exc:
retry_command = f'nanobot gateway --config "{config_path}"'
_print_runtime_config_validation_error(
exc,
config_path=config_path,
summary="Gateway configuration is invalid.",
path_prefix=("channels", "websocket"),
retry_command=retry_command,
)
raise typer.Exit(1) from exc
provider_error = _provider_setup_error(config)
if not provider_error:
return None
if bool(webui_config["enabled"]):
console.print(
Text(f"Provider/model setup is incomplete: {provider_error}", style="yellow")
)
console.print(
"Gateway will start so you can configure a provider and model "
"in WebUI Settings → Models."
)
browser_url = _webui_browser_url(config)
webui_url = browser_url.split("/#/", 1)[0]
console.print(Text(f"WebUI: {webui_url}", style="cyan"))
if browser_url != webui_url:
secret_key = (
"tokenIssueSecret"
if str(webui_config.get("tokenIssueSecret") or "").strip()
else "token"
)
console.print(
Text(
f"If prompted, enter the configured channels.websocket.{secret_key} "
f"value (see {config_path}).",
style="dim",
)
)
return provider_error
console.print(Text(f"Gateway cannot start: {provider_error}", style="red"))
console.print("Complete provider/model setup:")
_print_model_setup_steps(config_path)
raise typer.Exit(1)
def _prepare_webui_bundle_for_gateway(
config: Config,
*,
mode: BuildMode,
webui_static_dist: bool = True,
) -> None:
"""Refresh or warn about stale bundled WebUI assets before gateway startup."""
if not webui_static_dist or not _webui_channel_enabled(config):
return
def _print(message: str) -> None:
console.print(f"[yellow]{escape(message)}[/yellow]")
def _confirm(message: str) -> bool:
return typer.confirm(message, default=True)
try:
ensure_webui_bundle(
mode=mode,
confirm=_confirm if mode == "prompt" else None,
output=_print,
)
except WebUIBuildError as exc:
if mode == "warn":
console.print(f"[yellow]Warning: {escape(str(exc))}[/yellow]")
return
console.print(f"[red]Error: {escape(str(exc))}[/red]")
raise typer.Exit(1) from exc
def _host_for_local_browser(host: str) -> str:
"""Map bind hosts to a browser-openable local host."""
if host in {"0.0.0.0", ""}:
return "127.0.0.1"
if host == "::":
return "[::1]"
if ":" in host and not host.startswith("["):
return f"[{host}]"
return host
def _gateway_health_url(host: str, port: int) -> str:
"""Return a health URL that can be opened from this device."""
return f"http://{_host_for_local_browser(host)}:{port}/health"
def _gateway_health_bind_note(host: str) -> str:
"""Describe a non-local bind without presenting it as a usable URL."""
return "" if is_loopback_host(host) else f" [dim](listening on {host})[/dim]"
def _webui_bootstrap_secret(config: Config) -> str:
ws_cfg = _webui_config_dict(config)
return str(ws_cfg.get("tokenIssueSecret") or ws_cfg.get("token") or "").strip()
def _webui_browser_url(config: Config) -> str:
from urllib.parse import quote
ws_cfg = _webui_config_dict(config)
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)
if not secret:
return base_url
return f"{base_url}/#/?bootstrapSecret={quote(secret, safe='')}"
def _webui_display_url(url: str) -> str:
marker = "bootstrapSecret="
if marker not in url:
return url
prefix, _ = url.split(marker, 1)
return f"{prefix}{marker}<redacted>"
def _ensure_local_webui_channel(
config: Config,
*,
port: int | None,
yes: bool,
) -> tuple[bool, bool]:
"""Enable the local WebUI channel with safe localhost defaults."""
from nanobot.channels.websocket.runtime import WebSocketConfig
current: Any = getattr(config.channels, "websocket", None) or {}
model = WebSocketConfig.model_validate(current)
changed = False
generated_secret = False
needs_enable = not model.enabled
needs_port = port is not None and model.port != port
needs_secret = not model.token_issue_secret.strip() and not model.token.strip()
if not needs_enable and not needs_port and not needs_secret:
return False, False
target_port = port if port is not None else model.port
console.print()
console.print("[bold]Local WebUI setup[/bold]")
console.print(f" URL: [cyan]http://127.0.0.1:{target_port}[/cyan]")
console.print(" Bind: [cyan]127.0.0.1 only[/cyan] (not exposed to your LAN)")
console.print(" Auth: generated WebUI bootstrap secret stored in config")
console.print(
" LAN access requires an explicit host change plus a WebUI password in config."
)
_confirm_webui_action("Update the local WebUI channel in this config?", yes=yes)
if not model.enabled:
model.enabled = True
changed = True
if model.host != "127.0.0.1":
model.host = "127.0.0.1"
changed = True
if port is not None and model.port != port:
model.port = port
changed = True
if not model.websocket_requires_token:
model.websocket_requires_token = True
changed = True
if needs_secret:
import secrets
model.token_issue_secret = secrets.token_urlsafe(32)
changed = True
generated_secret = True
setattr(config.channels, "websocket", model.model_dump(by_alias=True, exclude_none=True))
return changed, generated_secret
def _warn_webui_bind_scope(config: Config) -> None:
ws_cfg = _webui_config_dict(config)
host = str(ws_cfg.get("host") or "127.0.0.1")
if host in {"127.0.0.1", "localhost", "::1"}:
return
console.print(
"[yellow]Warning: WebUI is configured to bind outside localhost. "
"Keep tokenIssueSecret set and use this only on trusted networks.[/yellow]"
)
def _wait_for_webui(url: str, *, timeout_s: float = 5.0) -> None:
"""Best-effort wait for the WebUI listener before opening a browser."""
import time
from urllib.parse import urlparse
parsed = urlparse(url)
host = parsed.hostname or "127.0.0.1"
port = parsed.port or (443 if parsed.scheme == "https" else 80)
deadline = time.monotonic() + timeout_s
while time.monotonic() < deadline:
if _tcp_endpoint_reachable(host, port, timeout_s=0.2):
return
time.sleep(0.1)
def _tcp_endpoint_reachable(host: str, port: int, *, timeout_s: float = 0.25) -> bool:
"""Return whether a local TCP endpoint accepts connections."""
import socket
try:
with socket.create_connection((host, port), timeout=timeout_s):
return True
except OSError:
return False
def _gateway_health_ready(host: str, port: int, *, timeout_s: float = 0.4) -> bool:
"""Return whether the nanobot gateway health endpoint responds OK."""
import json
import urllib.error
import urllib.request
browser_host = _host_for_local_browser(host)
try:
with urllib.request.urlopen(
f"http://{browser_host}:{port}/health",
timeout=timeout_s,
) as response:
if response.status != 200:
return False
body = response.read(1024)
except (OSError, urllib.error.URLError, TimeoutError, ValueError):
return False
try:
payload = json.loads(body.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError):
return False
return payload.get("status") == "ok"
def _webui_endpoint_reachable(url: str, *, timeout_s: float = 0.25) -> bool:
"""Return whether the WebUI URL's TCP endpoint is already listening."""
from urllib.parse import urlparse
parsed = urlparse(url)
host = parsed.hostname or "127.0.0.1"
port = parsed.port or (443 if parsed.scheme == "https" else 80)
return _tcp_endpoint_reachable(host, port, timeout_s=timeout_s)
def _print_foreground_port_conflict(
*,
webui_url: str,
gateway_host: str,
gateway_port: int,
) -> None:
console.print(
"[red]Error: nanobot cannot start because one of its local ports is already in use.[/red]"
)
console.print(f" WebUI: [cyan]{webui_url}[/cyan]")
console.print(
f" Gateway health: "
f"[cyan]http://{_host_for_local_browser(gateway_host)}:{gateway_port}/health[/cyan]"
)
console.print()
console.print("If this is an existing nanobot instance, use it or stop it first:")
console.print(" [cyan]nanobot gateway status[/cyan]")
console.print(" [cyan]nanobot gateway stop[/cyan]")
console.print(
"Or choose different ports with [cyan]--port[/cyan] "
"and [cyan]--gateway-port[/cyan]."
)
def _open_webui_browser(url: str, *, wait: bool = True) -> None:
"""Open the WebUI in the user's default browser, with a copyable fallback."""
import webbrowser
if wait:
_wait_for_webui(url)
display_url = _webui_display_url(url)
try:
webbrowser.open(url)
console.print(f"[green]✓[/green] Opened WebUI: [cyan]{display_url}[/cyan]")
except Exception as exc:
console.print(f"[yellow]Could not open browser ({exc}); visit {display_url}[/yellow]")
def _print_webui_foreground_lifecycle(*, attached: bool) -> None:
"""Explain how the browser and gateway lifecycles differ."""
console.print()
if attached:
console.print("[green]nanobot is attached to the existing gateway.[/green]")
else:
console.print("[green]nanobot is running in this terminal.[/green]")
console.print("[dim]Closing the browser does not stop channels or automations.[/dim]")
console.print("[dim]Press Ctrl+C here to stop nanobot.[/dim]")
def _attach_to_background_gateway(runtime: "GatewayRuntime") -> None:
"""Keep a foreground WebUI command attached to a managed gateway."""
_print_webui_foreground_lifecycle(attached=True)
try:
while runtime.status().running:
time.sleep(0.5)
except KeyboardInterrupt:
console.print("\n[yellow]Stopping nanobot...[/yellow]")
result = runtime.stop()
if result.ok or result.message == "gateway_not_running":
console.print("[green]Gateway stopped.[/green]")
return
console.print(f"[red]Gateway could not be stopped: {result.message}[/red]")
raise typer.Exit(1)
console.print("[yellow]Gateway stopped.[/yellow]")
def _gateway_instance_command(
subcommand: str,
*,
config_path: Path,
workspace: str | None,
) -> str:
"""Return a copyable gateway command for the same config/workspace instance."""
import shlex
parts = ["nanobot", "gateway", subcommand, "--config", str(config_path)]
if workspace:
workspace_path = str(Path(workspace).expanduser().resolve(strict=False))
parts.extend(["--workspace", workspace_path])
return " ".join(shlex.quote(part) for part in parts)
def _run_quick_start_for_webui(
config: Config,
*,
yes: bool,
config_path: Path,
) -> Config:
"""Offer the existing Quick Start flow when provider setup is missing."""
if yes:
console.print(
"[red]Error: provider/model setup is incomplete, and --yes cannot answer "
"provider credentials.[/red]"
)
console.print("Complete provider/model setup:")
_print_model_setup_steps(config_path)
raise typer.Exit(1)
console.print()
console.print("[yellow]Model provider setup is not ready.[/yellow]")
console.print(
"Quick Start will ask for provider, API key/base URL, model, and WebUI password."
)
_confirm_webui_action("Run Quick Start now?", yes=False)
from nanobot.cli.onboard import run_quick_start_onboard
try:
result = run_quick_start_onboard(config)
except RuntimeError as exc:
console.print(f"[red]Error: {exc}[/red]")
console.print(
"[yellow]Run `nanobot onboard --wizard` "
"after installing wizard dependencies.[/yellow]"
)
raise typer.Exit(1) from exc
if not result.should_save:
console.print("[yellow]Quick Start cancelled. No changes were saved.[/yellow]")
raise typer.Exit(1)
return result.config

2
nanobot/command/builtin.py
View File

@ -311,7 +311,7 @@ async def cmd_new(ctx: CommandContext) -> OutboundMessage:
loop.sessions.save(session)
loop.sessions.invalidate(session.key)
if snapshot and runtime is not None:
loop._schedule_background( # pyright: ignore[reportPrivateUsage]
loop.schedule_background(
loop.consolidator.archive( # pyright: ignore[reportUnknownMemberType]
snapshot,
runtime=runtime,

8
tests/agent/test_auto_compact.py
View File

@ -356,7 +356,7 @@ class TestAutoCompact:
loop.sessions.save(s2)
loop.consolidator.compact_idle_session = _make_fake_compact(loop)
loop.auto_compact.check_expired(loop._schedule_background, loop.runtime_for_session)
loop.auto_compact.check_expired(loop.schedule_background, loop.runtime_for_session)
await _drain_background_tasks(loop)
active_after = loop.sessions.get_or_create("cli:active")
@ -836,7 +836,7 @@ class TestProactiveAutoCompact:
async def _run_check_expired(loop, active_session_keys=()):
"""Helper: run check_expired via callback and wait for background tasks."""
loop.auto_compact.check_expired(
loop._schedule_background,
loop.schedule_background,
loop.runtime_for_session,
active_session_keys=active_session_keys,
)
@ -976,12 +976,12 @@ class TestProactiveAutoCompact:
loop.consolidator.compact_idle_session = _slow_compact
# First call starts archiving via callback
loop.auto_compact.check_expired(loop._schedule_background, loop.runtime_for_session)
loop.auto_compact.check_expired(loop.schedule_background, loop.runtime_for_session)
await started.wait()
assert archive_count == 1
# Second call should skip (key is in _archiving)
loop.auto_compact.check_expired(loop._schedule_background, loop.runtime_for_session)
loop.auto_compact.check_expired(loop.schedule_background, loop.runtime_for_session)
assert archive_count == 1
# Clean up

4
tests/agent/test_loop_consolidation_tokens.py
View File

@ -215,7 +215,7 @@ async def test_preflight_consolidation_receives_pending_summary(tmp_path) -> Non
return_value=(session, "Previous conversation summary: earlier context")
) # type: ignore[method-assign]
loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(return_value=None) # type: ignore[method-assign]
loop._schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
loop.schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
runtime = loop.llm_runtime()
await loop.process_direct("hello", session_key="cli:test", runtime=runtime)
@ -252,7 +252,7 @@ async def test_preflight_consolidation_before_llm_call(tmp_path, monkeypatch) ->
return LLMResponse(content="ok", tool_calls=[])
loop.provider.chat_with_retry = track_llm
loop.provider.chat_stream_with_retry = track_llm
loop._schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
loop.schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
session = loop.sessions.get_or_create("cli:test")
session.messages = [

2
tests/agent/test_loop_direct_websocket_status.py
View File

@ -33,7 +33,7 @@ def _make_loop(tmp_path):
WebuiTurnCoordinator(
bus=bus,
sessions=loop.sessions,
schedule_background=lambda coro: loop._schedule_background(coro),
schedule_background=lambda coro: loop.schedule_background(coro),
).subscribe(loop.runtime_events)
loop.turn_delivery_factory.route_policy = WebuiTurnRoutePolicy(loop.sessions)
loop.tools.get_definitions = MagicMock(return_value=[])

6
tests/agent/test_loop_progress.py
View File

@ -52,7 +52,7 @@ def _attach_webui_runtime_events(loop: AgentLoop, bus: MessageBus) -> None:
coordinator = WebuiTurnCoordinator(
bus=bus,
sessions=loop.sessions,
schedule_background=lambda coro: loop._schedule_background(coro),
schedule_background=lambda coro: loop.schedule_background(coro),
)
coordinator.subscribe(loop.runtime_events)
@ -1203,7 +1203,7 @@ class TestToolEventProgress:
elif hasattr(coro, "close"):
coro.close()
loop._schedule_background = schedule_background # type: ignore[method-assign]
loop.schedule_background = schedule_background # type: ignore[method-assign]
await loop._dispatch(InboundMessage(
channel="websocket",
@ -1249,7 +1249,7 @@ class TestToolEventProgress:
fake_title_after_turn,
)
scheduled: list[object] = []
loop._schedule_background = scheduled.append # type: ignore[method-assign]
loop.schedule_background = scheduled.append # type: ignore[method-assign]
await loop._dispatch(InboundMessage(
channel="websocket",

2
tests/agent/test_loop_save_turn.py
View File

@ -78,7 +78,7 @@ def _make_full_loop(tmp_path: Path) -> AgentLoop:
WebuiTurnCoordinator(
bus=loop.bus,
sessions=loop.sessions,
schedule_background=lambda coro: loop._schedule_background(coro),
schedule_background=lambda coro: loop.schedule_background(coro),
).subscribe(loop.runtime_events)
return loop

10
tests/agent/test_session_model_runtime.py
View File

@ -65,7 +65,7 @@ async def test_sessions_run_concurrently_with_isolated_model_presets(tmp_path) -
model_presets=presets,
preset_snapshot_loader=load_preset,
)
loop._schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
loop.schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
loop.set_session_model_preset("sdk:fast", "fast")
loop.set_session_model_preset("sdk:deep", "deep")
@ -116,7 +116,7 @@ async def test_removed_session_model_preset_falls_back_and_clears_metadata(tmp_p
model="base-model",
context_window_tokens=8_000,
)
loop._schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
loop.schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
session_key = "sdk:removed-preset"
session = loop.sessions.get_or_create(session_key)
session.metadata[SESSION_MODEL_PRESET_METADATA_KEY] = "removed"
@ -161,7 +161,7 @@ async def test_streamed_sdk_resolves_session_runtime_after_lock_admission(tmp_pa
model_presets=presets,
preset_snapshot_loader=load_preset,
)
loop._schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
loop.schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
session_key = "sdk:queued"
loop.set_session_model_preset(session_key, "fast")
@ -198,7 +198,7 @@ async def test_sdk_custom_model_preset_metadata_does_not_select_runtime(
model="base-model",
context_window_tokens=8_000,
)
loop._schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
loop.schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
bot = Nanobot(loop)
await bot.sessions.ingest(
@ -239,7 +239,7 @@ async def test_sdk_invalid_internal_model_preset_metadata_fails_explicitly(
model="base-model",
context_window_tokens=8_000,
)
loop._schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
loop.schedule_background = lambda coro: coro.close() # type: ignore[method-assign]
bot = Nanobot(loop)
await bot.sessions.ingest(

6
tests/agent/test_unified_session.py
View File

@ -246,7 +246,7 @@ class TestCmdNewUnifiedSession:
assert len(sessions.get_or_create("unified:default").messages) == 2
expected_snapshot = list(shared.messages)
# _schedule_background is a *sync* method that schedules a coroutine via
# schedule_background is a *sync* method that schedules a coroutine via
# asyncio.create_task(). Mirror that exactly so the coroutine is consumed
# and no RuntimeWarning is emitted.
admitted_runtime = MagicMock(name="admitted_runtime")
@ -255,8 +255,8 @@ class TestCmdNewUnifiedSession:
consolidator=SimpleNamespace(archive=AsyncMock(return_value=True)),
_cancel_active_tasks=AsyncMock(return_value=0),
llm_runtime=MagicMock(return_value=MagicMock()),
schedule_background=lambda coro: asyncio.ensure_future(coro),
)
loop._schedule_background = lambda coro: asyncio.ensure_future(coro)
msg = InboundMessage(
channel="telegram", sender_id="user1", chat_id="111", content="/new",
@ -303,8 +303,8 @@ class TestCmdNewUnifiedSession:
consolidator=SimpleNamespace(archive=AsyncMock(return_value=True)),
_cancel_active_tasks=AsyncMock(return_value=0),
runtime_for_session=MagicMock(return_value=MagicMock()),
schedule_background=lambda coro: asyncio.ensure_future(coro),
)
loop._schedule_background = lambda coro: asyncio.ensure_future(coro)
msg = InboundMessage(
channel="telegram", sender_id="user1", chat_id="111", content="/new",

56
tests/cli/test_cli_input.py
View File

@ -5,8 +5,8 @@ from unittest.mock import AsyncMock, MagicMock, call, patch
import pytest
from prompt_toolkit.formatted_text import HTML
from nanobot.cli import commands
from nanobot.cli import stream as stream_mod
from nanobot.cli import terminal
@pytest.fixture
@ -14,8 +14,8 @@ def mock_prompt_session():
"""Mock the global prompt session."""
mock_session = MagicMock()
mock_session.prompt_async = AsyncMock()
with patch("nanobot.cli.commands._PROMPT_SESSION", mock_session), \
patch("nanobot.cli.commands.patch_stdout"):
with patch("nanobot.cli.terminal._prompt_session", mock_session), \
patch("nanobot.cli.terminal.patch_stdout"):
yield mock_session
@ -24,7 +24,7 @@ async def test_read_interactive_input_async_returns_input(mock_prompt_session):
"""Test that _read_interactive_input_async returns the user input from prompt_session."""
mock_prompt_session.prompt_async.return_value = "hello world"
result = await commands._read_interactive_input_async()
result = await terminal._read_interactive_input_async()
assert result == "hello world"
mock_prompt_session.prompt_async.assert_called_once()
@ -38,23 +38,23 @@ async def test_read_interactive_input_async_handles_eof(mock_prompt_session):
mock_prompt_session.prompt_async.side_effect = EOFError()
with pytest.raises(KeyboardInterrupt):
await commands._read_interactive_input_async()
await terminal._read_interactive_input_async()
def test_init_prompt_session_creates_session():
"""Test that _init_prompt_session initializes the global session."""
# Ensure global is None before test
commands._PROMPT_SESSION = None
terminal._prompt_session = None
with patch("nanobot.cli.commands.PromptSession") as mock_session_cls, \
patch("nanobot.cli.commands.FileHistory"), \
with patch("nanobot.cli.terminal.PromptSession") as mock_session_cls, \
patch("nanobot.cli.terminal.FileHistory"), \
patch("pathlib.Path.home") as mock_home:
mock_home.return_value = MagicMock()
commands._init_prompt_session()
terminal._init_prompt_session()
assert commands._PROMPT_SESSION is not None
assert terminal._prompt_session is not None
mock_session_cls.assert_called_once()
_, kwargs = mock_session_cls.call_args
# Buffer is multiline-capable so Alt+Enter can insert newlines;
@ -68,7 +68,7 @@ def test_cli_key_bindings_enter_submits_and_alt_enter_newlines():
"""Enter submits the buffer; Alt+Enter inserts a newline."""
from prompt_toolkit.keys import Keys
kb = commands._build_cli_key_bindings()
kb = terminal._build_cli_key_bindings()
def _keys(binding):
return tuple(getattr(k, "value", k) for k in binding.keys)
@ -102,8 +102,8 @@ async def test_raw_lf_enter_still_submits_like_wsl_terminals():
with create_pipe_input() as pipe_input:
with create_app_session(input=pipe_input, output=DummyOutput()):
commands._init_prompt_session()
session = commands._PROMPT_SESSION
terminal._init_prompt_session()
session = terminal._prompt_session
pipe_input.send_text("hello\x0aworld\r")
result = await session.prompt_async("> ")
@ -119,8 +119,8 @@ async def test_alt_enter_inserts_newline_on_lf_terminals():
with create_pipe_input() as pipe_input:
with create_app_session(input=pipe_input, output=DummyOutput()):
commands._init_prompt_session()
session = commands._PROMPT_SESSION
terminal._init_prompt_session()
session = terminal._prompt_session
pipe_input.send_text("foo\x1b\x0abar\r")
result = await session.prompt_async("> ")
@ -136,8 +136,8 @@ async def test_csi_u_shift_enter_inserts_newline_not_raw_escape():
with create_pipe_input() as pipe_input:
with create_app_session(input=pipe_input, output=DummyOutput()):
commands._init_prompt_session()
session = commands._PROMPT_SESSION
terminal._init_prompt_session()
session = terminal._prompt_session
pipe_input.send_text("foo\x1b[13;2ubar\r")
result = await session.prompt_async("> ")
@ -173,10 +173,10 @@ def test_print_cli_progress_line_pauses_spinner_before_printing():
mock_console = MagicMock()
mock_console.status.return_value = spinner
with patch.object(commands.console, "print", side_effect=lambda *_args, **_kwargs: order.append("print")):
with patch.object(terminal.console, "print", side_effect=lambda *_args, **_kwargs: order.append("print")):
thinking = stream_mod.ThinkingSpinner(console=mock_console)
with thinking:
commands._print_cli_progress_line("tool running", thinking)
terminal._print_cli_progress_line("tool running", thinking)
assert order == ["start", "stop", "print", "start", "stop"]
@ -224,7 +224,7 @@ def test_print_cli_progress_line_opens_renderer_header_before_trace():
renderer.ensure_header.side_effect = lambda: order.append("header")
renderer.pause_spinner.return_value = nullcontext()
commands._print_cli_progress_line("tool running", None, renderer)
terminal._print_cli_progress_line("tool running", None, renderer)
assert order == ["header", "print"]
@ -235,7 +235,7 @@ def test_print_cli_progress_line_stops_live_before_trace():
renderer = stream_mod.StreamRenderer(show_spinner=False)
renderer._live = mock_live
commands._print_cli_progress_line("tool running", None, renderer)
terminal._print_cli_progress_line("tool running", None, renderer)
mock_live.stop.assert_called_once()
assert renderer._live is None
@ -254,10 +254,10 @@ async def test_print_interactive_progress_line_pauses_spinner_before_printing():
async def fake_print(_text: str) -> None:
order.append("print")
with patch("nanobot.cli.commands._print_interactive_line", side_effect=fake_print):
with patch("nanobot.cli.terminal._print_interactive_line", side_effect=fake_print):
thinking = stream_mod.ThinkingSpinner(console=mock_console)
with thinking:
await commands._print_interactive_progress_line("tool running", thinking)
await terminal._print_interactive_progress_line("tool running", thinking)
assert order == ["start", "stop", "print", "start", "stop"]
@ -269,7 +269,7 @@ def test_response_renderable_uses_text_for_explicit_plain_rendering():
"📊 Tokens: 20639 in / 29 out"
)
renderable = commands._response_renderable(
renderable = terminal._response_renderable(
status,
render_markdown=True,
metadata={"render_as": "text"},
@ -279,7 +279,7 @@ def test_response_renderable_uses_text_for_explicit_plain_rendering():
def test_response_renderable_preserves_normal_markdown_rendering():
renderable = commands._response_renderable("**bold**", render_markdown=True)
renderable = terminal._response_renderable("**bold**", render_markdown=True)
assert renderable.__class__.__name__ == "Markdown"
@ -287,7 +287,7 @@ def test_response_renderable_preserves_normal_markdown_rendering():
def test_response_renderable_without_metadata_keeps_markdown_path():
help_text = "🐈 nanobot commands:\n/status — Show bot status\n/help — Show available commands"
renderable = commands._response_renderable(help_text, render_markdown=True)
renderable = terminal._response_renderable(help_text, render_markdown=True)
assert renderable.__class__.__name__ == "Markdown"
@ -389,9 +389,9 @@ def test_render_interactive_ansi_force_terminal_follows_isatty():
captured["console"] = c
with patch.object(sys.stdout, "isatty", return_value=True):
commands._render_interactive_ansi(render_fn)
terminal._render_interactive_ansi(render_fn)
assert captured["console"]._force_terminal is True
with patch.object(sys.stdout, "isatty", return_value=False):
commands._render_interactive_ansi(render_fn)
terminal._render_interactive_ansi(render_fn)
assert captured["console"]._force_terminal is False

252
tests/cli/test_commands.py
View File

@ -17,6 +17,11 @@ from nanobot.agent.tools.registry import ToolRegistry
from nanobot.agent.turn_delivery import TurnDeliveryFactory
from nanobot.bus.events import InboundMessage, OutboundMessage
from nanobot.cli import commands as cli_commands
from nanobot.cli import gateway_runtime as cli_gateway_runtime
from nanobot.cli import provider as provider_commands
from nanobot.cli import terminal as cli_terminal
from nanobot.cli import webui as cli_webui
from nanobot.cli import webui_support as cli_webui_support
from nanobot.cli.commands import app
from nanobot.config.schema import Config
from nanobot.cron.service import CronJobSkippedError
@ -113,7 +118,7 @@ def test_gateway_signal_handler_first_signal_stops_and_second_forces() -> None:
task = asyncio.create_task(never.wait())
output: list[str] = []
restore = cli_commands._install_gateway_shutdown_handlers(
restore = cli_gateway_runtime._install_gateway_shutdown_handlers(
loop, shutdown_event, [task], output.append,
)
try:
@ -161,8 +166,8 @@ def test_interactive_tty_mode_restores_line_input(monkeypatch) -> None:
attrs[3] &= ~(termios.ISIG | termios.ICANON | termios.ECHO)
termios.tcsetattr(slave_fd, termios.TCSANOW, attrs)
monkeypatch.setattr(cli_commands.sys, "stdin", _Stdin())
cli_commands._ensure_interactive_tty_mode()
monkeypatch.setattr(cli_terminal.sys, "stdin", _Stdin())
cli_terminal._ensure_interactive_tty_mode()
restored = termios.tcgetattr(slave_fd)
assert restored[0] & termios.ICRNL
@ -179,24 +184,24 @@ def test_webui_restores_tty_before_loading_config(monkeypatch, tmp_path: Path) -
config_file = tmp_path / "config.json"
config_file.write_text("{}", encoding="utf-8")
calls: list[str] = []
original_resolve = cli_commands._resolve_webui_config_path
original_resolve = cli_webui._resolve_webui_config_path
monkeypatch.setattr(
cli_commands,
cli_terminal,
"_ensure_interactive_tty_mode",
lambda: calls.append("tty"),
)
monkeypatch.setattr(
cli_commands,
cli_webui,
"_resolve_webui_config_path",
lambda path: calls.append("config") or original_resolve(path),
)
_patch_webui_provider_ready(monkeypatch)
monkeypatch.setattr(cli_commands, "sync_workspace_templates", lambda _path: None)
monkeypatch.setattr(cli_commands, "_gateway_health_ready", lambda *_args, **_kwargs: False)
monkeypatch.setattr(cli_commands, "_webui_endpoint_reachable", lambda *_args, **_kwargs: False)
monkeypatch.setattr(cli_commands, "_tcp_endpoint_reachable", lambda *_args, **_kwargs: False)
monkeypatch.setattr(cli_commands, "_run_gateway", lambda *_args, **_kwargs: None)
monkeypatch.setattr(cli_webui, "sync_workspace_templates", lambda _path: None)
monkeypatch.setattr(cli_webui, "_gateway_health_ready", lambda *_args, **_kwargs: False)
monkeypatch.setattr(cli_webui, "_webui_endpoint_reachable", lambda *_args, **_kwargs: False)
monkeypatch.setattr(cli_webui, "_tcp_endpoint_reachable", lambda *_args, **_kwargs: False)
monkeypatch.setattr(cli_webui, "_run_gateway", lambda *_args, **_kwargs: None)
result = runner.invoke(app, ["webui", "--config", str(config_file), "--yes", "--no-open"])
@ -209,11 +214,11 @@ def test_disabled_dream_cursor_only_advances_when_behind(tmp_path) -> None:
store.append_history("first")
store.append_history("second")
cli_commands._advance_dream_cursor_if_behind(store)
cli_gateway_runtime._advance_dream_cursor_if_behind(store)
assert store.get_last_dream_cursor() == 2
store.set_last_dream_cursor(10)
cli_commands._advance_dream_cursor_if_behind(store)
cli_gateway_runtime._advance_dream_cursor_if_behind(store)
assert store.get_last_dream_cursor() == 10
@ -225,7 +230,7 @@ def test_commit_dream_changes_skips_noop_run(tmp_path) -> None:
store.git.auto_commit("initial")
store.git.auto_commit = MagicMock(wraps=store.git.auto_commit)
assert cli_commands._commit_dream_changes(store) is None
assert cli_gateway_runtime._commit_dream_changes(store) is None
store.git.auto_commit.assert_not_called()
@ -238,7 +243,7 @@ def test_commit_dream_changes_commits_real_edits(tmp_path) -> None:
store.write_memory("# Memory\n- Research notes")
store.git.auto_commit = MagicMock(wraps=store.git.auto_commit)
sha = cli_commands._commit_dream_changes(store)
sha = cli_gateway_runtime._commit_dream_changes(store)
assert sha is not None
store.git.auto_commit.assert_called_once()
@ -493,7 +498,7 @@ def test_openai_codex_oauth_default_matches_curated_flagship():
assert spec is not None
assert spec.builtin_models
assert cli_commands._OAUTH_PROVIDER_DEFAULT_MODELS["openai_codex"] == (
assert provider_commands._OAUTH_PROVIDER_DEFAULT_MODELS["openai_codex"] == (
spec.builtin_models[0].id
)
@ -671,16 +676,28 @@ def test_provider_login_rejects_unknown_provider():
assert "Unknown OAuth provider" in result.stdout
def test_provider_login_openai_codex_handles_missing_oauth_symbol(monkeypatch):
import oauth_cli_kit
monkeypatch.delattr(oauth_cli_kit, "get_token")
result = runner.invoke(app, ["provider", "login", "openai-codex"])
assert result.exit_code == 1
assert "oauth_cli_kit not installed" in result.stdout
assert result.exception is not None
def test_provider_login_can_set_openai_codex_as_main_provider(tmp_path):
config_path = tmp_path / "config.json"
called = False
original = cli_commands._LOGIN_HANDLERS["openai_codex"]
original = provider_commands._LOGIN_HANDLERS["openai_codex"]
def fake_login() -> None:
nonlocal called
called = True
cli_commands._LOGIN_HANDLERS["openai_codex"] = fake_login
provider_commands._LOGIN_HANDLERS["openai_codex"] = fake_login
try:
result = runner.invoke(
app,
@ -694,7 +711,7 @@ def test_provider_login_can_set_openai_codex_as_main_provider(tmp_path):
],
)
finally:
cli_commands._LOGIN_HANDLERS["openai_codex"] = original
provider_commands._LOGIN_HANDLERS["openai_codex"] = original
assert result.exit_code == 0
assert called is True
@ -709,8 +726,8 @@ def test_provider_login_can_set_openai_codex_as_main_provider(tmp_path):
def test_provider_login_can_set_github_copilot_as_main_provider(tmp_path):
config_path = tmp_path / "config.json"
original = cli_commands._LOGIN_HANDLERS["github_copilot"]
cli_commands._LOGIN_HANDLERS["github_copilot"] = lambda: None
original = provider_commands._LOGIN_HANDLERS["github_copilot"]
provider_commands._LOGIN_HANDLERS["github_copilot"] = lambda: None
try:
result = runner.invoke(
app,
@ -724,7 +741,7 @@ def test_provider_login_can_set_github_copilot_as_main_provider(tmp_path):
],
)
finally:
cli_commands._LOGIN_HANDLERS["github_copilot"] = original
provider_commands._LOGIN_HANDLERS["github_copilot"] = original
assert result.exit_code == 0
assert "Set github-copilot as the main provider" in result.stdout
@ -738,8 +755,8 @@ def test_provider_login_can_set_github_copilot_as_main_provider(tmp_path):
def test_provider_login_can_set_xai_grok_as_main_provider(tmp_path):
config_path = tmp_path / "config.json"
original = cli_commands._LOGIN_HANDLERS["xai_grok"]
cli_commands._LOGIN_HANDLERS["xai_grok"] = lambda: None
original = provider_commands._LOGIN_HANDLERS["xai_grok"]
provider_commands._LOGIN_HANDLERS["xai_grok"] = lambda: None
try:
result = runner.invoke(
app,
@ -753,7 +770,7 @@ def test_provider_login_can_set_xai_grok_as_main_provider(tmp_path):
],
)
finally:
cli_commands._LOGIN_HANDLERS["xai_grok"] = original
provider_commands._LOGIN_HANDLERS["xai_grok"] = original
assert result.exit_code == 0
assert "Set xai-grok as the main provider" in result.stdout
@ -768,8 +785,8 @@ def test_provider_login_can_set_xai_grok_as_main_provider(tmp_path):
def test_provider_login_model_implies_set_main_provider(tmp_path):
config_path = tmp_path / "config.json"
original = cli_commands._LOGIN_HANDLERS["github_copilot"]
cli_commands._LOGIN_HANDLERS["github_copilot"] = lambda: None
original = provider_commands._LOGIN_HANDLERS["github_copilot"]
provider_commands._LOGIN_HANDLERS["github_copilot"] = lambda: None
try:
result = runner.invoke(
app,
@ -784,7 +801,7 @@ def test_provider_login_model_implies_set_main_provider(tmp_path):
],
)
finally:
cli_commands._LOGIN_HANDLERS["github_copilot"] = original
provider_commands._LOGIN_HANDLERS["github_copilot"] = original
assert result.exit_code == 0
assert "Set github-copilot as the main provider" in result.stdout
@ -1470,12 +1487,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.commands.sync_workspace_templates") as mock_sync_templates, \
patch("nanobot.cli.agent.sync_workspace_templates") as mock_sync_templates, \
patch("nanobot.providers.factory.make_provider", return_value=_fake_provider()), \
patch("nanobot.cli.commands._print_agent_response") as mock_print_response, \
patch("nanobot.cli.terminal._print_agent_response") as mock_print_response, \
patch("nanobot.bus.queue.MessageBus"), \
patch("nanobot.cron.service.CronService"), \
patch("nanobot.cli.commands.AgentLoop.from_config") as mock_from_config:
patch("nanobot.cli.agent.AgentLoop.from_config") as mock_from_config:
agent_loop = MagicMock()
agent_loop.channels_config = None
agent_loop.process_direct = AsyncMock(
@ -1544,7 +1561,7 @@ 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.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
monkeypatch.setattr("nanobot.cron.service.CronService", lambda _store: object())
@ -1562,8 +1579,8 @@ def test_agent_config_sets_active_path(monkeypatch, tmp_path: Path) -> None:
async def close_mcp(self) -> None:
return None
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.commands._print_agent_response", lambda *_args, **_kwargs: None)
monkeypatch.setattr("nanobot.cli.agent.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)])
@ -1582,7 +1599,7 @@ 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.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
@ -1604,8 +1621,8 @@ def test_agent_uses_workspace_directory_for_cron_store(monkeypatch, tmp_path: Pa
return None
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.commands._print_agent_response", lambda *_args, **_kwargs: None)
monkeypatch.setattr("nanobot.cli.agent.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)])
@ -1631,7 +1648,7 @@ 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.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
monkeypatch.setattr("nanobot.config.paths.get_cron_dir", lambda: legacy_dir)
@ -1654,8 +1671,8 @@ def test_agent_workspace_override_does_not_migrate_legacy_cron(
return None
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.commands._print_agent_response", lambda *_args, **_kwargs: None)
monkeypatch.setattr("nanobot.cli.agent.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.terminal._print_agent_response", lambda *_args, **_kwargs: None)
result = runner.invoke(
app,
@ -1687,7 +1704,7 @@ 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.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.agent.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: _fake_provider())
monkeypatch.setattr("nanobot.bus.queue.MessageBus", lambda: object())
monkeypatch.setattr("nanobot.config.paths.get_cron_dir", lambda: legacy_dir)
@ -1710,9 +1727,9 @@ def test_agent_custom_config_workspace_does_not_migrate_legacy_cron(
return None
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.agent.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr(
"nanobot.cli.commands._print_agent_response", lambda *_args, **_kwargs: None
"nanobot.cli.terminal._print_agent_response", lambda *_args, **_kwargs: None
)
result = runner.invoke(app, ["agent", "-m", "hello", "-c", str(config_file)])
@ -1774,20 +1791,20 @@ def test_heartbeat_retains_recent_messages_by_default():
],
)
def test_heartbeat_has_active_tasks(content, expected):
from nanobot.cli.commands import _heartbeat_has_active_tasks
from nanobot.cli.gateway_runtime import _heartbeat_has_active_tasks
assert _heartbeat_has_active_tasks(content) is expected
def test_heartbeat_skips_bundled_template():
from nanobot.cli.commands import _heartbeat_has_active_tasks
from nanobot.cli.gateway_runtime import _heartbeat_has_active_tasks
from nanobot.utils.helpers import load_bundled_template
assert _heartbeat_has_active_tasks(load_bundled_template("HEARTBEAT.md")) is False
def test_heartbeat_target_skips_archived_webui_sessions():
from nanobot.cli.commands import _pick_heartbeat_target_from_sessions
from nanobot.cli.gateway_runtime import _pick_heartbeat_target_from_sessions
target = _pick_heartbeat_target_from_sessions(
enabled_channels=["websocket"],
@ -1802,7 +1819,7 @@ def test_heartbeat_target_skips_archived_webui_sessions():
def test_heartbeat_target_uses_last_channel_for_unified_session():
from nanobot.cli.commands import _pick_heartbeat_target_from_sessions
from nanobot.cli.gateway_runtime import _pick_heartbeat_target_from_sessions
from nanobot.session.keys import LAST_CHANNEL_METADATA_KEY, UNIFIED_SESSION_KEY
target = _pick_heartbeat_target_from_sessions(
@ -1824,7 +1841,7 @@ def test_heartbeat_target_uses_last_channel_for_unified_session():
],
)
def test_heartbeat_target_rejects_unroutable_unified_metadata(metadata):
from nanobot.cli.commands import _pick_heartbeat_target_from_sessions
from nanobot.cli.gateway_runtime import _pick_heartbeat_target_from_sessions
from nanobot.session.keys import UNIFIED_SESSION_KEY
target = _pick_heartbeat_target_from_sessions(
@ -1865,9 +1882,17 @@ def _patch_webui_provider_ready(monkeypatch) -> None:
def _patch_gateway_ports_free(monkeypatch) -> None:
monkeypatch.setattr("nanobot.cli.commands._gateway_health_ready", lambda *_a, **_kw: False)
monkeypatch.setattr("nanobot.cli.commands._tcp_endpoint_reachable", lambda *_a, **_kw: False)
monkeypatch.setattr("nanobot.cli.commands._webui_endpoint_reachable", lambda *_a, **_kw: False)
monkeypatch.setattr("nanobot.cli.webui._gateway_health_ready", lambda *_a, **_kw: False)
monkeypatch.setattr("nanobot.cli.webui._tcp_endpoint_reachable", lambda *_a, **_kw: False)
monkeypatch.setattr("nanobot.cli.webui._webui_endpoint_reachable", lambda *_a, **_kw: False)
monkeypatch.setattr(
"nanobot.cli.gateway_runtime._tcp_endpoint_reachable",
lambda *_a, **_kw: False,
)
monkeypatch.setattr(
"nanobot.cli.gateway_runtime._webui_endpoint_reachable",
lambda *_a, **_kw: False,
)
def _patch_cli_command_runtime(
@ -1894,6 +1919,14 @@ def _patch_cli_command_runtime(
"nanobot.cli.commands.sync_workspace_templates",
sync_templates or (lambda _path: None),
)
monkeypatch.setattr(
"nanobot.cli.webui.sync_workspace_templates",
sync_templates or (lambda _path: None),
)
monkeypatch.setattr(
"nanobot.cli.gateway_runtime.sync_workspace_templates",
sync_templates or (lambda _path: None),
)
monkeypatch.setattr(
"nanobot.providers.factory.make_provider",
provider_factory,
@ -1907,7 +1940,7 @@ def _patch_cli_command_runtime(
lambda _config_path=None: _test_provider_snapshot(provider_factory(config), config),
)
monkeypatch.setattr(
"nanobot.cli.commands._provider_setup_error",
"nanobot.cli.webui_support._provider_setup_error",
lambda _config: None,
)
_patch_gateway_ports_free(monkeypatch)
@ -2008,10 +2041,10 @@ def test_heartbeat_empty_response_still_retains_recent_messages(
session_manager=_FakeSessionManager,
cron_service=_FakeCron,
)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.gateway_runtime.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
monkeypatch.setattr("nanobot.cli.commands.read_webui_sidebar_state", lambda: {})
monkeypatch.setattr("nanobot.cli.commands.evaluate_response", _unexpected_evaluator)
monkeypatch.setattr("nanobot.cli.gateway_runtime.read_webui_sidebar_state", lambda: {})
monkeypatch.setattr("nanobot.cli.gateway_runtime.evaluate_response", _unexpected_evaluator)
result = runner.invoke(app, ["gateway", "--config", str(config_file)])
@ -2034,7 +2067,7 @@ def test_webui_yes_creates_config_and_enables_local_websocket(
seen: dict[str, object] = {}
_patch_webui_provider_ready(monkeypatch)
monkeypatch.setattr(
"nanobot.cli.commands.sync_workspace_templates",
"nanobot.cli.webui.sync_workspace_templates",
lambda path: seen.__setitem__("templates", path),
)
@ -2042,7 +2075,7 @@ def test_webui_yes_creates_config_and_enables_local_websocket(
seen["gateway_config"] = config
seen["gateway_kwargs"] = kwargs
monkeypatch.setattr("nanobot.cli.commands._run_gateway", _fake_run_gateway)
monkeypatch.setattr("nanobot.cli.webui._run_gateway", _fake_run_gateway)
result = runner.invoke(
app,
@ -2091,13 +2124,17 @@ def test_webui_yes_starts_first_run_without_provider_setup(monkeypatch, tmp_path
seen: dict[str, object] = {}
monkeypatch.setattr(
"nanobot.cli.commands._provider_setup_error",
"nanobot.cli.webui_support._provider_setup_error",
lambda _config: "No API key configured for provider 'custom'.",
)
monkeypatch.setattr(
"nanobot.cli.webui._provider_setup_error",
lambda _config: "No API key configured for provider 'custom'.",
)
_patch_gateway_ports_free(monkeypatch)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.webui.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr(
"nanobot.cli.commands._run_gateway",
"nanobot.cli.webui._run_gateway",
lambda config, **kwargs: seen.update(config=config, **kwargs),
)
@ -2135,7 +2172,7 @@ def test_webui_missing_runtime_env_fails_before_starting_gateway(
encoding="utf-8",
)
monkeypatch.setattr(
"nanobot.cli.commands._run_gateway",
"nanobot.cli.webui._run_gateway",
lambda *_args, **_kwargs: pytest.fail("gateway must not start with unresolved config"),
)
@ -2189,9 +2226,9 @@ def test_webui_background_starts_runtime_and_opens_browser(monkeypatch, tmp_path
config_file.write_text("{}")
seen: dict[str, object] = {}
_patch_webui_provider_ready(monkeypatch)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.webui.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr(
"nanobot.cli.commands._prepare_webui_bundle_for_gateway",
"nanobot.cli.webui._prepare_webui_bundle_for_gateway",
lambda *_args, **_kwargs: None,
)
@ -2213,7 +2250,7 @@ def test_webui_background_starts_runtime_and_opens_browser(monkeypatch, tmp_path
monkeypatch.setattr("nanobot.gateway.GatewayRuntime", _FakeRuntime)
monkeypatch.setattr(
"nanobot.cli.commands._open_webui_browser",
"nanobot.cli.webui._open_webui_browser",
lambda url: seen.__setitem__("opened_url", url),
)
@ -2256,7 +2293,7 @@ def test_open_webui_browser_redacts_bootstrap_secret(monkeypatch, capsys) -> Non
url = "http://127.0.0.1:8765/#/?bootstrapSecret=super-secret"
monkeypatch.setattr("webbrowser.open", lambda value: opened.append(value))
cli_commands._open_webui_browser(url, wait=False)
cli_webui_support._open_webui_browser(url, wait=False)
assert opened == [url]
output = _strip_ansi(capsys.readouterr().out)
@ -2275,9 +2312,9 @@ def test_webui_background_restarts_when_config_changes_and_gateway_is_running(
config_file.write_text("{}")
seen: dict[str, object] = {}
_patch_webui_provider_ready(monkeypatch)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.webui.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr(
"nanobot.cli.commands._prepare_webui_bundle_for_gateway",
"nanobot.cli.webui._prepare_webui_bundle_for_gateway",
lambda *_args, **_kwargs: None,
)
@ -2306,7 +2343,7 @@ def test_webui_background_restarts_when_config_changes_and_gateway_is_running(
monkeypatch.setattr("nanobot.gateway.GatewayRuntime", _FakeRuntime)
monkeypatch.setattr(
"nanobot.cli.commands._open_webui_browser",
"nanobot.cli.webui._open_webui_browser",
lambda url: seen.__setitem__("opened_url", url),
)
@ -2347,15 +2384,15 @@ def test_webui_foreground_attaches_to_existing_managed_gateway(monkeypatch, tmp_
config_file.write_text("{}")
seen: dict[str, object] = {}
_patch_webui_provider_ready(monkeypatch)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.commands._gateway_health_ready", lambda *_args, **_kwargs: True)
monkeypatch.setattr("nanobot.cli.commands._webui_endpoint_reachable", lambda *_args, **_kwargs: True)
monkeypatch.setattr("nanobot.cli.webui.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.webui._gateway_health_ready", lambda *_args, **_kwargs: True)
monkeypatch.setattr("nanobot.cli.webui._webui_endpoint_reachable", lambda *_args, **_kwargs: True)
monkeypatch.setattr(
"nanobot.cli.commands._open_webui_browser",
"nanobot.cli.webui._open_webui_browser",
lambda url, **kwargs: seen.update({"opened_url": url, "open_kwargs": kwargs}),
)
monkeypatch.setattr(
"nanobot.cli.commands._run_gateway",
"nanobot.cli.webui._run_gateway",
lambda *_args, **_kwargs: pytest.fail("existing gateway should be reused"),
)
@ -2368,7 +2405,7 @@ def test_webui_foreground_attaches_to_existing_managed_gateway(monkeypatch, tmp_
monkeypatch.setattr("nanobot.gateway.GatewayRuntime", _FakeRuntime)
monkeypatch.setattr(
"nanobot.cli.commands._attach_to_background_gateway",
"nanobot.cli.webui._attach_to_background_gateway",
lambda runtime: seen.__setitem__("attached_runtime", runtime),
)
@ -2401,9 +2438,9 @@ def test_attach_to_background_gateway_stops_on_ctrl_c(monkeypatch, capsys) -> No
def _interrupt(_seconds: float) -> None:
raise KeyboardInterrupt
monkeypatch.setattr("nanobot.cli.commands.time.sleep", _interrupt)
monkeypatch.setattr("nanobot.cli.webui_support.time.sleep", _interrupt)
cli_commands._attach_to_background_gateway(_FakeRuntime())
cli_webui_support._attach_to_background_gateway(_FakeRuntime())
assert stopped is True
output = capsys.readouterr().out
@ -2416,12 +2453,12 @@ def test_webui_foreground_does_not_claim_unmanaged_gateway(monkeypatch, tmp_path
config_file = tmp_path / "config.json"
config_file.write_text("{}")
_patch_webui_provider_ready(monkeypatch)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.commands._gateway_health_ready", lambda *_args: True)
monkeypatch.setattr("nanobot.cli.commands._webui_endpoint_reachable", lambda *_args: True)
monkeypatch.setattr("nanobot.cli.commands._open_webui_browser", lambda *_args, **_kwargs: None)
monkeypatch.setattr("nanobot.cli.webui.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.webui._gateway_health_ready", lambda *_args: True)
monkeypatch.setattr("nanobot.cli.webui._webui_endpoint_reachable", lambda *_args: True)
monkeypatch.setattr("nanobot.cli.webui._open_webui_browser", lambda *_args, **_kwargs: None)
monkeypatch.setattr(
"nanobot.cli.commands._attach_to_background_gateway",
"nanobot.cli.webui._attach_to_background_gateway",
lambda _runtime: pytest.fail("unmanaged gateway must not be attached"),
)
@ -2444,12 +2481,12 @@ def test_webui_foreground_refuses_occupied_webui_port(monkeypatch, tmp_path: Pat
config_file = tmp_path / "config.json"
config_file.write_text("{}")
_patch_webui_provider_ready(monkeypatch)
monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.commands._gateway_health_ready", lambda *_args, **_kwargs: False)
monkeypatch.setattr("nanobot.cli.commands._webui_endpoint_reachable", lambda *_args, **_kwargs: True)
monkeypatch.setattr("nanobot.cli.commands._tcp_endpoint_reachable", lambda *_args, **_kwargs: False)
monkeypatch.setattr("nanobot.cli.webui.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.webui._gateway_health_ready", lambda *_args, **_kwargs: False)
monkeypatch.setattr("nanobot.cli.webui._webui_endpoint_reachable", lambda *_args, **_kwargs: True)
monkeypatch.setattr("nanobot.cli.webui._tcp_endpoint_reachable", lambda *_args, **_kwargs: False)
monkeypatch.setattr(
"nanobot.cli.commands._run_gateway",
"nanobot.cli.webui._run_gateway",
lambda *_args, **_kwargs: pytest.fail("gateway should not start on occupied ports"),
)
@ -2596,9 +2633,9 @@ def test_gateway_unbound_agent_cron_is_skipped(
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.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.gateway_runtime.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: provider)
monkeypatch.setattr("nanobot.cli.commands._provider_setup_error", lambda _config: None)
monkeypatch.setattr("nanobot.cli.webui_support._provider_setup_error", lambda _config: None)
_patch_gateway_ports_free(monkeypatch)
monkeypatch.setattr(
"nanobot.providers.factory.build_provider_snapshot",
@ -2672,10 +2709,10 @@ def test_gateway_unbound_agent_cron_is_skipped(
raise AssertionError("unbound cron job must not be evaluated for delivery")
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.gateway_runtime.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _StopAfterCronSetup)
monkeypatch.setattr(
"nanobot.cli.commands.evaluate_response",
"nanobot.cli.gateway_runtime.evaluate_response",
_capture_evaluate_response,
)
@ -2724,9 +2761,9 @@ def test_gateway_bound_cron_runs_as_session_turn(
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.commands.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.cli.gateway_runtime.sync_workspace_templates", lambda _path: None)
monkeypatch.setattr("nanobot.providers.factory.make_provider", lambda _config: provider)
monkeypatch.setattr("nanobot.cli.commands._provider_setup_error", lambda _config: None)
monkeypatch.setattr("nanobot.cli.webui_support._provider_setup_error", lambda _config: None)
_patch_gateway_ports_free(monkeypatch)
monkeypatch.setattr(
"nanobot.providers.factory.build_provider_snapshot",
@ -2788,9 +2825,9 @@ def test_gateway_bound_cron_runs_as_session_turn(
raise AssertionError("bound cron must not use legacy response evaluator")
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCron)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.gateway_runtime.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _StopAfterCronSetup)
monkeypatch.setattr("nanobot.cli.commands.evaluate_response", _unexpected_evaluator)
monkeypatch.setattr("nanobot.cli.gateway_runtime.evaluate_response", _unexpected_evaluator)
result = runner.invoke(app, ["gateway", "--config", str(config_file)])
assert isinstance(result.exception, _StopGatewayError)
@ -2984,7 +3021,7 @@ def test_gateway_local_trigger_queue_submits_agent_turns(
self.runtime_resolver = MagicMock()
seen["agent"] = self
def _schedule_background(self, _coro) -> None:
def schedule_background(self, _coro) -> None:
return None
async def run(self) -> None:
@ -3016,7 +3053,7 @@ def test_gateway_local_trigger_queue_submits_agent_turns(
seen["local_trigger_queue_kwargs"] = kwargs
raise _StopGatewayError("stop")
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.gateway_runtime.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
monkeypatch.setattr(
"nanobot.triggers.local_runner.run_local_trigger_queue",
@ -3124,7 +3161,7 @@ def test_gateway_custom_config_workspace_does_not_migrate_legacy_cron(
def test_migrate_cron_store_moves_legacy_file(tmp_path: Path) -> None:
"""Legacy global jobs.json is moved into the workspace on first run."""
from nanobot.cli.commands import _migrate_cron_store
from nanobot.cli.runtime_config import _migrate_cron_store
legacy_dir = tmp_path / "global" / "cron"
legacy_dir.mkdir(parents=True)
@ -3145,7 +3182,7 @@ def test_migrate_cron_store_moves_legacy_file(tmp_path: Path) -> None:
def test_migrate_cron_store_skips_when_workspace_file_exists(tmp_path: Path) -> None:
"""Migration does not overwrite an existing workspace cron store."""
from nanobot.cli.commands import _migrate_cron_store
from nanobot.cli.runtime_config import _migrate_cron_store
legacy_dir = tmp_path / "global" / "cron"
legacy_dir.mkdir(parents=True)
@ -3312,7 +3349,7 @@ def test_gateway_health_endpoint_binds_and_serves_expected_responses(
message_bus=lambda: object(),
session_manager=lambda _workspace: object(),
)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.gateway_runtime.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCronService)
monkeypatch.setattr("asyncio.start_server", _fake_start_server)
@ -3363,13 +3400,14 @@ def test_gateway_health_endpoint_binds_and_serves_expected_responses(
async def read(self, _size: int) -> bytes:
nonlocal started
started += 1
if started == cli_commands._GATEWAY_HEALTH_MAX_CONNECTIONS:
if started == cli_gateway_runtime._GATEWAY_HEALTH_MAX_CONNECTIONS:
all_started.set()
await release.wait()
return b"GET /health HTTP/1.1\r\n\r\n"
active_writers = [
_FakeWriter() for _ in range(cli_commands._GATEWAY_HEALTH_MAX_CONNECTIONS)
_FakeWriter()
for _ in range(cli_gateway_runtime._GATEWAY_HEALTH_MAX_CONNECTIONS)
]
active_tasks = [
asyncio.create_task(health_handler(_BlockingReader(), writer))
@ -3394,7 +3432,11 @@ def test_gateway_health_endpoint_binds_and_serves_expected_responses(
async def read(self, _size: int) -> bytes:
await asyncio.Event().wait()
monkeypatch.setattr(cli_commands, "_GATEWAY_HEALTH_READ_TIMEOUT_SECONDS", 0.01)
monkeypatch.setattr(
cli_gateway_runtime,
"_GATEWAY_HEALTH_READ_TIMEOUT_SECONDS",
0.01,
)
timed_out_writer = _FakeWriter()
asyncio.run(health_handler(_NeverRespondingReader(), timed_out_writer))
assert timed_out_writer.closed is True
@ -3485,7 +3527,7 @@ def test_gateway_shutdown_lets_agent_task_own_mcp_cleanup(
message_bus=lambda: object(),
session_manager=lambda _workspace: object(),
)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.gateway_runtime.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCronService)
monkeypatch.setattr("asyncio.start_server", _fake_start_server)
@ -3601,12 +3643,12 @@ def test_gateway_shutdown_event_exits_forever_runtime_tasks(
message_bus=lambda: object(),
session_manager=lambda _workspace: object(),
)
monkeypatch.setattr("nanobot.cli.commands.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.cli.gateway_runtime.AgentLoop", _FakeAgentLoop)
monkeypatch.setattr("nanobot.channels.manager.ChannelManager", _FakeChannelManager)
monkeypatch.setattr("nanobot.cron.service.CronService", _FakeCronService)
monkeypatch.setattr("asyncio.start_server", _fake_start_server)
monkeypatch.setattr(
"nanobot.cli.commands._install_gateway_shutdown_handlers",
"nanobot.cli.gateway_runtime._install_gateway_shutdown_handlers",
_fake_install_shutdown_handlers,
)

2
tests/cli/test_config_diagnostics.py
View File

@ -413,7 +413,7 @@ def test_gateway_missing_provider_managed_start_for_webui_setup(
monkeypatch.setattr(GatewayRuntime, "start_background", fake_start_background)
monkeypatch.setattr(GatewayRuntime, "restart", fake_restart)
monkeypatch.setattr(
"nanobot.cli.commands.ensure_webui_bundle",
"nanobot.cli.webui_support.ensure_webui_bundle",
lambda **_kwargs: None,
)

42
tests/cli/test_interactive_retry_wait.py
View File

@ -4,7 +4,7 @@ from unittest.mock import patch
import pytest
from nanobot.bus.outbound_events import ProgressEvent, RetryWaitEvent
from nanobot.cli import commands
from nanobot.cli import terminal
@pytest.mark.asyncio
@ -22,8 +22,8 @@ async def test_interactive_retry_wait_is_rendered_as_progress_even_when_progress
async def fake_print(text: str, active_thinking: object | None, renderer=None) -> None:
calls.append((text, active_thinking))
with patch("nanobot.cli.commands._print_interactive_progress_line", side_effect=fake_print):
handled = await commands._maybe_print_interactive_progress(
with patch("nanobot.cli.terminal._print_interactive_progress_line", side_effect=fake_print):
handled = await terminal._maybe_print_interactive_progress(
msg,
thinking,
channels_config,
@ -46,8 +46,8 @@ async def test_reasoning_displayed_when_show_reasoning_enabled():
metadata={},
)
with patch("nanobot.cli.commands._print_cli_reasoning", side_effect=lambda t, th, r=None: calls.append(t)):
handled = await commands._maybe_print_interactive_progress(msg, None, channels_config)
with patch("nanobot.cli.terminal._print_cli_reasoning", side_effect=lambda t, th, r=None: calls.append(t)):
handled = await terminal._maybe_print_interactive_progress(msg, None, channels_config)
assert handled is True
assert calls == ["Let me think about this..."]
@ -66,8 +66,8 @@ async def test_reasoning_delta_displayed_when_show_reasoning_enabled():
metadata={},
)
with patch("nanobot.cli.commands._print_cli_reasoning", side_effect=lambda t, th, r=None: calls.append(t)):
handled = await commands._maybe_print_interactive_progress(msg, None, channels_config)
with patch("nanobot.cli.terminal._print_cli_reasoning", side_effect=lambda t, th, r=None: calls.append(t)):
handled = await terminal._maybe_print_interactive_progress(msg, None, channels_config)
assert handled is True
assert calls == ["I should search first."]
@ -79,10 +79,10 @@ async def test_reasoning_delta_buffers_until_sentence_boundary():
channels_config = SimpleNamespace(
send_progress=True, send_tool_hints=False, show_reasoning=True,
)
reasoning_buffer = commands._ReasoningBuffer()
reasoning_buffer = terminal._ReasoningBuffer()
with patch("nanobot.cli.commands._print_cli_reasoning", side_effect=lambda t, th, r=None: calls.append(t)):
first = await commands._maybe_print_interactive_progress(
with patch("nanobot.cli.terminal._print_cli_reasoning", side_effect=lambda t, th, r=None: calls.append(t)):
first = await terminal._maybe_print_interactive_progress(
SimpleNamespace(
content="The",
event=ProgressEvent(content="The", reasoning_delta=True),
@ -92,7 +92,7 @@ async def test_reasoning_delta_buffers_until_sentence_boundary():
channels_config,
reasoning_buffer=reasoning_buffer,
)
second = await commands._maybe_print_interactive_progress(
second = await terminal._maybe_print_interactive_progress(
SimpleNamespace(
content=" user asked.",
event=ProgressEvent(content=" user asked.", reasoning_delta=True),
@ -114,10 +114,10 @@ async def test_reasoning_end_flushes_buffered_delta():
channels_config = SimpleNamespace(
send_progress=True, send_tool_hints=False, show_reasoning=True,
)
reasoning_buffer = commands._ReasoningBuffer()
reasoning_buffer = terminal._ReasoningBuffer()
with patch("nanobot.cli.commands._print_cli_reasoning", side_effect=lambda t, th, r=None: calls.append(t)):
delta = await commands._maybe_print_interactive_progress(
with patch("nanobot.cli.terminal._print_cli_reasoning", side_effect=lambda t, th, r=None: calls.append(t)):
delta = await terminal._maybe_print_interactive_progress(
SimpleNamespace(
content="The user asked",
event=ProgressEvent(content="The user asked", reasoning_delta=True),
@ -127,7 +127,7 @@ async def test_reasoning_end_flushes_buffered_delta():
channels_config,
reasoning_buffer=reasoning_buffer,
)
end = await commands._maybe_print_interactive_progress(
end = await terminal._maybe_print_interactive_progress(
SimpleNamespace(
content="",
event=ProgressEvent(reasoning_end=True),
@ -155,8 +155,8 @@ async def test_reasoning_hidden_when_show_reasoning_disabled():
metadata={},
)
with patch("nanobot.cli.commands._print_cli_reasoning") as mock_reasoning:
handled = await commands._maybe_print_interactive_progress(msg, None, channels_config)
with patch("nanobot.cli.terminal._print_cli_reasoning") as mock_reasoning:
handled = await terminal._maybe_print_interactive_progress(msg, None, channels_config)
assert handled is True
mock_reasoning.assert_not_called()
@ -178,8 +178,8 @@ async def test_non_reasoning_progress_not_affected_by_show_reasoning():
async def fake_print(text: str, thinking=None, renderer=None):
calls.append(text)
with patch("nanobot.cli.commands._print_interactive_progress_line", side_effect=fake_print):
handled = await commands._maybe_print_interactive_progress(msg, None, channels_config)
with patch("nanobot.cli.terminal._print_interactive_progress_line", side_effect=fake_print):
handled = await terminal._maybe_print_interactive_progress(msg, None, channels_config)
assert handled is True
assert calls == ["working on it..."]
@ -200,10 +200,10 @@ async def test_reasoning_shown_when_send_progress_disabled():
)
with patch(
"nanobot.cli.commands._print_cli_reasoning",
"nanobot.cli.terminal._print_cli_reasoning",
side_effect=lambda t, th, r=None: calls.append(t),
):
handled = await commands._maybe_print_interactive_progress(msg, None, channels_config)
handled = await terminal._maybe_print_interactive_progress(msg, None, channels_config)
assert handled is True
assert calls == ["Let me think about this..."]

7
tests/cli/test_safe_file_history.py
View File

@ -3,7 +3,12 @@
Surrogate characters in CLI input must not crash history file writes.
"""
from nanobot.cli.commands import SafeFileHistory, _sanitize_surrogates
from nanobot.cli.commands import SafeFileHistory as LegacySafeFileHistory
from nanobot.cli.terminal import SafeFileHistory, _sanitize_surrogates
def test_commands_keeps_safe_file_history_import_compatible() -> None:
assert LegacySafeFileHistory is SafeFileHistory
class TestSanitizeSurrogates:

2
tests/command/test_router_dispatchable.py
View File

@ -108,7 +108,7 @@ class TestMidTurnCommandDispatchedDirectly:
))
loop.sessions.save = MagicMock()
loop.sessions.invalidate = MagicMock()
loop._schedule_background = MagicMock()
loop.schedule_background = MagicMock()
loop._cancel_active_tasks = AsyncMock(return_value=0)
return loop