""" RoboPark Session Scheduler - Backend API Manages robot fleet, LiveKit servers, and session orchestration """ import os import asyncio import base64 import contextvars import json import logging import re import secrets import hashlib import hmac import time import uuid from datetime import datetime, timedelta from typing import Any, Optional, List from contextlib import asynccontextmanager from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, Query, Header from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.responses import HTMLResponse from pydantic import BaseModel import httpx import aiosqlite # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Database path DB_PATH = os.path.join(os.getenv("SCHEDULER_DATA_DIR", "/app/data"), "scheduler.db") # Path to a ROBOVOICE-style settings file the scheduler reads for characters. # In production this should be a bind-mount to ROBOVOICE-main/settings.json (or .default.json). ROBOVOICE_SETTINGS_PATH = os.path.join(os.getenv("SCHEDULER_DATA_DIR", "/app/data"), "robovoice_settings.json") PRODUCTION_CONFIG_PATH = os.getenv( "ROBOPARK_PRODUCTION_CONFIG", os.path.join(os.path.dirname(os.path.abspath(__file__)), "production_config.json"), ) # Shared credential used only by the ROBOVOICE worker to refresh an active # RoboPark session. Device endpoints continue to require device tokens. If the # environment does not provide it, lifespan() creates/loads this local secret # file so a bare `robopark serve` is safe to run without manual secret wiring. ROBOPARK_AGENT_TOKEN = os.getenv("ROBOPARK_AGENT_TOKEN", "").strip() ROBOPARK_AGENT_TOKEN_FILE = os.getenv( "ROBOPARK_AGENT_TOKEN_FILE", os.path.join(os.getenv("SCHEDULER_DATA_DIR", "/app/data"), ".robopark-agent-token"), ) def _load_mesh_token() -> str: """Load the hub credential shared with unified robot runtimes.""" configured = os.getenv("ROBOPARK_MESH_TOKEN", "").strip() if configured: return configured try: with open(os.path.expanduser("~/.robopark/mesh.token"), "r", encoding="utf8") as handle: return handle.read().strip() except OSError: return "" ROBOPARK_MESH_TOKEN = _load_mesh_token() # ElevenLabs credentials used by the /api/voices/elevenlabs endpoint. We accept # both the long-form ELEVENLABS_API_KEY and a shorter ELEVENLABS_KEY alias so # operators can drop in whichever naming their tooling already exports. Either # may be set in the scheduler's environment. ELEVENLABS_API_KEY = ( os.getenv("ELEVENLABS_API_KEY", "").strip() or os.getenv("ELEVENLABS_KEY", "").strip() or os.getenv("ELEVEN_API_KEY", "").strip() ) ELEVENLABS_BASE_URL = os.getenv("ELEVENLABS_BASE_URL", "https://api.elevenlabs.io").rstrip("/") async def _get_elevenlabs_api_key() -> str: """Resolve the provider key without ever returning it through public APIs.""" if ELEVENLABS_API_KEY: return ELEVENLABS_API_KEY return (await get_setting("elevenlabs_api_key") or "").strip() # Optional base URL of a local Kokoro TTS server. When set, /api/voices/kokoro # will proxy its /v1/audio/voices endpoint to return the dynamic list of voices # the server actually has installed; otherwise we fall back to a curated static # list of the canonical Kokoro-82M voices. KOKORO_BASE_URL = os.getenv("KOKORO_BASE_URL", "").strip().rstrip("/") # Session end_reasons considered "abnormal" for drop-rate telemetry. A session # that ended with one of these reasons is counted as a dropped session; anything # that ended cleanly (e.g. "silence", "completed", "user_ended") is not. ABNORMAL_END_REASONS = { "error", "timeout", "crash", "server_error", "connection_lost", "lost", "failed", "disconnect", "unknown", } # ============================================================================= # MODELS # ============================================================================= class Robot(BaseModel): id: str name: str character_id: Optional[str] = None status: str = "idle" # idle, detecting, connecting, running current_session_id: Optional[str] = None connected_server_id: Optional[str] = None ip_address: Optional[str] = None last_heartbeat: Optional[datetime] = None total_sessions: int = 0 total_runtime_seconds: int = 0 # Added after the `robots` table's additive trigger_count/created_at # migration — response_model=Robot on GET /api/robots(/{id}) was # silently stripping both from every response since the DB row has # them but this model didn't declare them. trigger_count: int = 0 created_at: Optional[datetime] = None class LiveKitServer(BaseModel): id: str name: str url: str webhook_url: str # Internal URL for health/metrics api_key: Optional[str] = None api_secret: Optional[str] = None gpu_name: Optional[str] = None gpu_vram_mb: int = 0 max_sessions: int = 8 status: str = "unknown" # online, offline, maintenance class Session(BaseModel): id: str robot_id: str server_id: str room_name: str started_at: datetime ended_at: Optional[datetime] = None end_reason: Optional[str] = None duration_seconds: Optional[int] = None voice_config: Optional[dict] = None class GpuModel(BaseModel): name: str size_gb: float is_loaded: bool = False vram_used_mb: int = 0 class ServerMetrics(BaseModel): server_id: str gpu_utilization: int = 0 vram_used_mb: int = 0 vram_total_mb: int = 0 active_sessions: int = 0 models: List[GpuModel] = [] class SessionRequest(BaseModel): robot_id: str class WebhookEvent(BaseModel): event: str data: dict class ServiceStatus(BaseModel): name: str enabled: bool running: bool pid: Optional[int] = None uptime_seconds: Optional[float] = None failure_count: int = 0 last_exit_code: Optional[int] = None class Device(BaseModel): id: str name: str device_role: str = "combined" tailscale_ip: Optional[str] = None lan_ip: Optional[str] = None motor_server_url: Optional[str] = None character_id: Optional[str] = None livekit_url: Optional[str] = None video_device: Optional[str] = None audio_device: Optional[str] = None audio_output_device: Optional[str] = None greeting_phrases: List[str] = [] motor_registry: List[dict] = [] motor_sequences: List[dict] = [] greeting_motor_sequence_id: Optional[str] = None device_inventory: Optional[dict] = None production_mode: bool = False supervisor_status: Optional[List[ServiceStatus]] = None supervisor_status_at: Optional[str] = None status: str = "enrolled" # enrolled, online, offline, disabled last_heartbeat: Optional[datetime] = None enrolled_at: Optional[datetime] = None last_seen_ip: Optional[str] = None notes: Optional[str] = None created_at: Optional[datetime] = None class DeviceCreate(BaseModel): name: str device_role: str = "combined" tailscale_ip: Optional[str] = None lan_ip: Optional[str] = None motor_server_url: Optional[str] = None character_id: Optional[str] = None livekit_url: Optional[str] = None video_device: Optional[str] = None audio_device: Optional[str] = None greeting_phrases: Optional[List[str]] = None notes: Optional[str] = None enrollment_token: Optional[str] = None # if omitted, one is auto-generated class DeviceEnrollRequest(BaseModel): enrollment_token: str device_role: Optional[str] = None name: Optional[str] = None tailscale_ip: Optional[str] = None lan_ip: Optional[str] = None motor_server_url: Optional[str] = None livekit_url: Optional[str] = None character_id: Optional[str] = None video_device: Optional[str] = None audio_device: Optional[str] = None class DeviceBootstrapRequest(BaseModel): name: str device_role: str = "combined" character_id: Optional[str] = None lan_ip: Optional[str] = None tailscale_ip: Optional[str] = None livekit_url: Optional[str] = None motor_server_url: Optional[str] = None class DeviceEnrollResponse(BaseModel): device_id: str device_token: str scheduler_url: str class DeviceHeartbeat(BaseModel): status: Optional[str] = None device_role: Optional[str] = None ip: Optional[str] = None uptime_seconds: Optional[int] = None production_mode: Optional[bool] = None device_inventory: Optional[dict] = None livekit_url: Optional[str] = None motor_server_url: Optional[str] = None class PipelineEventPayload(BaseModel): stage: str status: str = "ok" message: Optional[str] = None session_id: Optional[str] = None source: str = "robot" details: Optional[dict] = None class TranscriptTurnPayload(BaseModel): role: str text: str speaker: Optional[str] = None is_final: bool = True sequence: Optional[int] = None source: str = "voice_agent" metadata: Optional[dict] = None class TranscriptLedgerTurn(BaseModel): role: str text: str speaker: Optional[str] = None timestamp: Optional[str] = None source_id: Optional[str] = None sequence: Optional[int] = None metadata: Optional[dict] = None class TranscriptLedgerIngest(BaseModel): model_config = {"protected_namespaces": ()} source: str engine: str engine_conversation_id: str robot_id: Optional[str] = None agent_id: Optional[str] = None character_id: Optional[str] = None model_provider: Optional[str] = None model_name: Optional[str] = None started_at: Optional[str] = None ended_at: Optional[str] = None duration_seconds: Optional[int] = None summary: Optional[str] = None language: Optional[str] = None successful: Optional[str] = None termination_reason: Optional[str] = None labels: List[str] = [] metadata: Optional[dict] = None turns: List[TranscriptLedgerTurn] = [] class TranscriptLabelsPayload(BaseModel): labels: List[str] class TranscriptBackfillPayload(BaseModel): days: int = 3 agent_ids: List[str] = [] robot_agents: Optional[dict[str, str]] = None class VoiceCallRequest(BaseModel): character_preset_id: str voice_stack_id: Optional[str] = None client_name: str = "RoboPark Operator" engine: str = "elevenlabs" agent_id: Optional[str] = None branch_id: Optional[str] = None robot_id: Optional[str] = None join_link_id: Optional[str] = None call_mode: str = "test" class DirectVoiceSessionStart(BaseModel): agent_id: str branch_id: Optional[str] = None trigger_reason: str = "motion" class DirectVoiceSessionEnd(BaseModel): reason: str = "completed" duration_seconds: Optional[float] = None error: Optional[str] = None class DurableCommandPayload(BaseModel): operation: str desired_revision: Optional[int] = None force: bool = False expires_in_seconds: int = 300 payload: Optional[dict] = None class DurableCommandAck(BaseModel): status: str result: Optional[dict] = None class DurableTelemetryBatch(BaseModel): events: List[dict] class RobotVoiceConfigurationPayload(BaseModel): character_id: Optional[str] = None engine_override: Optional[str] = None desired_configuration: dict desired_revision: Optional[int] = None class RobotVoiceRuntimeState(BaseModel): applied_revision: int desired_revision: int effective_engine: str = "elevenlabs" state: str session: Optional[dict] = None health: Optional[dict] = None class VoiceEngineConnectedPayload(BaseModel): conversation_id: str connection_type: str = "websocket" metadata: Optional[dict] = None class SupervisorStatusReport(BaseModel): services: List[ServiceStatus] = [] class SettingsPayload(BaseModel): production_mode: bool class LiveKitConfig(BaseModel): url: Optional[str] = None api_key: Optional[str] = None has_secret: bool = False class LiveKitTokenRequest(BaseModel): room: str identity: str name: Optional[str] = None can_publish: bool = True can_subscribe: bool = True ttl_seconds: int = 3600 class LiveKitTokenResponse(BaseModel): url: str token: str room: str identity: str expires_at: datetime # ============================================================================= # DATABASE # ============================================================================= async def _seed_voice_stacks_and_characters(db): """Reconcile deployable character bindings while preserving local runtime data.""" default_stacks = [ ( "english-default", "English Default", "speaches", "Systran/faster-whisper-small", "en", "ollama", "gemma3:27b", "elevenlabs", "21m00Tcm4TlvDq8ikWAM", "en", 1, 0.7, 20, 0, ), ( "hebrew-default", "Hebrew Default", "speaches", "ivrit-ai/whisper-large-v3-turbo-ct2", "he", "ollama", "gemma3:27b", "elevenlabs", "21m00Tcm4TlvDq8ikWAM", "he", 1, 0.7, 20, 0, ), ] for row in default_stacks: await db.execute( """INSERT OR IGNORE INTO voice_stacks (id, name, stt_provider, stt_model, stt_language, llm_provider, llm_model, tts_provider, tts_voice, tts_language, allow_interruptions, min_endpointing_delay, max_turns, wake_word_enabled) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", row, ) try: with open(PRODUCTION_CONFIG_PATH, "r", encoding="utf8") as handle: deployment = json.load(handle) except (OSError, json.JSONDecodeError) as exc: raise RuntimeError(f"RoboPark production config is unavailable: {PRODUCTION_CONFIG_PATH}") from exc if deployment.get("schema_version") != 1 or not deployment.get("characters"): raise RuntimeError("RoboPark production config has an unsupported schema or no characters") now = datetime.utcnow().isoformat() for preset in deployment["characters"]: if preset.get("default_engine") not in {"elevenlabs", "robovoice"}: raise RuntimeError(f"Invalid voice engine for character {preset.get('id')}") agent_id = preset.get("elevenlabs_agent_id") if preset["default_engine"] == "elevenlabs" and not agent_id: raise RuntimeError(f"Character {preset.get('id')} has no ElevenLabs agent binding") await db.execute( """INSERT INTO character_presets (id, name, description, system_prompt, voice_stack_id, motors, nx, ny, img, elevenlabs_agent_id, elevenlabs_branch_id, default_engine, robovoice_profile_id, system_prompt_revision, created_at, updated_at) VALUES (?, ?, ?, NULL, ?, '[]', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, description=excluded.description, voice_stack_id=excluded.voice_stack_id, nx=excluded.nx, ny=excluded.ny, img=excluded.img, elevenlabs_agent_id=excluded.elevenlabs_agent_id, elevenlabs_branch_id=excluded.elevenlabs_branch_id, default_engine=excluded.default_engine, robovoice_profile_id=excluded.robovoice_profile_id, system_prompt_revision=MAX(COALESCE(character_presets.system_prompt_revision, 1), excluded.system_prompt_revision), updated_at=excluded.updated_at""", ( preset["id"], preset["name"], preset.get("description"), preset.get("voice_stack_id", "english-default"), float(preset.get("nx", 0.5)), float(preset.get("ny", 0.5)), preset.get("img"), agent_id, preset.get("elevenlabs_branch_id"), preset["default_engine"], preset.get("robovoice_profile_id"), int(preset.get("system_prompt_revision", 1)), now, now, ), ) # Canonical aliases are migration instructions, not fuzzy runtime matching. # Prompt and motor data are intentionally not copied from the alias because # those fields may be stale or belong to a different character. for alias, canonical in deployment.get("aliases", {}).items(): for table, column in ( ("devices", "character_id"), ("robots", "character_id"), ("sessions", "character_id"), ("robot_voice_state", "character_id"), ("robot_voice_configs", "character_preset_id"), ): await db.execute( f"UPDATE {table} SET {column} = ? WHERE {column} = ?", (canonical, alias), ) await db.execute("DELETE FROM character_presets WHERE id = ?", (alias,)) await db.execute( """INSERT INTO settings (key, value, updated_at) VALUES ('voice_engine_default', ?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at""", (deployment.get("global", {}).get("voice_engine_default", "elevenlabs"), now), ) await db.execute( """INSERT INTO settings (key, value, updated_at) VALUES ('production_config_revision', ?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at""", (str(deployment.get("revision", 1)), now), ) await db.commit() async def init_db(): """Initialize SQLite database""" os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) async with aiosqlite.connect(DB_PATH) as db: await db.executescript(""" CREATE TABLE IF NOT EXISTS robots ( id TEXT PRIMARY KEY, name TEXT NOT NULL, character_id TEXT, status TEXT DEFAULT 'idle', current_session_id TEXT, connected_server_id TEXT, ip_address TEXT, last_heartbeat TEXT, total_sessions INTEGER DEFAULT 0, total_runtime_seconds INTEGER DEFAULT 0, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS livekit_servers ( id TEXT PRIMARY KEY, name TEXT NOT NULL, url TEXT NOT NULL, webhook_url TEXT NOT NULL, api_key TEXT, api_secret TEXT, gpu_name TEXT, gpu_vram_mb INTEGER DEFAULT 0, max_sessions INTEGER DEFAULT 8, status TEXT DEFAULT 'unknown', created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, robot_id TEXT, server_id TEXT, room_name TEXT, started_at TEXT, ended_at TEXT, end_reason TEXT, duration_seconds INTEGER, FOREIGN KEY (robot_id) REFERENCES robots(id), FOREIGN KEY (server_id) REFERENCES livekit_servers(id) ); CREATE TABLE IF NOT EXISTS metrics_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, server_id TEXT, gpu_utilization INTEGER, vram_used_mb INTEGER, active_sessions INTEGER, recorded_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS devices ( id TEXT PRIMARY KEY, name TEXT NOT NULL, device_role TEXT DEFAULT 'combined', tailscale_ip TEXT, lan_ip TEXT, motor_server_url TEXT, character_id TEXT, livekit_url TEXT, video_device TEXT, audio_device TEXT, audio_output_device TEXT, device_inventory TEXT, token_hash TEXT, enrollment_token_hash TEXT, status TEXT DEFAULT 'enrolled', last_heartbeat TEXT, enrolled_at TEXT, last_seen_ip TEXT, notes TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS previews ( robot_id TEXT PRIMARY KEY, room_name TEXT NOT NULL, server_id TEXT, expires_at TEXT NOT NULL, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS trigger_commands ( robot_id TEXT PRIMARY KEY, source TEXT NOT NULL DEFAULT 'dashboard', requested_at TEXT NOT NULL ); -- One pending remote-control command per (device, service): an -- operator clicking "restart" in the dashboard queues a row here; -- robot_supervisor.py polls GET .../supervisor-commands (same -- request that carries its status report) and executes+clears it. -- A second click before the robot polls just overwrites the row -- rather than queuing duplicates. CREATE TABLE IF NOT EXISTS supervisor_commands ( device_id TEXT NOT NULL, service_name TEXT NOT NULL, action TEXT NOT NULL DEFAULT 'restart', requested_at TEXT NOT NULL, PRIMARY KEY (device_id, service_name) ); -- ── Operator shell (C1 gap-fill) ── -- When the operator clicks "Tail logs" or runs a diagnostic -- command in the dashboard, we insert a row here. The robot -- supervisor's next status-report poll receives the row in -- its `commands` response, runs the request on a background -- thread, and POSTs the result back to /api/devices/{id}/ -- supervisor-output which updates the `result` column. The -- dashboard polls GET /api/robots/{id}/shell/result/{req_id} -- for the response, with an 8s budget. -- -- `kind` discriminates: tail_logs (read log file) vs -- shell_run (execute an allowlisted diagnostic command). CREATE TABLE IF NOT EXISTS shell_requests ( id TEXT PRIMARY KEY, device_id TEXT NOT NULL, service_name TEXT, kind TEXT NOT NULL, params TEXT NOT NULL DEFAULT '{}', requested_at TEXT NOT NULL, completed_at TEXT, result_ok INTEGER, result_payload TEXT ); CREATE INDEX IF NOT EXISTS idx_shell_requests_device ON shell_requests(device_id, requested_at); CREATE TABLE IF NOT EXISTS history_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT DEFAULT CURRENT_TIMESTAMP, category TEXT NOT NULL, entity_type TEXT, entity_id TEXT, action TEXT, actor TEXT, details TEXT ); CREATE TABLE IF NOT EXISTS management_commands ( command_id TEXT PRIMARY KEY, robot_id TEXT NOT NULL, operation TEXT NOT NULL, payload TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL, expires_at TEXT NOT NULL, desired_revision INTEGER, force INTEGER NOT NULL DEFAULT 0, state TEXT NOT NULL DEFAULT 'queued', delivered_at TEXT, acknowledged_at TEXT, started_at TEXT, completed_at TEXT, result TEXT ); CREATE INDEX IF NOT EXISTS idx_management_commands_robot_state ON management_commands(robot_id, state, created_at); CREATE TABLE IF NOT EXISTS robot_events ( event_id TEXT PRIMARY KEY, robot_id TEXT NOT NULL, sequence INTEGER NOT NULL, session_id TEXT, type TEXT NOT NULL, timestamp TEXT NOT NULL, payload TEXT NOT NULL, received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(robot_id, sequence) ); CREATE INDEX IF NOT EXISTS idx_robot_events_robot_sequence ON robot_events(robot_id, sequence); CREATE TABLE IF NOT EXISTS robot_voice_state ( robot_id TEXT PRIMARY KEY, character_id TEXT, engine_override TEXT, desired_revision INTEGER NOT NULL DEFAULT 1, applied_revision INTEGER NOT NULL DEFAULT 0, application_state TEXT NOT NULL DEFAULT 'pending', desired_configuration TEXT NOT NULL DEFAULT '{}', applied_configuration TEXT NOT NULL DEFAULT '{}', last_known_good_configuration TEXT NOT NULL DEFAULT '{}', updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS pipeline_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, robot_id TEXT NOT NULL, session_id TEXT, stage TEXT NOT NULL, status TEXT NOT NULL, message TEXT, source TEXT NOT NULL DEFAULT 'robot', details TEXT, timestamp TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_pipeline_robot_time ON pipeline_events(robot_id, timestamp DESC); CREATE TABLE IF NOT EXISTS session_transcripts ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, robot_id TEXT, role TEXT NOT NULL, speaker TEXT, text TEXT NOT NULL, is_final INTEGER NOT NULL DEFAULT 1, sequence INTEGER, source TEXT NOT NULL DEFAULT 'voice_agent', metadata TEXT, timestamp TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_transcript_session_time ON session_transcripts(session_id, timestamp, id); -- Scheduler-managed voice stacks (STT/LLM/TTS configuration) CREATE TABLE IF NOT EXISTS voice_stacks ( id TEXT PRIMARY KEY, name TEXT NOT NULL, stt_provider TEXT DEFAULT 'speaches', stt_model TEXT DEFAULT 'Systran/faster-whisper-small', stt_language TEXT DEFAULT 'en', llm_provider TEXT DEFAULT 'ollama', llm_model TEXT DEFAULT 'gemma3:27b', tts_provider TEXT DEFAULT 'elevenlabs', tts_voice TEXT DEFAULT '21m00Tcm4TlvDq8ikWAM', tts_language TEXT DEFAULT 'en', allow_interruptions INTEGER DEFAULT 1, min_endpointing_delay REAL DEFAULT 0.7, max_turns INTEGER DEFAULT 20, wake_word_enabled INTEGER DEFAULT 0, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); -- Scheduler-managed character presets (production park characters) CREATE TABLE IF NOT EXISTS character_presets ( id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT, system_prompt TEXT, voice_stack_id TEXT, motors TEXT, -- JSON array nx REAL DEFAULT 0.5, ny REAL DEFAULT 0.5, img TEXT, elevenlabs_agent_id TEXT, elevenlabs_branch_id TEXT, elevenlabs_first_message TEXT, elevenlabs_synced_at TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (voice_stack_id) REFERENCES voice_stacks(id) ); -- Per-robot/device effective voice config override. -- Falls back to the assigned character preset's voice stack. CREATE TABLE IF NOT EXISTS robot_voice_configs ( robot_id TEXT PRIMARY KEY, character_preset_id TEXT, voice_stack_id TEXT, override_config TEXT, -- JSON merge on top of voice_stack updated_at TEXT DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (robot_id) REFERENCES robots(id) ON DELETE CASCADE, FOREIGN KEY (character_preset_id) REFERENCES character_presets(id), FOREIGN KEY (voice_stack_id) REFERENCES voice_stacks(id) ); CREATE INDEX IF NOT EXISTS idx_history_category ON history_log(category); CREATE INDEX IF NOT EXISTS idx_history_entity ON history_log(entity_type, entity_id); """) await db.commit() # ---- Additive telemetry migrations (safe on existing DBs) ---- # trigger_count on robots (camera-trigger counter) and joined_at on # sessions (room-join timestamp for latency). SQLite's # ALTER TABLE ... ADD COLUMN raises if the column already exists, so # each one is guarded and idempotent. for _table, _column, _coldef in ( ("robots", "trigger_count", "INTEGER DEFAULT 0"), ("sessions", "joined_at", "TEXT"), ("devices", "video_device", "TEXT"), ("devices", "audio_device", "TEXT"), ("devices", "device_inventory", "TEXT"), ("devices", "audio_output_device", "TEXT"), ("devices", "device_role", "TEXT DEFAULT 'combined'"), ("devices", "voice_engine_override", "TEXT"), ("devices", "voice_runtime_state", "TEXT"), ("devices", "greeting_phrases", "TEXT DEFAULT '[]'"), ("devices", "motor_registry", "TEXT DEFAULT '[]'"), ("devices", "motor_sequences", "TEXT DEFAULT '[]'"), ("devices", "greeting_motor_sequence_id", "TEXT"), # Per-robot production mode: an additional, more specific gate on # top of the global settings.production_mode switch. When on for # a given device, that robot keeps re-arming its motion-triggered # session loop indefinitely (device_heartbeat below returns the # AND of global+per-device as the effective flag the Pi acts on) # and the dashboard keeps that robot's camera panel live. ("devices", "production_mode", "INTEGER DEFAULT 0"), # robot_supervisor.py's own periodic self-report: which services # it's managing on this robot (preview_agent, vision_app, # motor_server) and each one's up/down state, pid, uptime, and # failure count. JSON blob; see POST .../supervisor-status. ("devices", "supervisor_status", "TEXT"), ("devices", "supervisor_status_at", "TEXT"), # Silence-based session end (see POST /api/sessions/{id}/keepalive # and GET /api/sessions/{id}/status): tracks the last time some # caller signaled this session was still active. Defaults to # started_at at INSERT time; existing rows on an upgraded DB will # have NULL here until touched, so readers should fall back to # started_at (see the /status endpoint below). ("sessions", "last_activity_at", "TEXT"), ("sessions", "character_id", "TEXT"), ("sessions", "voice_stack_id", "TEXT"), ("sessions", "initiated_by", "TEXT"), ("sessions", "event_token_hash", "TEXT"), ("sessions", "source", "TEXT DEFAULT 'robovoice'"), ("sessions", "engine", "TEXT DEFAULT 'elevenlabs'"), ("sessions", "engine_agent_id", "TEXT"), ("sessions", "engine_conversation_id", "TEXT"), ("sessions", "model_provider", "TEXT"), ("sessions", "model_name", "TEXT"), ("sessions", "summary", "TEXT"), ("sessions", "language", "TEXT"), ("sessions", "successful", "TEXT"), ("sessions", "termination_reason", "TEXT"), ("sessions", "labels", "TEXT DEFAULT '[]'"), ("sessions", "metadata_json", "TEXT DEFAULT '{}'"), ("sessions", "ingested_at", "TEXT"), ("session_transcripts", "source_id", "TEXT"), ("session_transcripts", "metadata_json", "TEXT DEFAULT '{}'"), # Speaker tests are executed only by preview_agent, which can # release and restore the active media publisher around ALSA I/O. ("shell_requests", "claimed_at", "TEXT"), ): try: await db.execute(f"ALTER TABLE {_table} ADD COLUMN {_column} {_coldef}") except Exception: pass # column already present await db.commit() await db.execute("""CREATE UNIQUE INDEX IF NOT EXISTS idx_sessions_engine_conversation ON sessions(source, engine_conversation_id) WHERE engine_conversation_id IS NOT NULL""") await db.execute("""CREATE UNIQUE INDEX IF NOT EXISTS idx_transcript_source_id ON session_transcripts(session_id, source_id) WHERE source_id IS NOT NULL""") await db.execute("CREATE INDEX IF NOT EXISTS idx_sessions_ledger_time ON sessions(started_at DESC)") await db.execute("CREATE INDEX IF NOT EXISTS idx_sessions_ledger_filters ON sessions(robot_id, engine, engine_agent_id)") await db.commit() # ---- character_presets park-map coordinate / avatar migrations ---- for _table, _column, _coldef in ( ("character_presets", "nx", "REAL DEFAULT 0.5"), ("character_presets", "ny", "REAL DEFAULT 0.5"), ("character_presets", "img", "TEXT"), ("character_presets", "elevenlabs_agent_id", "TEXT"), ("character_presets", "elevenlabs_branch_id", "TEXT"), ("character_presets", "elevenlabs_first_message", "TEXT"), ("character_presets", "elevenlabs_synced_at", "TEXT"), ("character_presets", "default_engine", "TEXT DEFAULT 'elevenlabs'"), ("character_presets", "robovoice_profile_id", "TEXT"), ("character_presets", "system_prompt_revision", "INTEGER DEFAULT 1"), ): try: await db.execute(f"ALTER TABLE {_table} ADD COLUMN {_column} {_coldef}") except Exception: pass await db.commit() # ---- voice_stacks: options ROBOVOICE actually reads at runtime but # this table didn't expose yet (llm auth + sampling + wake-word tuning) ---- for _table, _column, _coldef in ( ("voice_stacks", "llm_api_key", "TEXT"), ("voice_stacks", "temperature", "REAL DEFAULT 0.7"), ("voice_stacks", "num_ctx", "INTEGER DEFAULT 8192"), ("voice_stacks", "wake_word_model", "TEXT"), ("voice_stacks", "wake_word_threshold", "REAL DEFAULT 0.5"), ("voice_stacks", "wake_word_timeout", "REAL DEFAULT 3.0"), ): try: await db.execute(f"ALTER TABLE {_table} ADD COLUMN {_column} {_coldef}") except Exception: pass await db.commit() # No demo robots are seeded. Actual robots enroll via the device # enrollment API or are added through the scheduler UI. This keeps the # fleet view honest — only real connected hardware appears. await db.commit() # Seed default voice stacks and production park character presets. await _seed_voice_stacks_and_characters(db) # Seed default enrollment token + production_mode flag (only on first run) async with db.execute("SELECT value FROM settings WHERE key = 'default_enrollment_token'") as c: row = await c.fetchone() if not row: token = secrets.token_urlsafe(24) token_hash = hashlib.sha256(token.encode()).hexdigest() await db.execute( "INSERT INTO settings (key, value) VALUES (?, ?)", ("default_enrollment_token", token_hash) ) logger.info("=" * 70) logger.info(f"DEFAULT ENROLLMENT TOKEN (Pi first-boot): {token}") logger.info("Use this on the Pi: --enrollment-token ") logger.info("Rotate or replace it in the UI under Settings.") logger.info("=" * 70) async with db.execute("SELECT value FROM settings WHERE key = 'production_mode'") as c: row = await c.fetchone() if not row: await db.execute( "INSERT INTO settings (key, value) VALUES (?, ?)", ("production_mode", "false") ) await db.execute( "INSERT OR IGNORE INTO settings(key, value) VALUES('voice_engine_default', 'elevenlabs')" ) await db.commit() logger.info("Database initialized") # ============================================================================= # APP SETUP # ============================================================================= @asynccontextmanager async def lifespan(app: FastAPI): await init_db() global ROBOPARK_AGENT_TOKEN if not ROBOPARK_AGENT_TOKEN: try: if os.path.isfile(ROBOPARK_AGENT_TOKEN_FILE): with open(ROBOPARK_AGENT_TOKEN_FILE, "r", encoding="utf-8") as f: ROBOPARK_AGENT_TOKEN = f.read().strip() if not ROBOPARK_AGENT_TOKEN: ROBOPARK_AGENT_TOKEN = secrets.token_urlsafe(32) os.makedirs(os.path.dirname(ROBOPARK_AGENT_TOKEN_FILE) or ".", exist_ok=True) with open(ROBOPARK_AGENT_TOKEN_FILE, "w", encoding="utf-8") as f: f.write(ROBOPARK_AGENT_TOKEN + "\n") try: os.chmod(ROBOPARK_AGENT_TOKEN_FILE, 0o600) except OSError: pass logger.info("Generated RoboPark agent secret in the scheduler data directory") except OSError as exc: logger.error("Could not load/create RoboPark agent secret: %s", exc) if not ROBOPARK_AGENT_TOKEN: logger.warning( "RoboPark agent secret is unavailable; voice-agent keepalive calls will 401." ) asyncio.create_task(metrics_collector()) asyncio.create_task(health_checker()) yield app = FastAPI( title="RoboPark Session Scheduler", description="Manages robot fleet and LiveKit server orchestration", version="1.0.0", lifespan=lifespan ) # CORS: env-driven allowlist. Set ROBOPARK_CORS_ORIGINS to a comma-separated # list of allowed origins; defaults to localhost only. A wildcard origin is # incompatible with credentialed requests, so credentials are only enabled # when an explicit (non-wildcard) allowlist is configured. _cors_env = os.getenv("ROBOPARK_CORS_ORIGINS", "http://localhost:3000,http://127.0.0.1:3000") _cors_origins = [o.strip() for o in _cors_env.split(",") if o.strip()] _allow_credentials = "*" not in _cors_origins app.add_middleware( CORSMiddleware, allow_origins=_cors_origins, allow_credentials=_allow_credentials, allow_methods=["*"], allow_headers=["*"], # X-Operator carries the per-action attribution. CORS preflight # must allow it or the dashboard's rpAction() fetches will fail. expose_headers=["X-Operator"], ) # ── Operator-actor middleware (C3) ── # Reads the X-Operator header on every request and stores a sanitized # version in a contextvar. The header is informational only — the # scheduler is a single-tenant surface; the actor is recorded in the # audit log so "who clicked End-session?" is answerable. The header # value is sanitized at this boundary (trimmed, control chars stripped, # capped at 64 chars) so the value reaching the database is always a # safe printable string. An absent or invalid header leaves the # contextvar unset; _log_history then uses the call-site default # (usually "operator"). class _OperatorActorMiddleware: def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope.get("type") == "http": raw = None # ASGI scope headers is a list of (name, value) tuples where # value is bytes — except when a header has been repeated, in # which case some servers emit (name, [b"a", b"b"]). Build a # case-insensitive map that flattens both shapes. for k, v in (scope.get("headers") or []): if not isinstance(k, (bytes, bytearray)): continue if k.lower() != b"x-operator": continue if isinstance(v, (list, tuple)): v = b", ".join(bytes(x) for x in v if isinstance(x, (bytes, bytearray))) if not isinstance(v, (bytes, bytearray)): continue raw = bytes(v) break actor = None if raw is not None: try: actor = _sanitize_actor_name(raw.decode("utf-8", errors="replace")) except Exception: actor = None token = _current_actor_override.set(actor) try: await self.app(scope, receive, send) finally: _current_actor_override.reset(token) else: await self.app(scope, receive, send) app.add_middleware(_OperatorActorMiddleware) # WebSocket clients for real-time updates ws_clients: List[WebSocket] = [] # ============================================================================= # SETTINGS HELPERS # ============================================================================= def _hash(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() async def get_setting(key: str, default: Optional[str] = None) -> Optional[str]: async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT value FROM settings WHERE key = ?", (key,)) as c: row = await c.fetchone() return row[0] if row else default async def set_setting(key: str, value: str) -> None: async with aiosqlite.connect(DB_PATH) as db: await db.execute( "INSERT INTO settings (key, value, updated_at) VALUES (?, ?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at", (key, value, datetime.utcnow().isoformat()), ) await db.commit() async def get_production_mode() -> bool: v = await get_setting("production_mode", "false") return (v or "false").lower() == "true" # ============================================================================= # WEBSOCKET # ============================================================================= @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() ws_clients.append(websocket) logger.info(f"WebSocket client connected. Total: {len(ws_clients)}") try: while True: await websocket.receive_text() except WebSocketDisconnect: ws_clients.remove(websocket) logger.info(f"WebSocket client disconnected. Total: {len(ws_clients)}") async def broadcast(event_type: str, data: dict): """Broadcast update to all connected WebSocket clients""" if not ws_clients: return message = {"type": event_type, "data": data, "timestamp": datetime.utcnow().isoformat()} for client in ws_clients.copy(): try: await client.send_json(message) except: ws_clients.remove(client) # ============================================================================= # ROBOT ENDPOINTS # ============================================================================= CONNECTING_SESSION_LEASE_SECONDS = max( 15, int(os.getenv("ROBOPARK_CONNECTING_SESSION_LEASE_SECONDS", "45")) ) ACTIVE_SESSION_LEASE_SECONDS = max( CONNECTING_SESSION_LEASE_SECONDS, int(os.getenv("ROBOPARK_ACTIVE_SESSION_LEASE_SECONDS", "180")), ) async def _reap_robot_session_if_stale(db, robot_id: str, current_session_id: Optional[str], robot_status: Optional[str] = None) -> Optional[str]: """Release an invalid or expired session pointer without operator action.""" if not current_session_id: return None async with db.execute( "SELECT id, started_at, last_activity_at, ended_at FROM sessions WHERE id = ?", (current_session_id,), ) as cursor: session = await cursor.fetchone() reason = None if not session: reason = "missing_session" elif session["ended_at"]: reason = "ended_session" else: timestamp = session["last_activity_at"] or session["started_at"] try: activity_age = (datetime.utcnow() - datetime.fromisoformat(timestamp)).total_seconds() except (TypeError, ValueError): activity_age = ACTIVE_SESSION_LEASE_SECONDS + 1 lease = ( CONNECTING_SESSION_LEASE_SECONDS if robot_status == "connecting" else ACTIVE_SESSION_LEASE_SECONDS ) if activity_age >= lease: reason = "stale_connecting" if robot_status == "connecting" else "stale_activity" if not reason: return None now = datetime.utcnow().isoformat() if session and not session["ended_at"]: await db.execute( "UPDATE sessions SET ended_at = ?, end_reason = ? WHERE id = ? AND ended_at IS NULL", (now, reason, current_session_id), ) await db.execute( "UPDATE robots SET status = 'idle', current_session_id = NULL, connected_server_id = NULL " "WHERE id = ? AND current_session_id = ?", (robot_id, current_session_id), ) logger.warning( "Automatically released %s session %s for robot %s", reason, current_session_id, robot_id, ) return reason @app.get("/api/robots", response_model=List[Robot]) async def list_robots(): """List all robots""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM robots ORDER BY name") as cursor: rows = await cursor.fetchall() return [dict(row) for row in rows] @app.get("/api/robots/{robot_id}", response_model=Robot) async def get_robot(robot_id: str): """Get single robot""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM robots WHERE id = ?", (robot_id,)) as cursor: row = await cursor.fetchone() if not row: raise HTTPException(404, "Robot not found") return dict(row) @app.post("/api/robots/{robot_id}/heartbeat") async def robot_heartbeat(robot_id: str, ip_address: Optional[str] = None): """Robot heartbeat - call every 5 seconds""" async with aiosqlite.connect(DB_PATH) as db: await db.execute( "UPDATE robots SET last_heartbeat = ?, ip_address = COALESCE(?, ip_address) WHERE id = ?", (datetime.utcnow().isoformat(), ip_address, robot_id) ) await db.commit() await broadcast("robot_heartbeat", {"robot_id": robot_id}) await _log_history("robot", "robot", robot_id, "heartbeat", "robot", f"ip={ip_address}") return {"status": "ok"} @app.post("/api/robots/{robot_id}/request-session") async def request_session(robot_id: str, authorization: Optional[str] = Header(default=None)): """Robot requests a session after scene detection. Requires a valid enrolled-device Bearer token. The response never contains the LiveKit api_key/api_secret; instead the scheduler mints a short-lived JWT the robot uses to join the room.""" if not await _authorize_fleet(authorization): raise HTTPException(401, "Invalid or missing device token") if not await get_production_mode(): raise HTTPException(403, "production_mode_off") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row # Check robot exists and is idle; auto-create a robot row for an enrolled # device that has not yet appeared in the robots table. async with db.execute("SELECT * FROM robots WHERE id = ?", (robot_id,)) as cursor: robot = await cursor.fetchone() if not robot: async with db.execute("SELECT name FROM devices WHERE id = ?", (robot_id,)) as dc: device_row = await dc.fetchone() if device_row: # OR IGNORE means this name is only ever written on first # creation of the robots row. PATCH /api/devices/{device_id} # now keeps robots.name in sync on every rename going forward, # so a stale robots.name here is only possible if a session # request race wins before that PATCH sync runs — not an # ongoing drift concern. await db.execute( "INSERT OR IGNORE INTO robots (id, name, status) VALUES (?, ?, 'idle')", (robot_id, device_row["name"]), ) await db.commit() robot = { "id": robot_id, "name": device_row["name"], "status": "idle", "current_session_id": None, } else: raise HTTPException(404, "Robot not found") stale_reason = await _reap_robot_session_if_stale( db, robot_id, robot["current_session_id"], robot["status"] ) if stale_reason: await db.commit() async with db.execute("SELECT * FROM robots WHERE id = ?", (robot_id,)) as cursor: robot = await cursor.fetchone() if robot["current_session_id"] or robot["status"] != "idle": raise HTTPException(409, f"Robot is {robot['status']}, not idle") # Find least-loaded online server async with db.execute(""" SELECT s.*, (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) as active FROM livekit_servers s WHERE s.status = 'online' ORDER BY active ASC LIMIT 1 """) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(503, "No available servers") if server["active"] >= server["max_sessions"]: raise HTTPException(503, "All servers at capacity") # Create session. Room name MUST start with "robopark-" — that's the # literal prefix voice_agent.py checks to enable production-mode # (character/voice/motor resolution, unlimited idle timeout). session_id = f"session_{robot_id}_{int(datetime.utcnow().timestamp())}" room_name = f"robopark-{robot_id}-{int(datetime.utcnow().timestamp())}" _now_iso = datetime.utcnow().isoformat() await db.execute(""" INSERT INTO sessions (id, robot_id, server_id, room_name, started_at, last_activity_at) VALUES (?, ?, ?, ?, ?, ?) """, (session_id, robot_id, server["id"], room_name, _now_iso, _now_iso)) # Update robot status + count this as a camera trigger. request-session # is the natural trigger point ("robot requests a session after scene # detection"), so the trigger_count populates here with no client change. # A client that wants to report a camera-open that does NOT reach # request-session can call POST /api/robots/{robot_id}/trigger instead # (it should not call both for the same detection, to avoid double count). await db.execute(""" UPDATE robots SET status = 'connecting', current_session_id = ?, connected_server_id = ?, trigger_count = COALESCE(trigger_count, 0) + 1 WHERE id = ? """, (session_id, server["id"], robot_id)) await db.commit() await broadcast("session_requested", { "robot_id": robot_id, "server_id": server["id"], "session_id": session_id }) await _log_history("session", "robot", robot_id, "requested", "robot", f"session_id={session_id}, server={server['id']}, room={room_name}") # Mint a short-lived LiveKit token so the robot can join the room without # ever receiving the raw api_key/api_secret. try: from livekit.api import AccessToken, VideoGrants except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") lk_key = server["api_key"] or "devkey" lk_secret = server["api_secret"] or "secret" grants = VideoGrants( room=room_name, room_join=True, can_publish=True, can_subscribe=True, can_publish_data=True, ) token = ( AccessToken(lk_key, lk_secret) .with_identity(robot_id) .with_name(robot_id) .with_ttl(timedelta(hours=1)) .with_grants(grants) .to_jwt() ) return { "session_id": session_id, "server_id": server["id"], "server_url": server["url"], "room_name": room_name, "token": token, } @app.post("/api/robots/{robot_id}/end-session") async def end_session(robot_id: str, reason: str = "silence"): """Robot ends session""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM robots WHERE id = ?", (robot_id,)) as cursor: robot = await cursor.fetchone() if not robot or not robot["current_session_id"]: raise HTTPException(400, "No active session") current_session_id = robot["current_session_id"] # Update session async with db.execute("SELECT * FROM sessions WHERE id = ?", (current_session_id,)) as cursor: session = await cursor.fetchone() if session: started = datetime.fromisoformat(session["started_at"]) duration = int((datetime.utcnow() - started).total_seconds()) await db.execute(""" UPDATE sessions SET ended_at = ?, end_reason = ?, duration_seconds = ? WHERE id = ? """, (datetime.utcnow().isoformat(), reason, duration, session["id"])) # Update robot stats await db.execute(""" UPDATE robots SET status = 'idle', current_session_id = NULL, connected_server_id = NULL, total_sessions = total_sessions + 1, total_runtime_seconds = total_runtime_seconds + ? WHERE id = ? """, (duration, robot_id)) await db.commit() await broadcast("session_ended", {"robot_id": robot_id, "reason": reason}) await _log_history("session", "robot", robot_id, "session_ended", "robot", f"reason={reason}, session_id={current_session_id}") return {"status": "ok"} @app.post("/api/robots/{robot_id}/simulate-trigger") async def simulate_trigger(robot_id: str): """Queue a dashboard motion event for the enrolled robot.""" if not await get_production_mode(): return {"queued": False, "reason": "production_mode_off"} async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT id, current_session_id, status FROM robots WHERE id = ?", (robot_id,) ) as c: robot = await c.fetchone() if not robot: async with db.execute("SELECT id, name, character_id FROM devices WHERE id = ?", (robot_id,)) as c: device = await c.fetchone() if not device: raise HTTPException(404, "Robot not found") await db.execute( "INSERT OR IGNORE INTO robots (id, name, character_id, status) VALUES (?, ?, ?, 'idle')", (device["id"], device["name"], device["character_id"]), ) await db.commit() robot = {"id": device["id"], "current_session_id": None, "status": "idle"} if not await _effective_device_production_mode(db, robot_id): return {"queued": False, "reason": "device_production_mode_off"} stale_reason = await _reap_robot_session_if_stale( db, robot_id, robot["current_session_id"], robot["status"] ) if robot["current_session_id"] and not stale_reason: return {"queued": False, "reason": "session_active"} await db.execute( "INSERT INTO trigger_commands (robot_id, source, requested_at) VALUES (?, ?, ?) " "ON CONFLICT(robot_id) DO UPDATE SET source = excluded.source, requested_at = excluded.requested_at", (robot_id, "dashboard", datetime.utcnow().isoformat()), ) await db.commit() await broadcast("robot_triggered", {"robot_id": robot_id, "source": "dashboard"}) await _log_history("robot", "robot", robot_id, "simulate_trigger", "operator") return { "queued": True, "source": "dashboard", "recovered_stale_session": bool(stale_reason), "recovery_reason": stale_reason, } @app.post("/api/robots/{robot_id}/simulate-stop") async def simulate_stop(robot_id: str): """Stop a simulated or production conversation and release the robot.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT current_session_id FROM robots WHERE id = ?", (robot_id,) ) as c: robot = await c.fetchone() if not robot: raise HTTPException(404, "Robot not found") await db.execute("DELETE FROM trigger_commands WHERE robot_id = ?", (robot_id,)) await db.commit() stopped = False if robot["current_session_id"]: try: await end_session(robot_id, "operator_stop") stopped = True except HTTPException: pass await _log_history("robot", "robot", robot_id, "simulate_stop", "operator") return {"stopped": stopped} @app.post("/api/robots/{robot_id}/recover") async def recover_robot(robot_id: str): """Clear this robot's stale session/trigger state and return it to idle. This is intentionally scoped to one robot. It does not restart workers or touch other robots, and is safe to repeat after a failed LiveKit teardown. """ now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT id, current_session_id FROM robots WHERE id = ?", (robot_id,)) as c: robot = await c.fetchone() if not robot: raise HTTPException(404, "Robot not found") await db.execute( "UPDATE sessions SET ended_at = COALESCE(ended_at, ?), end_reason = COALESCE(end_reason, 'operator_recovery') " "WHERE robot_id = ? AND ended_at IS NULL", (now, robot_id), ) await db.execute("DELETE FROM trigger_commands WHERE robot_id = ?", (robot_id,)) await db.execute( "UPDATE robots SET status = 'idle', current_session_id = NULL, connected_server_id = NULL WHERE id = ?", (robot_id,), ) await db.commit() await broadcast("robot_recovered", {"robot_id": robot_id}) await _log_history("robot", "robot", robot_id, "recovered", "operator") return {"status": "ok", "robot_id": robot_id, "previous_session_id": robot["current_session_id"]} @app.post("/api/robots/{robot_id}/trigger") async def robot_trigger(robot_id: str, authorization: Optional[str] = Header(default=None)): """Record a camera trigger for a robot. Called when the robot's camera detects a scene / opens a session, to keep a per-robot trigger counter. Requires a valid enrolled-device Bearer token (same fleet auth as /request-session). NOTE: /request-session already increments trigger_count as the natural trigger point. Use this endpoint only to report a camera-open that does NOT proceed to /request-session; do not call both for the same detection or the trigger will be counted twice.""" if not await _authorize_fleet(authorization): raise HTTPException(401, "Invalid or missing device token") async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute( "UPDATE robots SET trigger_count = COALESCE(trigger_count, 0) + 1 WHERE id = ?", (robot_id,), ) await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Robot not found") async with db.execute("SELECT trigger_count FROM robots WHERE id = ?", (robot_id,)) as c: row = await c.fetchone() await broadcast("robot_triggered", {"robot_id": robot_id}) await _log_history("robot", "robot", robot_id, "trigger", "robot", f"trigger_count={row[0] if row else None}") return {"status": "ok", "trigger_count": row[0] if row else None} # ============================================================================= # SERVER ENDPOINTS # ============================================================================= @app.get("/api/servers") async def list_servers(): """List all LiveKit servers with current load""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute(""" SELECT s.*, (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) as active_sessions FROM livekit_servers s ORDER BY s.name """) as cursor: return [dict(row) for row in await cursor.fetchall()] @app.post("/api/servers") async def add_server(server: LiveKitServer): """Add a new LiveKit server""" async with aiosqlite.connect(DB_PATH) as db: await db.execute(""" INSERT OR REPLACE INTO livekit_servers (id, name, url, webhook_url, api_key, api_secret, gpu_name, gpu_vram_mb, max_sessions, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (server.id, server.name, server.url, server.webhook_url, server.api_key, server.api_secret, server.gpu_name, server.gpu_vram_mb, server.max_sessions, server.status)) await db.commit() await broadcast("server_added", {"server_id": server.id}) await _log_history("server", "server", server.id, "added", "operator", f"name={server.name}, url={server.url}") return {"status": "ok"} @app.delete("/api/servers/{server_id}") async def remove_server(server_id: str): """Remove a LiveKit server""" async with aiosqlite.connect(DB_PATH) as db: await db.execute("DELETE FROM livekit_servers WHERE id = ?", (server_id,)) await db.commit() await broadcast("server_removed", {"server_id": server_id}) await _log_history("server", "server", server_id, "removed", "operator") return {"status": "ok"} @app.get("/api/servers/{server_id}/metrics") async def get_server_metrics(server_id: str) -> ServerMetrics: """Get real-time metrics from a LiveKit server""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM livekit_servers WHERE id = ?", (server_id,)) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(404, "Server not found") # Fetch metrics from server try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.get(f"{server['webhook_url']}/metrics") data = resp.json() # Get models models_resp = await client.get(f"{server['webhook_url']}/models") models_data = models_resp.json() models = [] for m in models_data.get("models", []): models.append(GpuModel( name=m["name"], size_gb=m.get("size", 0) / (1024**3), is_loaded=True, vram_used_mb=int(m.get("size", 0) / (1024**2)) )) return ServerMetrics( server_id=server_id, gpu_utilization=data.get("gpu_utilization", 0), vram_used_mb=data.get("vram_used_mb", 0), vram_total_mb=data.get("vram_total_mb", 0), active_sessions=data.get("active_sessions", 0), models=models ) except Exception as e: logger.error(f"Failed to get metrics from {server_id}: {e}") return ServerMetrics(server_id=server_id) @app.post("/api/servers/{server_id}/models/{model_name}/load") async def load_model(server_id: str, model_name: str): """Load a model on a server""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT webhook_url FROM livekit_servers WHERE id = ?", (server_id,)) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(404, "Server not found") try: async with httpx.AsyncClient(timeout=120.0) as client: resp = await client.post(f"{server['webhook_url']}/models/{model_name}/load") await broadcast("model_loaded", {"server_id": server_id, "model": model_name}) return resp.json() except Exception as e: raise HTTPException(500, f"Failed to load model: {e}") @app.post("/api/servers/{server_id}/models/{model_name}/unload") async def unload_model(server_id: str, model_name: str): """Unload a model from a server""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT webhook_url FROM livekit_servers WHERE id = ?", (server_id,)) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(404, "Server not found") try: async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post(f"{server['webhook_url']}/models/{model_name}/unload") await broadcast("model_unloaded", {"server_id": server_id, "model": model_name}) return resp.json() except Exception as e: raise HTTPException(500, f"Failed to unload model: {e}") # ============================================================================= # SESSION ENDPOINTS # ============================================================================= @app.post("/api/devices/{device_id}/direct-voice/sessions") async def create_direct_voice_session( device_id: str, payload: DirectVoiceSessionStart, authorization: Optional[str] = Header(default=None), ): """Register a robot-local ElevenLabs session without creating another provider call.""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") now = datetime.utcnow() nonce = secrets.token_urlsafe(8).replace("-", "").replace("_", "") session_id = f"direct_{now.strftime('%Y%m%d%H%M%S')}_{nonce}" event_token = secrets.token_urlsafe(32) async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM devices WHERE id = ?", (device_id,)) as cursor: device = await cursor.fetchone() if not device: raise HTTPException(404, "Device not found") character = None if device["character_id"]: async with db.execute("SELECT * FROM character_presets WHERE id = ?", (device["character_id"],)) as cursor: character = await cursor.fetchone() if not character: async with db.execute( "SELECT * FROM character_presets WHERE elevenlabs_agent_id = ? ORDER BY updated_at DESC LIMIT 1", (payload.agent_id.strip(),), ) as cursor: character = await cursor.fetchone() stack = None if character and character["voice_stack_id"]: async with db.execute("SELECT * FROM voice_stacks WHERE id = ?", (character["voice_stack_id"],)) as cursor: stack = await cursor.fetchone() metadata = { "hardware_direct": True, "trigger_reason": payload.trigger_reason[:40], "branch_id": (payload.branch_id or "")[:160] or None, } await db.execute(""" INSERT INTO sessions (id, robot_id, room_name, started_at, last_activity_at, character_id, voice_stack_id, initiated_by, event_token_hash, source, engine, engine_agent_id, model_provider, model_name, labels, metadata_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( session_id, device_id, f"elevenlabs-direct-{nonce}", now.isoformat(), now.isoformat(), character["id"] if character else None, character["voice_stack_id"] if character else None, payload.trigger_reason[:80], _hash(event_token), "robot_hardware", "elevenlabs", payload.agent_id.strip(), stack["llm_provider"] if stack else None, stack["llm_model"] if stack else None, json.dumps(["robot-hardware", "elevenlabs", payload.trigger_reason[:40]]), json.dumps(metadata, separators=(",", ":")), )) await db.commit() await _insert_pipeline_event(device_id, PipelineEventPayload( stage="voice_worker", status="running", message="Direct ElevenLabs hardware session starting", session_id=session_id, source="robot_hardware", details={"agent_id": payload.agent_id}, )) await _log_history("session", "session", session_id, "direct_voice_started", "robot", f"device={device_id}, agent={payload.agent_id}") await broadcast("voice_call_started", {"session_id": session_id, "robot_id": device_id, "engine": "elevenlabs", "source": "robot_hardware"}) return {"session_id": session_id, "event_token": event_token} @app.post("/api/devices/{device_id}/direct-voice/sessions/{session_id}/end") async def end_direct_voice_session( device_id: str, session_id: str, payload: DirectVoiceSessionEnd, authorization: Optional[str] = Header(default=None), ): if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") now = datetime.utcnow() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT started_at FROM sessions WHERE id=? AND robot_id=?", (session_id, device_id)) as cursor: row = await cursor.fetchone() if not row: raise HTTPException(404, "Session not found") duration = payload.duration_seconds if duration is None: try: duration = max(0.0, (now - datetime.fromisoformat(row["started_at"])).total_seconds()) except Exception: duration = 0.0 reason = (payload.reason or ("error" if payload.error else "completed"))[:80] await db.execute("""UPDATE sessions SET ended_at=?, end_reason=?, duration_seconds=?, termination_reason=?, successful=?, last_activity_at=? WHERE id=?""", ( now.isoformat(), reason, int(duration), reason, "failure" if payload.error else "success", now.isoformat(), session_id, )) await db.commit() await _insert_pipeline_event(device_id, PipelineEventPayload( stage="session_ended", status="failed" if payload.error else "ok", message=(payload.error or "Direct ElevenLabs hardware session disposed")[:500], session_id=session_id, source="robot_hardware", details={"duration_seconds": duration}, )) await _log_history("session", "session", session_id, "direct_voice_ended", "robot", f"reason={reason}, duration={duration:.1f}") await broadcast("session_ended", {"session_id": session_id, "robot_id": device_id, "reason": reason}) return {"ok": True, "session_id": session_id, "duration_seconds": duration} class WebClientSessionRequest(BaseModel): """Request body for web client session""" client_name: Optional[str] = "Web User" class WebClientSessionResponse(BaseModel): """Response for web client session""" session_id: str server_id: str server_url: str room_name: str api_key: str api_secret: str async def _select_available_server(db): async with db.execute(""" SELECT s.*, (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) AS active FROM livekit_servers s WHERE s.status = 'online' ORDER BY active ASC LIMIT 1 """) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(503, "No available servers") if server["active"] >= server["max_sessions"]: raise HTTPException(503, "All servers at capacity") return server @app.post("/api/sessions/voice-call") async def create_voice_call(payload: VoiceCallRequest): """Create a browser voice call using a frozen character and voice stack. Unlike the legacy web-client endpoint this never exposes LiveKit API credentials. The event token only authorizes transcript writes for this session and is stored as a hash. """ async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT * FROM character_presets WHERE id = ?", (payload.character_preset_id,) ) as cursor: character = await cursor.fetchone() if not character: raise HTTPException(404, "Character preset not found") stack_id = payload.voice_stack_id or character["voice_stack_id"] if not stack_id: raise HTTPException(422, "Character has no voice stack") async with db.execute("SELECT * FROM voice_stacks WHERE id = ?", (stack_id,)) as cursor: stack = await cursor.fetchone() if not stack: raise HTTPException(404, "Voice stack not found") engine = (payload.engine or "elevenlabs").strip().lower() if engine not in {"elevenlabs", "robovoice"}: raise HTTPException(422, "engine must be elevenlabs or robovoice") bound_agent_id = (character["elevenlabs_agent_id"] or "").strip() requested_agent_id = (payload.agent_id or "").strip() if engine == "elevenlabs" and not bound_agent_id: raise HTTPException(422, "Selected character has no bound ElevenLabs agent") if engine == "elevenlabs" and requested_agent_id and requested_agent_id != bound_agent_id: raise HTTPException(409, "Requested ElevenLabs agent does not match the selected character binding") agent_id = bound_agent_id if engine == "elevenlabs" else requested_agent_id branch_id = (payload.branch_id or character["elevenlabs_branch_id"] or "").strip() join_link_id = (payload.join_link_id or "").strip()[:80] call_mode = (payload.call_mode or "test").strip().lower() if call_mode not in {"test", "production"}: raise HTTPException(422, "call_mode must be test or production") server = await _select_available_server(db) if engine == "robovoice" else None now = datetime.utcnow() nonce = secrets.token_urlsafe(8).replace("-", "").replace("_", "") session_id = f"voice_{now.strftime('%Y%m%d%H%M%S')}_{nonce}" room_name = f"robopark-voice-{nonce}" if engine == "robovoice" else f"elevenlabs-{nonce}" event_token = secrets.token_urlsafe(32) requested_robot_id = (payload.robot_id or "").strip()[:120] robot_id = "operator-test" robot_video_ref = None if call_mode == "production": if not requested_robot_id: raise HTTPException(422, "Production calls require a registered robot") async with db.execute( "SELECT id, name FROM devices WHERE id = ? OR lower(name) = lower(?) ORDER BY last_heartbeat DESC LIMIT 1", (requested_robot_id, requested_robot_id), ) as cursor: robot_device = await cursor.fetchone() if not robot_device: raise HTTPException(404, f"Production robot '{requested_robot_id}' is not registered") robot_id = robot_device["id"] robot_video_ref = robot_device["name"] or robot_device["id"] metadata = { "control_center": True, "character_id": character["id"], "voice_stack_id": stack_id, "elevenlabs_branch_id": branch_id or None, "join_link_id": join_link_id or None, "call_mode": call_mode, } await db.execute(""" INSERT INTO sessions (id, robot_id, server_id, room_name, started_at, last_activity_at, character_id, voice_stack_id, initiated_by, event_token_hash, source, engine, engine_agent_id, model_provider, model_name, labels, metadata_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( session_id, robot_id, server["id"] if server else None, room_name, now.isoformat(), now.isoformat(), character["id"], stack_id, (payload.client_name or "RoboPark Operator")[:80], _hash(event_token), "control_center", engine, agent_id or None, stack["llm_provider"], stack["llm_model"], json.dumps(["control-center", engine, call_mode] + (["managed-link"] if join_link_id else [])), json.dumps(metadata, separators=(",", ":")), )) await db.commit() signed_url = None if engine == "elevenlabs": key = await _get_elevenlabs_api_key() if not key: await end_session(session_id, "configuration_error") raise HTTPException(503, "ElevenLabs API key is not configured") params = {"agent_id": agent_id} if branch_id: params["branch_id"] = branch_id async with httpx.AsyncClient(timeout=20.0, headers={"xi-api-key": key, "Accept": "application/json"}) as client: response = await client.get( f"{ELEVENLABS_BASE_URL}/v1/convai/conversation/get-signed-url", params=params, ) if response.status_code >= 400: await end_session(session_id, "provider_auth_error") raise HTTPException(502, f"ElevenLabs signed URL failed ({response.status_code}): {response.text[:300]}") signed_url = response.json().get("signed_url") if not signed_url: await end_session(session_id, "provider_auth_error") raise HTTPException(502, "ElevenLabs returned no signed URL") await _log_history( "session", "session", session_id, "voice_call_started", (payload.client_name or "operator")[:64], f"engine=elevenlabs, character={character['id']}, agent={agent_id}, robot={robot_id}", ) await broadcast("voice_call_started", { "session_id": session_id, "character_id": character["id"], "engine": engine, "agent_id": agent_id, "robot_id": robot_id, }) return { "session_id": session_id, "engine": engine, "signed_url": signed_url, "connection_type": "websocket", "event_token": event_token, "agent_id": agent_id, "branch_id": branch_id or None, "join_link_id": join_link_id or None, "call_mode": call_mode, "robot_id": robot_id, "robot_video_ref": robot_video_ref, "character": {"id": character["id"], "name": character["name"], "img": character["img"]}, "voice_stack_id": stack_id, } try: from livekit.api import AccessToken, VideoGrants except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") identity = f"operator:{nonce}" token = ( AccessToken(server["api_key"] or "devkey", server["api_secret"] or "secret") .with_identity(identity) .with_name((payload.client_name or "RoboPark Operator")[:80]) .with_ttl(timedelta(hours=2)) .with_grants(VideoGrants( room=room_name, room_join=True, can_publish=True, can_subscribe=True, can_publish_data=True, )) .to_jwt() ) await _log_history( "session", "session", session_id, "voice_call_started", (payload.client_name or "operator")[:64], f"character={character['id']}, voice_stack={stack_id}", ) await broadcast("voice_call_started", {"session_id": session_id, "character_id": character["id"]}) return { "session_id": session_id, "engine": engine, "server_id": server["id"], "server_url": server["url"], "room_name": room_name, "identity": identity, "token": token, "event_token": event_token, "call_mode": call_mode, "robot_id": robot_id, "robot_video_ref": robot_video_ref, "character": {"id": character["id"], "name": character["name"], "img": character["img"]}, "voice_stack_id": stack_id, } @app.post("/api/sessions/{session_id}/engine-connected") async def voice_engine_connected( session_id: str, payload: VoiceEngineConnectedPayload, session_token: Optional[str] = Header(default=None, alias="X-RoboPark-Session-Token"), ): if not await _session_event_authorized(session_id, session_token, None, None): raise HTTPException(401, "Invalid or missing session event token") conversation_id = (payload.conversation_id or "").strip()[:240] if not conversation_id: raise HTTPException(422, "conversation_id is required") now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT metadata_json, joined_at FROM sessions WHERE id = ?", (session_id,)) as cursor: row = await cursor.fetchone() if not row: raise HTTPException(404, "Session not found") try: metadata = json.loads(row["metadata_json"] or "{}") except Exception: metadata = {} metadata.update(payload.metadata or {}) metadata["connection_type"] = payload.connection_type[:40] await db.execute(""" UPDATE sessions SET engine_conversation_id = ?, joined_at = COALESCE(joined_at, ?), last_activity_at = ?, metadata_json = ? WHERE id = ? """, (conversation_id, now, now, json.dumps(metadata, separators=(",", ":"))[:12000], session_id)) await db.commit() await broadcast("voice_engine_connected", { "session_id": session_id, "conversation_id": conversation_id, "connection_type": payload.connection_type, }) return {"status": "ok", "conversation_id": conversation_id, "joined_at": now} @app.post("/api/sessions/web-client", response_model=WebClientSessionResponse) async def create_web_client_session(req: Optional[WebClientSessionRequest] = None): """Create a session for a web client (not tied to a robot)""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row # Find least-loaded online server async with db.execute(""" SELECT s.*, (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) as active FROM livekit_servers s WHERE s.status = 'online' ORDER BY active ASC LIMIT 1 """) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(503, "No available servers") if server["active"] >= server["max_sessions"]: raise HTTPException(503, "All servers at capacity") # Create session session_id = f"web_{int(datetime.utcnow().timestamp())}" room_name = f"robopark_web_{int(datetime.utcnow().timestamp())}" _now_iso = datetime.utcnow().isoformat() await db.execute(""" INSERT INTO sessions (id, robot_id, server_id, room_name, started_at, last_activity_at) VALUES (?, ?, ?, ?, ?, ?) """, (session_id, "web-client", server["id"], room_name, _now_iso, _now_iso)) await db.commit() await broadcast("web_session_started", { "session_id": session_id, "server_id": server["id"], }) return WebClientSessionResponse( session_id=session_id, server_id=server["id"], server_url=server["url"], room_name=room_name, api_key=server["api_key"] or "devkey", api_secret=server["api_secret"] or "secret", ) @app.post("/api/sessions/{session_id}/end") async def end_web_session(session_id: str, reason: str = "disconnect"): """End a web client session""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)) as cursor: session = await cursor.fetchone() if not session: return {"status": "not_found"} if session["ended_at"]: return {"status": "already_ended"} started = datetime.fromisoformat(session["started_at"]) duration = int((datetime.utcnow() - started).total_seconds()) await db.execute(""" UPDATE sessions SET ended_at = ?, end_reason = ?, duration_seconds = ? WHERE id = ? """, (datetime.utcnow().isoformat(), reason, duration, session_id)) await db.commit() await broadcast("session_ended", {"session_id": session_id, "reason": reason}) await _log_history("session", "session", session_id, "ended", "operator/web", f"reason={reason}, duration={duration}") return {"status": "ok"} def _session_latency_ms(started_at: Optional[str], joined_at: Optional[str]) -> Optional[int]: """Milliseconds between session request (started_at) and room join (joined_at).""" if not started_at or not joined_at: return None try: s = datetime.fromisoformat(started_at) j = datetime.fromisoformat(joined_at) return int((j - s).total_seconds() * 1000) except Exception: return None @app.post("/api/sessions/{session_id}/joined") async def session_joined(session_id: str, authorization: Optional[str] = Header(default=None), session_token: Optional[str] = Header(default=None, alias="X-RoboPark-Session-Token")): """Mark the moment a robot successfully joined the LiveKit room for a session. Used to compute session latency = joined_at - started_at (the session row's started_at is set when the session is requested). Requires a valid enrolled-device Bearer token. Idempotent: the first join wins; later calls leave the original join time in place.""" if not await _session_event_authorized(session_id, session_token, authorization, None): raise HTTPException(401, "Invalid or missing device token") now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)) as c: session = await c.fetchone() if not session: raise HTTPException(404, "Session not found") if session["joined_at"]: if not session["ended_at"]: await db.execute( "UPDATE sessions SET last_activity_at = ? WHERE id = ?", (now, session_id), ) await db.execute( "UPDATE robots SET status = 'running' WHERE id = ? AND current_session_id = ?", (session["robot_id"], session_id), ) await db.commit() return { "status": "already_joined", "joined_at": session["joined_at"], "latency_ms": _session_latency_ms(session["started_at"], session["joined_at"]), } await db.execute( "UPDATE sessions SET joined_at = ?, last_activity_at = ? WHERE id = ?", (now, now, session_id), ) await db.execute( "UPDATE robots SET status = 'running' WHERE id = ? AND current_session_id = ?", (session["robot_id"], session_id), ) await db.commit() latency_ms = _session_latency_ms(session["started_at"], now) await broadcast("session_joined", {"session_id": session_id, "latency_ms": latency_ms}) await _log_history("session", "session", session_id, "joined", "robot", f"latency_ms={latency_ms}") return {"status": "ok", "joined_at": now, "latency_ms": latency_ms} @app.post("/api/sessions/{session_id}/keepalive") async def session_keepalive(session_id: str, authorization: Optional[str] = Header(default=None), agent_token: Optional[str] = Header(default=None, alias="X-RoboPark-Agent-Token"), session_token: Optional[str] = Header(default=None, alias="X-RoboPark-Session-Token")): """Silence-based session extension signal. INTEGRATION POINT (out of scope here — lives in a different repo): the VOICE AGENT is the one process that actually knows, via VAD, whether the user/agent are still talking. While a motion-triggered session's conversation is active, the voice agent should call this endpoint periodically (e.g. every few seconds) to bump last_activity_at. The robot side (preview_agent.py) has no independent way to know if a conversation is ongoing — it only knows this via what it reads back from here through GET /api/sessions/{session_id}/status. If nothing ever calls this endpoint, last_activity_at simply stays at started_at and the session will be torn down by preview_agent.py's silence timeout shortly after it starts — that's an intentionally conservative default, not a bug. preview_agent.py additionally enforces vision_session_seconds as a hard ceiling regardless of activity, so a misbehaving/stuck caller can never hold the publisher forever even if it calls this endpoint continuously. """ device_authorized = await _authorize_fleet(authorization) agent_authorized = bool( ROBOPARK_AGENT_TOKEN and agent_token and hmac.compare_digest(agent_token.strip(), ROBOPARK_AGENT_TOKEN) ) session_authorized = await _session_event_authorized(session_id, session_token, None, None) if not device_authorized and not agent_authorized and not session_authorized: raise HTTPException(401, "Invalid or missing device token") now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute( "UPDATE sessions SET last_activity_at = ? WHERE id = ? AND ended_at IS NULL", (now, session_id), ) await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Session not found or already ended") return {"status": "ok", "last_activity_at": now} @app.post("/api/sessions/{session_id}/pipeline-events") async def voice_pipeline_event(session_id: str, payload: PipelineEventPayload, authorization: Optional[str] = Header(default=None), agent_token: Optional[str] = Header(default=None, alias="X-RoboPark-Agent-Token"), session_token: Optional[str] = Header(default=None, alias="X-RoboPark-Session-Token")): """ROBOVOICE callback for worker, STT, LLM and TTS stage timing.""" device_authorized = await _authorize_fleet(authorization) agent_authorized = bool( ROBOPARK_AGENT_TOKEN and agent_token and hmac.compare_digest(agent_token.strip(), ROBOPARK_AGENT_TOKEN) ) session_authorized = await _session_event_authorized(session_id, session_token, None, None) if not device_authorized and not agent_authorized and not session_authorized: raise HTTPException(401, "Invalid or missing agent token") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT robot_id FROM sessions WHERE id = ?", (session_id,)) as c: session = await c.fetchone() if not session: raise HTTPException(404, "Session not found") payload.session_id = session_id payload.source = "control_center" if session_authorized else "voice_agent" return {"status": "ok", **(await _insert_pipeline_event(session["robot_id"], payload))} @app.get("/api/sessions/{session_id}/status") async def get_session_status(session_id: str): """Lightweight session status for polling — used by preview_agent.py to decide whether a motion-triggered session is still "active" (recent last_activity_at) or has gone silent and should be torn down. See POST /api/sessions/{session_id}/keepalive for who updates last_activity_at.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)) as c: session = await c.fetchone() if not session: raise HTTPException(404, "Session not found") return { "session_id": session["id"], "started_at": session["started_at"], "ended_at": session["ended_at"], # Existing rows created before this column existed (or that predate # any keepalive call) have NULL here — fall back to started_at. "last_activity_at": session["last_activity_at"] or session["started_at"], } # ── Back-compat: legacy web-client session token (ported from the deployed OLD # scheduler so voice_client.html / the old frontend keep working after the # fork reconcile). Raw-PyJWT HS256, publish-capable "Web User" token. ── @app.get("/api/sessions/{session_id}/token") async def get_session_token(session_id: str): """Generate a LiveKit access token for a session""" import jwt async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute(""" SELECT s.*, srv.api_key, srv.api_secret FROM sessions s JOIN livekit_servers srv ON s.server_id = srv.id WHERE s.id = ? """, (session_id,)) as cursor: session = await cursor.fetchone() if not session: raise HTTPException(404, "Session not found") if session["ended_at"]: raise HTTPException(400, "Session already ended") api_key = session["api_key"] or "devkey" api_secret = session["api_secret"] or "secret" room_name = session["room_name"] # Generate token token = jwt.encode( { "name": "Web User", "video": { "room": room_name, "roomJoin": True, "canPublish": True, "canPublishData": True, "canSubscribe": True }, "iss": api_key, "nbf": 0, "exp": int(datetime.utcnow().timestamp()) + 3600, # 1 hour "sub": f"web_user_{int(datetime.utcnow().timestamp())}" }, api_secret, algorithm="HS256" ) return {"token": token, "room_name": room_name} @app.get("/api/sessions") async def list_sessions(active_only: bool = False, limit: int = 50): """List enriched sessions for the operator history and live views.""" limit = max(1, min(limit, 1000)) async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row base = """SELECT s.*, s.id AS session_id, cp.name AS character_name, cp.img AS character_img, CASE WHEN s.joined_at IS NOT NULL THEN CAST((julianday(s.joined_at) - julianday(s.started_at)) * 86400000 AS INTEGER) END AS avg_latency_ms, (SELECT COUNT(*) FROM session_transcripts t WHERE t.session_id = s.id AND t.is_final = 1) AS transcript_count, (SELECT COUNT(*) FROM pipeline_events p WHERE p.session_id = s.id AND p.status IN ('failed','blocked')) AS error_count, (SELECT COUNT(*) FROM pipeline_events p WHERE p.session_id = s.id) AS pipeline_event_count FROM sessions s LEFT JOIN character_presets cp ON cp.id = COALESCE( s.character_id, (SELECT character_id FROM devices d WHERE d.id = s.robot_id), (SELECT character_id FROM robots r WHERE r.id = s.robot_id) )""" if active_only: query = base + " WHERE s.ended_at IS NULL ORDER BY s.started_at DESC" else: query = base + " ORDER BY s.started_at DESC LIMIT ?" async with db.execute(query, () if active_only else (limit,)) as cursor: return [dict(row) for row in await cursor.fetchall()] @app.get("/api/sessions/health") async def session_health(window_minutes: int = 60): """Aggregate browser and robot-hardware voice health for operator oversight.""" window_minutes = max(5, min(window_minutes, 1440)) cutoff = (datetime.utcnow() - timedelta(minutes=window_minutes)).isoformat() stale_cutoff = (datetime.utcnow() - timedelta(minutes=3)).isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("""SELECT COUNT(*) AS total, SUM(CASE WHEN ended_at IS NULL THEN 1 ELSE 0 END) AS active, SUM(CASE WHEN ended_at IS NULL AND COALESCE(last_activity_at, started_at) < ? THEN 1 ELSE 0 END) AS stale, SUM(CASE WHEN successful = 'failure' OR end_reason IN ('error','failed') THEN 1 ELSE 0 END) AS failed, SUM(CASE WHEN source = 'robot_hardware' THEN 1 ELSE 0 END) AS direct_hardware, SUM(CASE WHEN source != 'robot_hardware' THEN 1 ELSE 0 END) AS browser_managed, SUM(CASE WHEN NOT EXISTS ( SELECT 1 FROM session_transcripts t WHERE t.session_id = sessions.id AND t.is_final = 1 ) THEN 1 ELSE 0 END) AS transcriptless FROM sessions WHERE started_at >= ?""", (stale_cutoff, cutoff)) as cursor: row = dict(await cursor.fetchone()) async with db.execute("""SELECT id AS session_id, robot_id, source, engine_agent_id AS agent_id, successful, end_reason, started_at, last_activity_at FROM sessions WHERE started_at >= ? AND (successful = 'failure' OR end_reason IN ('error','failed')) ORDER BY started_at DESC LIMIT 10""", (cutoff,)) as cursor: failures = [dict(item) for item in await cursor.fetchall()] return { "window_minutes": window_minutes, "total": int(row.get("total") or 0), "active": int(row.get("active") or 0), "stale": int(row.get("stale") or 0), "failed": int(row.get("failed") or 0), "direct_hardware": int(row.get("direct_hardware") or 0), "browser_managed": int(row.get("browser_managed") or 0), "transcriptless": int(row.get("transcriptless") or 0), "recent_failures": failures, } def _ledger_labels(value: Any) -> list[str]: if isinstance(value, str): try: value = json.loads(value) except Exception: value = [part.strip() for part in value.split(",")] if not isinstance(value, list): return [] return list(dict.fromkeys( str(item).strip().lower()[:48] for item in value if str(item).strip() ))[:20] def _ledger_json(value: Any, fallback: Any) -> Any: if isinstance(value, (dict, list)): return value try: return json.loads(value or "") except Exception: return fallback @app.get("/api/transcript-ledger") async def transcript_ledger( page: int = 1, page_size: int = 30, robot_id: Optional[str] = None, engine: Optional[str] = None, agent_id: Optional[str] = None, character_id: Optional[str] = None, voice_stack_id: Optional[str] = None, model: Optional[str] = None, language: Optional[str] = None, label: Optional[str] = None, search: Optional[str] = None, date_from: Optional[str] = None, date_to: Optional[str] = None, has_transcript: Optional[bool] = None, ): """Search the durable, cross-engine transcript ledger with server pagination.""" page = max(1, page) page_size = max(1, min(page_size, 100)) clauses: list[str] = [] params: list[Any] = [] for column, value in ( ("s.robot_id", robot_id), ("s.engine", engine), ("s.engine_agent_id", agent_id), ("s.language", language), ): if value: clauses.append(f"{column} = ?") params.append(value.strip()) if model: clauses.append("(s.model_name LIKE ? OR vs.llm_model LIKE ?)") params.extend([f"%{model.strip()}%", f"%{model.strip()}%"]) if character_id: clauses.append("COALESCE(s.character_id, (SELECT character_id FROM devices d WHERE d.id = s.robot_id), (SELECT character_id FROM robots r WHERE r.id = s.robot_id)) = ?") params.append(character_id.strip()) if voice_stack_id: clauses.append("COALESCE(s.voice_stack_id, cp.voice_stack_id) = ?") params.append(voice_stack_id.strip()) if label: clauses.append("s.labels LIKE ?") params.append(f'%"{label.strip().lower()}"%') if search: needle = f"%{search.strip()}%" clauses.append("(s.summary LIKE ? OR s.id LIKE ? OR s.robot_id LIKE ? OR EXISTS (SELECT 1 FROM session_transcripts tx WHERE tx.session_id = s.id AND tx.text LIKE ?))") params.extend([needle, needle, needle, needle]) if date_from: clauses.append("s.started_at >= ?") params.append(date_from) if date_to: clauses.append("s.started_at <= ?") params.append(date_to) if has_transcript is not None: clauses.append(("EXISTS" if has_transcript else "NOT EXISTS") + " (SELECT 1 FROM session_transcripts ht WHERE ht.session_id = s.id AND ht.is_final = 1)") where = " WHERE " + " AND ".join(clauses) if clauses else "" joins = """ FROM sessions s LEFT JOIN character_presets cp ON cp.id = COALESCE(s.character_id, (SELECT character_id FROM devices d WHERE d.id = s.robot_id), (SELECT character_id FROM robots r WHERE r.id = s.robot_id)) LEFT JOIN voice_stacks vs ON vs.id = COALESCE(s.voice_stack_id, cp.voice_stack_id)""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT COUNT(*)" + joins + where, params) as cursor: total = int((await cursor.fetchone())[0]) query = """SELECT s.*, s.id AS session_id, cp.name AS character_name, cp.img AS character_img, vs.name AS voice_stack_name, COALESCE(s.model_name, vs.llm_model) AS effective_model, (SELECT COUNT(*) FROM session_transcripts t WHERE t.session_id = s.id AND t.is_final = 1) AS transcript_count, (SELECT text FROM session_transcripts t WHERE t.session_id = s.id AND t.is_final = 1 ORDER BY timestamp, id LIMIT 1) AS first_turn""" query += joins + where + " ORDER BY s.started_at DESC, s.id DESC LIMIT ? OFFSET ?" async with db.execute(query, [*params, page_size, (page - 1) * page_size]) as cursor: items = [dict(row) for row in await cursor.fetchall()] async with db.execute("""SELECT DISTINCT robot_id FROM sessions WHERE robot_id IS NOT NULL ORDER BY robot_id""") as cursor: robots = [row[0] for row in await cursor.fetchall()] async with db.execute("""SELECT DISTINCT engine FROM sessions WHERE engine IS NOT NULL ORDER BY engine""") as cursor: engines = [row[0] for row in await cursor.fetchall()] async with db.execute("""SELECT DISTINCT engine_agent_id FROM sessions WHERE engine_agent_id IS NOT NULL ORDER BY engine_agent_id""") as cursor: agents = [row[0] for row in await cursor.fetchall()] async with db.execute("SELECT id, name FROM character_presets ORDER BY name") as cursor: characters = [dict(row) for row in await cursor.fetchall()] async with db.execute("SELECT id, name FROM voice_stacks ORDER BY name") as cursor: voice_stacks = [dict(row) for row in await cursor.fetchall()] async with db.execute("SELECT labels FROM sessions WHERE labels IS NOT NULL AND labels != '[]'") as cursor: persisted_labels = [row[0] for row in await cursor.fetchall()] labels: set[str] = set() for stored in persisted_labels: labels.update(_ledger_labels(stored)) for item in items: item.pop("event_token_hash", None) item["labels"] = _ledger_labels(item.get("labels")) item["metadata"] = _ledger_json(item.pop("metadata_json", None), {}) return { "items": items, "total": total, "page": page, "page_size": page_size, "pages": max(1, (total + page_size - 1) // page_size), "filters": { "robots": robots, "engines": engines, "agents": agents, "characters": characters, "voice_stacks": voice_stacks, "labels": sorted(labels), }, } async def _ingest_ledger_session(payload: TranscriptLedgerIngest) -> tuple[str, int]: source = re.sub(r"[^a-z0-9_.-]", "-", payload.source.strip().lower())[:40] or "unknown" engine = re.sub(r"[^a-z0-9_.-]", "-", payload.engine.strip().lower())[:40] or source conversation_id = payload.engine_conversation_id.strip()[:240] if not conversation_id: raise HTTPException(422, "engine_conversation_id is required") session_id = f"{source}_{hashlib.sha256(conversation_id.encode()).hexdigest()[:24]}" now = datetime.utcnow().isoformat() started_at = payload.started_at or now ended_at = payload.ended_at if not ended_at and payload.duration_seconds is not None: try: ended_at = (datetime.fromisoformat(started_at.replace("Z", "+00:00")) + timedelta(seconds=payload.duration_seconds)).isoformat() except Exception: ended_at = None labels = _ledger_labels(payload.labels) async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row await db.execute("""INSERT INTO sessions (id, robot_id, room_name, started_at, ended_at, duration_seconds, character_id, source, engine, engine_agent_id, engine_conversation_id, model_provider, model_name, summary, language, successful, termination_reason, labels, metadata_json, ingested_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO UPDATE SET robot_id=COALESCE(excluded.robot_id, sessions.robot_id), ended_at=COALESCE(excluded.ended_at, sessions.ended_at), duration_seconds=COALESCE(excluded.duration_seconds, sessions.duration_seconds), character_id=COALESCE(excluded.character_id, sessions.character_id), engine_agent_id=COALESCE(excluded.engine_agent_id, sessions.engine_agent_id), model_provider=COALESCE(excluded.model_provider, sessions.model_provider), model_name=COALESCE(excluded.model_name, sessions.model_name), summary=COALESCE(excluded.summary, sessions.summary), language=COALESCE(excluded.language, sessions.language), successful=COALESCE(excluded.successful, sessions.successful), termination_reason=COALESCE(excluded.termination_reason, sessions.termination_reason), labels=excluded.labels, metadata_json=excluded.metadata_json, ingested_at=excluded.ingested_at""", ( session_id, payload.robot_id, conversation_id, started_at, ended_at, payload.duration_seconds, payload.character_id, source, engine, payload.agent_id, conversation_id, payload.model_provider, payload.model_name, payload.summary, payload.language, payload.successful, payload.termination_reason, json.dumps(labels), json.dumps(payload.metadata or {}, separators=(",", ":")), now, )) async with db.execute("SELECT id FROM sessions WHERE source = ? AND engine_conversation_id = ?", (source, conversation_id)) as cursor: session_id = (await cursor.fetchone())[0] inserted = 0 for index, turn in enumerate(payload.turns): role = (turn.role or "").strip().lower() if role == "agent": role = "assistant" if role not in {"user", "assistant", "system", "tool"} or not turn.text.strip(): continue source_id = (turn.source_id or f"{conversation_id}:{index}")[:300] cursor = await db.execute("""INSERT OR IGNORE INTO session_transcripts (session_id, robot_id, role, speaker, text, is_final, sequence, source, metadata, metadata_json, source_id, timestamp) VALUES (?, ?, ?, ?, ?, 1, ?, ?, ?, ?, ?, ?)""", ( session_id, payload.robot_id, role, (turn.speaker or "")[:120] or None, turn.text.strip()[:20000], turn.sequence if turn.sequence is not None else index, source, json.dumps(turn.metadata or {}, separators=(",", ":")), json.dumps(turn.metadata or {}, separators=(",", ":")), source_id, turn.timestamp or started_at, )) inserted += max(0, cursor.rowcount) await db.commit() return session_id, inserted @app.post("/api/transcript-ledger/ingest") async def ingest_transcript_ledger( payload: TranscriptLedgerIngest, authorization: Optional[str] = Header(default=None), agent_token: Optional[str] = Header(default=None, alias="X-RoboPark-Agent-Token"), ): if not await _authorize_fleet(authorization) and not ( ROBOPARK_AGENT_TOKEN and agent_token and hmac.compare_digest(agent_token.strip(), ROBOPARK_AGENT_TOKEN) ): raise HTTPException(401, "Fleet authorization required") session_id, inserted = await _ingest_ledger_session(payload) return {"ok": True, "session_id": session_id, "turns_inserted": inserted} @app.put("/api/transcript-ledger/{session_id}/labels") async def set_transcript_labels( session_id: str, payload: TranscriptLabelsPayload, authorization: Optional[str] = Header(default=None), ): if not await _authorize_fleet(authorization): raise HTTPException(401, "Fleet authorization required") labels = _ledger_labels(payload.labels) async with aiosqlite.connect(DB_PATH) as db: cursor = await db.execute("UPDATE sessions SET labels = ? WHERE id = ?", (json.dumps(labels), session_id)) await db.commit() if cursor.rowcount != 1: raise HTTPException(404, "Session not found") return {"ok": True, "session_id": session_id, "labels": labels} def _elevenlabs_turns(detail: dict, started_at: str) -> list[TranscriptLedgerTurn]: turns: list[TranscriptLedgerTurn] = [] for index, item in enumerate(detail.get("transcript") or []): role = str(item.get("role") or "").lower() text = str(item.get("message") or item.get("text") or "").strip() if not text: continue timestamp = started_at offset = item.get("time_in_call_secs") if isinstance(offset, (int, float)): try: timestamp = (datetime.fromisoformat(started_at.replace("Z", "+00:00")) + timedelta(seconds=float(offset))).isoformat() except Exception: pass turns.append(TranscriptLedgerTurn( role="assistant" if role in {"agent", "assistant"} else role, text=text, speaker=item.get("agent_metadata", {}).get("agent_id") if role == "agent" else None, timestamp=timestamp, sequence=index, source_id=f"{detail.get('conversation_id')}:{index}", metadata={ "time_in_call_secs": offset, "feedback": item.get("feedback"), "tool_calls": item.get("tool_calls") or [], }, )) return turns @app.post("/api/transcript-ledger/backfill") async def backfill_transcript_ledger( payload: TranscriptBackfillPayload, authorization: Optional[str] = Header(default=None), ): """Normalize local history and import full ElevenLabs transcripts for a bounded window.""" if not await _authorize_fleet(authorization): raise HTTPException(401, "Fleet authorization required") days = max(1, min(payload.days, 31)) cutoff = datetime.utcnow() - timedelta(days=days) async with aiosqlite.connect(DB_PATH) as db: local_cursor = await db.execute("""UPDATE sessions SET source=COALESCE(NULLIF(source, ''), 'robovoice'), engine=COALESCE(NULLIF(engine, ''), 'robovoice'), labels=COALESCE(labels, '[]'), metadata_json=COALESCE(metadata_json, '{}'), ingested_at=COALESCE(ingested_at, ?) WHERE started_at >= ?""", (datetime.utcnow().isoformat(), cutoff.isoformat())) await db.execute("""UPDATE session_transcripts SET metadata_json=COALESCE(metadata_json, metadata, '{}') WHERE timestamp >= ?""", (cutoff.isoformat(),)) await db.commit() normalized = max(0, local_cursor.rowcount) agent_ids = list(dict.fromkeys( item.strip() for item in payload.agent_ids if re.fullmatch(r"agent_[A-Za-z0-9]{8,80}", item.strip()) )) key = await _get_elevenlabs_api_key() imported = 0 turns_inserted = 0 errors: list[dict[str, str]] = [] robot_by_agent = {agent: robot for robot, agent in (payload.robot_agents or {}).items()} if agent_ids and not key: errors.append({"source": "elevenlabs", "error": "ElevenLabs API key is not configured on the scheduler"}) if key: async with httpx.AsyncClient(timeout=30.0, headers={"xi-api-key": key}) as client: for agent_id in agent_ids: cursor_value: Optional[str] = None while True: params: dict[str, Any] = { "agent_id": agent_id, "page_size": 100, "summary_mode": "include", "call_start_after_unix": int(cutoff.timestamp()), } if cursor_value: params["cursor"] = cursor_value try: response = await client.get(f"{ELEVENLABS_BASE_URL}/v1/convai/conversations", params=params) response.raise_for_status() page_data = response.json() except Exception as exc: errors.append({"source": agent_id, "error": str(exc)[:300]}) break conversations = page_data.get("conversations") or [] for summary in conversations: conversation_id = str(summary.get("conversation_id") or "") if not conversation_id: continue try: detail_response = await client.get(f"{ELEVENLABS_BASE_URL}/v1/convai/conversations/{conversation_id}") detail_response.raise_for_status() detail = detail_response.json() start_unix = detail.get("metadata", {}).get("start_time_unix_secs") or summary.get("start_time_unix_secs") started_at = datetime.utcfromtimestamp(int(start_unix)).isoformat() if start_unix else datetime.utcnow().isoformat() duration = int(summary.get("call_duration_secs") or detail.get("metadata", {}).get("call_duration_secs") or 0) session_id, new_turns = await _ingest_ledger_session(TranscriptLedgerIngest( source="elevenlabs", engine="elevenlabs", engine_conversation_id=conversation_id, robot_id=robot_by_agent.get(agent_id), agent_id=agent_id, started_at=started_at, duration_seconds=duration, summary=summary.get("transcript_summary") or detail.get("analysis", {}).get("transcript_summary"), language=summary.get("main_language") or detail.get("metadata", {}).get("main_language"), successful=str(summary.get("call_successful") or detail.get("analysis", {}).get("call_successful") or ""), termination_reason=summary.get("termination_reason") or detail.get("metadata", {}).get("termination_reason"), labels=["backfilled", "elevenlabs"], metadata={"agent_id": agent_id, "analysis": detail.get("analysis") or {}, "source_summary": summary}, turns=_elevenlabs_turns({**detail, "conversation_id": conversation_id}, started_at), )) imported += 1 turns_inserted += new_turns except Exception as exc: errors.append({"source": conversation_id, "error": str(exc)[:300]}) cursor_value = page_data.get("next_cursor") if not page_data.get("has_more") or not cursor_value: break await _log_history("transcript", "system", "ledger", "backfill", "operator", f"days={days}, local={normalized}, imported={imported}, turns={turns_inserted}, errors={len(errors)}") return { "ok": not errors, "days": days, "local_sessions_normalized": normalized, "elevenlabs_sessions_imported": imported, "turns_inserted": turns_inserted, "errors": errors[:50], } async def _session_event_authorized(session_id: str, session_token: Optional[str], authorization: Optional[str], agent_token: Optional[str]) -> bool: if await _authorize_fleet(authorization): return True if ROBOPARK_AGENT_TOKEN and agent_token and hmac.compare_digest(agent_token.strip(), ROBOPARK_AGENT_TOKEN): return True if not session_token: return False async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT event_token_hash FROM sessions WHERE id = ?", (session_id,)) as cursor: row = await cursor.fetchone() return bool(row and row[0] and hmac.compare_digest(row[0], _hash(session_token.strip()))) @app.post("/api/sessions/{session_id}/transcript") async def append_session_transcript( session_id: str, payload: TranscriptTurnPayload, authorization: Optional[str] = Header(default=None), agent_token: Optional[str] = Header(default=None, alias="X-RoboPark-Agent-Token"), session_token: Optional[str] = Header(default=None, alias="X-RoboPark-Session-Token"), ): if not await _session_event_authorized(session_id, session_token, authorization, agent_token): raise HTTPException(401, "Invalid or missing session event token") role = (payload.role or "").strip().lower() if role not in {"user", "assistant", "system", "tool"}: raise HTTPException(422, "role must be user, assistant, system, or tool") text = (payload.text or "").strip() if not text: raise HTTPException(422, "text is required") if len(text) > 20000: raise HTTPException(422, "text exceeds 20000 characters") now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT robot_id, ended_at FROM sessions WHERE id = ?", (session_id,)) as cursor: session = await cursor.fetchone() if not session: raise HTTPException(404, "Session not found") cur = await db.execute(""" INSERT INTO session_transcripts (session_id, robot_id, role, speaker, text, is_final, sequence, source, metadata, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( session_id, session["robot_id"], role, (payload.speaker or "")[:120] or None, text, int(payload.is_final), payload.sequence, (payload.source or "unknown")[:40], json.dumps(payload.metadata or {}, separators=(",", ":"))[:8000], now, )) if not session["ended_at"]: await db.execute("UPDATE sessions SET last_activity_at = ? WHERE id = ?", (now, session_id)) await db.commit() await broadcast("transcript_turn", {"session_id": session_id, "role": role, "is_final": payload.is_final}) return {"status": "ok", "id": cur.lastrowid, "timestamp": now} @app.post("/api/sessions/by-room/{room_name}/transcript") async def append_room_transcript( room_name: str, payload: TranscriptTurnPayload, authorization: Optional[str] = Header(default=None), agent_token: Optional[str] = Header(default=None, alias="X-RoboPark-Agent-Token"), ): """Persist a ROBOVOICE turn without coupling the worker to session ID formats.""" async with aiosqlite.connect(DB_PATH) as db: async with db.execute( "SELECT id FROM sessions WHERE room_name = ? ORDER BY started_at DESC LIMIT 1", (room_name,), ) as cursor: row = await cursor.fetchone() if not row: raise HTTPException(404, "No session bound to this room") return await append_session_transcript( row[0], payload, authorization=authorization, agent_token=agent_token, session_token=None ) @app.get("/api/sessions/{session_id}/detail") async def session_detail(session_id: str): async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute(""" SELECT s.*, s.id AS session_id, cp.name AS character_name, cp.img AS character_img, vs.name AS voice_stack_name FROM sessions s LEFT JOIN character_presets cp ON cp.id = COALESCE( s.character_id, (SELECT character_id FROM devices d WHERE d.id = s.robot_id), (SELECT character_id FROM robots r WHERE r.id = s.robot_id) ) LEFT JOIN voice_stacks vs ON vs.id = COALESCE(s.voice_stack_id, cp.voice_stack_id) WHERE s.id = ? """, (session_id,)) as cursor: session = await cursor.fetchone() if not session: raise HTTPException(404, "Session not found") async with db.execute( "SELECT * FROM session_transcripts WHERE session_id = ? ORDER BY COALESCE(sequence, id), timestamp, id", (session_id,), ) as cursor: transcript = [dict(r) for r in await cursor.fetchall()] async with db.execute( "SELECT * FROM pipeline_events WHERE session_id = ? ORDER BY timestamp, id", (session_id,) ) as cursor: pipeline = [dict(r) for r in await cursor.fetchall()] for turn in transcript: try: turn["metadata"] = json.loads(turn.get("metadata") or "{}") except Exception: turn["metadata"] = {} turn["is_final"] = bool(turn.get("is_final")) for event in pipeline: try: event["details"] = json.loads(event.get("details") or "{}") except Exception: event["details"] = {} data = dict(session) data.pop("event_token_hash", None) data["latency_ms"] = _session_latency_ms(data.get("started_at"), data.get("joined_at")) data["transcript"] = transcript data["pipeline_events"] = pipeline data["summary"] = { "turns": len([t for t in transcript if t["is_final"]]), "user_turns": len([t for t in transcript if t["is_final"] and t["role"] == "user"]), "assistant_turns": len([t for t in transcript if t["is_final"] and t["role"] == "assistant"]), "pipeline_errors": len([e for e in pipeline if e["status"] in {"failed", "blocked"}]), } return data @app.get("/api/sessions/stats") async def session_stats(): """Get session statistics""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row today = datetime.utcnow().date().isoformat() stats = {} # Today's sessions async with db.execute( "SELECT COUNT(*) as count FROM sessions WHERE started_at LIKE ?", (f"{today}%",) ) as cursor: row = await cursor.fetchone() stats["today_sessions"] = row["count"] # Active sessions async with db.execute( "SELECT COUNT(*) as count FROM sessions WHERE ended_at IS NULL" ) as cursor: row = await cursor.fetchone() stats["active_sessions"] = row["count"] # Average duration async with db.execute( "SELECT AVG(duration_seconds) as avg FROM sessions WHERE duration_seconds IS NOT NULL" ) as cursor: row = await cursor.fetchone() stats["avg_duration_seconds"] = int(row["avg"] or 0) # Total sessions async with db.execute("SELECT COUNT(*) as count FROM sessions") as cursor: row = await cursor.fetchone() stats["total_sessions"] = row["count"] return stats # ============================================================================= # WEBHOOK RECEIVER # ============================================================================= @app.post("/api/webhooks/livekit") async def livekit_webhook(event: WebhookEvent): """Receive events from LiveKit servers""" logger.info(f"Webhook received: {event.event} - {event.data}") if event.event == "session_started": robot_id = event.data.get("robot_id") if robot_id: async with aiosqlite.connect(DB_PATH) as db: await db.execute( "UPDATE robots SET status = 'running' WHERE id = ?", (robot_id,) ) await db.commit() await broadcast("session_started", event.data) elif event.event == "session_ended": robot_id = event.data.get("robot_id") if robot_id: # Use the end_session logic try: await end_session(robot_id, event.data.get("reason", "unknown")) except: pass return {"status": "ok"} # ============================================================================= # BACKGROUND TASKS # ============================================================================= async def metrics_collector(): """Collect metrics from all servers periodically""" while True: try: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM livekit_servers WHERE status = 'online'") as cursor: servers = await cursor.fetchall() all_metrics = [] for server in servers: try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.get(f"{server['webhook_url']}/metrics") if resp.status_code == 404: # Bare LiveKit servers do not expose the agent metrics endpoint. logger.debug(f"Metrics endpoint unavailable for {server['id']}") continue resp.raise_for_status() data = resp.json() # Store in history await db.execute(""" INSERT INTO metrics_history (server_id, gpu_utilization, vram_used_mb, active_sessions) VALUES (?, ?, ?, ?) """, (server["id"], data.get("gpu_utilization", 0), data.get("vram_used_mb", 0), data.get("active_sessions", 0))) all_metrics.append({ "server_id": server["id"], **data }) except Exception as e: logger.warning(f"Failed to collect metrics from {server['id']}: {e}") await db.commit() if all_metrics: await broadcast("metrics_update", {"servers": all_metrics}) except Exception as e: logger.error(f"Metrics collector error: {e}") await asyncio.sleep(5) # Collect every 5 seconds async def health_checker(): """Check server health periodically""" while True: try: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM livekit_servers") as cursor: servers = await cursor.fetchall() for server in servers: try: async with httpx.AsyncClient(timeout=5.0) as client: resp = await client.get(f"{server['webhook_url']}/health") if resp.status_code == 200: new_status = "online" else: # A bare LiveKit server (no sidecar webhook # service in front of it, e.g. `--dev` mode # with no separate health endpoint) has no # /health route and returns 404 here even # though it's genuinely up — LiveKit's own # root path returns "OK" instead. Fall back # to that before declaring it offline. resp2 = await client.get(server["webhook_url"]) new_status = "online" if resp2.status_code == 200 else "offline" except: new_status = "offline" if new_status != server["status"]: await db.execute( "UPDATE livekit_servers SET status = ? WHERE id = ?", (new_status, server["id"]) ) await broadcast("server_status_changed", { "server_id": server["id"], "status": new_status }) # Mark devices offline if heartbeat is stale (> 30s) cutoff = (datetime.utcnow() - timedelta(seconds=30)).isoformat() async with db.execute( "SELECT id, status FROM devices WHERE status = 'online' AND (last_heartbeat IS NULL OR last_heartbeat < ?)", (cutoff,), ) as c: stale = await c.fetchall() for (did, _st) in stale: await db.execute("UPDATE devices SET status = 'offline' WHERE id = ?", (did,)) await broadcast("device_status_changed", {"device_id": did, "status": "offline"}) # Release stale robot session pointers with the same lease rules # used by the trigger path. Failed joins cannot require manual # operator recovery or block an unattended robot indefinitely. async with db.execute( "SELECT id, status, current_session_id FROM robots WHERE current_session_id IS NOT NULL" ) as c: session_owners = await c.fetchall() for owner in session_owners: reason = await _reap_robot_session_if_stale( db, owner["id"], owner["current_session_id"], owner["status"] ) if reason: await broadcast("session_expired", { "session_id": owner["current_session_id"], "robot_id": owner["id"], "reason": reason, }) # Reap old orphan rows that are no longer owned by a robot. session_cutoff = ( datetime.utcnow() - timedelta(seconds=ACTIVE_SESSION_LEASE_SECONDS) ).isoformat() async with db.execute( "SELECT id, robot_id FROM sessions " "WHERE ended_at IS NULL AND COALESCE(last_activity_at, started_at) < ? " "AND NOT EXISTS (SELECT 1 FROM robots r WHERE r.current_session_id = sessions.id)", (session_cutoff,), ) as c: stale_sessions = await c.fetchall() for session in stale_sessions: await db.execute( "UPDATE sessions SET ended_at = ?, end_reason = 'stale_orphan' " "WHERE id = ? AND ended_at IS NULL", (datetime.utcnow().isoformat(), session["id"]), ) await db.execute( "UPDATE robots SET status = 'idle', current_session_id = NULL, " "connected_server_id = NULL WHERE id = ? AND current_session_id = ?", (session["robot_id"], session["id"]), ) await broadcast("session_expired", { "session_id": session["id"], "robot_id": session["robot_id"], "reason": "stale_orphan", }) await db.commit() except Exception as e: logger.error(f"Health checker error: {e}") await asyncio.sleep(10) # Check every 10 seconds # ============================================================================= # SETTINGS ENDPOINTS # ============================================================================= @app.get("/api/settings") async def get_settings(): """Public settings with secret values reduced to status only.""" pm = await get_production_mode() has_default_token = bool(await get_setting("default_enrollment_token")) elevenlabs_key = await _get_elevenlabs_api_key() return { "production_mode": pm, "default_enrollment_token_set": has_default_token, "agent_token_configured": bool(ROBOPARK_AGENT_TOKEN), "agent_token_source": "environment" if os.getenv("ROBOPARK_AGENT_TOKEN", "").strip() else "scheduler_secret_file", "elevenlabs_configured": bool(elevenlabs_key), "elevenlabs_source": "environment" if ELEVENLABS_API_KEY else ("scheduler_store" if elevenlabs_key else None), } @app.post("/api/settings/provider-secret") async def set_provider_secret(payload: dict): """Persist a provider credential while only exposing configured status.""" provider = str(payload.get("provider") or "").strip().lower() api_key = str(payload.get("api_key") or "").strip() if provider != "elevenlabs": raise HTTPException(400, "unsupported provider") if len(api_key) < 8: raise HTTPException(400, "ElevenLabs API key must be at least 8 characters") await set_setting("elevenlabs_api_key", api_key) await _log_history("settings", "provider_secret", provider, "updated", "operator") return {"ok": True, "provider": provider, "configured": True, "source": "scheduler_store"} @app.put("/api/settings") async def update_settings(payload: SettingsPayload): """Update scheduler settings (production mode, etc.).""" await set_setting("production_mode", "true" if payload.production_mode else "false") if not payload.production_mode: async with aiosqlite.connect(DB_PATH) as db: async with db.execute( "SELECT id FROM robots WHERE current_session_id IS NOT NULL" ) as c: active_robot_ids = [row[0] for row in await c.fetchall()] await db.execute("DELETE FROM trigger_commands") await db.execute("DELETE FROM previews") await db.commit() for robot_id in active_robot_ids: try: await end_session(robot_id, "production_mode_off") except HTTPException: pass await broadcast("settings_changed", {"production_mode": payload.production_mode}) await _log_history("settings", "settings", "production_mode", "updated", "operator", f"value={payload.production_mode}") return {"status": "ok", "production_mode": payload.production_mode} @app.post("/api/settings/enrollment-token/rotate") async def rotate_enrollment_token(): """Generate a fresh default enrollment token (rotates the secret).""" token = secrets.token_urlsafe(24) await set_setting("default_enrollment_token", _hash(token)) logger.info("Default enrollment token rotated") await _log_history("settings", "settings", "default_enrollment_token", "rotated", "operator") return {"enrollment_token": token} @app.post("/api/settings/agent-token/provision") async def provision_agent_token(payload: Optional[dict] = None): """Return the worker secret once so the UI/CLI can provision ROBOVOICE. The value is never included in normal settings responses or logs. Rotation is explicit because existing workers would otherwise lose connectivity. """ global ROBOPARK_AGENT_TOKEN rotate = bool((payload or {}).get("rotate")) if rotate or not ROBOPARK_AGENT_TOKEN: ROBOPARK_AGENT_TOKEN = secrets.token_urlsafe(32) os.makedirs(os.path.dirname(ROBOPARK_AGENT_TOKEN_FILE) or ".", exist_ok=True) temp_path = ROBOPARK_AGENT_TOKEN_FILE + ".tmp" with open(temp_path, "w", encoding="utf-8") as f: f.write(ROBOPARK_AGENT_TOKEN + "\n") os.replace(temp_path, ROBOPARK_AGENT_TOKEN_FILE) try: os.chmod(ROBOPARK_AGENT_TOKEN_FILE, 0o600) except OSError: pass await _log_history("settings", "settings", "agent_token", "rotated", "operator") return { "agent_token": ROBOPARK_AGENT_TOKEN, "warning": "Shown once by this response. Store it only in the ROBOVOICE worker environment.", } @app.get("/api/history") async def get_history(category: Optional[str] = None, limit: int = 200): """Audit / history log for devices, robots, sessions, previews and settings.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row if category: async with db.execute( "SELECT * FROM history_log WHERE category = ? ORDER BY timestamp DESC LIMIT ?", (category, limit), ) as c: rows = await c.fetchall() else: async with db.execute( "SELECT * FROM history_log ORDER BY timestamp DESC LIMIT ?", (limit,), ) as c: rows = await c.fetchall() return [dict(r) for r in rows] # ============================================================================= # LIVEKIT CONFIG + TOKEN ISSUANCE # ============================================================================= # # The scheduler is the single source of truth for LiveKit credentials. # The Pi UI (or any client) calls /api/livekit/token to receive a short-lived # JWT for a given room + identity, signed with the LiveKit API key/secret # that the scheduler was started with. # # Required env vars (set in docker-compose or the host): # LIVEKIT_URL e.g. ws://localhost:7880 # LIVEKIT_API_KEY e.g. devkey # LIVEKIT_API_SECRET e.g. # # These can also be set via the API (PUT /api/livekit/config) and are stored # in the settings table; env vars take precedence at startup. def _lk_settings() -> tuple[Optional[str], Optional[str], Optional[str]]: """Return (url, api_key, api_secret). Env overrides stored settings.""" url = os.getenv("LIVEKIT_URL") or None key = os.getenv("LIVEKIT_API_KEY") or None sec = os.getenv("LIVEKIT_API_SECRET") or None return url, key, sec async def _lk_config_async() -> tuple[str, str, str]: """Async lookup that falls back to settings table for any missing env value.""" url, key, sec = _lk_settings() if not url: url = await get_setting("livekit_url") if not key: key = await get_setting("livekit_api_key") if not sec: sec = await get_setting("livekit_api_secret") if not (url and key and sec): raise HTTPException( 503, "LiveKit is not configured. Set LIVEKIT_URL / LIVEKIT_API_KEY / " "LIVEKIT_API_SECRET env vars, or PUT /api/livekit/config.", ) return url, key, sec @app.get("/api/livekit/config", response_model=LiveKitConfig) async def get_livekit_config(): """Show the current LiveKit configuration. Secret is never returned.""" url, key, sec = _lk_settings() has_secret = bool(sec) if not url: url = await get_setting("livekit_url") if not key: key = await get_setting("livekit_api_key") if not has_secret: has_secret = bool(await get_setting("livekit_api_secret")) return LiveKitConfig(url=url, api_key=key, has_secret=has_secret) @app.put("/api/livekit/config") async def update_livekit_config(payload: LiveKitConfig): """Persist LiveKit config. If api_key is omitted, the existing key is kept. If api_secret is None, it is left as-is; if it is the empty string it is cleared.""" if payload.url is not None: await set_setting("livekit_url", payload.url) if payload.api_key is not None: await set_setting("livekit_api_key", payload.api_key) # Note: there's no separate endpoint to set the secret via JSON to avoid # accidental overwrites; use POST /api/livekit/config/secret for that. await broadcast("livekit_config_changed", {}) await _log_history("livekit", "settings", "livekit_config", "updated", "operator", f"url_set={payload.url is not None}, key_set={payload.api_key is not None}") return {"status": "ok"} @app.post("/api/livekit/config/secret") async def set_livekit_secret(payload: dict): secret = payload.get("api_secret") if not isinstance(secret, str) or len(secret) < 8: raise HTTPException(400, "api_secret must be a string of length >= 8") await set_setting("livekit_api_secret", secret) await broadcast("livekit_config_changed", {}) await _log_history("livekit", "settings", "livekit_secret", "updated", "operator") return {"status": "ok"} @app.post("/api/livekit/token", response_model=LiveKitTokenResponse) async def issue_livekit_token(payload: LiveKitTokenRequest, authorization: Optional[str] = Header(default=None)): """Sign a short-lived LiveKit access token for a given room + identity. Pi clients (or the Pi UI) use this to join a room in the browser. Requires a valid enrolled-device Bearer token and production_mode to be ON.""" if not await _authorize_fleet(authorization): raise HTTPException(401, "Invalid or missing device token") if not await get_production_mode(): raise HTTPException(403, "Production mode is disabled") if not payload.room or not payload.identity: raise HTTPException(400, "room and identity are required") if not re.match(r"^[A-Za-z0-9_\-=]{1,128}$", payload.room): raise HTTPException(400, "invalid room name") if not re.match(r"^[A-Za-z0-9_\-:.@]{1,128}$", payload.identity): raise HTTPException(400, "invalid identity") url, key, sec = await _lk_config_async() ttl = max(60, min(int(payload.ttl_seconds), 6 * 3600)) try: from livekit.api import AccessToken, VideoGrants except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") grants = VideoGrants( room=payload.room, room_join=True, can_publish=payload.can_publish, can_subscribe=payload.can_subscribe, can_publish_data=True, ) at = AccessToken(key, sec) \ .with_identity(payload.identity) \ .with_name(payload.name or payload.identity) \ .with_ttl(timedelta(seconds=ttl)) \ .with_grants(grants) token = at.to_jwt() expires = datetime.utcnow() + timedelta(seconds=ttl) return LiveKitTokenResponse( url=url, token=token, room=payload.room, identity=payload.identity, expires_at=expires, ) @app.get("/api/robots/{robot_id}/stream") async def robot_stream(robot_id: str, authorization: Optional[str] = Header(default=None)): """Mint a SUBSCRIBE-ONLY LiveKit token so an operator dashboard can watch a robot's live camera + audio for its CURRENT session. The Pi already publishes a 'cam' (video) and 'mic' (audio) track into its session room (pi-client livekit_bridge _publish_cam/_publish_mic), so a viewer just needs a subscribe-only token for that room. Returns {"active": false} when the robot has no live session (nothing to view yet). Token is signed with the robot's ASSIGNED server key (the same key that signed the publisher) so it validates against the room the robot publishes into. Subscribe-only, short TTL, and only while production_mode is ON. This sits at the same (open) tier as the other /api/robots read routes; to tighten, add `if not await _authorize_fleet(authorization): raise HTTPException(401, ...)` like /api/livekit/token.""" if not await get_production_mode(): return {"active": False, "reason": "production_mode_off"} async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row resolved_id = await _resolve_robot_reference(db, robot_id) if not resolved_id: raise HTTPException(404, "Robot not found") async with db.execute("SELECT * FROM robots WHERE id = ?", (resolved_id,)) as c: robot = await c.fetchone() if not robot: raise HTTPException(404, "Robot not found") if not robot["current_session_id"]: return {"active": False, "reason": "idle"} async with db.execute("SELECT * FROM sessions WHERE id = ?", (robot["current_session_id"],)) as c: session = await c.fetchone() if not session or session["ended_at"]: return {"active": False, "reason": "idle"} server = None if session["server_id"]: async with db.execute("SELECT * FROM livekit_servers WHERE id = ?", (session["server_id"],)) as c: server = await c.fetchone() room_name = session["room_name"] if server: url, lk_key, lk_secret = server["url"], (server["api_key"] or "devkey"), (server["api_secret"] or "secret") else: url, lk_key, lk_secret = await _lk_config_async() try: from livekit.api import AccessToken, VideoGrants except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") identity = f"viewer:{resolved_id}:{int(datetime.utcnow().timestamp())}" grants = VideoGrants( room=room_name, room_join=True, can_publish=False, can_publish_data=False, can_subscribe=True, ) token = ( AccessToken(lk_key, lk_secret) .with_identity(identity) .with_name("operator-dashboard") .with_ttl(timedelta(minutes=15)) .with_grants(grants) .to_jwt() ) return { "active": True, "url": url, "token": token, "room": room_name, "identity": identity, } # ============================================================================= # ON-DEMAND CAMERA PREVIEW (operator-initiated, independent of scene detection) # ============================================================================= # Flow: dashboard POSTs /preview/start -> a preview room is assigned with a short # TTL and the operator gets a SUBSCRIBE-only viewer token. The robot polls # GET /preview/agent; while a preview is active it receives a PUBLISH token and # opens its cam into the same room. The dashboard POSTs /preview/keepalive while # watching; closing it POSTs /preview/stop. If keepalives stop, the preview # auto-expires and the robot's next poll returns {active:false} so it stops the # cam. Cam therefore runs ONLY while an operator is watching. PREVIEW_TTL_SECONDS = 45 # preview auto-expires this long after the last keepalive DEFAULT_VISION_PORT = 5000 # RoboVisionAI_PI's default Flask port (/video_feed, /api/motion/*) async def _mirror_device_to_robot(db, device_id: str): """Ensure a `robots` row exists for an enrolled device, so id-keyed routes (/stream, /preview/*) work even before the device's first real voice session (device_request_session normally creates this mirror, but only on first request-session — operator "Go live" should work before that too).""" async with db.execute("SELECT * FROM robots WHERE id = ?", (device_id,)) as c: robot = await c.fetchone() if robot: return robot async with db.execute("SELECT * FROM devices WHERE id = ?", (device_id,)) as c: device = await c.fetchone() if not device: return None await db.execute( "INSERT OR IGNORE INTO robots (id, name, character_id) VALUES (?, ?, ?)", (device_id, device["name"] or device_id, device["character_id"]), ) await db.commit() async with db.execute("SELECT * FROM robots WHERE id = ?", (device_id,)) as c: return await c.fetchone() async def _resolve_robot_reference(db, robot_ref: str) -> Optional[str]: """Resolve a display name to its active enrolled device id when needed.""" async with db.execute("SELECT id FROM devices WHERE id = ?", (robot_ref,)) as c: row = await c.fetchone() if row: return row["id"] async with db.execute( """SELECT id FROM devices WHERE lower(name) = lower(?) ORDER BY CASE lower(status) WHEN 'online' THEN 0 WHEN 'running' THEN 1 ELSE 2 END, last_heartbeat DESC, enrolled_at DESC LIMIT 1""", (robot_ref,), ) as c: row = await c.fetchone() if row: return row["id"] async with db.execute( "SELECT id FROM robots WHERE id = ? OR lower(name) = lower(?) LIMIT 1", (robot_ref, robot_ref), ) as c: row = await c.fetchone() return row["id"] if row else None async def _pick_preview_server(db, prefer_id: Optional[str] = None): """Return a LiveKit server row for a preview: a preferred one if still present, else the least-loaded online server, else any configured server.""" if prefer_id: async with db.execute("SELECT * FROM livekit_servers WHERE id = ?", (prefer_id,)) as c: row = await c.fetchone() if row: return row async with db.execute(""" SELECT s.* FROM livekit_servers s WHERE s.status = 'online' OR s.status IS NULL OR s.status = 'unknown' ORDER BY (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) ASC LIMIT 1 """) as c: row = await c.fetchone() if row: return row async with db.execute("SELECT * FROM livekit_servers LIMIT 1") as c: return await c.fetchone() def _preview_active(row) -> bool: if not row: return False try: return datetime.fromisoformat(row["expires_at"]) > datetime.utcnow() except Exception: return False def _mint_lk(server, room, identity, name, publish: bool, ttl_min: int): """Mint a LiveKit token via livekit-api, matching the other routes' pattern.""" from livekit.api import AccessToken, VideoGrants grants = VideoGrants( room=room, room_join=True, can_publish=publish, can_publish_data=publish, can_subscribe=not publish, ) return ( AccessToken(server["api_key"] or "devkey", server["api_secret"] or "secret") .with_identity(identity).with_name(name) .with_ttl(timedelta(minutes=ttl_min)).with_grants(grants).to_jwt() ) @app.post("/api/robots/{robot_id}/preview/start") async def preview_start(robot_id: str): """Operator dashboard: begin an on-demand live camera preview for a robot. Returns a SUBSCRIBE-only viewer token + the preview room.""" if not await get_production_mode(): return {"active": False, "reason": "production_mode_off"} async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row robot_id = await _resolve_robot_reference(db, robot_id) if not robot_id: raise HTTPException(404, "Robot not found") if not await _mirror_device_to_robot(db, robot_id): raise HTTPException(404, "Robot not found") async with db.execute("SELECT * FROM previews WHERE robot_id = ?", (robot_id,)) as c: existing = await c.fetchone() server = await _pick_preview_server(db, existing["server_id"] if existing else None) if not server: raise HTTPException(503, "No LiveKit server configured") room_name = (existing["room_name"] if existing else None) or f"preview_{robot_id}" expires = (datetime.utcnow() + timedelta(seconds=PREVIEW_TTL_SECONDS)).isoformat() await db.execute(""" INSERT INTO previews (robot_id, room_name, server_id, expires_at) VALUES (?, ?, ?, ?) ON CONFLICT(robot_id) DO UPDATE SET room_name=excluded.room_name, server_id=excluded.server_id, expires_at=excluded.expires_at """, (robot_id, room_name, server["id"], expires)) await db.commit() try: identity = f"viewer:{robot_id}:{int(datetime.utcnow().timestamp())}" token = _mint_lk(server, room_name, identity, "operator-preview", publish=False, ttl_min=5) except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") await _log_history("preview", "robot", robot_id, "start", "operator", f"room={room_name}, server={server['id']}") return {"active": True, "url": server["url"], "token": token, "room": room_name, "identity": identity, "expires_in": PREVIEW_TTL_SECONDS} @app.post("/api/robots/{robot_id}/preview/keepalive") async def preview_keepalive(robot_id: str): """Dashboard pings while the operator is watching, to keep the preview alive.""" expires = (datetime.utcnow() + timedelta(seconds=PREVIEW_TTL_SECONDS)).isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row robot_id = await _resolve_robot_reference(db, robot_id) if not robot_id: return {"active": False} cur = await db.execute("UPDATE previews SET expires_at = ? WHERE robot_id = ?", (expires, robot_id)) await db.commit() if cur.rowcount == 0: return {"active": False} await _log_history("preview", "robot", robot_id, "keepalive", "operator") return {"active": True, "expires_in": PREVIEW_TTL_SECONDS} @app.post("/api/robots/{robot_id}/preview/stop") async def preview_stop(robot_id: str): """Dashboard closed / operator stopped watching -> end the preview now.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row robot_id = await _resolve_robot_reference(db, robot_id) if not robot_id: return {"stopped": False} await db.execute("DELETE FROM previews WHERE robot_id = ?", (robot_id,)) await db.commit() await _log_history("preview", "robot", robot_id, "stop", "operator") return {"stopped": True} @app.get("/api/robots/{robot_id}/preview/agent") async def preview_agent(robot_id: str, authorization: Optional[str] = Header(default=None)): """The ROBOT polls this. While an operator preview is active it returns a PUBLISH token + room so the Pi opens its cam and joins; otherwise {active:false} so the Pi stops publishing. Enrolled-device auth.""" fleet_authorized = await _authorize_fleet(authorization) event_authorized = False if session_token: async with aiosqlite.connect(DB_PATH) as auth_db: async with auth_db.execute("SELECT event_token_hash FROM sessions WHERE id = ?", (session_id,)) as c: auth_row = await c.fetchone() event_authorized = bool( auth_row and auth_row[0] and hmac.compare_digest(auth_row[0], _hash(session_token.strip())) ) if not fleet_authorized and not event_authorized: raise HTTPException(401, "Invalid or missing device token") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row robot_id = await _resolve_robot_reference(db, robot_id) if not robot_id: return {"active": False} async with db.execute("SELECT * FROM previews WHERE robot_id = ?", (robot_id,)) as c: row = await c.fetchone() if not _preview_active(row): return {"active": False} # If the robot is in a real scene-detection session, ROBOVOICE owns the # camera device — don't have the preview agent fight it for /dev/video0. # The operator sees the live session cam via /stream instead. async with db.execute("SELECT current_session_id FROM robots WHERE id = ?", (robot_id,)) as c: rob = await c.fetchone() if rob and rob["current_session_id"]: return {"active": False, "reason": "in_session"} server = await _pick_preview_server(db, row["server_id"]) if not server: return {"active": False} try: token = _mint_lk(server, row["room_name"], robot_id, robot_id, publish=True, ttl_min=2) except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") await _log_history("preview", "robot", robot_id, "agent_token", "pi", f"room={row['room_name']}") return {"active": True, "url": server["url"], "token": token, "room": row["room_name"]} # ============================================================================= # DEVICE ENDPOINTS (Pi fleet) # ============================================================================= def _device_row_to_model(row) -> Device: """Convert a sqlite Row (or dict) to a Device, normalizing fields.""" d = dict(row) d.pop("token_hash", None) d.pop("enrollment_token_hash", None) inventory = d.get("device_inventory") if isinstance(inventory, str): try: d["device_inventory"] = json.loads(inventory) except json.JSONDecodeError: d["device_inventory"] = None greetings = d.get("greeting_phrases") if isinstance(greetings, str): try: d["greeting_phrases"] = [str(p) for p in json.loads(greetings) if str(p).strip()] except (json.JSONDecodeError, TypeError): d["greeting_phrases"] = [] elif not isinstance(greetings, list): d["greeting_phrases"] = [] for field in ("motor_registry", "motor_sequences"): value = d.get(field) if isinstance(value, str): try: d[field] = json.loads(value) except (json.JSONDecodeError, TypeError): d[field] = [] elif not isinstance(value, list): d[field] = [] sup = d.get("supervisor_status") if isinstance(sup, str): try: d["supervisor_status"] = json.loads(sup) except json.JSONDecodeError: d["supervisor_status"] = None return Device(**d) @app.get("/api/devices", response_model=List[Device]) async def list_devices(): """List all enrolled Pi devices.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( """SELECT * FROM devices ORDER BY lower(name), CASE lower(status) WHEN 'online' THEN 0 WHEN 'running' THEN 1 WHEN 'detecting' THEN 2 WHEN 'connecting' THEN 3 ELSE 4 END, last_heartbeat DESC, enrolled_at DESC, created_at DESC""" ) as cursor: return [_device_row_to_model(r) for r in await cursor.fetchall()] @app.get("/api/devices/by-room/{room_name}") async def get_device_by_room(room_name: str): """Look up the device bound to a LiveKit room. Resolves via the sessions table (robust to any room-name format the scheduler generates — request-session's robopark--, etc.) rather than parsing the name, since a plain prefix-strip broke the moment request-session started including a trailing timestamp. Falls back to a prefix-strip for the standing-room convention (robopark-, no trailing timestamp, no sessions row — e.g. robopark-pi-client's always-on room when production_mode is on).""" device_id = None async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT robot_id FROM sessions WHERE room_name = ? ORDER BY started_at DESC LIMIT 1", (room_name,), ) as c: session = await c.fetchone() if session: device_id = session["robot_id"] elif room_name.startswith("robopark-"): device_id = room_name[len("robopark-"):] if not device_id: raise HTTPException(404, "No device bound to this room") async with db.execute("SELECT * FROM devices WHERE id = ?", (device_id,)) as c: row = await c.fetchone() if not row: raise HTTPException(404, "No device bound to this room") return _device_row_to_model(row) @app.get("/api/devices/{device_id}", response_model=Device) async def get_device(device_id: str): async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM devices WHERE id = ?", (device_id,)) as c: row = await c.fetchone() if not row: raise HTTPException(404, "Device not found") return _device_row_to_model(row) @app.get("/api/characters/{character_id}/deployment") async def get_character_deployment(character_id: str): """Return every hardware node bound to a character, grouped by ownership role.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( """SELECT * FROM devices WHERE lower(character_id) = lower(?) ORDER BY last_heartbeat DESC, enrolled_at DESC, created_at DESC""", (character_id,), ) as cursor: devices = [_device_row_to_model(row) for row in await cursor.fetchall()] grouped = {"voice_vision": [], "motor": [], "combined": [], "control": []} for device in devices: role = device.device_role if device.device_role in grouped else "combined" grouped[role].append(device.model_dump()) return { "character_id": character_id, "voice_vision_boxes": grouped["voice_vision"], "motor_nodes": grouped["motor"], "combined_nodes": grouped["combined"], "control_nodes": grouped["control"], "device_count": len(devices), } class DeviceUpdate(BaseModel): name: Optional[str] = None device_role: Optional[str] = None character_id: Optional[str] = None motor_server_url: Optional[str] = None livekit_url: Optional[str] = None video_device: Optional[str] = None audio_device: Optional[str] = None audio_output_device: Optional[str] = None greeting_phrases: Optional[List[str]] = None motor_registry: Optional[List[dict]] = None motor_sequences: Optional[List[dict]] = None greeting_motor_sequence_id: Optional[str] = None tailscale_ip: Optional[str] = None lan_ip: Optional[str] = None notes: Optional[str] = None status: Optional[str] = None production_mode: Optional[bool] = None @app.patch("/api/devices/{device_id}", response_model=Device) async def update_device(device_id: str, payload: DeviceUpdate): """Partial update for a device (character binding, addresses, notes, status).""" fields = {k: v for k, v in payload.model_dump(exclude_none=True).items()} if "device_role" in fields and fields["device_role"] not in {"combined", "voice_vision", "motor", "control"}: raise HTTPException(422, "device_role must be combined, voice_vision, motor, or control") if "greeting_phrases" in fields: fields["greeting_phrases"] = json.dumps( [str(p).strip() for p in fields["greeting_phrases"] if str(p).strip()][:20] ) if "production_mode" in fields: fields["production_mode"] = 1 if fields["production_mode"] else 0 if not fields: return await get_device(device_id) set_clause = ", ".join(f"{k} = ?" for k in fields) values = list(fields.values()) + [device_id] async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute(f"UPDATE devices SET {set_clause} WHERE id = ?", values) await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Device not found") # Keep the legacy `robots` table's own `name` column in sync when this # PATCH actually changes the device name. Without this, robots.name # only ever gets set once (at first-creation, via INSERT OR IGNORE) # and silently goes stale on every subsequent rename — the dashboard # reads from /api/robots, not /api/devices, so it would keep showing # the old name forever. Only touch robots.name when a name was part # of this update (don't unconditionally overwrite it on unrelated # field updates), and only if a robots row for this id exists. if "name" in fields: await db.execute( "UPDATE robots SET name = ? WHERE id = ?", (fields["name"], device_id), ) await db.commit() await broadcast("device_updated", {"device_id": device_id, "fields": list(fields.keys())}) await _log_history("device", "device", device_id, "updated", "operator", f"fields={','.join(fields.keys())}") return await get_device(device_id) @app.post("/api/devices") async def create_device(payload: DeviceCreate): """Manually register a device from the UI. Returns the device record and a one-time enrollment token (if one wasn't provided). BUG FIXED: this had response_model=Device, which made FastAPI silently strip the enrollment_token key from the response below (Device has no such field) — the docstring's promised token was never actually reaching the caller. Dropping response_model here since the return value is deliberately Device-plus-one-extra-field, not just Device.""" if not payload.name or not payload.name.strip(): raise HTTPException(400, "name is required") if payload.device_role not in {"combined", "voice_vision", "motor", "control"}: raise HTTPException(422, "device_role must be combined, voice_vision, motor, or control") device_id = f"dev_{secrets.token_hex(4)}" # Resolve enrollment token: use provided one, else generate a fresh one. if payload.enrollment_token: enrollment_token = payload.enrollment_token else: enrollment_token = secrets.token_urlsafe(24) enrollment_hash = _hash(enrollment_token) now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO devices (id, name, device_role, tailscale_ip, lan_ip, motor_server_url, character_id, livekit_url, video_device, audio_device, enrollment_token_hash, greeting_phrases, status, enrolled_at, notes, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'enrolled', ?, ?, ?)""", ( device_id, payload.name.strip(), payload.device_role, payload.tailscale_ip, payload.lan_ip, payload.motor_server_url, payload.character_id, payload.livekit_url, payload.video_device, payload.audio_device, enrollment_hash, json.dumps(payload.greeting_phrases or []), now, payload.notes, now, ), ) await db.commit() logger.info(f"Device {device_id} ({payload.name}) created") await broadcast("device_added", {"device_id": device_id, "name": payload.name}) await _log_history("device", "device", device_id, "created", "operator", f"name={payload.name.strip()}") device = await get_device(device_id) return { **device.model_dump(), "enrollment_token": enrollment_token, # returned ONCE, not stored plaintext } @app.delete("/api/devices/{device_id}") async def delete_device(device_id: str): async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT name FROM devices WHERE id = ?", (device_id,)) as c: row = await c.fetchone() if not row: raise HTTPException(404, "Device not found") name = row["name"] await db.execute("DELETE FROM devices WHERE id = ?", (device_id,)) await db.commit() await broadcast("device_removed", {"device_id": device_id}) await _log_history("device", "device", device_id, "deleted", "operator", f"name={name}") return {"status": "deleted", "device_id": device_id} @app.post("/api/devices/{device_id}/token/rotate") async def rotate_device_token(device_id: str): """Issue a new device token; the old one is invalidated.""" new_token = secrets.token_urlsafe(32) new_hash = _hash(new_token) async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute("UPDATE devices SET token_hash = ? WHERE id = ?", (new_hash, device_id)) await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Device not found") await broadcast("device_token_rotated", {"device_id": device_id}) await _log_history("device", "device", device_id, "token_rotated", "operator") return {"device_id": device_id, "device_token": new_token} async def _authorize_device(device_id: str, authorization: Optional[str]) -> bool: """Verify Bearer token matches stored hash for this device.""" if not authorization or not authorization.lower().startswith("bearer "): return False token = authorization.split(" ", 1)[1].strip() async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT token_hash FROM devices WHERE id = ?", (device_id,)) as c: row = await c.fetchone() if not row or not row[0]: return False return hmac.compare_digest(row[0], _hash(token)) def _authorize_mesh_bootstrap(presented: Optional[str]) -> bool: """Authenticate bootstrap before a device-scoped token exists.""" return bool( ROBOPARK_MESH_TOKEN and presented and hmac.compare_digest(ROBOPARK_MESH_TOKEN, presented.strip()) ) async def _log_history(category: str, entity_type: Optional[str] = None, entity_id: Optional[str] = None, action: Optional[str] = None, actor: Optional[str] = None, details: Optional[str] = None): """Append an immutable audit/history record. The actor argument is overridden by `_current_actor_override` (a contextvar set by the FastAPI middleware from the X-Operator header) when present. The middleware is the only writer to the contextvar, so an untrusted caller cannot impersonate another operator by setting it directly. Falls back to the call-site `actor` argument when no override is set (e.g. system events).""" final_actor = _current_actor_override.get() or actor try: async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO history_log (timestamp, category, entity_type, entity_id, action, actor, details) VALUES (?, ?, ?, ?, ?, ?, ?)""", (datetime.utcnow().isoformat(), category, entity_type, entity_id, action, final_actor, details), ) await db.commit() except Exception as e: logger.warning(f"history log failed: {e}") # Per-request operator name. Set by the operator_actor_middleware (which # reads the X-Operator header) and read by _log_history. Sanitized at # the middleware boundary so the value reaching the database is always # a safe printable string. _current_actor_override: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar("current_actor", default=None) def _sanitize_actor_name(name: Optional[str]) -> Optional[str]: """Trim, strip control chars, cap at 64 chars. Returns None if the result is empty so the caller can fall back to the default actor.""" if not name: return None s = str(name).strip() if not s: return None # Allow printable ASCII + tab; strip everything else. Newlines and # CR would split a single history row into two display lines, so # we collapse them to a single space. out = [] for c in s: if c == "\t" or c == " ": out.append(c) elif 0x20 <= ord(c) < 0x7f: out.append(c) else: # collapse any other whitespace-ish char to a space if c in ("\n", "\r", "\v", "\f"): out.append(" ") # drop anything else silently cleaned = "".join(out).strip() if not cleaned: return None if len(cleaned) > 64: cleaned = cleaned[:64] return cleaned async def _authorize_fleet(authorization: Optional[str]) -> bool: """Verify a Bearer token matches ANY enrolled device token (fleet auth). Used by endpoints that hand out LiveKit access but are keyed by robot_id (not device_id), so we can't scope to a single device row. Mirrors the hash + constant-time compare used by _authorize_device.""" if not authorization or not authorization.lower().startswith("bearer "): return False token_hash = _hash(authorization.split(" ", 1)[1].strip()) async with aiosqlite.connect(DB_PATH) as db: async with db.execute( "SELECT token_hash FROM devices WHERE token_hash IS NOT NULL" ) as c: rows = await c.fetchall() return any(r[0] and hmac.compare_digest(r[0], token_hash) for r in rows) @app.post("/api/devices/bootstrap", response_model=DeviceEnrollResponse) async def bootstrap_device( payload: DeviceBootstrapRequest, x_robopark_mesh_token: Optional[str] = Header(default=None), ): """Create or recover one scheduler identity for a mesh-authenticated robot.""" if not _authorize_mesh_bootstrap(x_robopark_mesh_token): raise HTTPException(401, "Invalid or missing RoboPark mesh token") name = payload.name.strip() if not name: raise HTTPException(400, "name is required") if payload.device_role not in {"combined", "voice_vision", "motor", "control"}: raise HTTPException(422, "device_role must be combined, voice_vision, motor, or control") new_token = secrets.token_urlsafe(32) new_hash = _hash(new_token) now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( """SELECT id FROM devices WHERE lower(name) = lower(?) AND COALESCE(device_role, 'combined') = ? ORDER BY CASE lower(status) WHEN 'online' THEN 0 WHEN 'running' THEN 1 ELSE 2 END, last_heartbeat DESC, enrolled_at DESC LIMIT 1""", (name, payload.device_role), ) as cursor: existing = await cursor.fetchone() if existing: device_id = existing["id"] await db.execute( """UPDATE devices SET token_hash = ?, status = 'enrolled', lan_ip = COALESCE(?, lan_ip), tailscale_ip = COALESCE(?, tailscale_ip), livekit_url = COALESCE(?, livekit_url), motor_server_url = COALESCE(?, motor_server_url), character_id = COALESCE(?, character_id), device_role = ?, enrolled_at = COALESCE(enrolled_at, ?) WHERE id = ?""", (new_hash, payload.lan_ip, payload.tailscale_ip, payload.livekit_url, payload.motor_server_url, payload.character_id, payload.device_role, now, device_id), ) else: device_id = f"dev_{secrets.token_hex(4)}" await db.execute( """INSERT INTO devices (id, name, device_role, lan_ip, tailscale_ip, livekit_url, motor_server_url, character_id, token_hash, status, enrolled_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'enrolled', ?, ?)""", (device_id, name, payload.device_role, payload.lan_ip, payload.tailscale_ip, payload.livekit_url, payload.motor_server_url, payload.character_id, new_hash, now, now), ) await db.execute( "INSERT OR IGNORE INTO robots (id, name, status) VALUES (?, ?, 'idle')", (device_id, name), ) await db.execute("UPDATE robots SET name = ? WHERE id = ?", (name, device_id)) await db.commit() await broadcast("device_bootstrapped", {"device_id": device_id, "name": name}) await _log_history("device", "device", device_id, "bootstrapped", "mesh", f"name={name}") scheduler_url = os.getenv("SCHEDULER_PUBLIC_URL", f"http://localhost:{os.getenv('SCHEDULER_PORT', '8080')}") return DeviceEnrollResponse(device_id=device_id, device_token=new_token, scheduler_url=scheduler_url) @app.post("/api/devices/enroll", response_model=DeviceEnrollResponse) async def enroll_device(payload: DeviceEnrollRequest): """First-boot enrollment for a Pi. Requires production_mode = true and a valid enrollment token. Returns a long-lived device_token the Pi will use for heartbeats.""" if not await get_production_mode(): raise HTTPException(403, "Production mode is disabled. Enable it in Settings before enrolling devices.") if not payload.enrollment_token: raise HTTPException(400, "enrollment_token is required") if payload.device_role is not None and payload.device_role not in {"combined", "voice_vision", "motor", "control"}: raise HTTPException(422, "device_role must be combined, voice_vision, motor, or control") token_hash = _hash(payload.enrollment_token) # Two paths: # 1. Token matches a pre-registered device row -> reuse that device, rotate its token. # 2. Token matches the GLOBAL default enrollment token -> mint a NEW device on the fly. async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT * FROM devices WHERE enrollment_token_hash = ?", (token_hash,) ) as c: existing = await c.fetchone() default_token_hash = await get_setting("default_enrollment_token") matches_default = default_token_hash and hmac.compare_digest(default_token_hash, token_hash) if not existing and not matches_default: raise HTTPException(401, "Invalid enrollment token") if existing: device_id = existing["id"] new_token = secrets.token_urlsafe(32) new_hash = _hash(new_token) now = datetime.utcnow().isoformat() await db.execute( """UPDATE devices SET token_hash = ?, status = 'enrolled', enrolled_at = COALESCE(enrolled_at, ?), tailscale_ip = COALESCE(?, tailscale_ip), lan_ip = COALESCE(?, lan_ip), motor_server_url = COALESCE(?, motor_server_url), livekit_url = COALESCE(?, livekit_url), character_id = COALESCE(?, character_id), video_device = COALESCE(?, video_device), audio_device = COALESCE(?, audio_device), device_role = COALESCE(?, device_role), name = COALESCE(?, name) WHERE id = ?""", ( new_hash, now, payload.tailscale_ip, payload.lan_ip, payload.motor_server_url, payload.livekit_url, payload.character_id, payload.video_device, payload.audio_device, payload.device_role, payload.name, device_id, ), ) # OR IGNORE: only fires the first time this robots row is created. # PATCH /api/devices/{device_id} now propagates renames into # robots.name afterwards, so this is purely a first-creation # concern, not an ongoing sync path. await db.execute( "INSERT OR IGNORE INTO robots (id, name, character_id, status) VALUES (?, ?, ?, 'idle')", (device_id, existing["name"] or payload.name or device_id, existing["character_id"]), ) else: # First-boot enroll via the global default token -> create a new device. # # BUG FIXED: this used to persist enrollment_token_hash=token_hash # (the hash of the SHARED default token) onto the new device row. # Since the "existing" lookup above matches on enrollment_token_hash, # every subsequent enroll using that same shared default token would # match THIS row and take the "existing" reuse/rotate path instead of # minting an independent device — silently renaming and re-keying # whichever device happened to enroll first, and invalidating its # live device_token. Devices minted via the default token must keep # enrollment_token_hash NULL so the default token stays reusable # across an entire fleet, as its name implies. device_id = f"dev_{secrets.token_hex(4)}" new_token = secrets.token_urlsafe(32) new_hash = _hash(new_token) now = datetime.utcnow().isoformat() await db.execute( """INSERT INTO devices (id, name, device_role, tailscale_ip, lan_ip, motor_server_url, character_id, livekit_url, video_device, audio_device, token_hash, enrollment_token_hash, status, enrolled_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'enrolled', ?, ?)""", ( device_id, (payload.name or f"Pi-{device_id[-4:]}").strip(), payload.device_role or "combined", payload.tailscale_ip, payload.lan_ip, payload.motor_server_url, payload.character_id, payload.livekit_url, payload.video_device, payload.audio_device, new_hash, None, now, now, ), ) # OR IGNORE: only fires the first time this robots row is created. # PATCH /api/devices/{device_id} now propagates renames into # robots.name afterwards, so this is purely a first-creation # concern, not an ongoing sync path. await db.execute( "INSERT OR IGNORE INTO robots (id, name, character_id, status) VALUES (?, ?, ?, 'idle')", (device_id, (payload.name or f"Pi-{device_id[-4:]}").strip(), payload.character_id), ) await db.commit() await broadcast("device_enrolled", {"device_id": device_id}) logger.info(f"Device {device_id} enrolled") await _log_history("device", "device", device_id, "enrolled", "pi", f"name={(payload.name or device_id).strip()}") # Discover our own scheduler URL (best effort) scheduler_url = os.getenv("SCHEDULER_PUBLIC_URL", f"http://localhost:{os.getenv('SCHEDULER_PORT', '8080')}") return DeviceEnrollResponse(device_id=device_id, device_token=new_token, scheduler_url=scheduler_url) async def _effective_device_production_mode(db, device_id: str) -> bool: """Global master switch AND this robot's own toggle. Used everywhere a robot decides whether to auto-trigger/stay in its motion-triggered loop.""" if not await get_production_mode(): return False db.row_factory = aiosqlite.Row async with db.execute("SELECT production_mode FROM devices WHERE id = ?", (device_id,)) as c: row = await c.fetchone() return bool(row["production_mode"]) if row else False @app.get("/api/devices/{device_id}/trigger-command") async def device_trigger_command(device_id: str, authorization: Optional[str] = Header(default=None)): """Authenticated robot poll for dashboard-simulated motion events.""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") async with aiosqlite.connect(DB_PATH) as db: if not await _effective_device_production_mode(db, device_id): return {"trigger": False, "production_mode": False} db.row_factory = aiosqlite.Row async with db.execute( "SELECT source FROM trigger_commands WHERE robot_id = ?", (device_id,) ) as c: command = await c.fetchone() if not command: return {"trigger": False, "production_mode": True} await db.execute("DELETE FROM trigger_commands WHERE robot_id = ?", (device_id,)) await db.commit() return {"trigger": True, "source": command["source"], "production_mode": True} @app.post("/api/devices/{device_id}/heartbeat") async def device_heartbeat(device_id: str, payload: DeviceHeartbeat, authorization: Optional[str] = Header(default=None)): if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") if payload.device_role is not None and payload.device_role not in {"combined", "voice_vision", "motor", "control"}: raise HTTPException(422, "device_role must be combined, voice_vision, motor, or control") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( """UPDATE devices SET status = ?, last_heartbeat = ?, lan_ip = COALESCE(?, lan_ip), last_seen_ip = COALESCE(?, last_seen_ip), device_inventory = COALESCE(?, device_inventory), device_role = COALESCE(?, device_role), livekit_url = COALESCE(?, livekit_url), motor_server_url = COALESCE(?, motor_server_url) WHERE id = ?""", ( payload.status or "online", datetime.utcnow().isoformat(), payload.ip, payload.ip, json.dumps(payload.device_inventory) if payload.device_inventory is not None else None, payload.device_role, payload.livekit_url, payload.motor_server_url, device_id, ), ) as cur: await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Device not found") effective_pm = await _effective_device_production_mode(db, device_id) await broadcast("device_heartbeat", {"device_id": device_id, "status": payload.status}) await _log_history("device", "device", device_id, "heartbeat", "pi", f"status={payload.status or 'online'}, ip={payload.ip}") # Effective production mode = the global master switch AND this specific # robot's own toggle. A robot with its own toggle off never auto-triggers # sessions from motion, even while the fleet-wide switch is on. return { "status": "ok", "production_mode": effective_pm, "server_time": datetime.utcnow().isoformat(), } COMMAND_STATES = { "queued", "delivered", "acknowledged", "running", "succeeded", "failed", "expired", "superseded", } COMMAND_OPERATIONS = { "voice.trigger", "voice.configuration.stage", "voice.configuration.apply_when_idle", "voice.restart_when_idle", "voice.stop_when_idle", "voice.restart", "voice.stop", "voice.configuration.apply", "voice.configuration.rollback", "voice.media.release_stale", } @app.post("/api/robots/{robot_id}/commands") async def create_management_command(robot_id: str, payload: DurableCommandPayload): if payload.operation not in COMMAND_OPERATIONS: raise HTTPException(422, "operation is not allowed") if payload.force and payload.operation not in { "voice.restart", "voice.stop", "voice.configuration.apply", "voice.media.release_stale", }: raise HTTPException(422, "force is only valid for an explicit destructive operation") command_id = str(uuid.uuid4()) created = datetime.utcnow() expires = created + timedelta(seconds=max(10, min(payload.expires_in_seconds, 86400))) async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT id FROM devices WHERE id = ?", (robot_id,)) as cursor: if not await cursor.fetchone(): raise HTTPException(404, "Robot device not found") await db.execute( "UPDATE management_commands SET state='superseded', completed_at=? " "WHERE robot_id=? AND operation=? AND state='queued'", (created.isoformat(), robot_id, payload.operation), ) await db.execute( """INSERT INTO management_commands (command_id, robot_id, operation, payload, created_at, expires_at, desired_revision, force, state) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'queued')""", ( command_id, robot_id, payload.operation, json.dumps(payload.payload or {}, separators=(",", ":")), created.isoformat(), expires.isoformat(), payload.desired_revision, int(payload.force), ), ) await db.commit() await _log_history( "voice_command", "device", robot_id, payload.operation, "operator", f"command_id={command_id}, force={payload.force}, desired_revision={payload.desired_revision}", ) return {"command_id": command_id, "robot_id": robot_id, "state": "queued", "expires_at": expires.isoformat()} @app.get("/api/robots/{robot_id}/commands") async def list_management_commands(robot_id: str, limit: int = 100): async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT * FROM management_commands WHERE robot_id=? ORDER BY created_at DESC LIMIT ?", (robot_id, max(1, min(limit, 500))), ) as cursor: rows = await cursor.fetchall() return [{**dict(row), "payload": json.loads(row["payload"] or "{}"), "result": json.loads(row["result"] or "null")} for row in rows] @app.get("/api/devices/{device_id}/commands/pending") async def pending_management_commands(device_id: str, authorization: Optional[str] = Header(default=None)): if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row await db.execute( "UPDATE management_commands SET state='expired', completed_at=? " "WHERE robot_id=? AND state IN ('queued','delivered') AND expires_at < ?", (now, device_id, now), ) async with db.execute( "SELECT * FROM management_commands WHERE robot_id=? AND state IN ('queued','delivered') " "ORDER BY created_at LIMIT 20", (device_id,), ) as cursor: rows = await cursor.fetchall() ids = [row["command_id"] for row in rows] if ids: await db.executemany( "UPDATE management_commands SET state='delivered', delivered_at=COALESCE(delivered_at, ?) " "WHERE command_id=? AND state='queued'", [(now, command_id) for command_id in ids], ) await db.commit() return {"commands": [{**dict(row), "payload": json.loads(row["payload"] or "{}")} for row in rows]} @app.post("/api/devices/{device_id}/commands/{command_id}/ack") async def acknowledge_management_command(device_id: str, command_id: str, payload: DurableCommandAck, authorization: Optional[str] = Header(default=None)): if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") if payload.status not in COMMAND_STATES - {"queued", "delivered", "expired", "superseded"}: raise HTTPException(422, "invalid robot acknowledgement state") now = datetime.utcnow().isoformat() columns = { "acknowledged": "acknowledged_at", "running": "started_at", "succeeded": "completed_at", "failed": "completed_at", } timestamp_column = columns[payload.status] async with aiosqlite.connect(DB_PATH) as db: cursor = await db.execute( f"UPDATE management_commands SET state=?, {timestamp_column}=?, result=? " "WHERE command_id=? AND robot_id=? AND state NOT IN ('expired','superseded','succeeded','failed')", (payload.status, now, json.dumps(payload.result or {}), command_id, device_id), ) await db.commit() if cursor.rowcount == 0: async with db.execute( "SELECT state FROM management_commands WHERE command_id=? AND robot_id=?", (command_id, device_id), ) as existing_cursor: existing = await existing_cursor.fetchone() if not existing: raise HTTPException(404, "Command not found") return {"command_id": command_id, "state": existing[0], "duplicate": True} return {"command_id": command_id, "state": payload.status} @app.post("/api/devices/{device_id}/telemetry/events") async def ingest_durable_robot_events(device_id: str, payload: DurableTelemetryBatch, authorization: Optional[str] = Header(default=None)): if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") if len(payload.events) > 1000: raise HTTPException(413, "A telemetry batch may contain at most 1000 events") highest = 0 accepted = 0 async with aiosqlite.connect(DB_PATH) as db: for event in sorted(payload.events, key=lambda item: int(item.get("sequence") or 0)): event_id = str(event.get("event_id") or "") sequence = int(event.get("sequence") or 0) event_type = str(event.get("type") or "")[:100] if not event_id or sequence <= 0 or not event_type: raise HTTPException(422, "event_id, positive sequence, and type are required") timestamp = event.get("timestamp") if isinstance(timestamp, (int, float)): timestamp = datetime.utcfromtimestamp(timestamp).isoformat() timestamp = str(timestamp or datetime.utcnow().isoformat()) event_payload = event.get("payload") or {} encoded = json.dumps(event_payload, separators=(",", ":")) cursor = await db.execute( "INSERT OR IGNORE INTO robot_events(event_id, robot_id, sequence, session_id, type, timestamp, payload) " "VALUES (?, ?, ?, ?, ?, ?, ?)", (event_id, device_id, sequence, event.get("session_id"), event_type, timestamp, encoded), ) accepted += max(0, cursor.rowcount) highest = max(highest, sequence) if event_type == "transcript.final" and event.get("session_id"): session_id = str(event["session_id"]) await db.execute( """INSERT OR IGNORE INTO sessions (id, robot_id, started_at, source, engine, initiated_by, metadata_json) VALUES (?, ?, ?, 'robot_hardware', 'elevenlabs', 'local_supervisor', '{}')""", (session_id, device_id, timestamp), ) await db.execute( """INSERT OR IGNORE INTO session_transcripts (session_id, robot_id, role, text, is_final, sequence, source, source_id, metadata_json, timestamp) VALUES (?, ?, ?, ?, 1, ?, 'robot_hardware', ?, '{}', ?)""", ( session_id, device_id, str(event_payload.get("role") or "user"), str(event_payload.get("text") or ""), sequence, event_id, timestamp, ), ) await db.commit() return {"ok": True, "accepted": accepted, "acknowledged_sequence": highest} @app.post("/api/devices/{device_id}/voice-runtime-state") async def report_voice_runtime_state(device_id: str, payload: RobotVoiceRuntimeState, authorization: Optional[str] = Header(default=None)): if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") now = datetime.utcnow().isoformat() encoded = json.dumps(payload.model_dump(), separators=(",", ":")) async with aiosqlite.connect(DB_PATH) as db: await db.execute( "UPDATE devices SET voice_runtime_state=?, last_heartbeat=?, status='online' WHERE id=?", (encoded, now, device_id), ) await db.execute( """INSERT INTO robot_voice_state (robot_id, desired_revision, applied_revision, application_state, applied_configuration, last_known_good_configuration, updated_at) VALUES (?, ?, ?, ?, '{}', '{}', ?) ON CONFLICT(robot_id) DO UPDATE SET applied_revision=excluded.applied_revision, application_state=excluded.application_state, updated_at=excluded.updated_at""", ( device_id, payload.desired_revision, payload.applied_revision, "waiting_for_idle" if payload.desired_revision != payload.applied_revision else "applied", now, ), ) await db.commit() return {"ok": True, "desired_revision": payload.desired_revision, "applied_revision": payload.applied_revision} async def _effective_voice_configuration(robot_id: str, session_engine: Optional[str] = None) -> dict: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT d.character_id, d.voice_engine_override, cp.default_engine, cp.elevenlabs_agent_id, " "cp.elevenlabs_branch_id, cp.robovoice_profile_id FROM devices d " "LEFT JOIN character_presets cp ON cp.id=d.character_id WHERE d.id=?", (robot_id,), ) as cursor: row = await cursor.fetchone() if not row: raise HTTPException(404, "Robot device not found") async with db.execute("SELECT value FROM settings WHERE key='voice_engine_default'") as cursor: setting = await cursor.fetchone() async with db.execute("SELECT * FROM robot_voice_state WHERE robot_id=?", (robot_id,)) as cursor: state = await cursor.fetchone() global_engine = str((setting[0] if setting else "elevenlabs") or "elevenlabs") candidates = ( (session_engine, "session"), (row["voice_engine_override"], "robot"), (row["default_engine"], "character"), (global_engine, "global"), ) engine, source = next((str(value), source) for value, source in candidates if value) if engine not in {"elevenlabs", "robovoice"}: engine, source = "elevenlabs", "global" state_data = dict(state) if state else {} return { "robot_id": robot_id, "character_id": row["character_id"], "effective_engine": engine, "source": source, "fallback_available": engine == "elevenlabs", "elevenlabs": {"agent_id": row["elevenlabs_agent_id"], "branch_id": row["elevenlabs_branch_id"]}, "robovoice": {"profile_id": row["robovoice_profile_id"]}, "desired_revision": int(state_data.get("desired_revision") or 1), "applied_revision": int(state_data.get("applied_revision") or 0), "application_state": state_data.get("application_state") or "pending", "desired_configuration": json.loads(state_data.get("desired_configuration") or "{}"), "applied_configuration": json.loads(state_data.get("applied_configuration") or "{}"), } @app.get("/api/robots/{robot_id}/voice-configuration") async def get_robot_voice_configuration(robot_id: str, session_engine: Optional[str] = None): return await _effective_voice_configuration(robot_id, session_engine) @app.post("/api/robots/{robot_id}/voice-configuration/stage") async def stage_robot_voice_configuration(robot_id: str, payload: RobotVoiceConfigurationPayload): if payload.engine_override not in {None, "elevenlabs", "robovoice"}: raise HTTPException(422, "engine_override must be elevenlabs, robovoice, or null") if any(word in json.dumps(payload.desired_configuration).casefold() for word in ("api_key", "password", "secret")): raise HTTPException(422, "credentials cannot be staged through this endpoint") current = await _effective_voice_configuration(robot_id) revision = max(int(payload.desired_revision or 0), int(current["desired_revision"]) + 1) now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO robot_voice_state (robot_id, character_id, engine_override, desired_revision, applied_revision, application_state, desired_configuration, updated_at) VALUES (?, ?, ?, ?, ?, 'staged', ?, ?) ON CONFLICT(robot_id) DO UPDATE SET character_id=excluded.character_id, engine_override=excluded.engine_override, desired_revision=excluded.desired_revision, application_state='staged', desired_configuration=excluded.desired_configuration, updated_at=excluded.updated_at""", ( robot_id, payload.character_id or current["character_id"], payload.engine_override, revision, current["applied_revision"], json.dumps(payload.desired_configuration, separators=(",", ":")), now, ), ) await db.execute("UPDATE devices SET voice_engine_override=? WHERE id=?", (payload.engine_override, robot_id)) await db.commit() command = await create_management_command(robot_id, DurableCommandPayload( operation="voice.configuration.stage", desired_revision=revision, payload={**payload.desired_configuration, "desired_revision": revision}, )) return {**(await _effective_voice_configuration(robot_id)), "command": command} PIPELINE_STAGES = ( "robot_online", "camera_ready", "microphone_ready", "speaker_ready", "motion_detected", "scheduler_session", "livekit_join", "camera_published", "microphone_published", "voice_worker", "stt_listening", "llm_response", "tts_subscribed", "playback_started", "motor_sequence", "session_ended", ) PIPELINE_STATUSES = {"pending", "running", "ok", "failed", "blocked", "skipped"} PIPELINE_STAGE_LABELS = { "robot_online": "Robot online", "camera_ready": "Camera detected", "microphone_ready": "Microphone detected", "speaker_ready": "Speaker detected", "motion_detected": "Waiting for motion", "scheduler_session": "Allocating session", "livekit_join": "Joining LiveKit", "camera_published": "Publishing camera", "microphone_published": "Publishing microphone", "voice_worker": "Starting voice worker", "stt_listening": "Listening to visitor", "llm_response": "Preparing response", "tts_subscribed": "Preparing robot voice", "playback_started": "Speaking through robot", "motor_sequence": "Running motor sequence", "session_ended": "Closing session", } def _pipeline_current_state(events: list[dict], *, active: bool, production_mode: bool, online: bool) -> dict: """Derive one operator-facing live stage from persistent pipeline events.""" if not online: return { "stage": "robot_online", "status": "blocked", "label": PIPELINE_STAGE_LABELS["robot_online"], "message": "Robot heartbeat is stale or missing", "since": None, } if not active: return { "stage": "motion_detected", "status": "waiting" if production_mode else "paused", "label": "Armed - waiting for motion" if production_mode else "Production paused", "message": "Motion detection is armed" if production_mode else "Enable production mode to arm this robot", "since": None, } scoped = [event for event in events if event.get("session_id")] for event in reversed(scoped): if event.get("status") in {"failed", "blocked"}: return { "stage": event["stage"], "status": event["status"], "label": PIPELINE_STAGE_LABELS.get(event["stage"], event["stage"]), "message": event.get("message") or "Pipeline requires attention", "since": event.get("timestamp"), } conversational = { "voice_worker", "stt_listening", "llm_response", "tts_subscribed", "playback_started", "motor_sequence", } latest = next((event for event in reversed(scoped) if event.get("stage") in conversational), None) if latest: stage = latest["stage"] status = latest.get("status") or "running" if status in {"ok", "skipped"}: stage = { "voice_worker": "stt_listening", "stt_listening": "llm_response", "llm_response": "tts_subscribed", "tts_subscribed": "playback_started", "playback_started": "stt_listening", "motor_sequence": "stt_listening", }.get(stage, stage) status = "active" return { "stage": stage, "status": status, "label": PIPELINE_STAGE_LABELS.get(stage, stage), "message": latest.get("message") or PIPELINE_STAGE_LABELS.get(stage, stage), "since": latest.get("timestamp"), } startup = ( "scheduler_session", "livekit_join", "camera_published", "microphone_published", "voice_worker", ) completed = {event.get("stage") for event in scoped if event.get("status") in {"ok", "skipped"}} stage = next((candidate for candidate in startup if candidate not in completed), "voice_worker") latest_startup = next((event for event in reversed(scoped) if event.get("stage") in startup), None) return { "stage": stage, "status": "active", "label": PIPELINE_STAGE_LABELS.get(stage, stage), "message": (latest_startup or {}).get("message") or "Production conversation is starting", "since": (latest_startup or {}).get("timestamp"), } async def _insert_pipeline_event(device_id: str, payload: PipelineEventPayload) -> dict: if payload.stage not in PIPELINE_STAGES: raise HTTPException(422, f"Unknown pipeline stage: {payload.stage}") if payload.status not in PIPELINE_STATUSES: raise HTTPException(422, f"Unknown pipeline status: {payload.status}") message = (payload.message or "")[:500] details = json.dumps(payload.details or {}, separators=(",", ":"))[:4000] now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute( """INSERT INTO pipeline_events (robot_id, session_id, stage, status, message, source, details, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", (device_id, payload.session_id, payload.stage, payload.status, message, (payload.source or "robot")[:40], details, now), ) await db.commit() await broadcast("pipeline_event", {"robot_id": device_id, "stage": payload.stage, "status": payload.status}) return {"id": cur.lastrowid, "timestamp": now} @app.post("/api/devices/{device_id}/pipeline-events") async def device_pipeline_event(device_id: str, payload: PipelineEventPayload, authorization: Optional[str] = Header(default=None)): """Receive real pipeline stage transitions from the enrolled robot.""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") result = await _insert_pipeline_event(device_id, payload) if payload.stage == "playback_started" and payload.status == "ok" and payload.session_id: await _queue_configured_greeting_sequence(device_id, payload.session_id) return {"status": "ok", **result} @app.get("/api/robots/{robot_id}/pipeline-status") async def robot_pipeline_status(robot_id: str, limit: int = 80): """One operator-facing snapshot for remote camera-to-playback testing.""" limit = max(10, min(limit, 200)) async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT * FROM devices WHERE id = ? OR lower(name) = lower(?) ORDER BY last_heartbeat DESC LIMIT 1", (robot_id, robot_id), ) as c: device = await c.fetchone() if not device: raise HTTPException(404, "Robot device not found") device_id = device["id"] telemetry = await _robot_telemetry(db, device_id) async with db.execute( "SELECT * FROM sessions WHERE robot_id = ? ORDER BY started_at DESC LIMIT 1", (device_id,) ) as c: latest_session = await c.fetchone() async with db.execute( "SELECT * FROM pipeline_events WHERE robot_id = ? ORDER BY timestamp DESC LIMIT ?", (device_id, limit), ) as c: rows = await c.fetchall() try: inventory = json.loads(device["device_inventory"] or "{}") except (TypeError, ValueError): inventory = {} cameras = inventory.get("video") or inventory.get("cameras") or [] inputs = inventory.get("audio_input") or inventory.get("audio_inputs") or inventory.get("inputs") or [] outputs = inventory.get("audio_output") or inventory.get("audio_outputs") or inventory.get("outputs") or [] cameras = [d for d in cameras if str(d.get("id", "")).lower() not in ("auto", "none")] inputs = [d for d in inputs if str(d.get("id", "")).lower() not in ("default", "none")] outputs = [d for d in outputs if str(d.get("id", "")).lower() not in ("default", "none")] heartbeat_age = telemetry.get("heartbeat_age_seconds") if telemetry else None prereqs = { "robot_online": heartbeat_age is not None and heartbeat_age <= 30, "camera_ready": bool(cameras), "microphone_ready": bool(inputs), "speaker_ready": bool(outputs), } events = [] for row in reversed(rows): item = dict(row) try: item["details"] = json.loads(item.get("details") or "{}") except (TypeError, ValueError): item["details"] = {} events.append(item) blockers = [] if not prereqs["robot_online"]: blockers.append("Robot heartbeat is stale or missing") if not bool(device["production_mode"]): blockers.append("Robot production mode is off") if telemetry and telemetry.get("readiness_code") not in ("ready", "in_session"): blockers.append(telemetry.get("readiness_message") or telemetry["readiness_code"]) if not prereqs["camera_ready"]: blockers.append("Camera inventory has not been reported") if not prereqs["microphone_ready"]: blockers.append("Microphone inventory has not been reported") if not prereqs["speaker_ready"]: blockers.append("Speaker inventory has not been reported") active = bool(latest_session and not latest_session["ended_at"]) active_session_id = latest_session["id"] if active else None current_events = [ event for event in events if not active_session_id or event.get("session_id") == active_session_id ] current_stage = _pipeline_current_state( current_events, active=active, production_mode=bool(device["production_mode"]), online=prereqs["robot_online"], ) return { "robot_id": device_id, "robot_name": device["name"], "overall": "blocked" if blockers else ("running" if active else "ready"), "blockers": list(dict.fromkeys(blockers)), "prerequisites": prereqs, "telemetry": telemetry, "latest_session": dict(latest_session) if latest_session else None, "events": events, "stages": list(PIPELINE_STAGES), "current_stage": current_stage, } @app.get("/api/robots/{robot_id}/pipeline-history") async def robot_pipeline_history(robot_id: str, page: int = 1, page_size: int = 6): """Return persistent, paged production runs for one robot.""" page = max(1, page) page_size = max(1, min(page_size, 25)) async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT id, name FROM devices WHERE id = ? OR lower(name) = lower(?) " "ORDER BY last_heartbeat DESC LIMIT 1", (robot_id, robot_id), ) as cursor: device = await cursor.fetchone() if not device: raise HTTPException(404, "Robot device not found") device_id = device["id"] async with db.execute( "SELECT COUNT(*) FROM sessions WHERE robot_id = ?", (device_id,) ) as cursor: total = int((await cursor.fetchone())[0]) pages = max(1, (total + page_size - 1) // page_size) page = min(page, pages) offset = (page - 1) * page_size async with db.execute( """SELECT s.*, s.id AS session_id, (SELECT COUNT(*) FROM session_transcripts t WHERE t.session_id = s.id AND t.is_final = 1) AS transcript_count, (SELECT COUNT(*) FROM pipeline_events p WHERE p.session_id = s.id) AS pipeline_event_count, (SELECT COUNT(*) FROM pipeline_events p WHERE p.session_id = s.id AND p.status IN ('failed','blocked')) AS error_count FROM sessions s WHERE s.robot_id = ? ORDER BY s.started_at DESC LIMIT ? OFFSET ?""", (device_id, page_size, offset), ) as cursor: sessions = [dict(row) for row in await cursor.fetchall()] events_by_session = {item["session_id"]: [] for item in sessions} session_ids = list(events_by_session) if session_ids: placeholders = ",".join("?" for _ in session_ids) async with db.execute( f"SELECT * FROM pipeline_events WHERE session_id IN ({placeholders}) ORDER BY timestamp, id", session_ids, ) as cursor: for row in await cursor.fetchall(): event = dict(row) try: event["details"] = json.loads(event.get("details") or "{}") except (TypeError, ValueError): event["details"] = {} events_by_session[event["session_id"]].append(event) for session in sessions: session["events"] = events_by_session.get(session["session_id"], []) return { "robot_id": device_id, "robot_name": device["name"], "items": sessions, "page": page, "page_size": page_size, "pages": pages, "total": total, } @app.post("/api/robots/{robot_id}/pipeline-test/start") async def start_pipeline_test(robot_id: str): snapshot = await robot_pipeline_status(robot_id) hard_blockers = [b for b in snapshot["blockers"] if "inventory" not in b.lower()] if hard_blockers: return {"queued": False, "reason": hard_blockers[0], "blockers": snapshot["blockers"]} result = await simulate_trigger(snapshot["robot_id"]) return {**result, "robot_id": snapshot["robot_id"], "blockers": snapshot["blockers"]} @app.post("/api/robots/{robot_id}/pipeline-test/stop") async def stop_pipeline_test(robot_id: str): snapshot = await robot_pipeline_status(robot_id) return await simulate_stop(snapshot["robot_id"]) @app.post("/api/devices/{device_id}/supervisor-status") async def device_supervisor_status(device_id: str, payload: SupervisorStatusReport, authorization: Optional[str] = Header(default=None)): """robot_supervisor.py's periodic self-report: which services it's managing on this robot and each one's current up/down state. Reported independently of device_heartbeat (which comes from preview_agent.py -- if preview_agent itself is down, it can't report that fact, only the supervisor watching it from outside can).""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") blob = json.dumps([s.model_dump() for s in payload.services]) now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row cur = await db.execute( "UPDATE devices SET supervisor_status = ?, supervisor_status_at = ? WHERE id = ?", (blob, now, device_id), ) await db.commit() if cur.rowcount == 0: raise HTTPException(404, "Device not found") # Piggyback any pending remote-control commands (operator clicked # "restart" on a service) onto this same request/response cycle -- # the robot already calls this every 10s, no separate poll needed. # Consumed immediately: read then delete, so a command fires once. # Backward-compat: rows may or may not have a `kind` column on # older DBs. Default to the original "supervisor_action" kind # (treated as restart / start / stop on the service) when missing. pending = [] try: async with db.execute( "SELECT service_name, action, kind, params, id FROM shell_requests " "WHERE device_id = ? AND completed_at IS NULL AND kind != 'speaker_test' " "ORDER BY requested_at", (device_id,), ) as c: shell_rows = [dict(r) for r in await c.fetchall()] for r in shell_rows: pending.append({ "service_name": r.get("service_name") or "", "action": "run", "kind": r.get("kind") or "shell_run", "params": json.loads(r["params"]) if r.get("params") else {}, "id": r["id"], }) except Exception: shell_rows = [] try: async with db.execute( "SELECT service_name, action FROM supervisor_commands WHERE device_id = ?", (device_id,) ) as c: for r in await c.fetchall(): pending.append({"service_name": r["service_name"], "action": r["action"], "kind": "supervisor_action"}) except Exception: pass if pending: # Mark shell requests as "in flight" so the dashboard can # distinguish "queued" from "still queued" (the row is read # but not yet completed). supervisor_commands rows are # one-shot and are deleted to mirror the previous behavior. try: ids = [p["id"] for p in pending if p.get("id")] if ids: placeholders = ",".join("?" for _ in ids) await db.execute( f"UPDATE shell_requests SET requested_at = requested_at " f"WHERE id IN ({placeholders})", ids, ) except Exception: pass try: await db.execute("DELETE FROM supervisor_commands WHERE device_id = ?", (device_id,)) except Exception: pass await db.commit() await broadcast("supervisor_status", {"device_id": device_id, "services": blob}) return {"status": "ok", "commands": pending} @app.post("/api/devices/{device_id}/services/{service_name}/{action}") async def control_device_service(device_id: str, service_name: str, action: str): """Queue an allowlisted service action for one enrolled robot.""" if action not in {"start", "stop", "restart"}: raise HTTPException(422, "Service action must be start, stop, or restart") if not re.fullmatch(r"(?:runtime|mesh|vision|preview|conversation|motor|screen|robopark-[a-z0-9-]+\.service)", service_name): raise HTTPException(422, "Service is not in the RoboPark control allowlist") now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: cur = await db.execute("SELECT 1 FROM devices WHERE id = ?", (device_id,)) if not await cur.fetchone(): raise HTTPException(404, "Device not found") await db.execute( """INSERT INTO supervisor_commands (device_id, service_name, action, requested_at) VALUES (?, ?, ?, ?) ON CONFLICT(device_id, service_name) DO UPDATE SET action=excluded.action, requested_at=excluded.requested_at""", (device_id, service_name, action, now), ) await db.commit() await _log_history("device", "device", device_id, f"service_{action}_queued", "operator", f"service={service_name}") return {"status": "queued", "service": service_name, "action": action} # ── Operator shell (C1) ── # Endpoints the dashboard calls to tail logs and run a small allowlisted # set of diagnostic commands on a specific robot. Commands are queued in # the `shell_requests` table; the robot's supervisor picks them up on its # next status-report poll, runs them, and POSTs results back to # /api/devices/{id}/supervisor-output. The dashboard polls # /api/robots/{id}/shell/result/{request_id} with an 8s budget. class ShellRequest(BaseModel): kind: str # "tail_logs" | "shell_run" service: Optional[str] = None params: Optional[dict] = None async def _queue_shell_request_async(device_id: str, kind: str, service: Optional[str], params: dict) -> str: """Insert a shell request row and return its id. Picked up by the robot's supervisor on its next status-report poll.""" request_id = secrets.token_urlsafe(12) now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO shell_requests (id, device_id, service_name, kind, params, requested_at) VALUES (?, ?, ?, ?, ?, ?)""", (request_id, device_id, service, kind, json.dumps(params or {}), now), ) await db.commit() return request_id @app.post("/api/robots/{robot_id}/shell/tail") async def robot_shell_tail(robot_id: str, payload: ShellRequest): """Queue a tail-logs request for a robot. Returns a request_id the dashboard polls to get the result. robot_id == device_id in this codebase (the dashboard's "robot" id is the same value as /api/devices rows).""" device_id = robot_id params = dict(payload.params or {}) if "lines" not in params: params["lines"] = 200 request_id = await _queue_shell_request_async(device_id, "tail_logs", payload.service, params) return {"status": "queued", "request_id": request_id, "device_id": device_id} @app.post("/api/robots/{robot_id}/shell/exec") async def robot_shell_exec(robot_id: str, payload: ShellRequest): """Queue a shell-run request for a robot. Returns a request_id the dashboard polls to get the result.""" device_id = robot_id params = dict(payload.params or {}) if not params.get("name"): raise HTTPException(400, "params.name is required (e.g. 'uptime', 'free', 'ps', 'tail')") request_id = await _queue_shell_request_async(device_id, "shell_run", payload.service, params) return {"status": "queued", "request_id": request_id, "device_id": device_id} @app.get("/api/robots/{robot_id}/shell/result/{request_id}") async def robot_shell_result(robot_id: str, request_id: str): """Return the result of a previously-queued shell request. The dashboard polls this with an 8s budget; we reply as soon as the robot has reported the result back.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM shell_requests WHERE id = ?", (request_id,)) as c: r = await c.fetchone() if not r or r["device_id"] != robot_id: raise HTTPException(404, "No such shell request for this robot") out = { "request_id": r["id"], "device_id": r["device_id"], "service": r["service_name"], "kind": r["kind"], "params": json.loads(r["params"]) if r["params"] else {}, "requested_at": r["requested_at"], "completed": r["completed_at"] is not None, "completed_at": r["completed_at"], "result": None, } if r["completed_at"] is not None: try: out["result"] = json.loads(r["result_payload"]) if r["result_payload"] else {"ok": bool(r["result_ok"])} except Exception: out["result"] = {"ok": bool(r["result_ok"]), "error": "(result payload not parseable)"} return out # Long-poll variant: holds the connection open until the result arrives # (or the 8s budget elapses), so the dashboard can wait without polling # in a tight loop. @app.get("/api/robots/{robot_id}/shell/wait/{request_id}") async def robot_shell_wait(robot_id: str, request_id: str, timeout: float = 8.0): deadline = asyncio.get_event_loop().time() + max(0.5, min(15.0, timeout)) while True: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM shell_requests WHERE id = ?", (request_id,)) as c: r = await c.fetchone() if not r or r["device_id"] != robot_id: raise HTTPException(404, "No such shell request for this robot") if r["completed_at"] is not None: try: payload = json.loads(r["result_payload"]) if r["result_payload"] else {"ok": bool(r["result_ok"])} except Exception: payload = {"ok": bool(r["result_ok"]), "error": "(result payload not parseable)"} return {"request_id": request_id, "completed": True, "result": payload} if asyncio.get_event_loop().time() >= deadline: return {"request_id": request_id, "completed": False, "timed_out": True} await asyncio.sleep(0.4) @app.post("/api/devices/{device_id}/supervisor-output") async def device_supervisor_output(device_id: str, payload: dict, authorization: Optional[str] = Header(default=None)): """Robot supervisor posts back the result of a tail_logs or shell_run request. We just record it on the shell_requests row; the dashboard picks it up via /api/robots/{id}/shell/result/{request_id}.""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") request_id = payload.get("request_id") if not request_id: raise HTTPException(400, "request_id is required") now = datetime.utcnow().isoformat() result_payload = payload.get("payload") or {} ok = 1 if result_payload.get("ok") else 0 async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT device_id FROM shell_requests WHERE id = ?", (request_id,)) as c: r = await c.fetchone() if not r: raise HTTPException(404, "No such shell request") if r["device_id"] != device_id: raise HTTPException(403, "request_id does not belong to this device") await db.execute( """UPDATE shell_requests SET completed_at = ?, result_ok = ?, result_payload = ? WHERE id = ?""", (now, ok, json.dumps(result_payload), request_id), ) await db.commit() kind = payload.get("kind") or "shell" service = payload.get("service") or "" await _log_history("device", "device", device_id, f"shell_{kind}_completed", "operator", f"service={service}, request_id={request_id}, ok={bool(ok)}") if kind == "motor_sequence" and result_payload.get("session_id"): await _insert_pipeline_event(device_id, PipelineEventPayload( stage="motor_sequence", status="ok" if ok else "failed", message=(f"Motor sequence {result_payload.get('sequence_id')} completed" if ok else f"Motor sequence failed: {result_payload.get('error', 'unknown error')}")[:500], session_id=result_payload.get("session_id"), source="robot_supervisor", details={"request_id": request_id, "duration_ms": result_payload.get("duration_ms")}, )) if kind == "motor_discover" and ok and isinstance(result_payload.get("registry"), list): device = await _resolve_device_by_id_or_name(device_id) model = _device_row_to_model(device) discovered_ids = { str(item.get("id")) for item in result_payload["registry"] if isinstance(item, dict) } reconciled_sequences = [] for sequence in model.motor_sequences: clean_steps = [ step for step in (sequence.get("steps") or []) if str(step.get("motor_id")) in discovered_ids ] reconciled_sequences.append({**sequence, "steps": clean_steps}) registry, sequences, greeting = _sanitize_motor_profile(MotorProfilePayload( registry=result_payload["registry"], sequences=reconciled_sequences, greeting_sequence_id=model.greeting_motor_sequence_id, )) result_payload.update({ "registry": registry, "sequences": sequences, "greeting_sequence_id": greeting, "persisted": True, }) async with aiosqlite.connect(DB_PATH) as db: await db.execute( "UPDATE devices SET motor_registry=?, motor_sequences=?, greeting_motor_sequence_id=? WHERE id=?", (json.dumps(registry), json.dumps(sequences), greeting, device_id), ) await db.execute( "UPDATE shell_requests SET result_payload=? WHERE id=?", (json.dumps(result_payload), request_id), ) await db.commit() character_id = await _sync_character_motor_sequences(device_id, sequences) result_payload["character_id"] = character_id result_payload["allowed_motor_sequences"] = [ item["id"] for item in sequences if item.get("steps") ] await _log_history( "motor", "device", device_id, "registry_discovered", "robot_supervisor", f"motors={len(registry)}", ) return {"status": "ok"} # Endpoint the dashboard uses to list the available shell commands # (allowlist) so the UI can populate a picker instead of typing names. # The list mirrors the SHELL_ALLOWLIST in robot_supervisor.py — that # file is the authoritative source for what's actually allowed at # run time. Keep these two lists in sync when adding/removing commands. SHELL_DASHBOARD_ALLOWLIST = [ {"name": "uptime", "argv": ["uptime"]}, {"name": "free", "argv": ["free", "-m"]}, {"name": "df", "argv": ["df", "-h"]}, {"name": "ps", "argv": ["ps", "aux"]}, {"name": "uname", "argv": ["uname", "-a"]}, {"name": "date", "argv": ["date"]}, {"name": "whoami", "argv": ["whoami"]}, {"name": "hostname", "argv": ["hostname"]}, {"name": "ip", "argv": ["ip", "addr"]}, {"name": "ss", "argv": ["ss", "-tlnp"]}, {"name": "os-release", "argv": ["cat", "/etc/os-release"]}, {"name": "ls-logs", "argv": ["ls", "-la", ""]}, {"name": "systemctl-status", "argv": ["systemctl", "status", "", "--no-pager", "-n", "30"]}, {"name": "journalctl", "argv": ["journalctl", "-u", "", "-n", "200", "--no-pager"]}, {"name": "tail", "argv": ["tail", "-n", "", ""]}, ] SHELL_DASHBOARD_OUTPUT_MAX = 64 * 1024 SHELL_DASHBOARD_TIMEOUT = 8.0 @app.get("/api/shell/allowlist") async def shell_allowlist(): """Return the list of shell command names the robot's supervisor supports. The dashboard uses this to populate the picker; the robot side does the actual allowlist enforcement at run time.""" return { "commands": [ {"name": s["name"], "argv": s["argv"]} for s in SHELL_DASHBOARD_ALLOWLIST ], "max_output_bytes": SHELL_DASHBOARD_OUTPUT_MAX, "timeout_seconds": SHELL_DASHBOARD_TIMEOUT, } # ── Speaker roundtrip test (C2) ── # The dashboard's "Test speaker" button hits this. We queue a # speaker_test request on the robot's shell_requests table; the # robot's next status-report poll runs the round-trip (play tone, # record mic) and POSTs the result back via /supervisor-output. # The dashboard polls the same /shell/wait endpoint used for the # other shell commands (the supervisor treats speaker_test as # just another request kind). class SpeakerTestRequest(BaseModel): mode: str = "tone" text: Optional[str] = None frequency: Optional[float] = None duration: Optional[float] = None amplitude: Optional[float] = None threshold_db: Optional[float] = None output: Optional[str] = None input: Optional[str] = None output_name: Optional[str] = None input_name: Optional[str] = None sample_rate: Optional[int] = None playback_only: Optional[bool] = None async def _cached_speaker_tts(robot_id: str, text: str) -> dict: """Generate configured ElevenLabs PCM once, then reuse it for device tests.""" phrase = " ".join((text or "Hello, this is a RoboPark speaker test.").split())[:240] config = await _resolve_robot_voice_config(robot_id) stack = config.voice_stack if config else None provider = (stack.tts_provider if stack else "elevenlabs").lower() voice_id = stack.tts_voice if stack else "21m00Tcm4TlvDq8ikWAM" if provider != "elevenlabs": raise HTTPException(400, f"Cached voice test currently requires ElevenLabs; selected stack uses {provider}") elevenlabs_key = await _get_elevenlabs_api_key() if not elevenlabs_key: raise HTTPException(503, "ElevenLabs is not configured; open Control Center > Setup > Voice provider secret") if not voice_id: raise HTTPException(400, "Selected voice stack has no ElevenLabs voice id") cache_dir = os.path.join(os.path.dirname(DB_PATH), "tts-test-cache") os.makedirs(cache_dir, exist_ok=True) cache_key = hashlib.sha256(f"{provider}|{voice_id}|{phrase}".encode("utf8")).hexdigest() cache_path = os.path.join(cache_dir, cache_key + ".pcm") cache_hit = os.path.isfile(cache_path) if cache_hit: with open(cache_path, "rb") as handle: pcm = handle.read() else: async with httpx.AsyncClient(timeout=20.0) as client: response = await client.post( f"{ELEVENLABS_BASE_URL}/v1/text-to-speech/{voice_id}", params={"output_format": "pcm_24000"}, headers={"xi-api-key": elevenlabs_key, "Accept": "audio/pcm"}, json={"text": phrase, "model_id": "eleven_flash_v2_5"}, ) if response.status_code != 200: detail = response.text[:240] or "no response body" raise HTTPException(502, f"ElevenLabs speaker test failed ({response.status_code}): {detail}") pcm = response.content if len(pcm) < 480: raise HTTPException(502, "ElevenLabs returned empty speaker-test audio") temp_path = cache_path + ".tmp" with open(temp_path, "wb") as handle: handle.write(pcm) os.replace(temp_path, cache_path) return { "mode": "tts", "text": phrase, "tts_provider": provider, "tts_voice": voice_id, "sample_rate": 24000, "audio_pcm_base64": base64.b64encode(pcm).decode("ascii"), "cache_hit": cache_hit, } @app.post("/api/robots/{robot_id}/shell/speaker-test") async def robot_speaker_test(robot_id: str, payload: SpeakerTestRequest): """Queue a speaker + mic round-trip test on the robot. Returns a request_id; poll /api/robots/{id}/shell/wait/{rid} for the result.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT id, audio_device, audio_output_device, device_inventory FROM devices " "WHERE id = ? OR lower(name) = lower(?) " "ORDER BY CASE WHEN id = ? THEN 0 ELSE 1 END LIMIT 1", (robot_id, robot_id, robot_id), ) as cursor: device = await cursor.fetchone() if not device: raise HTTPException(404, f"Robot device not found: {robot_id}") device_id = device["id"] params = payload.model_dump(exclude_none=True) # Scheduler persistence is authoritative. Dashboard state can be stale # while heartbeat polling rebuilds the drawer, so never let it silently # route a test to a different device than the one the operator saved. if device["audio_device"] not in (None, ""): params["input"] = str(device["audio_device"]) if device["audio_output_device"] not in (None, ""): params["output"] = str(device["audio_output_device"]) try: inventory = json.loads(device["device_inventory"] or "{}") except (TypeError, ValueError): inventory = {} for field, inventory_key, name_key in ( ("input", "audio_input", "input_name"), ("output", "audio_output", "output_name"), ): selected = str(params.get(field, "")) for item in inventory.get(inventory_key, []) or []: if str(item.get("id", "")) == selected: params[name_key] = str(item.get("name") or selected) break if payload.mode == "tts": # Cached voice is a speaker test. Do not also seize the microphone, # which may be owned by the live preview/conversation publisher. params.setdefault("playback_only", True) params.update(await _cached_speaker_tts(robot_id, payload.text or "")) elif payload.mode != "tone": raise HTTPException(422, "mode must be tone or tts") request_id = await _queue_shell_request_async(device_id, "speaker_test", None, params) return {"status": "queued", "request_id": request_id, "device_id": device_id} @app.get("/api/devices/{device_id}/shell/next-speaker-test") async def next_device_speaker_test(device_id: str, authorization: Optional[str] = Header(default=None)): """Allow the always-running preview agent to execute audio tests.""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") now = datetime.utcnow() stale_before = (now - timedelta(seconds=45)).isoformat() async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row await db.execute("BEGIN IMMEDIATE") async with db.execute( "SELECT id, params FROM shell_requests WHERE device_id = ? AND kind = 'speaker_test' " "AND completed_at IS NULL AND (claimed_at IS NULL OR claimed_at < ?) " "ORDER BY requested_at LIMIT 1", (device_id, stale_before), ) as cursor: row = await cursor.fetchone() if row: claimed = await db.execute( "UPDATE shell_requests SET claimed_at = ? WHERE id = ? AND completed_at IS NULL " "AND (claimed_at IS NULL OR claimed_at < ?)", (now.isoformat(), row["id"], stale_before), ) if claimed.rowcount != 1: row = None await db.commit() return {"request": {"id": row["id"], "params": json.loads(row["params"] or "{}")} if row else None} # ── Incident acknowledgement (C4) ── # The dashboard's Incidents sub-tab lets an operator acknowledge an # active incident (a robot stuck in stuck_session, server_unavailable, # etc). The acknowledgement is stored client-side (localStorage) so # each operator on their own browser gets their own list; this # endpoint is the "central log" mirror so the ack also shows up in # the audit history with the operator's name (via X-Operator). class IncidentAck(BaseModel): robot_id: str code: str message: Optional[str] = None @app.post("/api/incidents/ack") async def incidents_ack(payload: IncidentAck, x_operator: Optional[str] = Header(default=None, alias="X-Operator")): """Log an operator ack of an active incident. The dashboard stores its own list; this endpoint mirrors the ack in the audit log so the action is visible centrally with the operator's name.""" actor = _sanitize_actor_name(x_operator) or 'operator' await _log_history("incident", "robot", payload.robot_id, "acked", actor, f"code={payload.code}, message={payload.message or ''}") return {"status": "ok", "robot_id": payload.robot_id, "code": payload.code, "by": actor, "at": datetime.utcnow().isoformat()} @app.get("/api/devices/{device_id}/config") async def device_config(device_id: str, authorization: Optional[str] = Header(default=None)): """Pi polls this to learn current production_mode + assigned character, etc.""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid device token") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT id, name, character_id, motor_server_url, livekit_url, video_device, audio_device, audio_output_device, device_inventory, greeting_phrases, motor_registry, motor_sequences, greeting_motor_sequence_id FROM devices WHERE id = ?", (device_id,), ) as c: row = await c.fetchone() if not row: raise HTTPException(404, "Device not found") effective_pm = await _effective_device_production_mode(db, device_id) return { "device_id": row["id"], "name": row["name"], "character_id": row["character_id"], "motor_server_url": row["motor_server_url"], "livekit_url": row["livekit_url"], "video_device": row["video_device"], "audio_device": row["audio_device"], "audio_output_device": row["audio_output_device"], "greeting_phrases": _device_row_to_model(row).greeting_phrases, "motor_registry": _device_row_to_model(row).motor_registry, "motor_sequences": _device_row_to_model(row).motor_sequences, "greeting_motor_sequence_id": row["greeting_motor_sequence_id"], "device_inventory": json.loads(row["device_inventory"]) if row["device_inventory"] else None, "production_mode": effective_pm, } class MotorProfilePayload(BaseModel): registry: List[dict] = [] sequences: List[dict] = [] greeting_sequence_id: Optional[str] = None async def _resolve_device_by_id_or_name(robot_id: str): async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( """SELECT * FROM devices WHERE id=? OR lower(name)=lower(?) ORDER BY CASE WHEN id=? THEN 0 ELSE 1 END, CASE lower(status) WHEN 'online' THEN 0 WHEN 'running' THEN 1 WHEN 'detecting' THEN 2 WHEN 'connecting' THEN 3 ELSE 4 END, last_heartbeat DESC, enrolled_at DESC, created_at DESC LIMIT 1""", (robot_id, robot_id, robot_id), ) as cursor: device = await cursor.fetchone() if not device: raise HTTPException(404, "Robot device not found") return device def _sanitize_motor_profile(payload: MotorProfilePayload) -> tuple[list, list, Optional[str]]: import re registry, ids, gpios = [], set(), set() for raw in payload.registry[:32]: motor_id = str(raw.get("id") or raw.get("name") or "").strip().lower() if not re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,31}", motor_id): raise HTTPException(422, f"Invalid motor id: {motor_id!r}") gpio = int(raw.get("gpio", -1)) if gpio < 2 or gpio > 27: raise HTTPException(422, f"GPIO {gpio} is outside BCM 2..27") if motor_id in ids or gpio in gpios: raise HTTPException(422, "Motor ids and GPIO assignments must be unique") ids.add(motor_id); gpios.add(gpio) registry.append({"id": motor_id, "name": str(raw.get("name") or motor_id)[:60], "gpio": gpio, "active_high": bool(raw.get("active_high", True)), "max_duration_ms": max(50, min(10000, int(raw.get("max_duration_ms", 3000))))}) sequences, sequence_ids = [], set() for raw in payload.sequences[:32]: sequence_id = str(raw.get("id") or raw.get("name") or "").strip().lower() if not re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,31}", sequence_id) or sequence_id in sequence_ids: raise HTTPException(422, f"Invalid or duplicate sequence id: {sequence_id!r}") sequence_ids.add(sequence_id) steps = [] total_duration_ms = 0 for step in (raw.get("steps") or [])[:40]: motor_id = str(step.get("motor_id") or "") if motor_id not in ids: raise HTTPException(422, f"Sequence {sequence_id} references unknown motor {motor_id}") motor = next(item for item in registry if item["id"] == motor_id) duration = max(50, min(motor["max_duration_ms"], int(step.get("duration_ms", 500)))) delay = max(0, min(30000, int(step.get("delay_ms", 0)))) total_duration_ms += delay + duration if total_duration_ms > 60000: raise HTTPException(422, f"Sequence {sequence_id} exceeds the 60 second safety limit") steps.append({"motor_id": motor_id, "delay_ms": delay, "duration_ms": duration}) sequences.append({"id": sequence_id, "name": str(raw.get("name") or sequence_id)[:80], "steps": steps}) greeting = payload.greeting_sequence_id or None if greeting and greeting not in sequence_ids: raise HTTPException(422, "Greeting sequence does not exist") return registry, sequences, greeting async def _sync_character_motor_sequences(device_id: str, sequences: list) -> Optional[str]: """Point the assigned character at this robot's validated sequence IDs.""" sequence_ids = [str(item["id"]) for item in sequences if item.get("id") and item.get("steps")] async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT COALESCE(rvc.character_preset_id, d.character_id) AS character_id " "FROM devices d LEFT JOIN robot_voice_configs rvc ON rvc.robot_id=d.id WHERE d.id=?", (device_id,), ) as cursor: row = await cursor.fetchone() character_id = row["character_id"] if row else None if character_id: await db.execute( "UPDATE character_presets SET motors=?, updated_at=? WHERE id=?", (json.dumps(sequence_ids), datetime.utcnow().isoformat(), character_id), ) await db.commit() return character_id @app.get("/api/robots/{robot_id}/motor-profile") async def robot_motor_profile(robot_id: str): device = await _resolve_device_by_id_or_name(robot_id) model = _device_row_to_model(device) return {"robot_id": model.id, "motor_server_url": model.motor_server_url, "registry": model.motor_registry, "sequences": model.motor_sequences, "greeting_sequence_id": model.greeting_motor_sequence_id} @app.put("/api/robots/{robot_id}/motor-profile") async def save_robot_motor_profile(robot_id: str, payload: MotorProfilePayload): device = await _resolve_device_by_id_or_name(robot_id) registry, sequences, greeting = _sanitize_motor_profile(payload) async with aiosqlite.connect(DB_PATH) as db: await db.execute("UPDATE devices SET motor_registry=?, motor_sequences=?, greeting_motor_sequence_id=? WHERE id=?", (json.dumps(registry), json.dumps(sequences), greeting, device["id"])) await db.commit() character_id = await _sync_character_motor_sequences(device["id"], sequences) await _log_history("motor", "device", device["id"], "profile_saved", "operator", f"motors={len(registry)}, sequences={len(sequences)}") return {"status": "ok", "robot_id": device["id"], "registry": registry, "sequences": sequences, "greeting_sequence_id": greeting, "character_id": character_id, "allowed_motor_sequences": [item["id"] for item in sequences if item.get("steps")]} async def _queue_motor_sequence(device_id: str, sequence_id: str, session_id: Optional[str] = None) -> dict: device = await _resolve_device_by_id_or_name(device_id) model = _device_row_to_model(device) sequence = next((item for item in model.motor_sequences if item.get("id") == sequence_id), None) if not sequence: raise HTTPException(404, "Motor sequence not found") if not sequence.get("steps"): raise HTTPException(409, "Motor sequence has no steps") if not model.motor_server_url: raise HTTPException(409, "Robot has no motor_server_url") request_id = await _queue_shell_request_async(model.id, "motor_sequence", "motor_server", { "motor_server_url": model.motor_server_url, "sequence_id": sequence_id, "steps": sequence.get("steps") or [], "registry": model.motor_registry, "session_id": session_id, }) if session_id: await _insert_pipeline_event(model.id, PipelineEventPayload(stage="motor_sequence", status="running", message=f"Motor sequence {sequence_id} queued", session_id=session_id, source="scheduler")) return {"queued": True, "request_id": request_id, "device_id": model.id, "sequence_id": sequence_id} async def _queue_configured_greeting_sequence(device_id: str, session_id: str) -> None: try: device = await _resolve_device_by_id_or_name(device_id) sequence_id = device["greeting_motor_sequence_id"] async with aiosqlite.connect(DB_PATH) as db: async with db.execute( "SELECT 1 FROM pipeline_events WHERE robot_id=? AND session_id=? " "AND stage='motor_sequence' LIMIT 1", (device["id"], session_id), ) as cursor: if await cursor.fetchone(): return if sequence_id: await _queue_motor_sequence(device["id"], sequence_id, session_id) else: await _insert_pipeline_event(device["id"], PipelineEventPayload( stage="motor_sequence", status="skipped", message="No greeting motor sequence assigned", session_id=session_id, source="scheduler", )) except Exception as exc: logger.warning("Greeting motor sequence was not queued for %s: %s", device_id, exc) @app.post("/api/robots/{robot_id}/motor-sequences/{sequence_id}/run") async def run_robot_motor_sequence(robot_id: str, sequence_id: str): return await _queue_motor_sequence(robot_id, sequence_id) @app.post("/api/devices/by-room/{room_name}/motor-sequences/{sequence_id}/run") async def run_room_motor_sequence( room_name: str, sequence_id: str, authorization: Optional[str] = Header(default=None), agent_token: Optional[str] = Header(default=None, alias="X-RoboPark-Agent-Token"), ): """Allow the active voice worker to invoke a saved robot sequence by room.""" fleet_authorized = await _authorize_fleet(authorization) agent_authorized = bool( ROBOPARK_AGENT_TOKEN and agent_token and hmac.compare_digest(agent_token.strip(), ROBOPARK_AGENT_TOKEN) ) if not fleet_authorized and not agent_authorized: raise HTTPException(401, "Invalid or missing agent token") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT id, robot_id FROM sessions WHERE room_name=? AND ended_at IS NULL " "ORDER BY started_at DESC LIMIT 1", (room_name,), ) as cursor: session = await cursor.fetchone() if not session: raise HTTPException(404, "No active session bound to this room") if agent_authorized and not fleet_authorized: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT COALESCE(rvc.character_preset_id, d.character_id) AS character_id " "FROM devices d LEFT JOIN robot_voice_configs rvc ON rvc.robot_id=d.id WHERE d.id=?", (session["robot_id"],), ) as cursor: character_row = await cursor.fetchone() character_id = character_row["character_id"] if character_row else None allowed = [] if character_id: async with db.execute("SELECT motors FROM character_presets WHERE id=?", (character_id,)) as cursor: preset = await cursor.fetchone() if preset and preset["motors"]: try: allowed = [str(item) for item in json.loads(preset["motors"]) if isinstance(item, str)] except Exception: allowed = [] if sequence_id not in allowed: raise HTTPException( 403, f"Character {character_id or 'unassigned'} is not allowed to run motor sequence {sequence_id}", ) return await _queue_motor_sequence(session["robot_id"], sequence_id, session["id"]) @app.post("/api/robots/{robot_id}/motors/stop") async def stop_robot_motors(robot_id: str): device = await _resolve_device_by_id_or_name(robot_id) model = _device_row_to_model(device) if not model.motor_server_url: raise HTTPException(409, "Robot has no motor_server_url") request_id = await _queue_shell_request_async(model.id, "motor_sequence", "motor_server", { "motor_server_url": model.motor_server_url, "stop_all": True, }) await _log_history("motor", "device", model.id, "emergency_stop_queued", "operator", request_id) return {"queued": True, "request_id": request_id, "device_id": model.id} @app.post("/api/robots/{robot_id}/motors/{motor_id}/test") async def test_robot_motor(robot_id: str, motor_id: str, duration_ms: int = 300): device = await _resolve_device_by_id_or_name(robot_id) model = _device_row_to_model(device) motor = next((item for item in model.motor_registry if item.get("id") == motor_id), None) if not motor or not model.motor_server_url: raise HTTPException(404, "Motor or motor server not configured") duration_ms = max(50, min(int(motor.get("max_duration_ms", 3000)), duration_ms)) request_id = await _queue_shell_request_async(model.id, "motor_sequence", "motor_server", { "motor_server_url": model.motor_server_url, "sequence_id": f"test-{motor_id}", "registry": model.motor_registry, "steps": [{"motor_id": motor_id, "delay_ms": 0, "duration_ms": duration_ms}], }) return {"queued": True, "request_id": request_id, "device_id": model.id} @app.post("/api/robots/{robot_id}/motors/discover") async def discover_robot_motors(robot_id: str): """Import relays already registered by the robot-local motor server. This is read-only on the robot. It never scans or pulses unregistered GPIO. The dashboard validates and persists the returned registry before testing. """ device = await _resolve_device_by_id_or_name(robot_id) model = _device_row_to_model(device) if not model.motor_server_url: raise HTTPException(409, "Robot has no motor_server_url") request_id = await _queue_shell_request_async(model.id, "motor_discover", "motor_server", { "motor_server_url": model.motor_server_url, }) return {"queued": True, "request_id": request_id, "device_id": model.id} @app.post("/api/robots/{robot_id}/motors/test-all") async def test_all_robot_motors(robot_id: str, duration_ms: int = 300, pause_ms: int = 200): device = await _resolve_device_by_id_or_name(robot_id) model = _device_row_to_model(device) if not model.motor_registry or not model.motor_server_url: raise HTTPException(409, "Robot motor registry or motor server is not configured") duration_ms = max(50, min(1000, int(duration_ms))) pause_ms = max(0, min(2000, int(pause_ms))) request_id = await _queue_shell_request_async(model.id, "motor_sequence", "motor_server", { "motor_server_url": model.motor_server_url, "sequence_id": "registered-gpio-test", "registry": model.motor_registry, "steps": [], "test_all": True, "duration_ms": duration_ms, "pause_ms": pause_ms, }) return {"queued": True, "request_id": request_id, "device_id": model.id, "motor_count": len(model.motor_registry)} @app.post("/api/devices/{device_id}/request-session") async def device_request_session(device_id: str, authorization: Optional[str] = Header(default=None)): """Enrolled DEVICE requests a session (presence-triggered LiveKit join). This is the device-keyed sibling of POST /api/robots/{robot_id}/request-session. The pi-client is an enrolled *device*, but sessions + assignment are keyed by *robot*. Rather than link the two fleet models with a new schema, we MIRROR the device onto a robot row whose id == device_id (additive + idempotent via INSERT OR IGNORE). Everything downstream (least-loaded server selection, session creation, telemetry) then reuses the existing robot machinery unchanged. Auth: the specific enrolled-device Bearer token (_authorize_device), so a device can only request a session for itself. Response mirrors the robot path plus is safe to feed straight into the Pi's join path: {session_id, server_id, server_url, room_name, token}. A session row is created with started_at so latency (joined_at - started_at) computes once POST /api/sessions/{session_id}/joined is called. NOTE: unlike the robot path, this does NOT hard-fail when the mirrored robot is not 'idle'. Presence can re-trigger while a prior session's end-session was never delivered; blocking would strand the device. Each call creates a fresh session and bumps trigger_count (the natural camera-trigger point).""" if not await _authorize_device(device_id, authorization): raise HTTPException(401, "Invalid or missing device token") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row # Resolve the enrolled device. async with db.execute("SELECT * FROM devices WHERE id = ?", (device_id,)) as c: device = await c.fetchone() if not device: raise HTTPException(404, "Device not found") if not await get_production_mode(): raise HTTPException(403, "production_mode_off") if not bool(device["production_mode"]): raise HTTPException(403, "device_production_mode_off") # Mirror device -> robot identity (id == device_id). Additive + idempotent: # if the robot row already exists it is left untouched. # OR IGNORE: only fires the first time this robots row is created. # PATCH /api/devices/{device_id} now propagates renames into # robots.name afterwards, so this is purely a first-creation # concern, not an ongoing sync path. await db.execute( "INSERT OR IGNORE INTO robots (id, name, character_id) VALUES (?, ?, ?)", (device_id, device["name"] or device_id, device["character_id"]), ) # A robot can only physically be in one room at a time -- it always # disconnects from its previous room before joining a new one (see # preview_agent.py's _apply_state). But this endpoint deliberately # never hard-fails on a prior un-ended session (see docstring), so # without this, repeated presence-triggers (e.g. continuous motion # detection) each leave the previous session's row with # ended_at IS NULL forever -- permanently consuming a slot against # the server's max_sessions capacity until every slot leaks away and # every future request-session 503s. Close them out here instead. await db.execute( """UPDATE sessions SET ended_at = ?, end_reason = 'superseded' WHERE robot_id = ? AND ended_at IS NULL""", (datetime.utcnow().isoformat(), device_id), ) # Find least-loaded online server (identical selection to the robot path). async with db.execute(""" SELECT s.*, (SELECT COUNT(*) FROM sessions WHERE server_id = s.id AND ended_at IS NULL) as active FROM livekit_servers s WHERE s.status = 'online' ORDER BY active ASC LIMIT 1 """) as cursor: server = await cursor.fetchone() if not server: raise HTTPException(503, "No available servers") if server["active"] >= server["max_sessions"]: raise HTTPException(503, "All servers at capacity") ts = int(datetime.utcnow().timestamp()) session_id = f"session_{device_id}_{ts}" # Room name MUST start with "robopark-" — the literal prefix # voice_agent.py checks to enable production-mode (character/voice/ # motor resolution via GET /api/devices/by-room, unlimited idle # timeout instead of the 30s default). room_name = f"robopark-{device_id}-{ts}" _now_iso = datetime.utcnow().isoformat() await db.execute(""" INSERT INTO sessions (id, robot_id, server_id, room_name, started_at, last_activity_at) VALUES (?, ?, ?, ?, ?, ?) """, (session_id, device_id, server["id"], room_name, _now_iso, _now_iso)) # Mark the mirrored robot connecting + count the camera trigger, exactly # as the robot path does (request-session is the natural trigger point). await db.execute(""" UPDATE robots SET status = 'connecting', current_session_id = ?, connected_server_id = ?, trigger_count = COALESCE(trigger_count, 0) + 1 WHERE id = ? """, (session_id, server["id"], device_id)) await db.commit() await broadcast("session_requested", { "robot_id": device_id, "device_id": device_id, "server_id": server["id"], "session_id": session_id, }) # Mint a short-lived room-scoped JWT so the device joins the assigned room # without ever receiving the raw api_key/api_secret. Identity keeps the # existing pi: convention so the agent side sees the same participant. try: from livekit.api import AccessToken, VideoGrants except ImportError: raise HTTPException(503, "livekit-api not installed on scheduler") lk_key = server["api_key"] or "devkey" lk_secret = server["api_secret"] or "secret" identity = f"pi:{device_id}" grants = VideoGrants( room=room_name, room_join=True, can_publish=True, can_subscribe=True, can_publish_data=True, ) token = ( AccessToken(lk_key, lk_secret) .with_identity(identity) .with_name(device["name"] or device_id) .with_ttl(timedelta(hours=1)) .with_grants(grants) .to_jwt() ) # Resolve the per-robot voice/character config so the ROBOVOICE agent and the # robot client can apply it for this session. The agent fetches it via # /api/devices/by-room/{room_name}/voice-config; we include it here too so # the robot client can log/adjust local settings if needed. voice_config = await _resolve_robot_voice_config(device_id) return { "session_id": session_id, "server_id": server["id"], "server_url": device["livekit_url"] or server["url"], "room_name": room_name, "token": token, "voice_config": voice_config.model_dump() if voice_config else None, } # ============================================================================= # CHARACTER ENDPOINTS (read from ROBOVOICE settings.json) # ============================================================================= import json as _json class CharacterSummary(BaseModel): id: str name: Optional[str] = None description: Optional[str] = None tts_voice: Optional[str] = None motors: List[str] = [] class Character(CharacterSummary): system_prompt: Optional[str] = None raw: dict = {} # Scheduler-managed character presets (production park characters + custom). class CharacterPreset(BaseModel): id: str name: str description: Optional[str] = None system_prompt: Optional[str] = None voice_stack_id: Optional[str] = None motors: List[str] = [] nx: float = 0.5 ny: float = 0.5 img: Optional[str] = None elevenlabs_agent_id: Optional[str] = None elevenlabs_branch_id: Optional[str] = None elevenlabs_first_message: Optional[str] = None elevenlabs_synced_at: Optional[datetime] = None default_engine: str = "elevenlabs" robovoice_profile_id: Optional[str] = None system_prompt_revision: int = 1 created_at: Optional[datetime] = None updated_at: Optional[datetime] = None class CharacterPresetCreate(BaseModel): id: Optional[str] = None name: str description: Optional[str] = None system_prompt: Optional[str] = None voice_stack_id: Optional[str] = None motors: List[str] = [] nx: Optional[float] = None ny: Optional[float] = None img: Optional[str] = None elevenlabs_agent_id: Optional[str] = None elevenlabs_branch_id: Optional[str] = None elevenlabs_first_message: Optional[str] = None default_engine: str = "elevenlabs" robovoice_profile_id: Optional[str] = None system_prompt_revision: int = 1 class ElevenLabsAgentSync(BaseModel): direction: str = "pull" class VoiceStack(BaseModel): id: str name: str stt_provider: str = "speaches" stt_model: str = "Systran/faster-whisper-small" stt_language: str = "en" llm_provider: str = "ollama" llm_model: str = "gemma3:27b" llm_api_key: Optional[str] = None temperature: float = 0.7 num_ctx: int = 8192 tts_provider: str = "elevenlabs" tts_voice: str = "21m00Tcm4TlvDq8ikWAM" tts_language: str = "en" allow_interruptions: bool = True min_endpointing_delay: float = 0.7 max_turns: int = 20 wake_word_enabled: bool = False wake_word_model: Optional[str] = None wake_word_threshold: float = 0.5 wake_word_timeout: float = 3.0 created_at: Optional[datetime] = None updated_at: Optional[datetime] = None class VoiceStackCreate(BaseModel): id: Optional[str] = None name: str stt_provider: str = "speaches" stt_model: str = "Systran/faster-whisper-small" stt_language: str = "en" llm_provider: str = "ollama" llm_model: str = "gemma3:27b" llm_api_key: Optional[str] = None temperature: float = 0.7 num_ctx: int = 8192 tts_provider: str = "elevenlabs" tts_voice: str = "21m00Tcm4TlvDq8ikWAM" tts_language: str = "en" allow_interruptions: bool = True min_endpointing_delay: float = 0.7 max_turns: int = 20 wake_word_enabled: bool = False wake_word_model: Optional[str] = None wake_word_threshold: float = 0.5 wake_word_timeout: float = 3.0 # Effective configuration returned to the ROBOVOICE agent for a given room/robot. class RobotVoiceConfig(BaseModel): character_id: Optional[str] = None character_name: Optional[str] = None system_prompt: Optional[str] = None voice_stack_id: Optional[str] = None voice_stack: Optional[VoiceStack] = None greeting_phrases: List[str] = [] allowed_motor_sequences: List[str] = [] def _load_robovoice_settings() -> dict: """Read the ROBOVOICE settings file. Returns {} if missing/invalid.""" try: if not os.path.exists(ROBOVOICE_SETTINGS_PATH): return {} with open(ROBOVOICE_SETTINGS_PATH, "r", encoding="utf-8") as f: return _json.load(f) except Exception as e: log.warning(f"Could not read ROBOVOICE settings at {ROBOVOICE_SETTINGS_PATH}: {e}") return {} def _character_summary(c: dict) -> CharacterSummary: motors = c.get("motors") or [] if isinstance(motors, list) and motors and isinstance(motors[0], dict): motor_names = [m.get("name") for m in motors if m.get("name")] else: motor_names = [m for m in motors if isinstance(m, str)] voice = ( c.get("tts_voice") or c.get("elevenlabs_voice_id") or c.get("tts_voice_kokoro") or c.get("tts_voice_piper") ) return CharacterSummary( id=c.get("id") or c.get("name") or "unknown", name=c.get("name"), description=c.get("description"), tts_voice=voice, motors=motor_names, ) @app.get("/api/characters", response_model=List[CharacterSummary]) async def list_characters(): """List all characters available in the bound ROBOVOICE settings file.""" s = _load_robovoice_settings() chars = s.get("characters") or [] return [_character_summary(c) for c in chars] @app.get("/api/characters/{character_id}", response_model=Character) async def get_character(character_id: str): """Get a single character (incl. system_prompt and motor config) by id.""" s = _load_robovoice_settings() for c in s.get("characters") or []: if c.get("id") == character_id: summary = _character_summary(c) return Character( **summary.model_dump(), system_prompt=c.get("system_prompt"), raw=c, ) raise HTTPException(404, f"Character '{character_id}' not found in {ROBOVOICE_SETTINGS_PATH}") # ── Scheduler-managed character presets + voice stacks ───────────────────── async def _resolve_voice_stack_for_robot(db, robot_id: str) -> dict: """Return the effective voice-stack dict for a robot/device id. Resolution order: 1. robot_voice_configs.voice_stack_id override 2. character_presets.voice_stack_id assigned to the robot 3. 'english-default' fallback """ # 1. robot_voice_configs row async with db.execute( "SELECT character_preset_id, voice_stack_id, override_config FROM robot_voice_configs WHERE robot_id = ?", (robot_id,), ) as c: rvc = await c.fetchone() preset_id = None voice_stack_id = None override = {} if rvc: preset_id = rvc["character_preset_id"] voice_stack_id = rvc["voice_stack_id"] if rvc["override_config"]: try: override = _json.loads(rvc["override_config"]) except Exception: pass # 2. If no explicit voice_stack, fall back via character preset if not voice_stack_id: if not preset_id: async with db.execute( "SELECT character_id FROM robots WHERE id = ? UNION ALL SELECT character_id FROM devices WHERE id = ?", (robot_id, robot_id), ) as c: row = await c.fetchone() if row: preset_id = row["character_id"] if preset_id: async with db.execute( "SELECT voice_stack_id FROM character_presets WHERE id = ?", (preset_id,) ) as c: row = await c.fetchone() if row: voice_stack_id = row["voice_stack_id"] if not voice_stack_id: voice_stack_id = "english-default" async with db.execute( "SELECT * FROM voice_stacks WHERE id = ?", (voice_stack_id,) ) as c: stack = await c.fetchone() if not stack: return {} result = dict(stack) result.update(override) # SQLite stores booleans as 0/1; normalize for key in ("allow_interruptions", "wake_word_enabled"): if key in result: result[key] = bool(result[key]) return result _ELEVENLABS_NAME_CACHE: tuple[float, list[dict]] = (0.0, []) async def _auto_elevenlabs_voice(robot_name: Optional[str], current: Optional[str]) -> Optional[str]: """Resolve a custom ElevenLabs voice by robot name without overwriting a manual choice.""" if current and current != "21m00Tcm4TlvDq8ikWAM": return current if not robot_name: return current normalized = "".join(ch for ch in robot_name.lower() if ch.isalnum()) try: mapping = json.loads(os.getenv("ROBOPARK_ELEVENLABS_VOICE_MAP", "{}")) for key, voice_id in mapping.items(): if "".join(ch for ch in str(key).lower() if ch.isalnum()) in (normalized, "".join(ch for ch in robot_name.lower() if ch.isalnum())): return str(voice_id) except Exception: pass global _ELEVENLABS_NAME_CACHE now = time.monotonic() voices = _ELEVENLABS_NAME_CACHE[1] if now - _ELEVENLABS_NAME_CACHE[0] > 300 and ELEVENLABS_API_KEY: try: async with httpx.AsyncClient(timeout=8.0) as client: response = await client.get( f"{ELEVENLABS_BASE_URL}/v1/voices", headers={"xi-api-key": ELEVENLABS_API_KEY, "Accept": "application/json"}, ) if response.status_code == 200: voices = response.json().get("voices") or [] _ELEVENLABS_NAME_CACHE = (now, voices) except Exception as e: logger.debug(f"ElevenLabs voice auto-assignment unavailable: {e}") aliases = {normalized} if normalized == "vixenbmw": aliases.add("vixen") for voice in voices: voice_name = "".join(ch for ch in str(voice.get("name", "")).lower() if ch.isalnum()) if voice_name in aliases or any(alias in voice_name for alias in aliases): return str(voice.get("voice_id") or current) return current async def _resolve_robot_voice_config(robot_id: str) -> RobotVoiceConfig: """Public helper to resolve the effective voice config for a robot id.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row # Dashboard routes commonly use the human robot name ("bmw"), while # assignments are persisted against the enrolled device id. Resolve # that alias once so character, stack and greetings use one identity. async with db.execute( "SELECT id FROM devices WHERE id = ? OR lower(name) = lower(?) " "ORDER BY CASE WHEN id = ? THEN 0 ELSE 1 END LIMIT 1", (robot_id, robot_id, robot_id), ) as c: canonical_device = await c.fetchone() canonical_id = canonical_device["id"] if canonical_device else robot_id stack_dict = await _resolve_voice_stack_for_robot(db, canonical_id) # Determine character preset + name character_id = None character_name = None system_prompt = None allowed_motor_sequences = [] async with db.execute( "SELECT character_preset_id FROM robot_voice_configs WHERE robot_id = ?", (canonical_id,) ) as c: rvc = await c.fetchone() if rvc and rvc["character_preset_id"]: character_id = rvc["character_preset_id"] else: async with db.execute( "SELECT character_id FROM robots WHERE id = ? UNION ALL SELECT character_id FROM devices WHERE id = ?", (canonical_id, canonical_id), ) as c: row = await c.fetchone() if row: character_id = row["character_id"] if character_id: async with db.execute( "SELECT name, description, system_prompt, motors FROM character_presets WHERE id = ?", (character_id,) ) as c: row = await c.fetchone() if row: character_name = row["name"] system_prompt = row["system_prompt"] try: allowed_motor_sequences = [ str(item) for item in _json.loads(row["motors"] or "[]") if isinstance(item, str) ] except Exception: allowed_motor_sequences = [] if not character_name: async with db.execute( "SELECT name FROM robots WHERE id = ? UNION ALL SELECT name FROM devices WHERE id = ?", (canonical_id, canonical_id), ) as c: row = await c.fetchone() if row: character_name = row["name"] async with db.execute("SELECT greeting_phrases FROM devices WHERE id = ?", (canonical_id,)) as c: device_row = await c.fetchone() greeting_phrases = [] if device_row and device_row["greeting_phrases"]: try: greeting_phrases = [str(p) for p in _json.loads(device_row["greeting_phrases"]) if str(p).strip()] except Exception: greeting_phrases = [] if not greeting_phrases and (character_name or character_id): greeting_phrases = [f"Hello, I am {character_name or character_id}. How can I help you?"] if stack_dict: stack_dict["tts_voice"] = await _auto_elevenlabs_voice(character_name or character_id, stack_dict.get("tts_voice")) voice_stack = VoiceStack(**stack_dict) if stack_dict else None return RobotVoiceConfig( character_id=character_id, character_name=character_name, system_prompt=system_prompt, voice_stack_id=voice_stack.id if voice_stack else None, voice_stack=voice_stack, greeting_phrases=greeting_phrases, allowed_motor_sequences=allowed_motor_sequences, ) async def _resolve_session_voice_config(session_row) -> RobotVoiceConfig: """Resolve the immutable character/stack snapshot selected for a web call.""" character_id = session_row["character_id"] stack_id = session_row["voice_stack_id"] if not character_id: return await _resolve_robot_voice_config(session_row["robot_id"]) async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT name, system_prompt, voice_stack_id FROM character_presets WHERE id = ?", (character_id,), ) as cursor: character = await cursor.fetchone() stack_id = stack_id or (character["voice_stack_id"] if character else None) stack = None if stack_id: async with db.execute("SELECT * FROM voice_stacks WHERE id = ?", (stack_id,)) as cursor: stack_row = await cursor.fetchone() if stack_row: stack_data = dict(stack_row) stack_data["tts_voice"] = await _auto_elevenlabs_voice( character["name"] if character else character_id, stack_data.get("tts_voice"), ) stack = VoiceStack(**stack_data) return RobotVoiceConfig( character_id=character_id, character_name=character["name"] if character else character_id, system_prompt=character["system_prompt"] if character else None, voice_stack_id=stack.id if stack else stack_id, voice_stack=stack, greeting_phrases=[], ) @app.get("/api/voice-stacks", response_model=List[VoiceStack]) async def list_voice_stacks(): async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM voice_stacks ORDER BY name") as c: rows = await c.fetchall() return [VoiceStack(**dict(r)) for r in rows] @app.post("/api/voice-stacks", response_model=VoiceStack) async def create_voice_stack(payload: VoiceStackCreate): stack_id = payload.id or secrets.token_urlsafe(12) now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO voice_stacks (id, name, stt_provider, stt_model, stt_language, llm_provider, llm_model, llm_api_key, temperature, num_ctx, tts_provider, tts_voice, tts_language, allow_interruptions, min_endpointing_delay, max_turns, wake_word_enabled, wake_word_model, wake_word_threshold, wake_word_timeout, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( stack_id, payload.name, payload.stt_provider, payload.stt_model, payload.stt_language, payload.llm_provider, payload.llm_model, payload.llm_api_key, payload.temperature, payload.num_ctx, payload.tts_provider, payload.tts_voice, payload.tts_language, int(payload.allow_interruptions), payload.min_endpointing_delay, payload.max_turns, int(payload.wake_word_enabled), payload.wake_word_model, payload.wake_word_threshold, payload.wake_word_timeout, now, now, ), ) await db.commit() return await _get_voice_stack(stack_id) async def _get_voice_stack(stack_id: str) -> VoiceStack: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM voice_stacks WHERE id = ?", (stack_id,)) as c: row = await c.fetchone() if not row: raise HTTPException(404, "Voice stack not found") return VoiceStack(**dict(row)) @app.get("/api/voice-stacks/{stack_id}", response_model=VoiceStack) async def get_voice_stack(stack_id: str): return await _get_voice_stack(stack_id) @app.put("/api/voice-stacks/{stack_id}", response_model=VoiceStack) async def update_voice_stack(stack_id: str, payload: VoiceStackCreate): now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT id FROM voice_stacks WHERE id = ?", (stack_id,)) as c: if not await c.fetchone(): raise HTTPException(404, "Voice stack not found") await db.execute( """UPDATE voice_stacks SET name = ?, stt_provider = ?, stt_model = ?, stt_language = ?, llm_provider = ?, llm_model = ?, llm_api_key = ?, temperature = ?, num_ctx = ?, tts_provider = ?, tts_voice = ?, tts_language = ?, allow_interruptions = ?, min_endpointing_delay = ?, max_turns = ?, wake_word_enabled = ?, wake_word_model = ?, wake_word_threshold = ?, wake_word_timeout = ?, updated_at = ? WHERE id = ?""", ( payload.name, payload.stt_provider, payload.stt_model, payload.stt_language, payload.llm_provider, payload.llm_model, payload.llm_api_key, payload.temperature, payload.num_ctx, payload.tts_provider, payload.tts_voice, payload.tts_language, int(payload.allow_interruptions), payload.min_endpointing_delay, payload.max_turns, int(payload.wake_word_enabled), payload.wake_word_model, payload.wake_word_threshold, payload.wake_word_timeout, now, stack_id, ), ) await db.commit() return await _get_voice_stack(stack_id) @app.delete("/api/voice-stacks/{stack_id}") async def delete_voice_stack(stack_id: str): async with aiosqlite.connect(DB_PATH) as db: await db.execute("DELETE FROM voice_stacks WHERE id = ?", (stack_id,)) await db.commit() return {"ok": True} # ============================================================================= # TTS VOICE CATALOG # ============================================================================= # These endpoints let the dashboard populate its voice selector with the real # set of available voices per provider. ElevenLabs is fetched live from the # ElevenLabs REST API (using ELEVENLABS_API_KEY); Kokoro is served from a # curated static list (or proxied from a local Kokoro server if KOKORO_BASE_URL # is set). Both endpoints are public on the scheduler and require no auth — # the ElevenLabs key is only used server-side. class TtsVoice(BaseModel): id: str name: str language: Optional[str] = None gender: Optional[str] = None description: Optional[str] = None preview_url: Optional[str] = None category: Optional[str] = None labels: Optional[dict] = None class TtsVoiceList(BaseModel): provider: str source: str # "api", "static", or "proxy" count: int voices: List[TtsVoice] warning: Optional[str] = None # Canonical Kokoro-82M voice list. The id is the value written into # voice_stacks.tts_voice; name is what the dashboard shows in the dropdown. # Sources: https://huggingface.co/hexgrad/Kokoro-82M and the Kokoro repo # voices.json manifest (v1.0). Grouped by language for the dropdown label. KOKORO_STATIC_VOICES: List[dict] = [ # American English {"id": "af_heart", "name": "Heart (American English · Female)", "language": "en-US", "gender": "female"}, {"id": "af_bella", "name": "Bella (American English · Female)", "language": "en-US", "gender": "female"}, {"id": "af_nicole", "name": "Nicole (American English · Female)", "language": "en-US", "gender": "female"}, {"id": "af_sarah", "name": "Sarah (American English · Female)", "language": "en-US", "gender": "female"}, {"id": "af_sky", "name": "Sky (American English · Female)", "language": "en-US", "gender": "female"}, {"id": "am_adam", "name": "Adam (American English · Male)", "language": "en-US", "gender": "male"}, {"id": "am_michael", "name": "Michael (American English · Male)", "language": "en-US", "gender": "male"}, # British English {"id": "bf_emma", "name": "Emma (British English · Female)", "language": "en-GB", "gender": "female"}, {"id": "bf_isabella","name": "Isabella (British English · Female)", "language": "en-GB", "gender": "female"}, {"id": "bm_george", "name": "George (British English · Male)", "language": "en-GB", "gender": "male"}, {"id": "bm_lewis", "name": "Lewis (British English · Male)", "language": "en-GB", "gender": "male"}, # Spanish {"id": "ef_dora", "name": "Dora (Spanish · Female)", "language": "es", "gender": "female"}, {"id": "em_alex", "name": "Alex (Spanish · Male)", "language": "es", "gender": "male"}, {"id": "em_santa", "name": "Santa (Spanish · Male)", "language": "es", "gender": "male"}, # French {"id": "ff_siwis", "name": "Siwis (French · Female)", "language": "fr", "gender": "female"}, # Hindi {"id": "hf_alpha", "name": "Alpha (Hindi · Female)", "language": "hi", "gender": "female"}, {"id": "hf_beta", "name": "Beta (Hindi · Female)", "language": "hi", "gender": "female"}, {"id": "hm_omega", "name": "Omega (Hindi · Male)", "language": "hi", "gender": "male"}, {"id": "hm_psi", "name": "Psi (Hindi · Male)", "language": "hi", "gender": "male"}, # Italian {"id": "if_sara", "name": "Sara (Italian · Female)", "language": "it", "gender": "female"}, {"id": "im_nicola", "name": "Nicola (Italian · Male)", "language": "it", "gender": "male"}, # Japanese {"id": "jf_alpha", "name": "Alpha (Japanese · Female)", "language": "ja", "gender": "female"}, {"id": "jf_gongitsune","name": "Gongitsune (Japanese · Female)", "language": "ja", "gender": "female"}, {"id": "jf_nezumi", "name": "Nezumi (Japanese · Female)", "language": "ja", "gender": "female"}, {"id": "jf_tebukuro","name": "Tebukuro (Japanese · Female)", "language": "ja", "gender": "female"}, {"id": "jm_kumo", "name": "Kumo (Japanese · Male)", "language": "ja", "gender": "male"}, # Brazilian Portuguese {"id": "pf_dora", "name": "Dora (Brazilian Portuguese · Female)","language": "pt-BR", "gender": "female"}, {"id": "pm_alex", "name": "Alex (Brazilian Portuguese · Male)", "language": "pt-BR", "gender": "male"}, {"id": "pm_santa", "name": "Santa (Brazilian Portuguese · Male)", "language": "pt-BR", "gender": "male"}, # Mandarin Chinese {"id": "zf_xiaobei", "name": "Xiaobei (Mandarin · Female)", "language": "zh", "gender": "female"}, {"id": "zf_xiaoni", "name": "Xiaoni (Mandarin · Female)", "language": "zh", "gender": "female"}, {"id": "zf_xiaoxiao","name": "Xiaoxiao (Mandarin · Female)", "language": "zh", "gender": "female"}, {"id": "zf_xiaoyi", "name": "Xiaoyi (Mandarin · Female)", "language": "zh", "gender": "female"}, {"id": "zm_yunjian", "name": "Yunjian (Mandarin · Male)", "language": "zh", "gender": "male"}, {"id": "zm_yunxi", "name": "Yunxi (Mandarin · Male)", "language": "zh", "gender": "male"}, {"id": "zm_yunyang", "name": "Yunyang (Mandarin · Male)", "language": "zh", "gender": "male"}, {"id": "zm_yunze", "name": "Yunze (Mandarin · Male)", "language": "zh", "gender": "male"}, ] def _kokoro_static_payload() -> TtsVoiceList: voices = [TtsVoice(**v) for v in KOKORO_STATIC_VOICES] return TtsVoiceList(provider="kokoro", source="static", count=len(voices), voices=voices) def _normalize_elevenlabs_voices(raw: dict) -> List[TtsVoice]: """Translate ElevenLabs' /v1/voices response into our TtsVoice shape.""" out: List[TtsVoice] = [] for v in (raw.get("voices") or []): labels = v.get("labels") or {} # Build a friendly display name. ElevenLabs' "name" is usually the # human label (e.g. "Rachel"); we suffix the category so it's obvious # whether it's premade/cloned/etc. category = v.get("category") or labels.get("category") or "" desc_parts = [] if labels.get("accent"): desc_parts.append(str(labels["accent"])) if labels.get("gender"): desc_parts.append(str(labels["gender"]).lower()) if labels.get("age"): desc_parts.append(str(labels["age"]).lower()) if labels.get("use_case"): desc_parts.append("use: " + str(labels["use_case"])) if labels.get("descriptive"): desc_parts.append(str(labels["descriptive"])) description = ", ".join([p for p in desc_parts if p]) preview = v.get("preview_url") or "" out.append(TtsVoice( id=str(v.get("voice_id") or ""), name=str(v.get("name") or v.get("voice_id") or "voice"), language=labels.get("language") or labels.get("accent") or None, gender=labels.get("gender") or None, description=description or None, preview_url=preview or None, category=str(category) if category else None, labels=labels or None, )) # Sort: by category then by name for a stable, predictable dropdown order. out.sort(key=lambda x: ((x.category or ""), (x.name or "").lower())) return out async def _fetch_elevenlabs_voices() -> TtsVoiceList: """Call the ElevenLabs /v1/voices endpoint and normalize the result. On any failure (missing key, network error, non-2xx) we surface a clear warning alongside an empty list so the dashboard can show the error in the picker rather than silently showing zero voices.""" if not ELEVENLABS_API_KEY: return TtsVoiceList( provider="elevenlabs", source="api", count=0, voices=[], warning=( "ELEVENLABS_API_KEY is not set on the scheduler — set it in the " "scheduler's environment to fetch the live voice list." ), ) url = f"{ELEVENLABS_BASE_URL}/v1/voices" headers = {"xi-api-key": ELEVENLABS_API_KEY, "Accept": "application/json"} try: async with httpx.AsyncClient(timeout=10.0) as client: r = await client.get(url, headers=headers) except Exception as e: return TtsVoiceList( provider="elevenlabs", source="api", count=0, voices=[], warning=f"Could not reach ElevenLabs: {e}", ) if r.status_code != 200: # Try to extract a useful message from the response body. msg = (r.text or "").strip()[:200] return TtsVoiceList( provider="elevenlabs", source="api", count=0, voices=[], warning=f"ElevenLabs returned {r.status_code}: {msg or 'no body'}", ) try: payload = r.json() except Exception as e: return TtsVoiceList( provider="elevenlabs", source="api", count=0, voices=[], warning=f"ElevenLabs returned non-JSON body: {e}", ) voices = _normalize_elevenlabs_voices(payload) return TtsVoiceList(provider="elevenlabs", source="api", count=len(voices), voices=voices) async def _fetch_kokoro_voices() -> TtsVoiceList: """Return the Kokoro voice catalog. If KOKORO_BASE_URL is set we proxy to that server's /v1/audio/voices (or whatever path it exposes) and adapt the result. Otherwise we serve the curated static list so the dropdown is always populated.""" if not KOKORO_BASE_URL: return _kokoro_static_payload() url = f"{KOKORO_BASE_URL}/v1/audio/voices" try: async with httpx.AsyncClient(timeout=6.0) as client: r = await client.get(url) if r.status_code != 200: warning = f"Kokoro server returned {r.status_code}; falling back to static list" fallback = _kokoro_static_payload() return TtsVoiceList( provider=fallback.provider, source="static", count=fallback.count, voices=fallback.voices, warning=warning, ) data = r.json() # Kokoro's API shape varies by server; be permissive. raw_list = data.get("voices") if isinstance(data, dict) else data if not isinstance(raw_list, list): raw_list = [] voices: List[TtsVoice] = [] for v in raw_list: if isinstance(v, str): voices.append(TtsVoice(id=v, name=v)) elif isinstance(v, dict): voices.append(TtsVoice( id=str(v.get("id") or v.get("voice_id") or v.get("name") or ""), name=str(v.get("name") or v.get("id") or v.get("voice_id") or ""), language=v.get("language") or v.get("lang"), gender=v.get("gender"), )) if not voices: fallback = _kokoro_static_payload() return TtsVoiceList( provider=fallback.provider, source="static", count=fallback.count, voices=fallback.voices, warning="Kokoro server returned no voices; falling back to static list", ) return TtsVoiceList(provider="kokoro", source="proxy", count=len(voices), voices=voices) except Exception as e: fallback = _kokoro_static_payload() return TtsVoiceList( provider=fallback.provider, source="static", count=fallback.count, voices=fallback.voices, warning=f"Could not reach Kokoro server ({KOKORO_BASE_URL}): {e}", ) @app.get("/api/voices/elevenlabs", response_model=TtsVoiceList) async def list_elevenlabs_voices(): """Return the catalog of ElevenLabs voices for the dashboard picker. Proxies the call to ElevenLabs' /v1/voices endpoint using the scheduler's ELEVENLABS_API_KEY env var. Returns a TtsVoiceList with a `warning` field if the key is missing or the upstream call fails — the dashboard surfaces the warning inside the picker so the operator knows what to fix.""" return await _fetch_elevenlabs_voices() @app.get("/api/voices/kokoro", response_model=TtsVoiceList) async def list_kokoro_voices(): """Return the catalog of Kokoro voices for the dashboard picker. Served from a curated static list of the canonical Kokoro-82M voices. If KOKORO_BASE_URL is set we proxy to that local server's voices endpoint and surface the live set instead; the static list is the fallback.""" return await _fetch_kokoro_voices() @app.get("/api/deployment-config") async def get_deployment_config(): """Report the sanitized, deployable character binding revision and drift.""" try: with open(PRODUCTION_CONFIG_PATH, "r", encoding="utf8") as handle: deployment = json.load(handle) except (OSError, json.JSONDecodeError) as exc: raise HTTPException(500, "Production character manifest is unavailable") from exc async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT id, name, default_engine, elevenlabs_agent_id, elevenlabs_branch_id, robovoice_profile_id " "FROM character_presets" ) as cursor: actual = {row["id"]: dict(row) for row in await cursor.fetchall()} bindings = [] all_applied = True for desired in deployment.get("characters", []): current = actual.get(desired["id"]) applied = bool( current and current.get("default_engine") == desired.get("default_engine") and current.get("elevenlabs_agent_id") == desired.get("elevenlabs_agent_id") and current.get("elevenlabs_branch_id") == desired.get("elevenlabs_branch_id") and current.get("robovoice_profile_id") == desired.get("robovoice_profile_id") ) all_applied = all_applied and applied bindings.append({ "character_id": desired["id"], "name": desired["name"], "default_engine": desired["default_engine"], "elevenlabs_agent_id": desired.get("elevenlabs_agent_id"), "elevenlabs_branch_id": desired.get("elevenlabs_branch_id"), "robovoice_profile_id": desired.get("robovoice_profile_id"), "applied": applied, }) return { "schema_version": deployment.get("schema_version"), "revision": deployment.get("revision"), "global": deployment.get("global", {}), "aliases": deployment.get("aliases", {}), "all_applied": all_applied, "bindings": bindings, } @app.get("/api/character-presets", response_model=List[CharacterPreset]) async def list_character_presets(): async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM character_presets ORDER BY name") as c: rows = await c.fetchall() return [ CharacterPreset( id=r["id"], name=r["name"], description=r["description"], system_prompt=r["system_prompt"], voice_stack_id=r["voice_stack_id"], motors=_json.loads(r["motors"]) if r["motors"] else [], nx=float(r["nx"]) if r["nx"] is not None else 0.5, ny=float(r["ny"]) if r["ny"] is not None else 0.5, img=r["img"], elevenlabs_agent_id=r["elevenlabs_agent_id"], elevenlabs_branch_id=r["elevenlabs_branch_id"], elevenlabs_first_message=r["elevenlabs_first_message"], elevenlabs_synced_at=r["elevenlabs_synced_at"], default_engine=r["default_engine"] or "elevenlabs", robovoice_profile_id=r["robovoice_profile_id"], system_prompt_revision=int(r["system_prompt_revision"] or 1), created_at=r["created_at"], updated_at=r["updated_at"], ) for r in rows ] @app.post("/api/character-presets", response_model=CharacterPreset) async def create_character_preset(payload: CharacterPresetCreate): preset_id = payload.id or secrets.token_urlsafe(12) now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO character_presets (id, name, description, system_prompt, voice_stack_id, motors, nx, ny, img, elevenlabs_agent_id, elevenlabs_branch_id, elevenlabs_first_message, default_engine, robovoice_profile_id, system_prompt_revision, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( preset_id, payload.name, payload.description, payload.system_prompt, payload.voice_stack_id, _json.dumps(payload.motors or []), payload.nx if payload.nx is not None else 0.5, payload.ny if payload.ny is not None else 0.5, payload.img, payload.elevenlabs_agent_id, payload.elevenlabs_branch_id, payload.elevenlabs_first_message, payload.default_engine, payload.robovoice_profile_id, payload.system_prompt_revision, now, now, ), ) await db.commit() return await _get_character_preset(preset_id) async def _get_character_preset(preset_id: str) -> CharacterPreset: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT * FROM character_presets WHERE id = ?", (preset_id,)) as c: row = await c.fetchone() if not row: raise HTTPException(404, "Character preset not found") return CharacterPreset( id=row["id"], name=row["name"], description=row["description"], system_prompt=row["system_prompt"], voice_stack_id=row["voice_stack_id"], motors=_json.loads(row["motors"]) if row["motors"] else [], nx=float(row["nx"]) if row["nx"] is not None else 0.5, ny=float(row["ny"]) if row["ny"] is not None else 0.5, img=row["img"], elevenlabs_agent_id=row["elevenlabs_agent_id"], elevenlabs_branch_id=row["elevenlabs_branch_id"], elevenlabs_first_message=row["elevenlabs_first_message"], elevenlabs_synced_at=row["elevenlabs_synced_at"], default_engine=row["default_engine"] or "elevenlabs", robovoice_profile_id=row["robovoice_profile_id"], system_prompt_revision=int(row["system_prompt_revision"] or 1), created_at=row["created_at"], updated_at=row["updated_at"], ) @app.get("/api/character-presets/{preset_id}", response_model=CharacterPreset) async def get_character_preset(preset_id: str): return await _get_character_preset(preset_id) @app.put("/api/character-presets/{preset_id}", response_model=CharacterPreset) async def update_character_preset(preset_id: str, payload: CharacterPresetCreate): now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: async with db.execute("SELECT id FROM character_presets WHERE id = ?", (preset_id,)) as c: if not await c.fetchone(): raise HTTPException(404, "Character preset not found") await db.execute( """UPDATE character_presets SET name = ?, description = ?, system_prompt = ?, voice_stack_id = ?, motors = ?, nx = ?, ny = ?, img = ?, elevenlabs_agent_id = ?, elevenlabs_branch_id = ?, elevenlabs_first_message = ?, default_engine = ?, robovoice_profile_id = ?, system_prompt_revision = ?, updated_at = ? WHERE id = ?""", ( payload.name, payload.description, payload.system_prompt, payload.voice_stack_id, _json.dumps(payload.motors or []), payload.nx if payload.nx is not None else 0.5, payload.ny if payload.ny is not None else 0.5, payload.img, payload.elevenlabs_agent_id, payload.elevenlabs_branch_id, payload.elevenlabs_first_message, payload.default_engine, payload.robovoice_profile_id, payload.system_prompt_revision, now, preset_id, ), ) await db.commit() return await _get_character_preset(preset_id) async def _elevenlabs_agent(agent_id: str) -> dict: key = await _get_elevenlabs_api_key() if not key: raise HTTPException(503, "ElevenLabs API key is not configured") async with httpx.AsyncClient(timeout=20.0, headers={"xi-api-key": key}) as client: response = await client.get(f"{ELEVENLABS_BASE_URL}/v1/convai/agents/{agent_id}") if response.status_code >= 400: raise HTTPException(502, f"ElevenLabs agent lookup failed ({response.status_code}): {response.text[:300]}") return response.json() @app.get("/api/elevenlabs/agents") async def list_elevenlabs_agents(): key = await _get_elevenlabs_api_key() if not key: raise HTTPException(503, "ElevenLabs API key is not configured") async with httpx.AsyncClient(timeout=20.0, headers={"xi-api-key": key}) as client: response = await client.get(f"{ELEVENLABS_BASE_URL}/v1/convai/agents", params={"page_size": 100}) if response.status_code >= 400: raise HTTPException(502, f"ElevenLabs agent catalog failed ({response.status_code}): {response.text[:300]}") payload = response.json() agents = payload.get("agents") or payload.get("items") or [] return {"agents": [{"agent_id": a.get("agent_id") or a.get("id"), "name": a.get("name") or a.get("agent_id")} for a in agents]} @app.post("/api/character-presets/{preset_id}/elevenlabs-sync") async def sync_character_elevenlabs(preset_id: str, payload: ElevenLabsAgentSync): preset = await _get_character_preset(preset_id) agent_id = (preset.elevenlabs_agent_id or "").strip() if not agent_id: raise HTTPException(400, "Character has no ElevenLabs agent ID") remote = await _elevenlabs_agent(agent_id) config = remote.get("conversation_config") or {} remote_agent = config.get("agent") or {} remote_prompt = (remote_agent.get("prompt") or {}).get("prompt") or "" remote_first = remote_agent.get("first_message") or "" remote_voice = (config.get("tts") or {}).get("voice_id") or "" direction = payload.direction.strip().lower() now = datetime.utcnow().isoformat() if direction == "pull": async with aiosqlite.connect(DB_PATH) as db: await db.execute( "UPDATE character_presets SET name=?, system_prompt=?, elevenlabs_first_message=?, elevenlabs_synced_at=?, updated_at=? WHERE id=?", (remote.get("name") or preset.name, remote_prompt, remote_first, now, now, preset_id), ) if preset.voice_stack_id and remote_voice: await db.execute("UPDATE voice_stacks SET tts_provider='elevenlabs', tts_voice=?, updated_at=? WHERE id=?", (remote_voice, now, preset.voice_stack_id)) await db.commit() elif direction == "push": key = await _get_elevenlabs_api_key() body = {"name": preset.name, "conversation_config": {"agent": {"prompt": {"prompt": preset.system_prompt or ""}, "first_message": preset.elevenlabs_first_message or ""}}} async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row if preset.voice_stack_id: async with db.execute("SELECT tts_voice FROM voice_stacks WHERE id=?", (preset.voice_stack_id,)) as cursor: stack = await cursor.fetchone() if stack and stack["tts_voice"]: body["conversation_config"]["tts"] = {"voice_id": stack["tts_voice"]} async with httpx.AsyncClient(timeout=20.0, headers={"xi-api-key": key}) as client: response = await client.patch(f"{ELEVENLABS_BASE_URL}/v1/convai/agents/{agent_id}", json=body) if response.status_code >= 400: raise HTTPException(502, f"ElevenLabs agent sync failed ({response.status_code}): {response.text[:300]}") async with aiosqlite.connect(DB_PATH) as db: await db.execute("UPDATE character_presets SET elevenlabs_synced_at=?, updated_at=? WHERE id=?", (now, now, preset_id)) await db.commit() else: raise HTTPException(400, "direction must be pull or push") current = await _get_character_preset(preset_id) return {"ok": True, "direction": direction, "agent_id": agent_id, "character": current.model_dump(), "remote": {"name": remote.get("name"), "system_prompt": remote_prompt, "first_message": remote_first, "voice_id": remote_voice}} @app.delete("/api/character-presets/{preset_id}") async def delete_character_preset(preset_id: str): async with aiosqlite.connect(DB_PATH) as db: await db.execute("DELETE FROM character_presets WHERE id = ?", (preset_id,)) await db.commit() return {"ok": True} class PresetPosition(BaseModel): id: str nx: float ny: float class PresetPositionBatch(BaseModel): positions: List[PresetPosition] @app.put("/api/character-presets/positions") async def update_preset_positions(payload: PresetPositionBatch): """Batch-update (id, nx, ny) for the Park map. nx/ny are the normalized [0,1] pad coordinates the dashboard uses to place each robot on the iso park. Bounded + persisted so an operator can Ctrl+drag a robot on the Park map and have its position survive a refresh, server restart, or robot reconnect. Designed for small, infrequent updates (a single drag releases one robot, plus an occasional "reset all" tool). The endpoint intentionally ignores all other preset fields — only the coordinates move; voice stack, name, system prompt, etc. are owned by the character-preset editor and are not touched here. """ if not payload.positions: return {"updated": 0} now = datetime.utcnow().isoformat() updated = 0 async with aiosqlite.connect(DB_PATH) as db: for p in payload.positions: if not p.id: continue # Clamp nx/ny to the visible pad rectangle so a stray # drag can't park a robot off the island. A small inner # margin (0.04) keeps the avatar fully on the grass and # clear of the gold wall on the perimeter. nx = max(0.04, min(0.96, float(p.nx))) ny = max(0.04, min(0.96, float(p.ny))) cur = await db.execute( "UPDATE character_presets SET nx = ?, ny = ?, updated_at = ? WHERE id = ?", (nx, ny, now, p.id), ) updated += cur.rowcount or 0 await db.commit() return {"updated": updated} class RobotVoiceConfigUpdate(BaseModel): character_preset_id: Optional[str] = None voice_stack_id: Optional[str] = None override_config: Optional[dict] = None @app.get("/api/robots/{robot_id}/voice-config", response_model=RobotVoiceConfig) async def get_robot_voice_config(robot_id: str): return await _resolve_robot_voice_config(robot_id) @app.put("/api/robots/{robot_id}/voice-config", response_model=RobotVoiceConfig) async def update_robot_voice_config(robot_id: str, payload: RobotVoiceConfigUpdate): now = datetime.utcnow().isoformat() async with aiosqlite.connect(DB_PATH) as db: await db.execute( """INSERT INTO robot_voice_configs (robot_id, character_preset_id, voice_stack_id, override_config, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(robot_id) DO UPDATE SET character_preset_id = excluded.character_preset_id, voice_stack_id = excluded.voice_stack_id, override_config = excluded.override_config, updated_at = excluded.updated_at""", ( robot_id, payload.character_preset_id, payload.voice_stack_id, _json.dumps(payload.override_config) if payload.override_config else None, now, ), ) await db.commit() return await _resolve_robot_voice_config(robot_id) @app.get("/api/devices/by-room/{room_name}/voice-config", response_model=RobotVoiceConfig) async def get_device_voice_config_by_room(room_name: str): """ROBOVOICE agent calls this when it joins a room to fetch per-robot config.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute( "SELECT robot_id, character_id, voice_stack_id FROM sessions WHERE room_name = ? ORDER BY started_at DESC LIMIT 1", (room_name,), ) as c: session = await c.fetchone() if not session: if room_name.startswith("robopark-"): device_id = room_name[len("robopark-"):] if device_id: return await _resolve_robot_voice_config(device_id) raise HTTPException(404, "No session bound to this room") if session["character_id"]: return await _resolve_session_voice_config(session) return await _resolve_robot_voice_config(session["robot_id"]) # ============================================================================= # TELEMETRY (close-the-loop: trigger counts, latency, drop rate, error reasons) # ============================================================================= async def _robot_telemetry(db, robot_id: str) -> Optional[dict]: """Per-robot telemetry. `db` must have row_factory = aiosqlite.Row set. Returns trigger_count, sessions_total, avg_latency_ms, drop_rate, drops and an end_reason breakdown; None if the robot does not exist.""" async with db.execute( "SELECT trigger_count, name, status, current_session_id, connected_server_id, character_id FROM robots WHERE id = ?", (robot_id,) ) as c: robot = await c.fetchone() if not robot: return None trigger_count = robot["trigger_count"] or 0 async with db.execute( """SELECT status, production_mode, last_heartbeat, character_id, video_device, audio_device, audio_output_device, device_inventory, supervisor_status, supervisor_status_at FROM devices WHERE id = ?""", (robot_id,), ) as c: device = await c.fetchone() character_id = (device["character_id"] if device else None) or robot["character_id"] character_name = None character_img = None if character_id: async with db.execute( "SELECT name, img FROM character_presets WHERE id = ?", (character_id,) ) as c: character = await c.fetchone() if character: character_name = character["name"] character_img = character["img"] def parse_json_field(value, fallback): if not value: return fallback try: return json.loads(value) except (TypeError, ValueError, json.JSONDecodeError): return fallback device_inventory = parse_json_field(device["device_inventory"], {}) if device else {} supervisor_status = parse_json_field(device["supervisor_status"], []) if device else [] session = None current_pipeline_events = [] if robot["current_session_id"]: async with db.execute( "SELECT started_at, last_activity_at, ended_at FROM sessions WHERE id = ?", (robot["current_session_id"],), ) as c: session = await c.fetchone() async with db.execute( "SELECT * FROM pipeline_events WHERE robot_id = ? AND session_id = ? " "ORDER BY timestamp, id", (robot_id, robot["current_session_id"]), ) as c: current_pipeline_events = [dict(row) for row in await c.fetchall()] now = datetime.utcnow() session_age = None activity_age = None if session: try: session_age = max(0, int((now - datetime.fromisoformat(session["started_at"])).total_seconds())) activity_age = max(0, int((now - datetime.fromisoformat(session["last_activity_at"] or session["started_at"])).total_seconds())) except (TypeError, ValueError): pass heartbeat_age = None if device and device["last_heartbeat"]: try: heartbeat_age = max(0, int((now - datetime.fromisoformat(device["last_heartbeat"])).total_seconds())) except (TypeError, ValueError): pass server_status = None if robot["connected_server_id"]: async with db.execute( "SELECT status FROM livekit_servers WHERE id = ?", (robot["connected_server_id"],) ) as c: server_row = await c.fetchone() server_status = server_row["status"] if server_row else None readiness_code = "ready" readiness_message = "ready for motion trigger" if not device: readiness_code, readiness_message = "device_missing", "no enrolled device heartbeat" elif not await get_production_mode(): readiness_code, readiness_message = "production_off", "global production mode is off" elif not device["production_mode"]: readiness_code, readiness_message = "robot_production_off", "robot production mode is off" elif heartbeat_age is None or heartbeat_age > 30: readiness_code, readiness_message = "heartbeat_stale", f"robot heartbeat is {heartbeat_age or 'unknown'}s old" elif robot["connected_server_id"] and server_status != "online": readiness_code, readiness_message = "server_unavailable", f"LiveKit server {robot['connected_server_id']} is {server_status or 'missing'}" elif not robot["current_session_id"]: readiness_code, readiness_message = "ready", "ready for motion trigger" elif robot["status"] == "connecting" and session_age is not None and session_age > CONNECTING_SESSION_LEASE_SECONDS: readiness_code, readiness_message = "stuck_session", f"joining LiveKit for {session_age}s; automatic recovery pending" elif activity_age is not None and activity_age > ACTIVE_SESSION_LEASE_SECONDS: readiness_code, readiness_message = "stale_session", f"no session activity for {activity_age}s; automatic recovery pending" else: readiness_code, readiness_message = "in_session", "conversation session is active" # Pull the device's freshest known LAN address (heartbeat > enrollment) so # the dashboard can reach the robot directly for e.g. the vision overlay # feed, which the scheduler doesn't proxy. async with db.execute( "SELECT last_seen_ip, lan_ip FROM devices WHERE id = ?", (robot_id,) ) as c: dev = await c.fetchone() lan_ip = (dev["last_seen_ip"] or dev["lan_ip"]) if dev else None # Average latency over sessions that recorded a room join. async with db.execute( "SELECT started_at, joined_at FROM sessions WHERE robot_id = ? AND joined_at IS NOT NULL", (robot_id,), ) as c: latency_rows = await c.fetchall() latencies = [ ms for ms in ( _session_latency_ms(r["started_at"], r["joined_at"]) for r in latency_rows ) if ms is not None ] avg_latency_ms = int(sum(latencies) / len(latencies)) if latencies else None async with db.execute( "SELECT COUNT(*) AS n FROM sessions WHERE robot_id = ?", (robot_id,) ) as c: sessions_total = (await c.fetchone())["n"] # End-reason breakdown (surfaces "per-robot error reasons"); drops = the # subset whose reason is abnormal. async with db.execute( """SELECT end_reason, COUNT(*) AS n FROM sessions WHERE robot_id = ? AND end_reason IS NOT NULL GROUP BY end_reason""", (robot_id,), ) as c: end_reasons = {r["end_reason"]: r["n"] for r in await c.fetchall()} drops = sum(n for reason, n in end_reasons.items() if reason in ABNORMAL_END_REASONS) drop_rate = round(drops / sessions_total, 4) if sessions_total else 0.0 current_stage = _pipeline_current_state( current_pipeline_events, active=bool(robot["current_session_id"] and session and not session["ended_at"]), production_mode=bool(device["production_mode"]) if device else False, online=heartbeat_age is not None and heartbeat_age <= 30, ) return { "robot_id": robot_id, "name": robot["name"], "character_id": character_id, "character_name": character_name, "character_img": character_img, "lan_ip": lan_ip, "vision_port": DEFAULT_VISION_PORT, "trigger_count": trigger_count, "sessions_total": sessions_total, "avg_latency_ms": avg_latency_ms, "drop_rate": drop_rate, "drops": drops, "end_reasons": end_reasons, "status": robot["status"], "current_session_id": robot["current_session_id"], "connected_server_id": robot["connected_server_id"], "server_status": server_status, "session_age_seconds": session_age, "session_activity_age_seconds": activity_age, "heartbeat_age_seconds": heartbeat_age, "device_status": device["status"] if device else None, "video_device": device["video_device"] if device else None, "audio_device": device["audio_device"] if device else None, "audio_output_device": device["audio_output_device"] if device else None, "device_inventory": device_inventory, "supervisor_status": supervisor_status, "supervisor_status_at": device["supervisor_status_at"] if device else None, "production_mode": bool(device["production_mode"]) if device else False, "readiness_code": readiness_code, "readiness_message": readiness_message, "current_stage": current_stage, "current_stage_id": current_stage["stage"], "current_stage_label": current_stage["label"], "current_stage_status": current_stage["status"], } @app.get("/api/robots/{robot_id}/telemetry") async def robot_telemetry(robot_id: str): """Per-robot telemetry: camera-trigger count, average session latency (ms), drop rate, total sessions, and an end-reason breakdown.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row data = await _robot_telemetry(db, robot_id) if data is None: raise HTTPException(404, "Robot not found") return data @app.get("/api/telemetry") async def telemetry(): """Fleet-wide telemetry: per-robot trigger counts, latency, drop rates and session totals, plus fleet totals and the latest GPU metrics recorded per LiveKit server (surfaced from the previously write-only metrics_history).""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row async with db.execute("SELECT id FROM robots ORDER BY name") as c: robot_ids = [r["id"] for r in await c.fetchall()] robots = [] for rid in robot_ids: t = await _robot_telemetry(db, rid) if t is not None: robots.append(t) # Surface metrics_history (written by metrics_collector, never read # before): the latest sample per server. async with db.execute( """SELECT m.server_id, m.gpu_utilization, m.vram_used_mb, m.active_sessions, m.recorded_at FROM metrics_history m JOIN ( SELECT server_id, MAX(recorded_at) AS latest FROM metrics_history GROUP BY server_id ) latest ON m.server_id = latest.server_id AND m.recorded_at = latest.latest""" ) as c: server_metrics = [dict(r) for r in await c.fetchall()] totals = { "trigger_count": sum(r["trigger_count"] for r in robots), "sessions_total": sum(r["sessions_total"] for r in robots), "drops": sum(r["drops"] for r in robots), } totals["drop_rate"] = ( round(totals["drops"] / totals["sessions_total"], 4) if totals["sessions_total"] else 0.0 ) return { "robots": robots, "servers": server_metrics, "totals": totals, "abnormal_end_reasons": sorted(ABNORMAL_END_REASONS), } # ============================================================================= # DASHBOARD (Embedded) # ============================================================================= # ── Back-compat: legacy /voice page (ported verbatim from the deployed OLD # scheduler). Serves voice_client.html if present next to main.py, else an # inline fallback — matches the deployed container exactly (its Dockerfile # copies only main.py, so the fallback is what actually renders). ── @app.get("/voice", response_class=HTMLResponse) async def voice_client(): """Serve voice client page for remote access""" import os html_path = os.path.join(os.path.dirname(__file__), "voice_client.html") if os.path.exists(html_path): with open(html_path, "r") as f: return f.read() return """ Voice Client

Voice client HTML not found

""" @app.get("/", response_class=HTMLResponse) async def dashboard(): """Serve embedded dashboard""" return """ RoboPark Control Center
Legacy scheduler view. The canonical operator UI is the full RoboPark Control Center Park view on the mesh hub at port 47913. Open full Park Control Center
""" # ── Test robot client page (browser-based, zero install on robot laptop) ── @app.get("/robot", response_class=HTMLResponse) async def robot_client_page(): """Browser-based robot client for LAN testing. Opens the laptop's webcam + mic, enrolls with the scheduler, and joins the LiveKit room returned by /request-session. Useful for iterating on scene detection, session triggering, and dashboard streaming before deploying to real Pis. """ return r""" RoboPark Test Robot

RoboPark Test Robot

Idle — configure and start
""" if __name__ == "__main__": import uvicorn host = os.getenv("SCHEDULER_HOST", "0.0.0.0") port = int(os.getenv("SCHEDULER_PORT", "8080")) uvicorn.run(app, host=host, port=port)