chengyongruandXubin Ren
883776358e
maintainer edit: Align OpenAI-compatible API auth with the WebSocket channel boundary: loopback serve remains usable without a key, while wildcard binds still fail before agent initialization unless api.api_key is configured.
2525 lines
93 KiB
Python
2525 lines
93 KiB
Python
"""CLI commands for nanobot."""
|
||
|
||
import asyncio
|
||
import os
|
||
import select
|
||
import signal
|
||
import sys
|
||
from collections.abc import Callable, Iterable
|
||
from contextlib import nullcontext, suppress
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
# Force UTF-8 encoding for Windows console
|
||
if sys.platform == "win32":
|
||
if sys.stdout.encoding != "utf-8":
|
||
os.environ["PYTHONIOENCODING"] = "utf-8"
|
||
# Re-open stdout/stderr with UTF-8 encoding
|
||
with suppress(Exception):
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
||
|
||
# Keep console encoding setup before importing CLI UI/logging libraries.
|
||
import typer # noqa: E402
|
||
from loguru import logger # noqa: E402
|
||
|
||
# Remove default handler and re-add with unified nanobot format
|
||
logger.remove()
|
||
_log_handler_id = logger.add(
|
||
sys.stderr,
|
||
format=(
|
||
"<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
|
||
"<level>{level: <5}</level> | "
|
||
"<cyan>{extra[channel]}</cyan> | "
|
||
"<level>{message}</level>"
|
||
),
|
||
level="INFO",
|
||
colorize=None,
|
||
filter=lambda record: record["extra"].setdefault("channel", "-") or True,
|
||
)
|
||
|
||
|
||
def _set_nanobot_logs(enabled: bool) -> None:
|
||
if enabled:
|
||
logger.enable("nanobot")
|
||
else:
|
||
logger.disable("nanobot")
|
||
|
||
|
||
from prompt_toolkit import PromptSession, print_formatted_text # noqa: E402
|
||
from prompt_toolkit.application import run_in_terminal # noqa: E402
|
||
from prompt_toolkit.formatted_text import ANSI, HTML # noqa: E402
|
||
from prompt_toolkit.history import FileHistory # noqa: E402
|
||
from prompt_toolkit.key_binding import KeyBindings # noqa: E402
|
||
from prompt_toolkit.keys import Keys # noqa: E402
|
||
from prompt_toolkit.patch_stdout import patch_stdout # noqa: E402
|
||
from rich.console import Console # noqa: E402
|
||
from rich.markdown import Markdown # noqa: E402
|
||
from rich.markup import escape # noqa: E402
|
||
from rich.table import Table # noqa: E402
|
||
from rich.text import Text # noqa: E402
|
||
|
||
from nanobot import __logo__, __version__ # noqa: E402
|
||
from nanobot import optional_features as feature_support # noqa: E402
|
||
from nanobot.agent.loop import AgentLoop # noqa: E402
|
||
from nanobot.bus.outbound_events import ( # noqa: E402
|
||
ProgressEvent,
|
||
RetryWaitEvent,
|
||
StreamDeltaEvent,
|
||
StreamedResponseEvent,
|
||
StreamEndEvent,
|
||
outbound_event_from_message,
|
||
)
|
||
from nanobot.cli.gateway import create_gateway_app # noqa: E402
|
||
from nanobot.cli.stream import StreamRenderer, ThinkingSpinner # noqa: E402
|
||
from nanobot.config.paths import get_workspace_path, is_default_workspace # noqa: E402
|
||
from nanobot.config.schema import Config # noqa: E402
|
||
from nanobot.utils.evaluator import evaluate_response # noqa: E402
|
||
from nanobot.utils.helpers import sync_workspace_templates # noqa: E402
|
||
from nanobot.utils.restart import ( # noqa: E402
|
||
consume_restart_notice_from_env,
|
||
format_restart_completed_message,
|
||
should_show_cli_restart_notice,
|
||
)
|
||
from nanobot.webui.sidebar_state import read_webui_sidebar_state # noqa: E402
|
||
|
||
|
||
def _sanitize_surrogates(text: str) -> str:
|
||
"""Reconstruct surrogate pairs into real characters; replace lone surrogates.
|
||
|
||
On Windows, console input may produce lone surrogate code points (e.g.
|
||
``\\ud83d\\udc08`` for U+1F408). Round-tripping through UTF-16 reconstructs
|
||
paired surrogates into their actual characters and replaces unpaired ones
|
||
with U+FFFD.
|
||
"""
|
||
return text.encode("utf-16-le", errors="surrogatepass").decode("utf-16-le", errors="replace")
|
||
|
||
|
||
def _signal_name(signum: int) -> str:
|
||
with suppress(ValueError):
|
||
return signal.Signals(signum).name
|
||
return f"signal {signum}"
|
||
|
||
|
||
def _ensure_gateway_tty_signal_mode() -> None:
|
||
"""Keep foreground gateway Ctrl+C usable even 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)
|
||
lflag = attrs[3]
|
||
required = termios.ISIG | termios.ICANON | termios.ECHO
|
||
if (lflag & required) == required:
|
||
return
|
||
attrs[3] = lflag | required
|
||
termios.tcsetattr(fd, termios.TCSANOW, attrs)
|
||
termios.tcflush(fd, termios.TCIFLUSH)
|
||
logger.debug("Restored foreground gateway TTY signal mode")
|
||
|
||
|
||
def _install_gateway_shutdown_handlers(
|
||
loop: asyncio.AbstractEventLoop,
|
||
shutdown_event: asyncio.Event,
|
||
tasks: list[asyncio.Task],
|
||
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)
|
||
|
||
|
||
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))
|
||
|
||
|
||
app = typer.Typer(
|
||
name="nanobot",
|
||
context_settings={"help_option_names": ["-h", "--help"]},
|
||
help=f"{__logo__} nanobot - Personal AI Assistant",
|
||
no_args_is_help=True,
|
||
)
|
||
|
||
console = Console()
|
||
EXIT_COMMANDS = {"exit", "quit", "/exit", "/quit", ":q"}
|
||
_REASONING_SENTENCE_ENDINGS = (".", "!", "?", "。", "！", "？")
|
||
_REASONING_FLUSH_CHARS = 60
|
||
|
||
_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],
|
||
) -> 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 ":" 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"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CLI input: prompt_toolkit for editing, paste, history, and display
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_PROMPT_SESSION: PromptSession | None = None
|
||
_SAVED_TERM_ATTRS = None # original termios settings, restored on exit
|
||
|
||
|
||
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.
|
||
"""
|
||
kb = KeyBindings()
|
||
|
||
@kb.add("enter")
|
||
def _(event):
|
||
event.current_buffer.validate_and_handle()
|
||
|
||
@kb.add("escape", "enter") # Alt+Enter / Meta+Enter (ESC + CR, "\x1b\r")
|
||
def _(event):
|
||
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):
|
||
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) -> str:
|
||
"""Render Rich output to ANSI so prompt_toolkit can print it safely."""
|
||
ansi_console = Console(
|
||
force_terminal=sys.stdout.isatty(),
|
||
color_system=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 | 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 | None = None):
|
||
"""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 | None = None,
|
||
) -> None:
|
||
"""Print async interactive replies with prompt_toolkit-safe Rich styling."""
|
||
def _write() -> None:
|
||
content = response or ""
|
||
ansi = _render_interactive_ansi(
|
||
lambda c: (
|
||
c.print(),
|
||
c.print(f"[cyan]{__logo__} nanobot[/cyan]"),
|
||
c.print(_response_renderable(content, render_markdown, metadata)),
|
||
c.print(),
|
||
)
|
||
)
|
||
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
|
||
|
||
|
||
def version_callback(value: bool):
|
||
if value:
|
||
console.print(f"{__logo__} nanobot v{__version__}")
|
||
raise typer.Exit()
|
||
|
||
|
||
@app.callback()
|
||
def main(
|
||
version: bool = typer.Option(
|
||
None, "--version", "-v", callback=version_callback, is_eager=True
|
||
),
|
||
):
|
||
"""nanobot - Personal AI Assistant."""
|
||
pass
|
||
|
||
|
||
# ============================================================================
|
||
# Onboard / Setup
|
||
# ============================================================================
|
||
|
||
|
||
@app.command()
|
||
def onboard(
|
||
workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
|
||
config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
|
||
wizard: bool = typer.Option(False, "--wizard", help="Use interactive wizard"),
|
||
):
|
||
"""Initialize nanobot configuration and workspace."""
|
||
from nanobot.config.loader import get_config_path, load_config, save_config, set_config_path
|
||
from nanobot.config.schema import Config
|
||
|
||
if config:
|
||
config_path = Path(config).expanduser().resolve()
|
||
set_config_path(config_path)
|
||
console.print(f"[dim]Using config: {config_path}[/dim]")
|
||
else:
|
||
config_path = get_config_path()
|
||
|
||
def _apply_workspace_override(loaded: Config) -> Config:
|
||
if workspace:
|
||
loaded.agents.defaults.workspace = workspace
|
||
return loaded
|
||
|
||
# Create or update config
|
||
if config_path.exists():
|
||
if wizard:
|
||
config = _apply_workspace_override(load_config(config_path))
|
||
else:
|
||
console.print(f"[yellow]Config already exists at {config_path}[/yellow]")
|
||
console.print(
|
||
" [bold]y[/bold] = overwrite with defaults (existing values will be lost)"
|
||
)
|
||
console.print(
|
||
" [bold]N[/bold] = refresh config, keeping existing values and adding new fields"
|
||
)
|
||
if typer.confirm("Overwrite?"):
|
||
config = _apply_workspace_override(Config())
|
||
save_config(config, config_path)
|
||
console.print(f"[green]✓[/green] Config reset to defaults at {config_path}")
|
||
else:
|
||
config = _apply_workspace_override(load_config(config_path))
|
||
save_config(config, config_path)
|
||
console.print(
|
||
f"[green]✓[/green] Config refreshed at {config_path} (existing values preserved)"
|
||
)
|
||
else:
|
||
config = _apply_workspace_override(Config())
|
||
# In wizard mode, don't save yet - the wizard will handle saving if should_save=True
|
||
if not wizard:
|
||
save_config(config, config_path)
|
||
console.print(f"[green]✓[/green] Created config at {config_path}")
|
||
|
||
# Run interactive wizard if enabled
|
||
if wizard:
|
||
from nanobot.cli.onboard import run_onboard
|
||
|
||
try:
|
||
result = run_onboard(initial_config=config)
|
||
if not result.should_save:
|
||
console.print("[yellow]Configuration discarded. No changes were saved.[/yellow]")
|
||
return
|
||
|
||
config = result.config
|
||
save_config(config, config_path)
|
||
console.print(f"[green]✓[/green] Config saved at {config_path}")
|
||
except Exception as e:
|
||
console.print(f"[red]✗[/red] Error during configuration: {e}")
|
||
console.print("[yellow]Please run 'nanobot onboard' again to complete setup.[/yellow]")
|
||
raise typer.Exit(1)
|
||
_onboard_plugins(config_path)
|
||
|
||
# Create workspace, preferring the configured workspace path.
|
||
workspace_path = get_workspace_path(config.workspace_path)
|
||
if not workspace_path.exists():
|
||
workspace_path.mkdir(parents=True, exist_ok=True)
|
||
console.print(f"[green]✓[/green] Created workspace at {workspace_path}")
|
||
|
||
sync_workspace_templates(workspace_path)
|
||
|
||
agent_cmd = 'nanobot agent -m "Hello!"'
|
||
gateway_cmd = "nanobot gateway"
|
||
if config:
|
||
agent_cmd += f" --config {config_path}"
|
||
gateway_cmd += f" --config {config_path}"
|
||
|
||
console.print(f"\n{__logo__} nanobot is ready!")
|
||
console.print("\nNext steps:")
|
||
if wizard:
|
||
console.print(f" 1. Chat: [cyan]{agent_cmd}[/cyan]")
|
||
console.print(f" 2. Start gateway: [cyan]{gateway_cmd}[/cyan]")
|
||
else:
|
||
console.print(f" 1. Add your API key to [cyan]{config_path}[/cyan]")
|
||
console.print(" Get one at: https://openrouter.ai/keys")
|
||
console.print(f" 2. Chat: [cyan]{agent_cmd}[/cyan]")
|
||
console.print(
|
||
"\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]"
|
||
)
|
||
|
||
|
||
def _merge_missing_defaults(existing: Any, defaults: Any) -> Any:
|
||
"""Recursively fill in missing values from defaults without overwriting user config."""
|
||
if not isinstance(existing, dict) or not isinstance(defaults, dict):
|
||
return existing
|
||
|
||
merged = dict(existing)
|
||
for key, value in defaults.items():
|
||
if key not in merged:
|
||
merged[key] = value
|
||
else:
|
||
merged[key] = _merge_missing_defaults(merged[key], value)
|
||
return merged
|
||
|
||
|
||
def _onboard_plugins(config_path: Path) -> None:
|
||
"""Inject default config for all discovered channels (built-in + plugins)."""
|
||
import json
|
||
|
||
from nanobot.channels.registry import discover_all
|
||
|
||
all_channels = discover_all()
|
||
if not all_channels:
|
||
return
|
||
|
||
with open(config_path, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
|
||
channels = data.setdefault("channels", {})
|
||
for name, cls in all_channels.items():
|
||
if name not in channels:
|
||
channels[name] = cls.default_config()
|
||
else:
|
||
channels[name] = _merge_missing_defaults(channels[name], cls.default_config())
|
||
|
||
with open(config_path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||
|
||
|
||
def _print_enable_options(
|
||
extras: dict[str, list[str] | None],
|
||
builtin_channels: set[str],
|
||
plugin_channels: dict[str, Any],
|
||
config: Config,
|
||
) -> None:
|
||
table = Table(title="Available Features")
|
||
table.add_column("Name", style="cyan")
|
||
table.add_column("Type")
|
||
table.add_column("Enabled")
|
||
|
||
for item in sorted(builtin_channels | set(plugin_channels) | set(extras)):
|
||
is_channel = item in builtin_channels or item in plugin_channels
|
||
enabled = (
|
||
feature_support.channel_enabled(config, item)
|
||
if is_channel
|
||
else feature_support.extra_installed(item, extras[item])
|
||
)
|
||
table.add_row(
|
||
item,
|
||
"channel" if is_channel else "feature",
|
||
"[green]yes[/green]" if enabled else "[dim]no[/dim]",
|
||
)
|
||
|
||
console.print(table)
|
||
|
||
|
||
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 _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 load_config, resolve_config_env_vars, 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]")
|
||
|
||
try:
|
||
loaded = resolve_config_env_vars(load_config(config_path))
|
||
except ValueError as e:
|
||
console.print(f"[red]Error: {e}[/red]")
|
||
raise typer.Exit(1)
|
||
_warn_deprecated_config_keys(config_path)
|
||
if workspace:
|
||
loaded.agents.defaults.workspace = workspace
|
||
return loaded
|
||
|
||
|
||
def _read_trigger_cli_message(message: str | None) -> str:
|
||
"""Read a trigger message from an argument or stdin."""
|
||
if message and message.strip():
|
||
return message
|
||
try:
|
||
if not sys.stdin.isatty():
|
||
content = sys.stdin.read()
|
||
if content.strip():
|
||
return content
|
||
except Exception:
|
||
pass
|
||
console.print("[red]Error: trigger message is required[/red]")
|
||
raise typer.Exit(1)
|
||
|
||
|
||
def _warn_deprecated_config_keys(config_path: Path | None) -> None:
|
||
"""Hint users to remove obsolete keys from their config file."""
|
||
import json
|
||
|
||
from nanobot.config.loader import get_config_path
|
||
|
||
path = config_path or get_config_path()
|
||
try:
|
||
raw = json.loads(path.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
return
|
||
if "memoryWindow" in raw.get("agents", {}).get("defaults", {}):
|
||
console.print(
|
||
"[dim]Hint: `memoryWindow` in your config is no longer used "
|
||
"and can be safely removed.[/dim]"
|
||
)
|
||
|
||
|
||
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.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 ValueError as exc:
|
||
console.print(f"[red]Error: {exc}[/red]")
|
||
raise typer.Exit(1) from exc
|
||
_warn_deprecated_config_keys(display_path)
|
||
if workspace:
|
||
loaded.agents.defaults.workspace = workspace
|
||
return display_path, loaded
|
||
|
||
|
||
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
|
||
try:
|
||
interactive = sys.stdin.isatty()
|
||
except Exception:
|
||
interactive = False
|
||
if not interactive:
|
||
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 _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."""
|
||
from nanobot.config.loader import load_config
|
||
|
||
try:
|
||
return load_config(config_path)
|
||
except ValueError as e:
|
||
console.print(f"[red]Error: {e}[/red]")
|
||
raise typer.Exit(1) from e
|
||
|
||
|
||
def _provider_setup_error(config: Config) -> str | None:
|
||
"""Return the provider setup error, or None when the current model can start."""
|
||
from nanobot.config.loader import resolve_config_env_vars
|
||
from nanobot.providers.factory import build_provider_snapshot
|
||
|
||
try:
|
||
build_provider_snapshot(resolve_config_env_vars(config.model_copy(deep=True)))
|
||
except ValueError as exc:
|
||
return str(exc)
|
||
return None
|
||
|
||
|
||
def _webui_config_dict(config: Config) -> dict[str, Any]:
|
||
"""Return the current WebSocket config as a mutable alias-key dictionary."""
|
||
from nanobot.channels.websocket import WebSocketConfig
|
||
|
||
current = getattr(config.channels, "websocket", None) or {}
|
||
model = WebSocketConfig.model_validate(current)
|
||
return model.model_dump(by_alias=True, exclude_none=True)
|
||
|
||
|
||
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 _webui_browser_url(config: Config) -> str:
|
||
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)
|
||
return f"http://{host}:{port}"
|
||
|
||
|
||
def _ensure_local_webui_channel(config: Config, *, port: int | None, yes: bool) -> bool:
|
||
"""Enable the local WebUI channel with safe localhost defaults."""
|
||
from nanobot.channels.websocket import WebSocketConfig
|
||
|
||
current = getattr(config.channels, "websocket", None) or {}
|
||
model = WebSocketConfig.model_validate(current)
|
||
changed = False
|
||
|
||
needs_enable = not model.enabled
|
||
needs_port = port is not None and model.port != port
|
||
if not needs_enable and not needs_port:
|
||
return 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: localhost bootstrap issues short-lived WebSocket tokens")
|
||
console.print(
|
||
" LAN access requires an explicit host change plus a WebUI password in config."
|
||
)
|
||
_confirm_webui_action("Enable 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
|
||
|
||
setattr(config.channels, "websocket", model.model_dump(by_alias=True, exclude_none=True))
|
||
return changed
|
||
|
||
|
||
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 socket
|
||
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:
|
||
try:
|
||
with socket.create_connection((host, port), timeout=0.2):
|
||
return
|
||
except OSError:
|
||
time.sleep(0.1)
|
||
|
||
|
||
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)
|
||
try:
|
||
webbrowser.open(url)
|
||
console.print(f"[green]✓[/green] Opened WebUI: [cyan]{url}[/cyan]")
|
||
except Exception as exc:
|
||
console.print(f"[yellow]Could not open browser ({exc}); visit {url}[/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:
|
||
"""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. Run `nanobot webui` interactively or "
|
||
"`nanobot onboard --wizard`.[/red]"
|
||
)
|
||
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
|
||
|
||
|
||
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))
|
||
|
||
|
||
@app.command()
|
||
def trigger(
|
||
trigger_id: str = typer.Argument(..., help="Trigger ID returned by /trigger"),
|
||
message: str | None = typer.Argument(None, help="Message to deliver; stdin is used when omitted"),
|
||
workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
|
||
config: str | None = typer.Option(None, "--config", "-c", help="Config file path"),
|
||
):
|
||
"""Deliver a local trigger message to its bound chat session."""
|
||
from nanobot.triggers.local_store import (
|
||
LocalTriggerStore,
|
||
TriggerDisabledError,
|
||
TriggerNotFoundError,
|
||
TriggerStoreError,
|
||
)
|
||
|
||
runtime_config = _load_runtime_config(config, workspace)
|
||
content = _read_trigger_cli_message(message)
|
||
store = LocalTriggerStore(runtime_config.workspace_path)
|
||
try:
|
||
delivery = store.enqueue(trigger_id, content)
|
||
except (TriggerNotFoundError, TriggerDisabledError) as exc:
|
||
console.print(f"[red]Error: {exc}[/red]")
|
||
raise typer.Exit(1) from exc
|
||
except (TriggerStoreError, ValueError) as exc:
|
||
console.print(f"[red]Error: {exc}[/red]")
|
||
raise typer.Exit(1) from exc
|
||
console.print(f"[green]Queued[/green] {delivery.trigger_id} ({delivery.id})")
|
||
|
||
|
||
# ============================================================================
|
||
# OpenAI-Compatible API Server
|
||
# ============================================================================
|
||
|
||
|
||
@app.command()
|
||
def serve(
|
||
port: int | None = typer.Option(None, "--port", "-p", help="API server port"),
|
||
host: str | None = typer.Option(None, "--host", "-H", help="Bind address"),
|
||
timeout: float | None = typer.Option(None, "--timeout", "-t", help="Per-request timeout (seconds)"),
|
||
verbose: bool = typer.Option(False, "--verbose", "-v", help="Show nanobot runtime logs"),
|
||
workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
|
||
config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
|
||
):
|
||
"""Start the OpenAI-compatible API server (/v1/chat/completions)."""
|
||
try:
|
||
from aiohttp import web # noqa: F401
|
||
except ImportError:
|
||
console.print("[red]aiohttp is required. Install with: nanobot plugins enable api[/red]")
|
||
raise typer.Exit(1)
|
||
|
||
from nanobot.api.server import create_app
|
||
from nanobot.bus.queue import MessageBus
|
||
from nanobot.providers.image_generation import image_gen_provider_configs
|
||
from nanobot.session.manager import SessionManager
|
||
|
||
_set_nanobot_logs(verbose)
|
||
|
||
runtime_config = _load_runtime_config(config, workspace)
|
||
api_cfg = runtime_config.api
|
||
host = host if host is not None else api_cfg.host
|
||
port = port if port is not None else api_cfg.port
|
||
timeout = timeout if timeout is not None else api_cfg.timeout
|
||
api_key = api_cfg.api_key.strip() if api_cfg.api_key else ""
|
||
if host in {"0.0.0.0", "::"} and not api_key:
|
||
console.print(
|
||
"[red]Error: host is 0.0.0.0 (all interfaces) but api_key is not set. "
|
||
"Set api.api_key in config to prevent unauthenticated access.[/red]"
|
||
)
|
||
raise typer.Exit(1)
|
||
sync_workspace_templates(runtime_config.workspace_path)
|
||
bus = MessageBus()
|
||
session_manager = SessionManager(runtime_config.workspace_path)
|
||
try:
|
||
agent_loop = AgentLoop.from_config(
|
||
runtime_config, bus,
|
||
session_manager=session_manager,
|
||
image_generation_provider_configs=image_gen_provider_configs(runtime_config),
|
||
)
|
||
except ValueError as exc:
|
||
console.print(f"[red]Error: {exc}[/red]")
|
||
raise typer.Exit(1) from exc
|
||
|
||
model_name, preset_tag = _model_display(runtime_config)
|
||
console.print(f"{__logo__} Starting OpenAI-compatible API server")
|
||
console.print(f" [cyan]Endpoint[/cyan] : http://{host}:{port}/v1/chat/completions")
|
||
console.print(f" [cyan]Model[/cyan] : {model_name}{preset_tag}")
|
||
console.print(" [cyan]Session[/cyan] : api:default")
|
||
console.print(f" [cyan]Timeout[/cyan] : {timeout}s")
|
||
if host in {"0.0.0.0", "::"}:
|
||
console.print(
|
||
"[yellow]API is bound to all interfaces "
|
||
"(authentication required).[/yellow]"
|
||
)
|
||
console.print()
|
||
|
||
api_app = create_app(
|
||
agent_loop, model_name=model_name, request_timeout=timeout,
|
||
api_key=api_key,
|
||
)
|
||
|
||
async def on_startup(_app):
|
||
await agent_loop._connect_mcp()
|
||
|
||
async def on_cleanup(_app):
|
||
await agent_loop.close_mcp()
|
||
|
||
api_app.on_startup.append(on_startup)
|
||
api_app.on_cleanup.append(on_cleanup)
|
||
|
||
web.run_app(api_app, host=host, port=port, print=lambda msg: logger.info(msg))
|
||
|
||
|
||
# ============================================================================
|
||
# WebUI Launcher
|
||
# ============================================================================
|
||
|
||
|
||
@app.command()
|
||
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="Start gateway in the background"),
|
||
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 save_config
|
||
from nanobot.gateway import GatewayRuntime, GatewayRuntimePaths, GatewayStartOptions
|
||
|
||
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
|
||
|
||
provider_error = _provider_setup_error(setup_config)
|
||
if provider_error:
|
||
console.print(f"[dim]Provider check: {provider_error}[/dim]")
|
||
setup_config = _run_quick_start_for_webui(setup_config, yes=yes)
|
||
if workspace:
|
||
setup_config.agents.defaults.workspace = workspace
|
||
|
||
try:
|
||
changed_webui = _ensure_local_webui_channel(setup_config, port=port, yes=yes)
|
||
_warn_webui_bind_scope(setup_config)
|
||
webui_url = _webui_browser_url(setup_config)
|
||
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_url}[/cyan]")
|
||
console.print(f"Gateway health: [cyan]http://{runtime_config.gateway.host}:{effective_gateway_port}/health[/cyan]")
|
||
if no_open:
|
||
console.print("[dim]Browser opening disabled by --no-open.[/dim]")
|
||
|
||
if background:
|
||
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,
|
||
)
|
||
)
|
||
result = runtime.start_background(
|
||
GatewayStartOptions(
|
||
port=effective_gateway_port,
|
||
workspace=workspace_arg,
|
||
config_path=config_arg,
|
||
)
|
||
)
|
||
if not result.ok and result.message != "gateway_already_running":
|
||
console.print(f"[yellow]Gateway was not started: {result.message}[/yellow]")
|
||
console.print(f"Logs: {result.status.log_path}")
|
||
raise typer.Exit(1)
|
||
if 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]"
|
||
)
|
||
if not no_open:
|
||
_open_webui_browser(webui_url)
|
||
return
|
||
|
||
_run_gateway(
|
||
runtime_config,
|
||
port=effective_gateway_port,
|
||
open_browser_url=None if no_open else webui_url,
|
||
)
|
||
|
||
|
||
# ============================================================================
|
||
# Gateway / Server
|
||
# ============================================================================
|
||
|
||
|
||
def _run_gateway(
|
||
config: Config,
|
||
*,
|
||
port: int | None = None,
|
||
open_browser_url: str | None = None,
|
||
webui_static_dist: bool = True,
|
||
webui_runtime_surface: str = "browser",
|
||
webui_runtime_capabilities: dict[str, Any] | None = None,
|
||
health_server_enabled: bool = True,
|
||
) -> None:
|
||
"""Shared gateway runtime; ``open_browser_url`` opens a tab once channels are up."""
|
||
from nanobot.agent.tools.message import MessageTool
|
||
from nanobot.bus.queue import MessageBus
|
||
from nanobot.bus.runtime_events import RuntimeEventBus
|
||
from nanobot.channels.manager import ChannelManager
|
||
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 build_provider_snapshot, load_provider_snapshot
|
||
from nanobot.providers.image_generation import image_gen_provider_configs
|
||
from nanobot.session.manager import SessionManager
|
||
from nanobot.session.webui_turns import WebuiTurnCoordinator
|
||
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
|
||
|
||
console.print(f"{__logo__} Starting nanobot gateway version {__version__} on port {port}...")
|
||
sync_workspace_templates(config.workspace_path)
|
||
bus = MessageBus()
|
||
runtime_events = RuntimeEventBus()
|
||
try:
|
||
provider_snapshot = 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)
|
||
|
||
# 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_provider_snapshot,
|
||
runtime_events=runtime_events,
|
||
provider_signature=provider_snapshot.signature,
|
||
hooks=[TokenUsageHook(timezone_name=config.agents.defaults.timezone)],
|
||
local_trigger_store=trigger_store,
|
||
)
|
||
WebuiTurnCoordinator(
|
||
bus=bus,
|
||
sessions=session_manager,
|
||
schedule_background=lambda coro: agent._schedule_background(coro),
|
||
).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 = getattr(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, **_kwargs):
|
||
pass
|
||
|
||
# Dream is an internal job — run directly, not through the agent loop.
|
||
if job.name == "dream":
|
||
from nanobot.agent.memory import MemoryStore
|
||
|
||
dream_session_key = MemoryStore.dream_session_key
|
||
build_dream_commit_message = MemoryStore.build_dream_commit_message
|
||
prune_dream_sessions = MemoryStore.prune_dream_sessions
|
||
|
||
store = agent.context.memory
|
||
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()
|
||
resp = await agent.process_direct(
|
||
prompt,
|
||
session_key=key,
|
||
ephemeral=True,
|
||
tools=store.build_dream_tools(),
|
||
on_progress=_silent,
|
||
)
|
||
# Ground truth: the real file delta, not the LLM's self-report.
|
||
diff_body = store.dream_content_diff()
|
||
productive = bool(diff_body) or (
|
||
not store.git.is_initialized()
|
||
and MemoryStore.dream_run_completed(resp)
|
||
)
|
||
if productive:
|
||
store.set_last_dream_cursor(last_cursor)
|
||
logger.info("Dream cron job completed, cursor advanced to {}", last_cursor)
|
||
elif MemoryStore.dream_run_completed(resp):
|
||
logger.info(
|
||
"Dream cron job completed with no memory changes; "
|
||
"cursor not advanced",
|
||
)
|
||
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,
|
||
)
|
||
if store.git.is_initialized():
|
||
msg = build_dream_commit_message(
|
||
"dream: periodic memory consolidation", diff_body,
|
||
)
|
||
sha = store.git.auto_commit(msg)
|
||
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"Review the following HEARTBEAT.md and report any active tasks:\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)
|
||
response = resp.content if resp else ""
|
||
|
||
# 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 response:
|
||
return None
|
||
|
||
# Fail closed: stay silent on evaluator failure instead of notifying.
|
||
should_notify = await evaluate_response(
|
||
response, prompt, agent.provider, agent.model,
|
||
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:
|
||
model = getattr(agent, "model", None)
|
||
if isinstance(model, str):
|
||
stripped = model.strip()
|
||
return stripped or None
|
||
return None
|
||
|
||
# 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=getattr(agent, "pending_cron_job_ids_for_session", None),
|
||
webui_local_trigger_pending_ids=getattr(
|
||
agent,
|
||
"pending_local_trigger_ids_for_session",
|
||
None,
|
||
),
|
||
webui_static_dist=webui_static_dist,
|
||
webui_runtime_surface=webui_runtime_surface,
|
||
webui_runtime_capabilities=webui_runtime_capabilities,
|
||
)
|
||
|
||
def _pick_heartbeat_target() -> tuple[str, str]:
|
||
"""Pick a routable channel/chat target for heartbeat-triggered messages."""
|
||
sidebar_state = read_webui_sidebar_state()
|
||
return _pick_heartbeat_target_from_sessions(
|
||
enabled_channels=channels.enabled_channels,
|
||
sessions=session_manager.list_sessions(),
|
||
archived_keys=sidebar_state.get("archived_keys", []),
|
||
)
|
||
|
||
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()
|
||
if cron_status["jobs"] > 0:
|
||
console.print(f"[green]✓[/green] Cron: {cron_status['jobs']} 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):
|
||
"""Lightweight HTTP health endpoint on the gateway port."""
|
||
import json as _json
|
||
|
||
async def handle(reader, writer):
|
||
try:
|
||
data = await asyncio.wait_for(reader.read(4096), timeout=5)
|
||
except (asyncio.TimeoutError, ConnectionError):
|
||
writer.close()
|
||
return
|
||
|
||
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"})
|
||
resp = (
|
||
f"HTTP/1.0 200 OK\r\n"
|
||
f"Content-Type: application/json\r\n"
|
||
f"Content-Length: {len(body)}\r\n"
|
||
f"\r\n{body}"
|
||
)
|
||
else:
|
||
body = "Not Found"
|
||
resp = (
|
||
f"HTTP/1.0 404 Not Found\r\n"
|
||
f"Content-Type: text/plain\r\n"
|
||
f"Content-Length: {len(body)}\r\n"
|
||
f"\r\n{body}"
|
||
)
|
||
|
||
writer.write(resp.encode())
|
||
await writer.drain()
|
||
writer.close()
|
||
|
||
server = await asyncio.start_server(handle, host, health_port)
|
||
console.print(f"[green]✓[/green] Health endpoint: http://{host}:{health_port}/health")
|
||
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():
|
||
tasks: list[asyncio.Task] = []
|
||
shutdown_task: asyncio.Task | None = None
|
||
runtime_tasks: asyncio.Future | None = None
|
||
runtime_tasks_drained = False
|
||
shutdown_event = asyncio.Event()
|
||
_ensure_gateway_tty_signal_mode()
|
||
restore_shutdown_handlers = _install_gateway_shutdown_handlers(
|
||
asyncio.get_running_loop(),
|
||
shutdown_event,
|
||
tasks,
|
||
console.print,
|
||
)
|
||
try:
|
||
await cron.start()
|
||
tasks = [
|
||
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=getattr(agent, "submit_local_trigger_turn", 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
|
||
elif runtime_tasks is not None:
|
||
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()
|
||
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
|
||
await channels.stop_all()
|
||
# 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())
|
||
|
||
|
||
app.add_typer(
|
||
create_gateway_app(
|
||
console=console,
|
||
log_handler_id=_log_handler_id,
|
||
load_runtime_config=_load_runtime_config,
|
||
run_gateway=_run_gateway,
|
||
),
|
||
name="gateway",
|
||
)
|
||
|
||
|
||
# ============================================================================
|
||
# Agent Commands
|
||
# ============================================================================
|
||
|
||
|
||
@app.command()
|
||
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="Config file path"),
|
||
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.image_generation import image_gen_provider_configs
|
||
|
||
config = _load_runtime_config(config, workspace)
|
||
sync_workspace_templates(config.workspace_path)
|
||
|
||
bus = MessageBus()
|
||
|
||
# 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)
|
||
|
||
_set_nanobot_logs(logs)
|
||
|
||
try:
|
||
agent_loop = AgentLoop.from_config(
|
||
config, bus,
|
||
cron_service=cron,
|
||
image_generation_provider_configs=image_gen_provider_configs(config),
|
||
)
|
||
except ValueError as exc:
|
||
console.print(f"[red]Error: {exc}[/red]")
|
||
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):
|
||
_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):
|
||
reasoning_buffer = _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:
|
||
_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:
|
||
_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
|
||
_print_cli_progress_line(content, _thinking, renderer)
|
||
return _cli_progress
|
||
|
||
if message:
|
||
# Single message mode — direct call, no bus needed
|
||
async def run_once():
|
||
renderer = StreamRenderer(
|
||
render_markdown=markdown,
|
||
bot_name=config.agents.defaults.bot_name,
|
||
bot_icon=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
|
||
_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
|
||
_init_prompt_session()
|
||
_model, _preset_tag = _model_display(config)
|
||
_icon = 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, frame):
|
||
sig_name = signal.Signals(signum).name
|
||
_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():
|
||
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 = _ReasoningBuffer()
|
||
|
||
async def _consume_outbound():
|
||
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
|
||
_print_agent_response(
|
||
msg.content,
|
||
render_markdown=markdown,
|
||
metadata=msg.metadata,
|
||
**print_kwargs,
|
||
)
|
||
turn_done.set()
|
||
continue
|
||
|
||
if await _maybe_print_interactive_progress(
|
||
msg,
|
||
renderer,
|
||
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 _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:
|
||
_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 _read_interactive_input_async())
|
||
command = user_input.strip()
|
||
if not command:
|
||
continue
|
||
|
||
if _is_exit_command(command):
|
||
_restore_terminal()
|
||
console.print("\nGoodbye!")
|
||
break
|
||
|
||
turn_done.clear()
|
||
turn_response.clear()
|
||
reasoning_buffer.clear()
|
||
renderer = StreamRenderer(
|
||
render_markdown=markdown,
|
||
bot_name=config.agents.defaults.bot_name,
|
||
bot_icon=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
|
||
_print_agent_response(
|
||
content,
|
||
render_markdown=markdown,
|
||
metadata=meta,
|
||
**print_kwargs,
|
||
)
|
||
elif renderer and not renderer.streamed:
|
||
await renderer.close()
|
||
except KeyboardInterrupt:
|
||
_restore_terminal()
|
||
console.print("\nGoodbye!")
|
||
break
|
||
except EOFError:
|
||
_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())
|
||
|
||
|
||
# ============================================================================
|
||
# Channel Commands
|
||
# ============================================================================
|
||
|
||
|
||
channels_app = typer.Typer(help="Manage channels")
|
||
app.add_typer(channels_app, name="channels")
|
||
|
||
|
||
@channels_app.command("status")
|
||
def channels_status(
|
||
config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
|
||
):
|
||
"""Show channel status."""
|
||
from nanobot.channels.registry import discover_all
|
||
|
||
_, loaded = _load_inspection_config(config=config)
|
||
|
||
table = Table(title="Channel Status")
|
||
table.add_column("Channel", style="cyan")
|
||
table.add_column("Enabled")
|
||
|
||
for name, cls in sorted(discover_all().items()):
|
||
section = getattr(loaded.channels, name, None)
|
||
if section is None:
|
||
enabled = False
|
||
elif isinstance(section, dict):
|
||
enabled = section.get("enabled", False)
|
||
else:
|
||
enabled = getattr(section, "enabled", False)
|
||
table.add_row(
|
||
cls.display_name,
|
||
"[green]\u2713[/green]" if enabled else "[dim]\u2717[/dim]",
|
||
)
|
||
|
||
console.print(table)
|
||
|
||
|
||
@channels_app.command("login")
|
||
def channels_login(
|
||
channel_name: str = typer.Argument(..., help="Channel name (e.g. weixin, whatsapp)"),
|
||
force: bool = typer.Option(False, "--force", "-f", help="Force re-authentication even if already logged in"),
|
||
config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
|
||
):
|
||
"""Authenticate with a channel via QR code or other interactive login."""
|
||
from nanobot.channels.registry import discover_all
|
||
|
||
_, loaded = _load_inspection_config(config=config)
|
||
channel_cfg = getattr(loaded.channels, channel_name, None) or {}
|
||
|
||
# Validate channel exists
|
||
all_channels = discover_all()
|
||
if channel_name not in all_channels:
|
||
available = ", ".join(all_channels.keys())
|
||
console.print(f"[red]Unknown channel: {channel_name}[/red] Available: {available}")
|
||
raise typer.Exit(1)
|
||
|
||
console.print(f"{__logo__} {all_channels[channel_name].display_name} Login\n")
|
||
|
||
channel_cls = all_channels[channel_name]
|
||
channel = channel_cls(channel_cfg, bus=None)
|
||
|
||
success = asyncio.run(channel.login(force=force))
|
||
|
||
if not success:
|
||
raise typer.Exit(1)
|
||
|
||
|
||
# ============================================================================
|
||
# Plugin Commands
|
||
# ============================================================================
|
||
|
||
plugins_app = typer.Typer(help="Manage optional nanobot features")
|
||
app.add_typer(plugins_app, name="plugins")
|
||
|
||
|
||
@plugins_app.command("list")
|
||
def plugins_list(
|
||
config_path: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
|
||
):
|
||
"""List optional nanobot features."""
|
||
from nanobot.channels.registry import discover_channel_names, discover_plugins
|
||
from nanobot.config.loader import load_config, set_config_path
|
||
|
||
resolved_config_path = Path(config_path).expanduser().resolve() if config_path else None
|
||
if resolved_config_path is not None:
|
||
set_config_path(resolved_config_path)
|
||
|
||
_print_enable_options(
|
||
feature_support.optional_dependency_groups(),
|
||
set(discover_channel_names()),
|
||
discover_plugins(),
|
||
load_config(resolved_config_path),
|
||
)
|
||
|
||
|
||
@plugins_app.command("enable")
|
||
def plugins_enable(
|
||
name: str = typer.Argument(..., help="Feature name (e.g. weixin, matrix, pdf)"),
|
||
config_path: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
|
||
logs: bool = typer.Option(False, "--logs/--no-logs", help="Show optional package install logs"),
|
||
):
|
||
"""Enable a nanobot feature."""
|
||
from nanobot.config.loader import get_config_path, set_config_path
|
||
|
||
resolved_config_path = Path(config_path).expanduser().resolve() if config_path else None
|
||
if resolved_config_path is not None:
|
||
set_config_path(resolved_config_path)
|
||
resolved_config_path = resolved_config_path or get_config_path()
|
||
_set_nanobot_logs(logs)
|
||
|
||
try:
|
||
payload = feature_support.enable_optional_feature(
|
||
name,
|
||
config_path=resolved_config_path,
|
||
runner=feature_support.run_install_command,
|
||
)
|
||
except feature_support.OptionalFeatureError as exc:
|
||
console.print(f"[red]{escape(exc.message)}[/red]")
|
||
raise typer.Exit(1) from exc
|
||
|
||
message = payload.get("last_action", {}).get("message") or f"Enabled feature '{name}'"
|
||
console.print(f"[green]{escape(message)}[/green]")
|
||
|
||
|
||
@plugins_app.command("disable")
|
||
def plugins_disable(
|
||
name: str = typer.Argument(..., help="Channel name (e.g. telegram, matrix, slack)"),
|
||
config_path: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
|
||
):
|
||
"""Disable a nanobot channel feature."""
|
||
from nanobot.config.loader import get_config_path, set_config_path
|
||
|
||
resolved_config_path = Path(config_path).expanduser().resolve() if config_path else None
|
||
if resolved_config_path is not None:
|
||
set_config_path(resolved_config_path)
|
||
resolved_config_path = resolved_config_path or get_config_path()
|
||
|
||
try:
|
||
payload = feature_support.disable_optional_feature(name, config_path=resolved_config_path)
|
||
except feature_support.OptionalFeatureError as exc:
|
||
console.print(f"[red]{escape(exc.message)}[/red]")
|
||
raise typer.Exit(1) from exc
|
||
|
||
message = payload.get("last_action", {}).get("message") or f"Disabled channel '{name}'"
|
||
console.print(f"[green]{escape(message)}[/green] in {resolved_config_path}")
|
||
|
||
|
||
# ============================================================================
|
||
# Status Commands
|
||
# ============================================================================
|
||
|
||
|
||
@app.command()
|
||
def status(
|
||
config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
|
||
workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
|
||
):
|
||
"""Show nanobot status."""
|
||
config_path, loaded = _load_inspection_config(config=config, workspace=workspace)
|
||
workspace_path = loaded.workspace_path
|
||
|
||
console.print(f"{__logo__} nanobot Status\n")
|
||
|
||
console.print(f"Config: {config_path} {'[green]✓[/green]' if config_path.exists() else '[red]✗[/red]'}")
|
||
console.print(
|
||
f"Workspace: {workspace_path} "
|
||
f"{'[green]✓[/green]' if workspace_path.exists() else '[red]✗[/red]'}"
|
||
)
|
||
|
||
if config_path.exists():
|
||
from nanobot.providers.registry import PROVIDERS
|
||
|
||
_model, _preset_tag = _model_display(loaded)
|
||
console.print(f"Model: {_model}{_preset_tag}")
|
||
|
||
# Check API keys from registry
|
||
for spec in PROVIDERS:
|
||
p = getattr(loaded.providers, spec.name, None)
|
||
if p is None:
|
||
continue
|
||
if spec.is_oauth:
|
||
console.print(f"{spec.label}: [green]✓ (OAuth)[/green]")
|
||
elif spec.is_local:
|
||
# Local deployments show api_base instead of api_key
|
||
if p.api_base:
|
||
console.print(f"{spec.label}: [green]✓ {p.api_base}[/green]")
|
||
else:
|
||
console.print(f"{spec.label}: [dim]not set[/dim]")
|
||
else:
|
||
has_key = bool(p.api_key)
|
||
console.print(f"{spec.label}: {'[green]✓[/green]' if has_key else '[dim]not set[/dim]'}")
|
||
|
||
|
||
# ============================================================================
|
||
# OAuth Login
|
||
# ============================================================================
|
||
|
||
provider_app = typer.Typer(help="Manage providers")
|
||
app.add_typer(provider_app, name="provider")
|
||
|
||
|
||
_LOGIN_HANDLERS: dict[str, Callable[[], None]] = {}
|
||
_LOGOUT_HANDLERS: dict[str, Callable[[], None]] = {}
|
||
|
||
_PROVIDER_DISPLAY: dict[str, str] = {
|
||
"openai_codex": "OpenAI Codex",
|
||
"github_copilot": "GitHub Copilot",
|
||
}
|
||
|
||
_OAUTH_PROVIDER_DEFAULT_MODELS: dict[str, str] = {
|
||
"openai_codex": "openai-codex/gpt-5.4-mini",
|
||
"github_copilot": "github-copilot/gpt-5.4-mini",
|
||
}
|
||
|
||
|
||
def _register_login(name: str):
|
||
"""Register an OAuth login handler."""
|
||
def decorator(fn):
|
||
_LOGIN_HANDLERS[name] = fn
|
||
return fn
|
||
|
||
return decorator
|
||
|
||
|
||
def _register_logout(name: str):
|
||
"""Register an OAuth logout handler."""
|
||
def decorator(fn):
|
||
_LOGOUT_HANDLERS[name] = fn
|
||
return fn
|
||
return decorator
|
||
|
||
|
||
def _resolve_oauth_provider(provider: str):
|
||
"""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:
|
||
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
|
||
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', '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)
|
||
|
||
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', 'github-copilot')"),
|
||
):
|
||
"""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)
|
||
|
||
console.print(f"{__logo__} OAuth Logout - {spec.label}\n")
|
||
handler()
|
||
|
||
|
||
@_register_login("openai_codex")
|
||
def _login_openai_codex() -> None:
|
||
try:
|
||
from oauth_cli_kit import get_token, login_oauth_interactive
|
||
|
||
from nanobot.config.loader import load_config, resolve_config_env_vars
|
||
|
||
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)
|
||
|
||
|
||
@_register_logout("openai_codex")
|
||
def _logout_openai_codex() -> None:
|
||
"""Clear local OAuth credentials for OpenAI Codex."""
|
||
try:
|
||
from oauth_cli_kit.providers import OPENAI_CODEX_PROVIDER
|
||
from oauth_cli_kit.storage import FileTokenStorage
|
||
except ImportError:
|
||
console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
|
||
raise typer.Exit(1)
|
||
|
||
storage = FileTokenStorage(token_filename=OPENAI_CODEX_PROVIDER.token_filename)
|
||
_delete_oauth_files(storage.get_token_path(), _PROVIDER_DISPLAY["openai_codex"])
|
||
|
||
|
||
@_register_logout("github_copilot")
|
||
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]")
|
||
|
||
|
||
@_register_login("github_copilot")
|
||
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)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app()
|