diff --git a/nanobot/channels/websocket/runtime.py b/nanobot/channels/websocket/runtime.py index d3913b343..92f5f12fe 100644 --- a/nanobot/channels/websocket/runtime.py +++ b/nanobot/channels/websocket/runtime.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import ipaddress import json +import socket import ssl import uuid from contextlib import suppress @@ -13,7 +14,7 @@ from typing import TYPE_CHECKING, Any, Self, TypeGuard, cast from urllib.parse import urlsplit, urlunsplit from pydantic import Field, PrivateAttr, field_validator, model_validator -from websockets.asyncio.server import ServerConnection, serve, unix_serve +from websockets.asyncio.server import Server, ServerConnection, serve, unix_serve from websockets.exceptions import ConnectionClosed from websockets.http11 import Request as WsRequest @@ -55,6 +56,37 @@ if TYPE_CHECKING: # Plain HTTP WebUI routes also run through websockets.process_request. _WEBUI_HTTP_OPEN_TIMEOUT_S = 360.0 +_LISTENER_CHECK_INTERVAL_S = 0.5 +_LISTENER_STABLE_AFTER_S = 30.0 +_LISTENER_RESTART_BACKOFF_S = (1.0, 2.0, 4.0, 8.0, 16.0, 30.0) + +# A bind conflict or invalid address needs operator action and must not be +# retried forever. These errors can be caused by a transient local network +# interruption and are safe to retry at the channel boundary. +_RECOVERABLE_LISTENER_ERRNOS = { + getattr(socket, name) + for name in ( + "ECONNABORTED", + "ECONNRESET", + "EHOSTDOWN", + "EHOSTUNREACH", + "ENETDOWN", + "ENETRESET", + "ENETUNREACH", + "ETIMEDOUT", + ) + if hasattr(socket, name) +} +_RECOVERABLE_LISTENER_WINERRORS = { + 64, # ERROR_NETNAME_DELETED / "The specified network name is no longer available." + 995, # ERROR_OPERATION_ABORTED + 10050, # WSAENETDOWN + 10052, # WSAENETRESET + 10053, # WSAECONNABORTED + 10054, # WSAECONNRESET + 10060, # WSAETIMEDOUT + 10065, # WSAEHOSTUNREACH +} _ROUTING_ASSERTION_HEADERS = frozenset( @@ -295,6 +327,10 @@ def _parse_envelope(raw: str) -> dict[str, Any] | None: return envelope +class _ListenerUnavailableError(OSError): + """Raised when a previously bound listener loses its serving socket.""" + + class WebSocketChannel(BaseChannel): """Run a local WebSocket server; forward text/JSON messages to the message bus.""" @@ -322,6 +358,7 @@ class WebSocketChannel(BaseChannel): self._webui_connections = gateway.endpoint.webui_connections self._stop_event: asyncio.Event | None = None self._server_task: asyncio.Task[None] | None = None + self._server: Server | None = None self.gateway = gateway self._media = gateway.media @@ -500,14 +537,89 @@ class WebSocketChannel(BaseChannel): # -- Server lifecycle and connection ingress --------------------------- + @staticmethod + def _listener_is_serving(server: Server) -> bool: + """Return whether every bound socket still has a live listen capability.""" + try: + sockets = server.sockets + return bool(sockets) and server.is_serving() and all( + sock.fileno() >= 0 + and bool(sock.getsockopt(socket.SOL_SOCKET, socket.SO_ACCEPTCONN)) + for sock in sockets + ) + except OSError: + return False + + @staticmethod + def _is_recoverable_listener_error(error: Exception, *, was_serving: bool) -> bool: + if isinstance(error, _ListenerUnavailableError): + return True + if not isinstance(error, OSError): + return False + if was_serving: + return True + winerror = getattr(error, "winerror", None) + return ( + error.errno in _RECOVERABLE_LISTENER_ERRNOS + or winerror in _RECOVERABLE_LISTENER_WINERRORS + ) + + async def _wait_for_listener_loss(self, server: Server) -> None: + """Wait for shutdown or raise when the serving socket disappears.""" + assert self._stop_event is not None + while not self._stop_event.is_set(): + try: + await asyncio.wait_for( + self._stop_event.wait(), + timeout=_LISTENER_CHECK_INTERVAL_S, + ) + except TimeoutError: + if not self._listener_is_serving(server): + raise _ListenerUnavailableError( + "WebSocket listener is no longer accepting connections" + ) + + async def _close_server(self, server: Server, socket_path: str) -> None: + server.close() + try: + await server.wait_closed() + except OSError as exc: + self.logger.warning("WebSocket server close failed: {}", exc) + if socket_path: + with suppress(FileNotFoundError): + Path(socket_path).unlink() + + def _log_listener_ready(self, scheme: str) -> None: + self.logger.info( + "WebSocket server listening on {}", + ( + f"unix:{self.config.unix_socket_path}{self.config.path}" + if self.config.unix_socket_path + else f"{scheme}://{self.config.host}:{self.config.port}{self.config.path}" + ), + ) + if self.config.token_issue_path: + self.logger.info( + "WebSocket token issue route: {}", + ( + f"unix:{self.config.unix_socket_path}" + f"{_normalize_config_path(self.config.token_issue_path)}" + if self.config.unix_socket_path + else ( + f"{scheme}://{self.config.host}:{self.config.port}" + f"{_normalize_config_path(self.config.token_issue_path)}" + ) + ), + ) + async def start(self) -> None: from nanobot.utils.logging_bridge import redirect_lib_logging redirect_lib_logging("websockets", level="WARNING") ws_logger = websockets_server_logger() - self._running = True - self._stop_event = asyncio.Event() + stop_event = asyncio.Event() + self._stop_event = stop_event ssl_context = self._build_ssl_context() scheme = "wss" if ssl_context else "ws" @@ -521,71 +633,100 @@ class WebSocketChannel(BaseChannel): async def handler(connection: ServerConnection) -> None: await self._connection_loop(connection) - self.logger.info( - "WebSocket server listening on {}", - ( - f"unix:{self.config.unix_socket_path}{self.config.path}" - if self.config.unix_socket_path - else f"{scheme}://{self.config.host}:{self.config.port}{self.config.path}" - ), - ) - if self.config.token_issue_path: - self.logger.info( - "WebSocket token issue route: {}", - ( - f"unix:{self.config.unix_socket_path}{_normalize_config_path(self.config.token_issue_path)}" - if self.config.unix_socket_path - else ( - f"{scheme}://{self.config.host}:{self.config.port}" - f"{_normalize_config_path(self.config.token_issue_path)}" - ) - ), - ) - async def runner() -> None: socket_path = self.config.unix_socket_path - if socket_path: - path_obj = Path(socket_path) - path_obj.parent.mkdir(parents=True, exist_ok=True) - with suppress(FileNotFoundError): - path_obj.unlink() - server = await unix_serve( - handler, - socket_path, - process_request=process_request, - open_timeout=_WEBUI_HTTP_OPEN_TIMEOUT_S, - max_size=self.config.max_message_bytes, - ping_interval=self.config.ping_interval_s, - ping_timeout=self.config.ping_timeout_s, - logger=ws_logger, - ) - with suppress(OSError): - path_obj.chmod(0o600) - else: - server = await serve( - handler, - self.config.host, - self.config.port, - process_request=process_request, - open_timeout=_WEBUI_HTTP_OPEN_TIMEOUT_S, - max_size=self.config.max_message_bytes, - ping_interval=self.config.ping_interval_s, - ping_timeout=self.config.ping_timeout_s, - ssl=ssl_context, - logger=ws_logger, - ) - try: - assert self._stop_event is not None - await self._stop_event.wait() - finally: - server.close() - await server.wait_closed() - if socket_path: - with suppress(FileNotFoundError): - Path(socket_path).unlink() + failures = 0 + while not stop_event.is_set(): + server: Server | None = None + was_serving = False + started_at = 0.0 + try: + if socket_path: + path_obj = Path(socket_path) + path_obj.parent.mkdir(parents=True, exist_ok=True) + with suppress(FileNotFoundError): + path_obj.unlink() + server = await unix_serve( + handler, + socket_path, + process_request=process_request, + open_timeout=_WEBUI_HTTP_OPEN_TIMEOUT_S, + max_size=self.config.max_message_bytes, + ping_interval=self.config.ping_interval_s, + ping_timeout=self.config.ping_timeout_s, + logger=ws_logger, + ) + with suppress(OSError): + path_obj.chmod(0o600) + else: + server = await serve( + handler, + self.config.host, + self.config.port, + process_request=process_request, + open_timeout=_WEBUI_HTTP_OPEN_TIMEOUT_S, + max_size=self.config.max_message_bytes, + ping_interval=self.config.ping_interval_s, + ping_timeout=self.config.ping_timeout_s, + ssl=ssl_context, + logger=ws_logger, + ) - self._server_task = asyncio.create_task(runner()) - await self._server_task + self._server = server + was_serving = True + if not self._listener_is_serving(server): + raise _ListenerUnavailableError( + "WebSocket listener did not enter a serving state" + ) + self._running = True + started_at = asyncio.get_running_loop().time() + self._log_listener_ready(scheme) + await self._wait_for_listener_loss(server) + except asyncio.CancelledError: + raise + except Exception as exc: + self._running = False + if not self._is_recoverable_listener_error( + exc, + was_serving=was_serving, + ): + raise + uptime = ( + asyncio.get_running_loop().time() - started_at + if started_at + else 0.0 + ) + if uptime >= _LISTENER_STABLE_AFTER_S: + failures = 0 + delay = _LISTENER_RESTART_BACKOFF_S[ + min(failures, len(_LISTENER_RESTART_BACKOFF_S) - 1) + ] + failures += 1 + self.logger.warning( + "WebSocket listener failed ({}: {}); retrying in {:.1f}s", + type(exc).__name__, + exc, + delay, + ) + try: + await asyncio.wait_for(stop_event.wait(), timeout=delay) + except TimeoutError: + pass + finally: + self._running = False + if server is not None: + await self._close_server(server, socket_path) + if self._server is server: + self._server = None + + task = asyncio.create_task(runner()) + self._server_task = task + try: + await task + finally: + self._running = False + if self._server_task is task: + self._server_task = None async def _connection_loop(self, connection: ServerConnection) -> None: request = connection.request @@ -666,14 +807,15 @@ class WebSocketChannel(BaseChannel): # -- Outbound WebSocket events ----------------------------------------- async def stop(self) -> None: - if not self._running: + server_task = self._server_task + if not self._running and server_task is None: return self._running = False if self._stop_event: self._stop_event.set() - if self._server_task: + if server_task: try: - await self._server_task + await server_task except asyncio.CancelledError: current_task = asyncio.current_task() if current_task is not None and current_task.cancelling(): @@ -681,7 +823,8 @@ class WebSocketChannel(BaseChannel): self.logger.debug("server task was already cancelled during shutdown") except Exception as e: self.logger.warning("server task error during shutdown: {}", e) - self._server_task = None + if self._server_task is server_task: + self._server_task = None await self._commands.close() self._subs.clear() self._conn_chats.clear() diff --git a/nanobot/channels/websocket/tests/test_websocket_channel.py b/nanobot/channels/websocket/tests/test_websocket_channel.py index 16533020e..39c76003d 100644 --- a/nanobot/channels/websocket/tests/test_websocket_channel.py +++ b/nanobot/channels/websocket/tests/test_websocket_channel.py @@ -227,6 +227,7 @@ async def test_start_extends_http_open_timeout_for_slow_settings_routes( return Server() monkeypatch.setattr(websocket_module, "serve", fake_serve) + monkeypatch.setattr(channel, "_listener_is_serving", lambda _server: True) await channel.start() diff --git a/nanobot/channels/websocket/tests/test_websocket_runtime_lifecycle.py b/nanobot/channels/websocket/tests/test_websocket_runtime_lifecycle.py new file mode 100644 index 000000000..f0a23a902 --- /dev/null +++ b/nanobot/channels/websocket/tests/test_websocket_runtime_lifecycle.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import asyncio +import errno +from unittest.mock import MagicMock + +import pytest + +from nanobot.bus.queue import MessageBus +from nanobot.channels.websocket.runtime import WebSocketChannel + + +class _FakeSocket: + def __init__(self) -> None: + self.open = True + + def fileno(self) -> int: + return 1 if self.open else -1 + + def getsockopt(self, _level: int, _option: int) -> int: + return int(self.open) + + +class _FakeServer: + def __init__(self) -> None: + self.socket = _FakeSocket() + self.closed = False + + @property + def sockets(self) -> tuple[_FakeSocket, ...]: + return (self.socket,) + + def is_serving(self) -> bool: + return not self.closed + + def close(self) -> None: + self.closed = True + self.socket.open = False + + async def wait_closed(self) -> None: + return None + + +def _channel() -> WebSocketChannel: + gateway = MagicMock() + gateway.session_manager = None + return WebSocketChannel( + {"enabled": True, "allowFrom": ["*"]}, + MessageBus(), + gateway=gateway, + ) + + +@pytest.mark.asyncio +async def test_websocket_does_not_report_running_before_bind_succeeds(monkeypatch) -> None: + channel = _channel() + channel.logger = MagicMock() + bind_error = OSError(errno.EADDRINUSE, "address already in use") + + async def fail_bind(*_args, **_kwargs): + raise bind_error + + monkeypatch.setattr("nanobot.channels.websocket.runtime.serve", fail_bind) + + with pytest.raises(OSError) as exc_info: + await channel.start() + + assert exc_info.value is bind_error + assert channel.is_running is False + assert not any( + call.args and call.args[0] == "WebSocket server listening on {}" + for call in channel.logger.info.call_args_list + ) + + +@pytest.mark.asyncio +async def test_websocket_restarts_only_its_listener_after_serving_socket_is_lost( + monkeypatch, +) -> None: + channel = _channel() + first = _FakeServer() + second = _FakeServer() + servers = iter((first, second)) + bind_count = 0 + rebound = asyncio.Event() + + async def bind(*_args, **_kwargs): + nonlocal bind_count + bind_count += 1 + server = next(servers) + if bind_count == 2: + rebound.set() + return server + + monkeypatch.setattr("nanobot.channels.websocket.runtime.serve", bind) + monkeypatch.setattr( + "nanobot.channels.websocket.runtime._LISTENER_CHECK_INTERVAL_S", + 0.01, + ) + monkeypatch.setattr( + "nanobot.channels.websocket.runtime._LISTENER_RESTART_BACKOFF_S", + (0.05,), + ) + + start_task = asyncio.create_task(channel.start()) + try: + for _ in range(20): + if channel.is_running: + break + await asyncio.sleep(0) + assert channel.is_running is True + + first.socket.open = False + for _ in range(50): + if not channel.is_running: + break + await asyncio.sleep(0.005) + + assert channel.is_running is False + assert bind_count == 1 + await asyncio.wait_for(rebound.wait(), timeout=1) + assert channel.is_running is True + assert first.closed is True + finally: + await channel.stop() + await start_task + + assert second.closed is True diff --git a/nanobot/cli/gateway_runtime.py b/nanobot/cli/gateway_runtime.py index f2315ab08..9aef3e18d 100644 --- a/nanobot/cli/gateway_runtime.py +++ b/nanobot/cli/gateway_runtime.py @@ -246,6 +246,44 @@ def _print_gateway_health_endpoint(host: str, port: int) -> None: ) +def _gateway_readiness_payload(channels: Any) -> tuple[bool, dict[str, object]]: + """Describe process liveness separately from required WebSocket readiness.""" + channel_status: dict[str, Any] = {} + get_status = getattr(channels, "get_status", None) + if callable(get_status): + try: + raw_status = get_status() + if isinstance(raw_status, dict): + channel_status = cast(dict[str, Any], raw_status) + except Exception: + logger.exception("Gateway readiness could not read channel status") + + websocket = channel_status.get("websocket") + websocket_required = websocket is not None or "websocket" in getattr( + channels, + "enabled_channels", + (), + ) + if not websocket_required: + websocket_state = "disabled" + ready = True + elif isinstance(websocket, dict): + websocket_status = cast(dict[str, Any], websocket) + ready = websocket_status.get("running") is True + state = websocket_status.get("state") + websocket_state = str(state) if isinstance(state, str) else "unavailable" + else: + ready = False + websocket_state = "unavailable" + + return ready, { + "status": "ok" if ready else "degraded", + "process": "alive", + "ready": ready, + "websocket": websocket_state, + } + + async def _close_gateway_runtime( agent: AgentLoop, mcp_provider: MCPProvider, @@ -758,8 +796,9 @@ def _run_gateway( method, path = parts[0], parts[1] if method == "GET" and path == "/health": - body = _json.dumps({"status": "ok"}) - status = "200 OK" + ready, payload = _gateway_readiness_payload(channels) + body = _json.dumps(payload) + status = "200 OK" if ready else "503 Service Unavailable" content_type = "application/json" else: body = "Not Found" @@ -994,4 +1033,6 @@ def _run_gateway( restore_shutdown_handlers() with gateway_runtime.foreground_instance(gateway_start_options): + if health_server_enabled: + gateway_runtime.publish_health_host(config.gateway.host) asyncio.run(run()) diff --git a/nanobot/cli/tui_launcher.py b/nanobot/cli/tui_launcher.py index a76d13e00..4d4075709 100644 --- a/nanobot/cli/tui_launcher.py +++ b/nanobot/cli/tui_launcher.py @@ -64,6 +64,8 @@ _TUI_RELEASE_LIMITS = { } # Keep in sync with TUI_DETACH_EXIT_CODE in tui/src/index.ts. _TUI_DETACH_EXIT_CODE = 90 +_GATEWAY_READY_TIMEOUT_S = 20.0 +_GATEWAY_READY_POLL_S = 0.1 @dataclass(frozen=True) @@ -417,17 +419,52 @@ def _ensure_gateway( lease = GatewayClientLease(runtime, kind="tui") lease.acquire() try: + def ready(status: object) -> bool: + management_ready = getattr(status, "ready", None) + if not isinstance(management_ready, bool): + management_ready = _gateway_health_ready( + config.gateway.host, + config.gateway.port, + ) + return _webui_endpoint_reachable(base_url) and management_ready + + def wait_for_ready(log_path: object) -> _GatewayHandle: + deadline = time.monotonic() + _GATEWAY_READY_TIMEOUT_S + while time.monotonic() < deadline: + current = runtime.status() + if not current.running: + break + if current.port not in {None, config.gateway.port}: + break + if ready(current): + return _GatewayHandle(base_url=base_url, lease=lease) + time.sleep(_GATEWAY_READY_POLL_S) + + current = runtime.status() + if current.running: + raise TuiUnavailableError( + "local gateway process is running but its WebSocket/WebUI listener " + "is unavailable; channel recovery did not restore it. " + "Run `nanobot gateway status` and inspect logs at " + f"{log_path}; if it remains degraded, run `nanobot gateway restart`." + ) + raise TuiUnavailableError( + f"local gateway did not become ready; logs: {log_path}" + ) + status = runtime.status() - endpoint_reachable = _webui_endpoint_reachable(base_url) if status.running: if status.port not in {None, config.gateway.port}: raise TuiUnavailableError( "the matching gateway instance is running on a different port; " "restart it or use `nanobot agent --classic`" ) - if endpoint_reachable or not wait_until_ready: + if not wait_until_ready: return _GatewayHandle(base_url=base_url, lease=lease) - elif endpoint_reachable: + if ready(status): + return _GatewayHandle(base_url=base_url, lease=lease) + return wait_for_ready(status.log_path) + elif _webui_endpoint_reachable(base_url): raise TuiUnavailableError( "the configured gateway port belongs to a different nanobot instance; " "stop that instance or use `nanobot agent --classic`" @@ -442,26 +479,17 @@ def _ensure_gateway( f"logs: {result.status.log_path}" ) + if result.message == "gateway_already_running" and result.status.port not in { + None, + config.gateway.port, + }: + raise TuiUnavailableError( + "the matching gateway instance is running on a different port; " + "restart it or use `nanobot agent --classic`" + ) if not wait_until_ready: return _GatewayHandle(base_url=base_url, lease=lease) - - deadline = time.monotonic() + 20 - while time.monotonic() < deadline: - if _webui_endpoint_reachable(base_url): - current = runtime.status() - if current.running and current.port in {None, config.gateway.port}: - return _GatewayHandle(base_url=base_url, lease=lease) - break - if not runtime.status().running and not _gateway_health_ready( - config.gateway.host, - config.gateway.port, - ): - break - time.sleep(0.1) - - raise TuiUnavailableError( - f"local gateway did not become ready; logs: {result.status.log_path}" - ) + return wait_for_ready(result.status.log_path) except BaseException: lease.release(timeout_s=5) raise diff --git a/nanobot/gateway/runtime.py b/nanobot/gateway/runtime.py index d0f6fb1fb..641f5d64f 100644 --- a/nanobot/gateway/runtime.py +++ b/nanobot/gateway/runtime.py @@ -6,6 +6,7 @@ from __future__ import annotations import asyncio import hashlib +import http.client import json import os import subprocess @@ -38,6 +39,33 @@ GatewayLaunchMode = Literal["foreground", "background", "unknown"] GatewayLifetime = Literal["explicit", "on_demand"] +def _gateway_health_ready(host: str, port: int, *, timeout_s: float = 0.4) -> bool: + """Read readiness from the management listener without using proxy settings.""" + connect_host = "127.0.0.1" if host in {"", "0.0.0.0"} else "::1" if host == "::" else host + connection = http.client.HTTPConnection(connect_host, port, timeout=timeout_s) + try: + connection.request("GET", "/health") + response = connection.getresponse() + body = response.read(1024) + except (OSError, http.client.HTTPException, TimeoutError): + return False + finally: + connection.close() + if response.status != 200: + return False + try: + raw_payload = cast(object, json.loads(body.decode("utf-8"))) + except (UnicodeDecodeError, json.JSONDecodeError): + return False + if not isinstance(raw_payload, dict): + return False + payload = cast(dict[str, object], raw_payload) + return ( + payload.get("status") == "ok" + and payload.get("ready") is not False + ) + + def _default_config_path() -> Path: return (Path.home() / ".nanobot" / "config.json").resolve(strict=False) @@ -49,6 +77,7 @@ class GatewayStatus(ProcessStatus): launch_mode: GatewayLaunchMode = "unknown" lifetime: GatewayLifetime = "explicit" clients: int = 0 + ready: bool | None = None @dataclass(frozen=True) @@ -259,6 +288,18 @@ class GatewayRuntime(ManagedProcessRuntime[ProcessStartOptions]): raw_mode if raw_mode in {"foreground", "background"} else "unknown" ) lease = GatewayClientLease(self, kind="gateway-status").snapshot() + ready: bool | None = None + health_host = state.get("health_host") if state else None + if ( + process.running + and process.pid != os.getpid() + and isinstance(health_host, str) + and process.port is not None + ): + ready = _gateway_health_ready(health_host, process.port) + status_reason = process.reason + if ready is False and reason is None and status_reason == "running": + status_reason = "websocket_unavailable" return GatewayStatus( running=process.running, pid=process.pid, @@ -267,12 +308,22 @@ class GatewayRuntime(ManagedProcessRuntime[ProcessStartOptions]): started_at=process.started_at, port=process.port, command=process.command, - reason=process.reason, + reason=status_reason, launch_mode=launch_mode, lifetime="on_demand" if lease.auto_stop else "explicit", clients=lease.clients, + ready=ready, ) + def publish_health_host(self, host: str) -> None: + """Record the management bind host for out-of-process readiness diagnostics.""" + with self._lifecycle_lock(): + state = self._read_state() + if not state or not self._record_matches_process(state, os.getpid()): + return + state["health_host"] = host + self._write_state(state) + @contextmanager def foreground_instance(self, options: ProcessStartOptions) -> Generator[None]: """Publish this foreground gateway while it is available to local clients.""" diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index ea565b7bf..c50ade5d9 100644 --- a/tests/cli/test_commands.py +++ b/tests/cli/test_commands.py @@ -3618,7 +3618,12 @@ def test_gateway_health_endpoint_binds_and_serves_expected_responses( assert health_writer.closed is True assert "HTTP/1.0 200 OK" in health_response health_body = json.loads(health_response.split("\r\n\r\n", 1)[1]) - assert health_body == {"status": "ok"} + assert health_body == { + "status": "ok", + "process": "alive", + "ready": True, + "websocket": "disabled", + } missing_response, missing_writer = _call_handler("/missing") assert missing_writer.closed is True diff --git a/tests/cli/test_gateway_runtime.py b/tests/cli/test_gateway_runtime.py index 38adf624f..7cee95735 100644 --- a/tests/cli/test_gateway_runtime.py +++ b/tests/cli/test_gateway_runtime.py @@ -14,7 +14,11 @@ from contextlib import suppress from nanobot.agent.hook import AgentRunHookContext from nanobot.agent.tools.mcp import MCPProvider from nanobot.agent.tools.registry import ToolRegistry -from nanobot.cli.gateway_runtime import _close_gateway_runtime, _MCPReadinessHook +from nanobot.cli.gateway_runtime import ( + _close_gateway_runtime, + _gateway_readiness_payload, + _MCPReadinessHook, +) class _FakeAgent: @@ -65,6 +69,33 @@ class _TrackingMCPProvider(MCPProvider): self.connect_calls += 1 +def test_gateway_readiness_is_degraded_when_required_websocket_is_unavailable() -> None: + channels = type( + "Channels", + (), + { + "enabled_channels": ["websocket"], + "get_status": lambda self: { + "websocket": { + "enabled": True, + "running": False, + "state": "starting", + } + }, + }, + )() + + ready, payload = _gateway_readiness_payload(channels) + + assert ready is False + assert payload == { + "status": "degraded", + "process": "alive", + "ready": False, + "websocket": "starting", + } + + async def test_mcp_readiness_hook_delegates_to_application_provider() -> None: provider = _TrackingMCPProvider() hook = _MCPReadinessHook(provider) diff --git a/tests/cli/test_tui_launcher.py b/tests/cli/test_tui_launcher.py index 3d438c6eb..a4b7fede2 100644 --- a/tests/cli/test_tui_launcher.py +++ b/tests/cli/test_tui_launcher.py @@ -191,6 +191,78 @@ def test_launcher_terminates_the_tui_when_gateway_start_fails( assert terminated == [True] +def test_launcher_keeps_the_tui_alive_while_an_existing_gateway_recovers( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + config = Config() + events: list[str] = [] + status_calls = 0 + + class FakeRuntime: + def __init__(self, *, paths: object) -> None: + self.paths = paths + + def status(self) -> SimpleNamespace: + nonlocal status_calls + status_calls += 1 + return SimpleNamespace( + running=True, + port=config.gateway.port, + ready=False, + log_path=tmp_path / "gateway.log", + ) + + class FakeProcess: + def poll(self) -> None: + return None + + def terminate(self) -> None: + events.append("terminated") + + def wait(self, timeout: float | None = None) -> int: + assert timeout is None + events.append("waited") + return 0 + + monkeypatch.setattr("nanobot.gateway.GatewayRuntime", FakeRuntime) + monkeypatch.setattr("nanobot.cli.tui_launcher._resolve_tui_command", lambda: ["nanobot-tui"]) + monkeypatch.setattr( + "nanobot.cli.tui_launcher.subprocess.Popen", + lambda *args, **kwargs: FakeProcess(), + ) + monkeypatch.setattr( + "nanobot.cli.tui_launcher._webui_endpoint_reachable", + lambda _url: pytest.fail( + "launcher must not probe readiness for a live recovering gateway" + ), + ) + monkeypatch.setattr( + tui_launcher, + "time", + SimpleNamespace( + monotonic=lambda: pytest.fail( + "launcher must not wait for a live gateway to recover" + ), + sleep=lambda _seconds: pytest.fail( + "launcher must not sleep for gateway recovery" + ), + ), + ) + + result = launch_tui( + config, + config_path=tmp_path / "config.json", + workspace_override=None, + session_id=None, + theme="auto", + ) + + assert result == 0 + assert status_calls == 1 + assert events == ["waited"] + + def test_launcher_promotes_the_gateway_when_the_tui_detaches( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -684,6 +756,10 @@ def test_gateway_reuses_the_matching_managed_instance( monkeypatch.setattr("nanobot.gateway.GatewayRuntime", FakeRuntime) monkeypatch.setattr("nanobot.cli.tui_launcher._webui_endpoint_reachable", lambda _url: True) + monkeypatch.setattr( + "nanobot.cli.tui_launcher._gateway_health_ready", + lambda *_args, **_kwargs: True, + ) gateway = _ensure_gateway( config, @@ -694,21 +770,42 @@ def test_gateway_reuses_the_matching_managed_instance( assert gateway.base_url == "http://127.0.0.1:8765" -def test_gateway_reuse_can_return_before_the_webui_endpoint_is_ready( +def test_gateway_reuse_returns_a_degraded_live_gateway_without_waiting( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: config = Config() + status_calls = 0 class FakeRuntime: def __init__(self, *, paths: object) -> None: self.paths = paths def status(self) -> SimpleNamespace: - return SimpleNamespace(running=True, port=config.gateway.port) + nonlocal status_calls + status_calls += 1 + return SimpleNamespace( + running=True, + port=config.gateway.port, + ready=False, + log_path=tmp_path / "gateway.log", + ) monkeypatch.setattr("nanobot.gateway.GatewayRuntime", FakeRuntime) - monkeypatch.setattr("nanobot.cli.tui_launcher._webui_endpoint_reachable", lambda _url: False) + monkeypatch.setattr( + "nanobot.cli.tui_launcher._webui_endpoint_reachable", + lambda _url: pytest.fail("non-blocking reuse must not probe readiness"), + ) + monkeypatch.setattr( + tui_launcher, + "time", + SimpleNamespace( + monotonic=lambda: pytest.fail( + "non-blocking reuse must not enter the readiness wait" + ), + sleep=lambda _seconds: pytest.fail("non-blocking reuse must not sleep"), + ), + ) gateway = _ensure_gateway( config, @@ -719,6 +816,95 @@ def test_gateway_reuse_can_return_before_the_webui_endpoint_is_ready( assert gateway.base_url == "http://127.0.0.1:8765" assert gateway.lease is not None + assert status_calls == 1 + gateway.lease.release(wait_for_stop=False) + + +def test_gateway_reuse_waits_for_a_live_gateway_to_recover_its_webui_listener( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + config = Config() + endpoint_results = iter((False, False, True)) + + class FakeRuntime: + def __init__(self, *, paths: object) -> None: + self.paths = paths + + def status(self) -> SimpleNamespace: + return SimpleNamespace( + running=True, + port=config.gateway.port, + log_path=tmp_path / "gateway.log", + ) + + monkeypatch.setattr("nanobot.gateway.GatewayRuntime", FakeRuntime) + monkeypatch.setattr( + "nanobot.cli.tui_launcher._webui_endpoint_reachable", + lambda _url: next(endpoint_results), + ) + monkeypatch.setattr( + "nanobot.cli.tui_launcher._gateway_health_ready", + lambda *_args, **_kwargs: True, + ) + clock = iter((0.0, 0.0, 0.1)) + sleeps: list[float] = [] + monkeypatch.setattr( + tui_launcher, + "time", + SimpleNamespace(monotonic=lambda: next(clock), sleep=sleeps.append), + ) + + gateway = _ensure_gateway( + config, + config_path=tmp_path / "config.json", + workspace_override=None, + ) + + assert gateway.base_url == "http://127.0.0.1:8765" + assert gateway.lease is not None + assert sleeps == [tui_launcher._GATEWAY_READY_POLL_S] + gateway.lease.release(wait_for_stop=False) + + +def test_gateway_reuse_with_explicit_wait_rejects_a_live_but_unready_gateway( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + config = Config() + + class FakeRuntime: + def __init__(self, *, paths: object) -> None: + self.paths = paths + + def status(self) -> SimpleNamespace: + return SimpleNamespace( + running=True, + port=config.gateway.port, + log_path=tmp_path / "gateway.log", + ) + + monkeypatch.setattr("nanobot.gateway.GatewayRuntime", FakeRuntime) + monkeypatch.setattr( + "nanobot.cli.tui_launcher._webui_endpoint_reachable", + lambda _url: False, + ) + clock = iter((0.0, tui_launcher._GATEWAY_READY_TIMEOUT_S)) + monkeypatch.setattr( + tui_launcher, + "time", + SimpleNamespace( + monotonic=lambda: next(clock), + sleep=lambda _seconds: pytest.fail("expired readiness wait must not sleep"), + ), + ) + + with pytest.raises(TuiUnavailableError, match="process is running.*listener is unavailable"): + _ensure_gateway( + config, + config_path=tmp_path / "config.json", + workspace_override=None, + ) def test_gateway_started_for_tui_stops_when_its_last_lease_exits( @@ -767,6 +953,10 @@ def test_gateway_started_for_tui_stops_when_its_last_lease_exits( "nanobot.cli.tui_launcher._webui_endpoint_reachable", lambda _url: started, ) + monkeypatch.setattr( + "nanobot.cli.tui_launcher._gateway_health_ready", + lambda *_args, **_kwargs: started, + ) gateway = _ensure_gateway( config, diff --git a/tests/gateway/test_runtime.py b/tests/gateway/test_runtime.py index 67e07c6cc..8dff4c0cb 100644 --- a/tests/gateway/test_runtime.py +++ b/tests/gateway/test_runtime.py @@ -994,6 +994,37 @@ def test_status_keeps_live_state_when_identity_probe_is_temporarily_unavailable( assert runtime.paths.state_path.exists() +def test_status_distinguishes_live_process_from_degraded_gateway_readiness( + tmp_path, + monkeypatch, +): + runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux") + runtime.paths.run_dir.mkdir(parents=True) + runtime.paths.state_path.write_text( + json.dumps( + { + "pid": 12345, + "identity": 42, + "port": 18791, + "health_host": "127.0.0.1", + } + ), + encoding="utf-8", + ) + monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True) + monkeypatch.setattr(runtime, "_process_identity", lambda _pid: 42) + monkeypatch.setattr( + "nanobot.gateway.runtime._gateway_health_ready", + lambda _host, _port: False, + ) + + status = runtime.status() + + assert status.running is True + assert status.ready is False + assert status.reason == "websocket_unavailable" + + def test_stop_refuses_to_signal_a_process_when_identity_cannot_be_verified( tmp_path, monkeypatch,