+
+
+
diff --git a/images/nanobot_webui.png b/images/nanobot_webui.png new file mode 100644 index 00000000..b074281d Binary files /dev/null and b/images/nanobot_webui.png differ diff --git a/nanobot/channels/manager.py b/nanobot/channels/manager.py index c0622a27..f8f2c1d8 100644 --- a/nanobot/channels/manager.py +++ b/nanobot/channels/manager.py @@ -3,7 +3,8 @@ from __future__ import annotations import asyncio -from typing import Any +from pathlib import Path +from typing import TYPE_CHECKING, Any from loguru import logger @@ -13,6 +14,19 @@ from nanobot.channels.base import BaseChannel from nanobot.config.schema import Config from nanobot.utils.restart import consume_restart_notice_from_env, format_restart_completed_message +if TYPE_CHECKING: + from nanobot.session.manager import SessionManager + + +def _default_webui_dist() -> Path | None: + """Return the absolute path to the bundled webui dist directory if it exists.""" + try: + import nanobot.web as web_pkg # type: ignore[import-not-found] + except ImportError: + return None + candidate = Path(web_pkg.__file__).resolve().parent / "dist" + return candidate if candidate.is_dir() else None + # Retry delays for message sending (exponential backoff: 1s, 2s, 4s) _SEND_RETRY_DELAYS = (1, 2, 4) @@ -27,9 +41,16 @@ class ChannelManager: - Route outbound messages """ - def __init__(self, config: Config, bus: MessageBus): + def __init__( + self, + config: Config, + bus: MessageBus, + *, + session_manager: "SessionManager | None" = None, + ): self.config = config self.bus = bus + self._session_manager = session_manager self.channels: dict[str, BaseChannel] = {} self._dispatch_task: asyncio.Task | None = None @@ -55,7 +76,15 @@ class ChannelManager: if not enabled: continue try: - channel = cls(section, self.bus) + kwargs: dict[str, Any] = {} + # Only the WebSocket channel currently hosts the embedded webui + # surface; other channels stay oblivious to these knobs. + if cls.name == "websocket" and self._session_manager is not None: + kwargs["session_manager"] = self._session_manager + static_path = _default_webui_dist() + if static_path is not None: + kwargs["static_dist_path"] = static_path + channel = cls(section, self.bus, **kwargs) channel.transcription_provider = transcription_provider channel.transcription_api_key = transcription_key channel.transcription_api_base = transcription_base diff --git a/nanobot/channels/websocket.py b/nanobot/channels/websocket.py index 882793ae..639519a4 100644 --- a/nanobot/channels/websocket.py +++ b/nanobot/channels/websocket.py @@ -7,13 +7,15 @@ import email.utils import hmac import http import json +import mimetypes import re import secrets import ssl import time import uuid -from typing import Any, Self -from urllib.parse import parse_qs, urlparse +from pathlib import Path +from typing import TYPE_CHECKING, Any, Self +from urllib.parse import parse_qs, unquote, urlparse from loguru import logger from pydantic import Field, field_validator, model_validator @@ -28,6 +30,9 @@ from nanobot.bus.queue import MessageBus from nanobot.channels.base import BaseChannel from nanobot.config.schema import Base +if TYPE_CHECKING: + from nanobot.session.manager import SessionManager + def _strip_trailing_slash(path: str) -> str: if len(path) > 1 and path.endswith("/"): @@ -116,6 +121,18 @@ def _http_json_response(data: dict[str, Any], *, status: int = 200) -> Response: return Response(status, reason, headers, body) +def _read_webui_model_name() -> str | None: + """Return the configured default model for readonly webui display.""" + try: + from nanobot.config.loader import load_config + + model = load_config().agents.defaults.model.strip() + return model or None + except Exception as e: + logger.debug("webui bootstrap could not load model name: {}", e) + return None + + def _parse_request_path(path_with_query: str) -> tuple[str, dict[str, list[str]]]: """Parse normalized path and query parameters in one pass.""" parsed = urlparse("ws://x" + path_with_query) @@ -189,6 +206,78 @@ def _parse_envelope(raw: str) -> dict[str, Any] | None: return data +_LOCALHOSTS = frozenset({"127.0.0.1", "::1", "localhost"}) + +# Matches the legacy chat-id pattern but allows file-system-safe stems too, +# so the API can address sessions whose keys came from non-WebSocket channels. +_API_KEY_RE = re.compile(r"^[A-Za-z0-9_:.-]{1,128}$") + + +def _decode_api_key(raw_key: str) -> str | None: + """Decode a percent-encoded API path segment, then validate the result.""" + key = unquote(raw_key) + if _API_KEY_RE.match(key) is None: + return None + return key + + +def _is_localhost(connection: Any) -> bool: + """Return True if *connection* originated from the loopback interface.""" + addr = getattr(connection, "remote_address", None) + if not addr: + return False + host = addr[0] if isinstance(addr, tuple) else addr + if not isinstance(host, str): + return False + # ``::ffff:127.0.0.1`` is loopback in IPv6-mapped form. + if host.startswith("::ffff:"): + host = host[7:] + return host in _LOCALHOSTS + + +def _http_response( + body: bytes, + *, + status: int = 200, + content_type: str = "text/plain; charset=utf-8", + extra_headers: list[tuple[str, str]] | None = None, +) -> Response: + headers = [ + ("Date", email.utils.formatdate(usegmt=True)), + ("Connection", "close"), + ("Content-Length", str(len(body))), + ("Content-Type", content_type), + ] + if extra_headers: + headers.extend(extra_headers) + reason = http.HTTPStatus(status).phrase + return Response(status, reason, Headers(headers), body) + + +def _http_error(status: int, message: str | None = None) -> Response: + body = (message or http.HTTPStatus(status).phrase).encode("utf-8") + return _http_response(body, status=status) + + +def _bearer_token(headers: Any) -> str | None: + """Pull a Bearer token out of standard or query-style headers.""" + auth = headers.get("Authorization") or headers.get("authorization") + if auth and auth.lower().startswith("bearer "): + return auth[7:].strip() or None + return None + + +def _is_websocket_upgrade(request: WsRequest) -> bool: + """Detect an actual WS upgrade; plain HTTP GETs to the same path should fall through.""" + upgrade = request.headers.get("Upgrade") or request.headers.get("upgrade") + connection = request.headers.get("Connection") or request.headers.get("connection") + if not upgrade or "websocket" not in upgrade.lower(): + return False + if not connection or "upgrade" not in connection.lower(): + return False + return True + + def _issue_route_secret_matches(headers: Any, configured_secret: str) -> bool: """Return True if the token-issue HTTP request carries credentials matching ``token_issue_secret``.""" if not configured_secret: @@ -209,7 +298,14 @@ class WebSocketChannel(BaseChannel): name = "websocket" display_name = "WebSocket" - def __init__(self, config: Any, bus: MessageBus): + def __init__( + self, + config: Any, + bus: MessageBus, + *, + session_manager: "SessionManager | None" = None, + static_dist_path: Path | None = None, + ): if isinstance(config, dict): config = WebSocketConfig.model_validate(config) super().__init__(config, bus) @@ -220,9 +316,16 @@ class WebSocketChannel(BaseChannel): self._conn_chats: dict[Any, set[str]] = {} # connection -> default chat_id for legacy frames that omit routing. self._conn_default: dict[Any, str] = {} + # Single-use tokens consumed at WebSocket handshake. self._issued_tokens: dict[str, float] = {} + # Multi-use tokens for the embedded webui's REST surface; checked but not consumed. + self._api_tokens: dict[str, float] = {} self._stop_event: asyncio.Event | None = None self._server_task: asyncio.Task[None] | None = None + self._session_manager = session_manager + self._static_dist_path: Path | None = ( + static_dist_path.resolve() if static_dist_path is not None else None + ) # -- Subscription bookkeeping ------------------------------------------- @@ -324,6 +427,209 @@ class WebSocketChannel(BaseChannel): {"token": token_value, "expires_in": self.config.token_ttl_s} ) + # -- HTTP dispatch ------------------------------------------------------ + + async def _dispatch_http(self, connection: Any, request: WsRequest) -> Any: + """Route an inbound HTTP request to a handler or to the WS upgrade path.""" + got, query = _parse_request_path(request.path) + + # 1. Token issue endpoint (legacy, optional, gated by configured secret). + if self.config.token_issue_path: + issue_expected = _normalize_config_path(self.config.token_issue_path) + if got == issue_expected: + return self._handle_token_issue_http(connection, request) + + # 2. WebUI bootstrap: localhost-only, mints tokens for the embedded UI. + if got == "/webui/bootstrap": + return self._handle_webui_bootstrap(connection) + + # 3. REST surface for the embedded UI. + if got == "/api/sessions": + return self._handle_sessions_list(request) + + m = re.match(r"^/api/sessions/([^/]+)/messages$", got) + if m: + return self._handle_session_messages(request, m.group(1)) + + # NOTE: websockets' HTTP parser only accepts GET, so we cannot expose a + # true ``DELETE`` verb. The action is folded into the path instead. + m = re.match(r"^/api/sessions/([^/]+)/delete$", got) + if m: + return self._handle_session_delete(request, m.group(1)) + + # 4. WebSocket upgrade (the channel's primary purpose). Only run the + # handshake gate on requests that actually ask to upgrade; otherwise + # a bare ``GET /`` from the browser would be rejected as an + # unauthorized WS handshake instead of serving the SPA's index.html. + expected_ws = self._expected_path() + if got == expected_ws and _is_websocket_upgrade(request): + client_id = _query_first(query, "client_id") or "" + if len(client_id) > 128: + client_id = client_id[:128] + if not self.is_allowed(client_id): + return connection.respond(403, "Forbidden") + return self._authorize_websocket_handshake(connection, query) + + # 5. Static SPA serving (only if a build directory was wired in). + if self._static_dist_path is not None: + response = self._serve_static(got) + if response is not None: + return response + + return connection.respond(404, "Not Found") + + # -- HTTP route handlers ------------------------------------------------ + + def _check_api_token(self, request: WsRequest) -> bool: + """Validate a request against the API token pool (multi-use, TTL-bound).""" + self._purge_expired_api_tokens() + token = _bearer_token(request.headers) or _query_first( + _parse_query(request.path), "token" + ) + if not token: + return False + expiry = self._api_tokens.get(token) + if expiry is None or time.monotonic() > expiry: + self._api_tokens.pop(token, None) + return False + return True + + def _purge_expired_api_tokens(self) -> None: + now = time.monotonic() + for token_key, expiry in list(self._api_tokens.items()): + if now > expiry: + self._api_tokens.pop(token_key, None) + + def _handle_webui_bootstrap(self, connection: Any) -> Response: + if not _is_localhost(connection): + return _http_error(403, "webui bootstrap is localhost-only") + # Cap outstanding tokens to avoid runaway growth from a misbehaving client. + self._purge_expired_issued_tokens() + self._purge_expired_api_tokens() + if ( + len(self._issued_tokens) >= self._MAX_ISSUED_TOKENS + or len(self._api_tokens) >= self._MAX_ISSUED_TOKENS + ): + return _http_response( + json.dumps({"error": "too many outstanding tokens"}).encode("utf-8"), + status=429, + content_type="application/json; charset=utf-8", + ) + token = f"nbwt_{secrets.token_urlsafe(32)}" + expiry = time.monotonic() + float(self.config.token_ttl_s) + # Same string registered in both pools: the WS handshake consumes one copy + # while the REST surface keeps validating the other until TTL expiry. + self._issued_tokens[token] = expiry + self._api_tokens[token] = expiry + return _http_json_response( + { + "token": token, + "ws_path": self._expected_path(), + "expires_in": self.config.token_ttl_s, + "model_name": _read_webui_model_name(), + } + ) + + def _handle_sessions_list(self, request: WsRequest) -> Response: + if not self._check_api_token(request): + return _http_error(401, "Unauthorized") + if self._session_manager is None: + return _http_error(503, "session manager unavailable") + sessions = self._session_manager.list_sessions() + # The webui is only meaningful for websocket-channel chats — CLI / + # Slack / Lark / Discord sessions can't be resumed from the browser, + # so leaking them into the sidebar is just noise. Filter to the + # ``websocket:`` prefix and strip absolute paths on the way out. + cleaned = [ + {k: v for k, v in s.items() if k != "path"} + for s in sessions + if isinstance(s.get("key"), str) and s["key"].startswith("websocket:") + ] + return _http_json_response({"sessions": cleaned}) + + @staticmethod + def _is_webui_session_key(key: str) -> bool: + """Return True when *key* belongs to the webui's websocket-only surface.""" + return key.startswith("websocket:") + + def _handle_session_messages(self, request: WsRequest, key: str) -> Response: + if not self._check_api_token(request): + return _http_error(401, "Unauthorized") + if self._session_manager is None: + return _http_error(503, "session manager unavailable") + decoded_key = _decode_api_key(key) + if decoded_key is None: + return _http_error(400, "invalid session key") + # The embedded webui only understands websocket-channel sessions. Keep + # its read surface aligned with ``/api/sessions`` instead of letting a + # caller probe arbitrary CLI / Slack / Lark history by handcrafted URL. + if not self._is_webui_session_key(decoded_key): + return _http_error(404, "session not found") + data = self._session_manager.read_session_file(decoded_key) + if data is None: + return _http_error(404, "session not found") + return _http_json_response(data) + + def _handle_session_delete(self, request: WsRequest, key: str) -> Response: + if not self._check_api_token(request): + return _http_error(401, "Unauthorized") + if self._session_manager is None: + return _http_error(503, "session manager unavailable") + decoded_key = _decode_api_key(key) + if decoded_key is None: + return _http_error(400, "invalid session key") + # Same boundary as ``_handle_session_messages``: the webui may only + # mutate websocket sessions, and deletion really does unlink the local + # JSONL, so keep the blast radius narrow and explicit. + if not self._is_webui_session_key(decoded_key): + return _http_error(404, "session not found") + deleted = self._session_manager.delete_session(decoded_key) + return _http_json_response({"deleted": bool(deleted)}) + + def _serve_static(self, request_path: str) -> Response | None: + """Resolve *request_path* against the built SPA directory; SPA fallback to index.html.""" + assert self._static_dist_path is not None + rel = request_path.lstrip("/") + if not rel: + rel = "index.html" + # Reject path-traversal attempts and absolute targets. + if ".." in rel.split("/") or rel.startswith("/"): + return _http_error(403, "Forbidden") + candidate = (self._static_dist_path / rel).resolve() + try: + candidate.relative_to(self._static_dist_path) + except ValueError: + return _http_error(403, "Forbidden") + if not candidate.is_file(): + # SPA history-mode fallback: unknown routes serve index.html so the + # client-side router can render them. + index = self._static_dist_path / "index.html" + if index.is_file(): + candidate = index + else: + return None + try: + body = candidate.read_bytes() + except OSError as e: + logger.warning("websocket static: failed to read {}: {}", candidate, e) + return _http_error(500, "Internal Server Error") + ctype, _ = mimetypes.guess_type(candidate.name) + if ctype is None: + ctype = "application/octet-stream" + if ctype.startswith("text/") or ctype in {"application/javascript", "application/json"}: + ctype = f"{ctype}; charset=utf-8" + # Hash-named build assets are cache-friendly; index.html must stay fresh. + if candidate.name == "index.html": + cache = "no-cache" + else: + cache = "public, max-age=31536000, immutable" + return _http_response( + body, + status=200, + content_type=ctype, + extra_headers=[("Cache-Control", cache)], + ) + def _authorize_websocket_handshake(self, connection: Any, query: dict[str, list[str]]) -> Any: supplied = _query_first(query, "token") static_token = self.config.token.strip() @@ -355,24 +661,7 @@ class WebSocketChannel(BaseChannel): connection: ServerConnection, request: WsRequest, ) -> Any: - got, _ = _parse_request_path(request.path) - if self.config.token_issue_path: - issue_expected = _normalize_config_path(self.config.token_issue_path) - if got == issue_expected: - return self._handle_token_issue_http(connection, request) - - expected_ws = self._expected_path() - if got != expected_ws: - return connection.respond(404, "Not Found") - # Early reject before WebSocket upgrade to avoid unnecessary overhead; - # _handle_message() performs a second check as defense-in-depth. - query = _parse_query(request.path) - client_id = _query_first(query, "client_id") or "" - if len(client_id) > 128: - client_id = client_id[:128] - if not self.is_allowed(client_id): - return connection.respond(403, "Forbidden") - return self._authorize_websocket_handshake(connection, query) + return await self._dispatch_http(connection, request) async def handler(connection: ServerConnection) -> None: await self._connection_loop(connection) @@ -523,6 +812,7 @@ class WebSocketChannel(BaseChannel): self._conn_chats.clear() self._conn_default.clear() self._issued_tokens.clear() + self._api_tokens.clear() async def _safe_send_to(self, connection: Any, raw: str, *, label: str = "") -> None: """Send a raw frame to one connection, cleaning up on ConnectionClosed.""" @@ -550,6 +840,13 @@ class WebSocketChannel(BaseChannel): payload["media"] = msg.media if msg.reply_to: payload["reply_to"] = msg.reply_to + # Mark intermediate agent breadcrumbs (tool-call hints, generic + # progress strings) so WS clients can render them as subordinate + # trace rows rather than conversational replies. + if msg.metadata.get("_tool_hint"): + payload["kind"] = "tool_hint" + elif msg.metadata.get("_progress"): + payload["kind"] = "progress" raw = json.dumps(payload, ensure_ascii=False) for connection in conns: await self._safe_send_to(connection, raw, label=" ") diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index 5f043050..f3b72a72 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -636,6 +636,21 @@ def gateway( config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"), ): """Start the nanobot gateway.""" + if verbose: + import logging + + logging.basicConfig(level=logging.DEBUG) + cfg = _load_runtime_config(config, workspace) + _run_gateway(cfg, port=port) + + +def _run_gateway( + config: Config, + *, + port: int | None = None, + open_browser_url: str | None = None, +) -> None: + """Shared gateway runtime; ``open_browser_url`` opens a tab once channels are up.""" from nanobot.agent.loop import AgentLoop from nanobot.bus.queue import MessageBus from nanobot.channels.manager import ChannelManager @@ -644,12 +659,6 @@ def gateway( from nanobot.heartbeat.service import HeartbeatService from nanobot.session.manager import SessionManager - if verbose: - import logging - - logging.basicConfig(level=logging.DEBUG) - - config = _load_runtime_config(config, workspace) port = port if port is not None else config.gateway.port console.print(f"{__logo__} Starting nanobot gateway version {__version__} on port {port}...") @@ -749,8 +758,9 @@ def gateway( cron.on_job = on_cron_job - # Create channel manager - channels = ChannelManager(config, bus) + # 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) def _pick_heartbeat_target() -> tuple[str, str]: """Pick a routable channel/chat target for heartbeat-triggered messages.""" @@ -881,15 +891,43 @@ def gateway( )) console.print(f"[green]✓[/green] Dream: {dream_cfg.describe_schedule()}") + 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() + try: + await writer.wait_closed() + except Exception: + pass + 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(): try: await cron.start() await heartbeat.start() - await asyncio.gather( + tasks = [ agent.run(), channels.start_all(), _health_server(config.gateway.host, port), - ) + ] + if open_browser_url: + tasks.append(_open_browser_when_ready()) + await asyncio.gather(*tasks) except KeyboardInterrupt: console.print("\nShutting down...") except Exception: @@ -907,6 +945,68 @@ def gateway( asyncio.run(run()) +@app.command() +def web( + port: int | None = typer.Option(None, "--port", "-p", help="WebSocket port for the webui"), + workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"), + config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"), + open_browser: bool = typer.Option(True, "--open/--no-open", help="Open the browser when ready"), +): + """Start the gateway with the embedded webui and (by default) open a browser.""" + if verbose: + import logging + + logging.basicConfig(level=logging.DEBUG) + + cfg = _load_runtime_config(config, workspace) + + # Force the websocket channel on with token-gated auth so the webui is functional. + # ``--port`` applies to the webui's websocket/HTTP port, not the gateway's + # management port, since that's the only surface users visit. + ws_section = cfg.channels.websocket + if isinstance(ws_section, dict): + ws_section.setdefault("host", "127.0.0.1") + ws_section["enabled"] = True + ws_section["websocketRequiresToken"] = True + if port is not None: + ws_section["port"] = port + ws_host = ws_section.get("host", "127.0.0.1") + ws_port = ws_section.get("port", 8765) + ws_path = ws_section.get("path", "/") + else: + ws_section.enabled = True + if hasattr(ws_section, "websocket_requires_token"): + ws_section.websocket_requires_token = True + if port is not None: + ws_section.port = port + ws_host = getattr(ws_section, "host", "127.0.0.1") or "127.0.0.1" + ws_port = getattr(ws_section, "port", 8765) + ws_path = getattr(ws_section, "path", "/") or "/" + + # Confirm the bundled SPA exists before promising the user a browser launch. + from nanobot.channels.manager import _default_webui_dist + + dist = _default_webui_dist() + if dist is None: + console.print( + "[yellow]Warning: webui assets not found at nanobot/web/dist/. " + "Run `cd webui && bun install && bun run build` from a source checkout.[/yellow]" + ) + + scheme = "http" + # Browsers refuse cookies/JS on 0.0.0.0 — collapse to loopback for the visit URL. + visit_host = "127.0.0.1" if ws_host in {"0.0.0.0", "::"} else ws_host + open_url = f"{scheme}://{visit_host}:{ws_port}{ws_path if ws_path != '/' else ''}/" + + # The gateway's management port is separate from the webui port; leave it + # on its configured default so --port only moves the visible surface. + _run_gateway( + cfg, + open_browser_url=open_url if open_browser else None, + ) + + # ============================================================================ # Agent Commands # ============================================================================ diff --git a/nanobot/session/manager.py b/nanobot/session/manager.py index 2ed0624a..c91eabcb 100644 --- a/nanobot/session/manager.py +++ b/nanobot/session/manager.py @@ -106,15 +106,18 @@ class SessionManager: self.legacy_sessions_dir = get_legacy_sessions_dir() self._cache: dict[str, Session] = {} + @staticmethod + def safe_key(key: str) -> str: + """Public helper used by HTTP handlers to map an arbitrary key to a stable filename stem.""" + return safe_filename(key.replace(":", "_")) + def _get_session_path(self, key: str) -> Path: """Get the file path for a session.""" - safe_key = safe_filename(key.replace(":", "_")) - return self.sessions_dir / f"{safe_key}.jsonl" + return self.sessions_dir / f"{self.safe_key(key)}.jsonl" def _get_legacy_session_path(self, key: str) -> Path: """Legacy global session path (~/.nanobot/sessions/).""" - safe_key = safe_filename(key.replace(":", "_")) - return self.legacy_sessions_dir / f"{safe_key}.jsonl" + return self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl" def get_or_create(self, key: str) -> Session: """ @@ -209,6 +212,61 @@ class SessionManager: """Remove a session from the in-memory cache.""" self._cache.pop(key, None) + def delete_session(self, key: str) -> bool: + """Remove a session from disk and the in-memory cache. + + Returns True if a JSONL file was found and unlinked. + """ + path = self._get_session_path(key) + self.invalidate(key) + if not path.exists(): + return False + try: + path.unlink() + return True + except OSError as e: + logger.warning("Failed to delete session file {}: {}", path, e) + return False + + def read_session_file(self, key: str) -> dict[str, Any] | None: + """Load a session from disk without caching; intended for read-only HTTP endpoints. + + Returns ``{"key", "created_at", "updated_at", "metadata", "messages"}`` or + ``None`` when the session file does not exist or fails to parse. + """ + path = self._get_session_path(key) + if not path.exists(): + return None + try: + messages: list[dict[str, Any]] = [] + metadata: dict[str, Any] = {} + created_at: str | None = None + updated_at: str | None = None + stored_key: str | None = None + with open(path, encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + data = json.loads(line) + if data.get("_type") == "metadata": + metadata = data.get("metadata", {}) + created_at = data.get("created_at") + updated_at = data.get("updated_at") + stored_key = data.get("key") + else: + messages.append(data) + return { + "key": stored_key or key, + "created_at": created_at, + "updated_at": updated_at, + "metadata": metadata, + "messages": messages, + } + except Exception as e: + logger.warning("Failed to read session {}: {}", key, e) + return None + def list_sessions(self) -> list[dict[str, Any]]: """ List all sessions. diff --git a/nanobot/web/__init__.py b/nanobot/web/__init__.py new file mode 100644 index 00000000..7a08932f --- /dev/null +++ b/nanobot/web/__init__.py @@ -0,0 +1,6 @@ +"""Embedded web UI assets. + +The ``dist/`` subdirectory is populated by ``cd webui && bun run build`` and +is shipped in the wheel; it stays empty in source checkouts until that command +has been run. +""" diff --git a/pyproject.toml b/pyproject.toml index 1e55fed4..5878570a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -113,6 +113,7 @@ include = [ "nanobot/templates/**/*.md", "nanobot/skills/**/*.md", "nanobot/skills/**/*.sh", + "nanobot/web/dist/**/*", ] [tool.hatch.build.targets.wheel] diff --git a/tests/agent/test_session_delete.py b/tests/agent/test_session_delete.py new file mode 100644 index 00000000..aa3296d9 --- /dev/null +++ b/tests/agent/test_session_delete.py @@ -0,0 +1,65 @@ +"""Tests for SessionManager.delete_session and read_session_file.""" + +from pathlib import Path + +from nanobot.session.manager import Session, SessionManager + + +def _seed(workspace: Path, key: str = "telegram:abc") -> SessionManager: + sm = SessionManager(workspace) + session = Session(key=key) + session.add_message("user", "hello") + session.add_message("assistant", "hi back") + sm.save(session) + return sm + + +def test_delete_session_removes_file_and_invalidates_cache(tmp_path: Path) -> None: + sm = _seed(tmp_path, "telegram:abc") + file_path = sm._get_session_path("telegram:abc") + assert file_path.exists() + # Populate cache as a real consumer would. + cached = sm.get_or_create("telegram:abc") + assert cached.messages + + assert sm.delete_session("telegram:abc") is True + assert not file_path.exists() + # Subsequent get_or_create returns a fresh, empty Session (no stale cache). + fresh = sm.get_or_create("telegram:abc") + assert fresh.messages == [] + + +def test_delete_session_returns_false_when_missing(tmp_path: Path) -> None: + sm = SessionManager(tmp_path) + assert sm.delete_session("nope:none") is False + + +def test_read_session_file_returns_metadata_and_messages(tmp_path: Path) -> None: + sm = _seed(tmp_path, "telegram:abc") + data = sm.read_session_file("telegram:abc") + assert data is not None + assert data["key"] == "telegram:abc" + assert isinstance(data["messages"], list) + assert [m["role"] for m in data["messages"]] == ["user", "assistant"] + assert data["created_at"] + assert data["updated_at"] + + +def test_read_session_file_does_not_populate_cache(tmp_path: Path) -> None: + sm = _seed(tmp_path, "telegram:abc") + sm.invalidate("telegram:abc") + assert "telegram:abc" not in sm._cache + sm.read_session_file("telegram:abc") + assert "telegram:abc" not in sm._cache + + +def test_read_session_file_missing(tmp_path: Path) -> None: + sm = SessionManager(tmp_path) + assert sm.read_session_file("nope:none") is None + + +def test_safe_key_matches_internal_path(tmp_path: Path) -> None: + sm = SessionManager(tmp_path) + key = "telegram:abc/def" + expected = sm._get_session_path(key).name + assert SessionManager.safe_key(key) + ".jsonl" == expected diff --git a/tests/channels/test_websocket_http_routes.py b/tests/channels/test_websocket_http_routes.py new file mode 100644 index 00000000..51fd50f4 --- /dev/null +++ b/tests/channels/test_websocket_http_routes.py @@ -0,0 +1,381 @@ +"""End-to-end tests for the embedded webui's HTTP routes on the WebSocket channel.""" + +import asyncio +import functools +import json +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import httpx +import pytest + +from nanobot.channels.websocket import WebSocketChannel +from nanobot.session.manager import Session, SessionManager + +_PORT = 29900 + + +def _ch( + bus: Any, + *, + session_manager: SessionManager | None = None, + static_dist_path: Path | None = None, + port: int = _PORT, + **extra: Any, +) -> WebSocketChannel: + cfg: dict[str, Any] = { + "enabled": True, + "allowFrom": ["*"], + "host": "127.0.0.1", + "port": port, + "path": "/", + "websocketRequiresToken": False, + } + cfg.update(extra) + return WebSocketChannel( + cfg, + bus, + session_manager=session_manager, + static_dist_path=static_dist_path, + ) + + +@pytest.fixture() +def bus() -> MagicMock: + b = MagicMock() + b.publish_inbound = AsyncMock() + return b + + +async def _http_get( + url: str, headers: dict[str, str] | None = None +) -> httpx.Response: + return await asyncio.to_thread( + functools.partial(httpx.get, url, headers=headers or {}, timeout=5.0) + ) + + +def _seed_session(workspace: Path, key: str = "websocket:test") -> SessionManager: + sm = SessionManager(workspace) + s = Session(key=key) + s.add_message("user", "hi") + s.add_message("assistant", "hello back") + sm.save(s) + return sm + + +def _seed_many(workspace: Path, keys: list[str]) -> SessionManager: + sm = SessionManager(workspace) + for k in keys: + s = Session(key=k) + s.add_message("user", f"hi from {k}") + sm.save(s) + return sm + + +@pytest.mark.asyncio +async def test_bootstrap_returns_token_for_localhost( + bus: MagicMock, tmp_path: Path +) -> None: + sm = _seed_session(tmp_path) + channel = _ch(bus, session_manager=sm, port=29901) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + resp = await _http_get("http://127.0.0.1:29901/webui/bootstrap") + assert resp.status_code == 200 + body = resp.json() + assert body["token"].startswith("nbwt_") + assert body["ws_path"] == "/" + assert body["expires_in"] > 0 + assert isinstance(body.get("model_name"), str) + finally: + await channel.stop() + await server_task + + +@pytest.mark.asyncio +async def test_sessions_routes_require_bearer_token( + bus: MagicMock, tmp_path: Path +) -> None: + sm = _seed_session(tmp_path, key="websocket:abc") + channel = _ch(bus, session_manager=sm, port=29902) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + # Unauthenticated → 401. + deny = await _http_get("http://127.0.0.1:29902/api/sessions") + assert deny.status_code == 401 + + # Mint a token via bootstrap, then call the API with it. + boot = await _http_get("http://127.0.0.1:29902/webui/bootstrap") + token = boot.json()["token"] + auth = {"Authorization": f"Bearer {token}"} + + listing = await _http_get("http://127.0.0.1:29902/api/sessions", headers=auth) + assert listing.status_code == 200 + keys = [s["key"] for s in listing.json()["sessions"]] + assert "websocket:abc" in keys + # Server stays an opaque source: filesystem paths must not leak to the wire. + assert all("path" not in s for s in listing.json()["sessions"]) + + msgs = await _http_get( + "http://127.0.0.1:29902/api/sessions/websocket:abc/messages", + headers=auth, + ) + assert msgs.status_code == 200 + body = msgs.json() + assert body["key"] == "websocket:abc" + assert [m["role"] for m in body["messages"]] == ["user", "assistant"] + finally: + await channel.stop() + await server_task + + +@pytest.mark.asyncio +async def test_sessions_list_only_returns_websocket_sessions_by_default( + bus: MagicMock, tmp_path: Path +) -> None: + # Seed a realistic multi-channel disk state: CLI, Slack, Lark and + # websocket sessions all live in the same ``sessions/`` directory. + sm = _seed_many( + tmp_path, + [ + "cli:direct", + "slack:C123", + "lark:oc_abc", + "websocket:alpha", + "websocket:beta", + ], + ) + channel = _ch(bus, session_manager=sm, port=29906) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + boot = await _http_get("http://127.0.0.1:29906/webui/bootstrap") + token = boot.json()["token"] + auth = {"Authorization": f"Bearer {token}"} + + listing = await _http_get( + "http://127.0.0.1:29906/api/sessions", headers=auth + ) + assert listing.status_code == 200 + keys = {s["key"] for s in listing.json()["sessions"]} + # Only websocket-channel sessions are part of the webui surface; CLI / + # Slack / Lark rows would be non-resumable from the browser. + assert keys == {"websocket:alpha", "websocket:beta"} + finally: + await channel.stop() + await server_task + + +@pytest.mark.asyncio +async def test_session_delete_removes_file(bus: MagicMock, tmp_path: Path) -> None: + sm = _seed_session(tmp_path, key="websocket:doomed") + channel = _ch(bus, session_manager=sm, port=29903) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + boot = await _http_get("http://127.0.0.1:29903/webui/bootstrap") + token = boot.json()["token"] + auth = {"Authorization": f"Bearer {token}"} + + path = sm._get_session_path("websocket:doomed") + assert path.exists() + resp = await _http_get( + "http://127.0.0.1:29903/api/sessions/websocket:doomed/delete", + headers=auth, + ) + assert resp.status_code == 200 + assert resp.json()["deleted"] is True + assert not path.exists() + finally: + await channel.stop() + await server_task + + +@pytest.mark.asyncio +async def test_session_routes_accept_percent_encoded_websocket_keys( + bus: MagicMock, tmp_path: Path +) -> None: + sm = _seed_session(tmp_path, key="websocket:encoded-key") + channel = _ch(bus, session_manager=sm, port=29910) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + boot = await _http_get("http://127.0.0.1:29910/webui/bootstrap") + token = boot.json()["token"] + auth = {"Authorization": f"Bearer {token}"} + + msgs = await _http_get( + "http://127.0.0.1:29910/api/sessions/websocket%3Aencoded-key/messages", + headers=auth, + ) + assert msgs.status_code == 200 + assert msgs.json()["key"] == "websocket:encoded-key" + + path = sm._get_session_path("websocket:encoded-key") + assert path.exists() + deleted = await _http_get( + "http://127.0.0.1:29910/api/sessions/websocket%3Aencoded-key/delete", + headers=auth, + ) + assert deleted.status_code == 200 + assert deleted.json()["deleted"] is True + assert not path.exists() + finally: + await channel.stop() + await server_task + + +@pytest.mark.asyncio +async def test_session_routes_reject_non_websocket_keys( + bus: MagicMock, tmp_path: Path +) -> None: + sm = _seed_many( + tmp_path, + [ + "websocket:kept", + "cli:direct", + "slack:C123", + ], + ) + channel = _ch(bus, session_manager=sm, port=29909) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + boot = await _http_get("http://127.0.0.1:29909/webui/bootstrap") + token = boot.json()["token"] + auth = {"Authorization": f"Bearer {token}"} + + # The webui list already hides non-websocket sessions; handcrafted URLs + # should hit the same boundary rather than exposing or deleting them. + msgs = await _http_get( + "http://127.0.0.1:29909/api/sessions/cli:direct/messages", + headers=auth, + ) + assert msgs.status_code == 404 + + doomed = sm._get_session_path("slack:C123") + assert doomed.exists() + deny_delete = await _http_get( + "http://127.0.0.1:29909/api/sessions/slack:C123/delete", + headers=auth, + ) + assert deny_delete.status_code == 404 + assert doomed.exists() + finally: + await channel.stop() + await server_task + + +@pytest.mark.asyncio +async def test_session_routes_reject_invalid_key( + bus: MagicMock, tmp_path: Path +) -> None: + sm = _seed_session(tmp_path) + channel = _ch(bus, session_manager=sm, port=29904) + server_task = asyncio.create_task(channel.start()) + await asyncio.sleep(0.3) + try: + boot = await _http_get("http://127.0.0.1:29904/webui/bootstrap") + token = boot.json()["token"] + auth = {"Authorization": f"Bearer {token}"} + + # Invalid characters in the key -> regex match fails -> 404 + # (route doesn't match, falls through to channel 404). + resp = await _http_get( + "http://127.0.0.1:29904/api/sessions/bad%20key/messages", + headers=auth, + ) + assert resp.status_code in {400, 404} + finally: + await channel.stop() + await server_task + + +@pytest.mark.asyncio +async def test_static_serves_index_when_dist_present( + bus: MagicMock, tmp_path: Path +) -> None: + dist = tmp_path / "dist" + dist.mkdir() + (dist / "index.html").write_text("
+
+ {t("app.error.title")}
+{state.message}
++ {t("app.error.gatewayHint")} +
+
+ + Your conversations are persisted locally under the nanobot + workspace. Start typing and I'll open a new chat. +
+