axelray-devandXubin Ren
f7b027a295
Address review: avoid overwriting cursor on every restart when Dream is disabled. Now only advances if current cursor is behind the latest position, so repeated restarts don't permanently skip entries.
1901 lines
70 KiB
Python
1901 lines
70 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,
|
||
)
|
||
|
||
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.patch_stdout import patch_stdout # noqa: E402
|
||
from rich.console import Console # noqa: E402
|
||
from rich.markdown import Markdown # 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.agent.loop import AgentLoop # noqa: E402
|
||
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
|
||
|
||
|
||
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 _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=False, # Enter submits (single line mode)
|
||
)
|
||
|
||
|
||
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:
|
||
metadata = msg.metadata or {}
|
||
if metadata.get("_retry_wait"):
|
||
await _print_interactive_progress_line(msg.content, thinking, renderer)
|
||
return True
|
||
|
||
if not metadata.get("_progress"):
|
||
return False
|
||
|
||
reasoning_buffer = reasoning_buffer or _ReasoningBuffer()
|
||
|
||
if metadata.get("_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 = metadata.get("_tool_hint", False)
|
||
is_reasoning = metadata.get("_reasoning", False) or metadata.get("_reasoning_delta", False)
|
||
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 _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 _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 _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))
|
||
|
||
|
||
# ============================================================================
|
||
# 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: pip install 'nanobot-ai[api]'[/red]")
|
||
raise typer.Exit(1)
|
||
|
||
from loguru import logger
|
||
|
||
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
|
||
|
||
if verbose:
|
||
logger.enable("nanobot")
|
||
else:
|
||
logger.disable("nanobot")
|
||
|
||
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
|
||
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]Warning:[/yellow] API is bound to all interfaces. "
|
||
"Only do this behind a trusted network boundary, firewall, or reverse proxy."
|
||
)
|
||
console.print()
|
||
|
||
api_app = create_app(agent_loop, model_name=model_name, request_timeout=timeout)
|
||
|
||
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))
|
||
|
||
|
||
# ============================================================================
|
||
# 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.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)
|
||
|
||
# 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)
|
||
|
||
# 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)],
|
||
)
|
||
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
|
||
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,
|
||
)
|
||
if MemoryStore.dream_run_completed(resp):
|
||
store.set_last_dream_cursor(last_cursor)
|
||
logger.info("Dream cron job completed, 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,
|
||
)
|
||
if store.git.is_initialized():
|
||
msg = build_dream_commit_message(
|
||
"dream: periodic memory consolidation", resp,
|
||
)
|
||
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,
|
||
webui_runtime_model_name=_webui_runtime_model_name,
|
||
webui_cron_pending_job_ids=getattr(agent, "pending_cron_job_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")
|
||
latest = agent.context.memory.get_latest_cursor()
|
||
if agent.context.memory.get_last_dream_cursor() < latest:
|
||
agent.context.memory.set_last_dream_cursor(latest)
|
||
|
||
# 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
|
||
# 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(
|
||
config.gateway.host or "127.0.0.1", 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"),
|
||
]
|
||
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 loguru import logger
|
||
|
||
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)
|
||
|
||
if logs:
|
||
logger.enable("nanobot")
|
||
else:
|
||
logger.disable("nanobot")
|
||
|
||
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[tuple[str, dict]] = []
|
||
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)
|
||
|
||
if msg.metadata.get("_stream_delta"):
|
||
if renderer:
|
||
await renderer.on_delta(msg.content)
|
||
continue
|
||
if msg.metadata.get("_stream_end"):
|
||
if renderer:
|
||
await renderer.on_end(
|
||
resuming=msg.metadata.get("_resuming", False),
|
||
)
|
||
continue
|
||
if msg.metadata.get("_streamed"):
|
||
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.content, dict(msg.metadata or {})))
|
||
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:
|
||
content, meta = turn_response[0]
|
||
if content and not meta.get("_streamed"):
|
||
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_path: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
|
||
):
|
||
"""Show channel status."""
|
||
from nanobot.channels.registry import discover_all
|
||
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)
|
||
|
||
config = load_config(resolved_config_path)
|
||
|
||
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(config.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_path: 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
|
||
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)
|
||
|
||
config = load_config(resolved_config_path)
|
||
channel_cfg = getattr(config.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 channel plugins")
|
||
app.add_typer(plugins_app, name="plugins")
|
||
|
||
|
||
@plugins_app.command("list")
|
||
def plugins_list():
|
||
"""List all discovered channels (built-in and plugins)."""
|
||
from nanobot.channels.registry import discover_all, discover_channel_names
|
||
from nanobot.config.loader import load_config
|
||
|
||
config = load_config()
|
||
builtin_names = set(discover_channel_names())
|
||
all_channels = discover_all()
|
||
|
||
table = Table(title="Channel Plugins")
|
||
table.add_column("Name", style="cyan")
|
||
table.add_column("Source", style="magenta")
|
||
table.add_column("Enabled")
|
||
|
||
for name in sorted(all_channels):
|
||
cls = all_channels[name]
|
||
source = "builtin" if name in builtin_names else "plugin"
|
||
section = getattr(config.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,
|
||
source,
|
||
"[green]yes[/green]" if enabled else "[dim]no[/dim]",
|
||
)
|
||
|
||
console.print(table)
|
||
|
||
|
||
# ============================================================================
|
||
# Status Commands
|
||
# ============================================================================
|
||
|
||
|
||
@app.command()
|
||
def status():
|
||
"""Show nanobot status."""
|
||
from nanobot.config.loader import get_config_path, load_config
|
||
|
||
config_path = get_config_path()
|
||
config = load_config()
|
||
workspace = config.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} {'[green]✓[/green]' if workspace.exists() else '[red]✗[/red]'}")
|
||
|
||
if config_path.exists():
|
||
from nanobot.providers.registry import PROVIDERS
|
||
|
||
_model, _preset_tag = _model_display(config)
|
||
console.print(f"Model: {_model}{_preset_tag}")
|
||
|
||
# Check API keys from registry
|
||
for spec in PROVIDERS:
|
||
p = getattr(config.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",
|
||
}
|
||
|
||
|
||
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
|
||
|
||
|
||
@provider_app.command("login")
|
||
def provider_login(
|
||
provider: str = typer.Argument(..., help="OAuth provider (e.g. 'openai-codex', 'github-copilot')"),
|
||
):
|
||
"""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()
|
||
|
||
|
||
@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
|
||
|
||
token = None
|
||
with suppress(Exception):
|
||
token = get_token()
|
||
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),
|
||
)
|
||
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]GitHub Copilot provider unavailable. Ensure oauth-cli-kit is installed.[/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()
|