"""Connection supervisor. Owns the adapter's lifecycle so the web app never has to care whether the robot is up. One background task: disconnected -> connecting -> connected -> (health poll) -> disconnected -> ... Connection failures are normal, not exceptional: they are published as status events and retried with capped exponential backoff. The HTTP server keeps serving the dashboard throughout, which is the whole point of the offline-first design. """ from __future__ import annotations import asyncio import logging import time from typing import Any, Dict, Optional from ..config.settings import Settings from ..core.events import EventBus, EventType from .base import RobotAdapter, RobotError, RobotState from .factory import create_robot logger = logging.getLogger(__name__) class RobotManager: def __init__(self, settings: Settings, bus: EventBus) -> None: self._settings = settings self._bus = bus self._adapter: RobotAdapter = create_robot(settings) self._state = RobotState.DISCONNECTED self._error: Optional[str] = None self._error_code: Optional[str] = None self._latency_ms: Optional[float] = None self._connected_since: Optional[float] = None self._attempts = 0 self._task: Optional[asyncio.Task] = None self._wake = asyncio.Event() self._closing = False # -- accessors ----------------------------------------------------------- # @property def adapter(self) -> RobotAdapter: return self._adapter @property def state(self) -> RobotState: return self._state @property def is_connected(self) -> bool: return self._state == RobotState.CONNECTED and self._adapter.is_connected def status_dict(self) -> Dict[str, Any]: info = self._adapter.info return { "state": self._state.value, "connected": self.is_connected, "speaking": self._adapter.is_speaking, "robot": info.to_dict(), "latencyMs": round(self._latency_ms, 1) if self._latency_ms is not None else None, "uptimeSeconds": ( round(time.time() - self._connected_since, 1) if self._connected_since else None ), "attempts": self._attempts, "error": self._error, "errorCode": self._error_code, "configIssues": [issue.to_dict() for issue in self._settings.issues], } # -- lifecycle ----------------------------------------------------------- # async def start(self) -> None: self._closing = False self._task = asyncio.create_task(self._supervise(), name="robot-supervisor") self._publish() async def stop(self) -> None: self._closing = True self._wake.set() if self._task is not None: self._task.cancel() try: await self._task except (asyncio.CancelledError, Exception): pass self._task = None try: await self._adapter.disconnect() except Exception: # pragma: no cover logger.debug("adapter disconnect raised", exc_info=True) self._set_state(RobotState.DISCONNECTED) def request_reconnect(self) -> None: """Ask the supervisor to retry immediately (used by the API).""" self._wake.set() async def rebuild(self, settings: Settings) -> None: """Swap in a new configuration without restarting the process.""" await self.stop() self._settings = settings self._adapter = create_robot(settings) self._error = None self._error_code = None self._latency_ms = None self._attempts = 0 await self.start() # -- supervisor ---------------------------------------------------------- # async def _supervise(self) -> None: delay = self._settings.robot.reconnect_min_delay while not self._closing: try: if not self._adapter.is_connected: self._attempts += 1 self._set_state(RobotState.CONNECTING) started = time.perf_counter() await self._adapter.connect() self._latency_ms = (time.perf_counter() - started) * 1000 self._connected_since = time.time() self._error = None self._error_code = None delay = self._settings.robot.reconnect_min_delay self._set_state(RobotState.CONNECTED) logger.info("robot connected (%.0f ms)", self._latency_ms) await self._sleep_or_wake(self._settings.robot.health_interval) if self._closing: break started = time.perf_counter() healthy = await self._adapter.health_check() elapsed = (time.perf_counter() - started) * 1000 if healthy: # Smooth the latency reading so the UI badge does not flicker. self._latency_ms = ( elapsed if self._latency_ms is None else self._latency_ms * 0.7 + elapsed * 0.3 ) if self._state != RobotState.CONNECTED: self._set_state(RobotState.CONNECTED) else: self._publish() # keep latency fresh in the dashboard else: logger.warning("robot health check failed - reconnecting") self._connected_since = None self._error = "Lost connection to the robot." self._error_code = "unreachable" self._set_state(RobotState.DISCONNECTED) await self._safe_disconnect() except asyncio.CancelledError: raise except RobotError as exc: self._on_failure(exc.user_message, exc.code) delay = await self._backoff(delay) except Exception as exc: # never let the supervisor die logger.exception("unexpected supervisor error") self._on_failure("Unexpected robot error: {0}".format(exc), "internal") delay = await self._backoff(delay) def _on_failure(self, message: str, code: str) -> None: self._connected_since = None self._latency_ms = None self._error = message self._error_code = code self._set_state(RobotState.ERROR if code == "not_configured" else RobotState.DISCONNECTED) async def _backoff(self, delay: float) -> float: await self._sleep_or_wake(delay) return min(delay * 2, self._settings.robot.reconnect_max_delay) async def _sleep_or_wake(self, seconds: float) -> None: """Sleep, but return early if someone calls request_reconnect().""" try: await asyncio.wait_for(self._wake.wait(), timeout=max(0.1, seconds)) except asyncio.TimeoutError: return finally: self._wake.clear() async def _safe_disconnect(self) -> None: try: await self._adapter.disconnect() except Exception: # pragma: no cover logger.debug("disconnect during recovery raised", exc_info=True) # -- events -------------------------------------------------------------- # def _set_state(self, state: RobotState) -> None: changed = state != self._state self._state = state if changed: logger.info("robot state -> %s", state.value) self._publish() def _publish(self) -> None: self._bus.publish(EventType.ROBOT_STATUS, self.status_dict())