#!/usr/bin/env python3
"""
RoboPark Robot Supervisor — runs on each robot/satellite Pi (or a Windows
test rig), keeping every robot-side service alive without a human watching
a terminal.

Problem this solves: a real robot needs several independent processes —
preview_agent.py (LiveKit publish + heartbeat + motion webhook), optionally
a vision/motion-detection app (RoboVisionAI_PI's app_pi_clean.py or
equivalent), optionally a motor server — and today each one has to be
started by hand in its own terminal, with nothing bringing a crashed one
back. This supervisor is a single process management layer, not a merge of
those services into one program: each keeps its own hardware access,
failure mode, and restart behavior isolated from the others, while still
being "one thing to run" operationally.

Configuration: ~/.robopark/supervisor.json (see supervisor.example.json in
this directory for the schema). Services are opt-in — only preview_agent is
enabled by default, since vision_app/motor_server commands are specific to
each robot's actual hardware/codebase and have no safe generic default.

Usage:
    python robot_supervisor.py [--config PATH]

Auto-start on boot:
    Windows: scripts/install-robot-supervisor-windows.ps1 (Task Scheduler,
             runs at user logon — required for mic/camera access, which
             Windows restricts to an interactive session).
    Linux:   scripts/robopark-supervisor.service (systemd unit template).
"""
from __future__ import annotations

import argparse
import base64
import csv
import json
import logging
import math
import os
import re
import signal
import struct
import subprocess
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

logger = logging.getLogger("robopark.supervisor")

CONFIG_DIR = Path.home() / ".robopark"
DEFAULT_CONFIG_FILE = CONFIG_DIR / "supervisor.json"
LOG_DIR = CONFIG_DIR / "logs"

# A service that has run this long without exiting is considered stable —
# its failure/backoff count resets so a single crash after weeks of uptime
# doesn't get treated like a crash loop.
STABLE_UPTIME_SECONDS = 60.0
BACKOFF_BASE_SECONDS = 2.0
BACKOFF_MAX_SECONDS = 60.0
LOG_ROTATE_BYTES = 5 * 1024 * 1024  # rotate a service's log past 5MB
POLL_INTERVAL_SECONDS = 2.0
STATUS_REPORT_INTERVAL_SECONDS = 10.0


def _scheduler_headers(token: str) -> dict[str, str]:
    """Authenticate the enrolled device and, when present, the mesh proxy."""
    headers = {"Authorization": f"Bearer {token}"}
    mesh_token = os.getenv("ROBOPARK_MESH_TOKEN", "").strip()
    if mesh_token:
        headers["X-RoboPark-Mesh-Token"] = mesh_token
    return headers


def _motor_headers() -> dict[str, str]:
    token = (
        os.getenv("ROBOPARK_MOTOR_TOKEN", "").strip()
        or os.getenv("ROBOPARK_MESH_TOKEN", "").strip()
    )
    return {"X-RoboPark-Motor-Token": token} if token else {}


def _load_robopark_identity() -> Optional[tuple[str, str, str]]:
    """Reuse preview_agent.py's own enrollment files (same ~/.robopark/ dir)
    so the supervisor can report status to the scheduler without needing
    separate credentials. Returns (device_id, scheduler_url, token) or None
    if this robot hasn't been enrolled yet."""
    cfg_file = CONFIG_DIR / "preview_agent.json"
    token_file = CONFIG_DIR / "device_token"
    if not cfg_file.exists() or not token_file.exists():
        return None
    try:
        cfg = json.loads(cfg_file.read_text(encoding="utf8"))
        device_id = cfg.get("device_id")
        scheduler_url = cfg.get("scheduler_url", "http://localhost:8080")
        token = token_file.read_text(encoding="utf8").strip()
        if not device_id or not token:
            return None
        return device_id, scheduler_url, token
    except Exception:
        return None


def _systemd_service_statuses() -> list[dict]:
    """Report persistent RoboPark units even in command-only mode."""
    if sys.platform == "win32":
        try:
            listed = subprocess.run(
                ["schtasks", "/Query", "/FO", "CSV", "/NH"],
                capture_output=True, text=True, timeout=10, check=False,
            )
        except Exception as exc:
            logger.debug("could not enumerate Windows tasks: %s", exc)
            return []
        rows = []
        for columns in csv.reader(listed.stdout.splitlines()):
            task = columns[0].lstrip("\\") if columns else ""
            if not task.lower().startswith("robopark-"):
                continue
            status = columns[2].strip().lower() if len(columns) > 2 else "unknown"
            rows.append({
                "name": task,
                "enabled": status != "disabled",
                "running": status == "running",
                "pid": None,
                "uptime_seconds": None,
                "failure_count": 0,
                "last_exit_code": None,
            })
        return rows
    if sys.platform != "linux":
        return []
    try:
        listed = subprocess.run(
            ["systemctl", "list-unit-files", "robopark-*.service", "--no-legend", "--no-pager"],
            capture_output=True, text=True, timeout=5, check=False,
        )
    except Exception as exc:
        logger.debug("could not enumerate systemd services: %s", exc)
        return []
    rows = []
    for line in listed.stdout.splitlines():
        columns = line.strip().split()
        unit = columns[0] if columns else ""
        if not re.fullmatch(r"robopark-[a-z0-9-]+\.service", unit):
            continue
        show = subprocess.run(
            ["systemctl", "show", unit, "--property=ActiveState,MainPID,ExecMainStatus", "--value"],
            capture_output=True, text=True, timeout=5, check=False,
        )
        values = show.stdout.splitlines()
        active = values[0].strip() if values else "unknown"
        pid = int(values[1]) if len(values) > 1 and values[1].isdigit() and int(values[1]) else None
        exit_code = int(values[2]) if len(values) > 2 and values[2].lstrip("-").isdigit() else None
        rows.append({
            "name": unit,
            "enabled": any(value.startswith("enabled") for value in columns[1:]),
            "running": active == "active",
            "pid": pid,
            "uptime_seconds": None,
            "failure_count": 0,
            "last_exit_code": exit_code,
        })
    return rows


def _report_status(states: "list[ServiceState]", identity: tuple[str, str, str]) -> list[dict]:
    """POST current service status; returns any pending remote-control
    commands the scheduler had queued for this device (e.g. an operator
    clicking "restart" in the dashboard) -- delivered on this same
    request/response cycle rather than a separate poll."""
    device_id, scheduler_url, token = identity
    try:
        import httpx
    except ImportError:
        logger.debug("httpx not installed -- skipping status report (see requirements-robot.txt)")
        return []
    services = [s.status_dict() for s in states]
    known = {service["name"] for service in services}
    services.extend(service for service in _systemd_service_statuses() if service["name"] not in known)
    payload = {"services": services}
    url = f"{scheduler_url.rstrip('/')}/api/devices/{device_id}/supervisor-status"
    try:
        resp = httpx.post(url, json=payload, headers=_scheduler_headers(token), timeout=5.0)
        resp.raise_for_status()
        return resp.json().get("commands", [])
    except Exception as e:
        logger.debug(f"status report failed (scheduler unreachable?): {e}")
        return []


@dataclass
class ServiceSpec:
    name: str
    enabled: bool
    command: list[str]
    cwd: Optional[str] = None
    env: Optional[dict] = None


@dataclass
class ServiceState:
    spec: ServiceSpec
    proc: Optional[subprocess.Popen] = None
    log_file: Optional[object] = None
    started_at: float = 0.0
    failure_count: int = 0
    next_restart_at: float = 0.0
    last_exit_code: Optional[int] = None
    stopped: bool = False  # true once the supervisor is shutting down

    def status_dict(self) -> dict:
        running = self.proc is not None
        return {
            "name": self.spec.name,
            "enabled": self.spec.enabled,
            "running": running,
            "pid": self.proc.pid if running else None,
            "uptime_seconds": (time.monotonic() - self.started_at) if running else None,
            "failure_count": self.failure_count,
            "last_exit_code": self.last_exit_code,
        }


def _load_config(path: Path) -> tuple[list[ServiceSpec], Path]:
    if not path.exists():
        raise SystemExit(
            f"No config at {path}. Copy supervisor.example.json there and "
            f"edit it, or pass --config PATH."
        )
    data = json.loads(path.read_text(encoding="utf8"))
    services = [
        ServiceSpec(
            name=s["name"],
            enabled=bool(s.get("enabled", False)),
            command=list(s["command"]),
            cwd=s.get("cwd"),
            env=s.get("env"),
        )
        for s in data.get("services", [])
    ]
    log_dir = Path(data.get("log_dir", str(LOG_DIR))).expanduser()
    return services, log_dir


def _open_log(log_dir: Path, name: str):
    log_dir.mkdir(parents=True, exist_ok=True)
    path = log_dir / f"{name}.log"
    if path.exists() and path.stat().st_size > LOG_ROTATE_BYTES:
        rotated = log_dir / f"{name}.log.1"
        try:
            rotated.unlink(missing_ok=True)
            path.rename(rotated)
        except OSError as e:
            logger.warning(f"log rotate failed for {name}: {e}")
    return open(path, "a", encoding="utf8", buffering=1)


def _spawn(state: ServiceState, log_dir: Path) -> None:
    spec = state.spec
    state.log_file = _open_log(log_dir, spec.name)
    env = os.environ.copy()
    if spec.env:
        env.update(spec.env)
    banner = f"\n=== supervisor: starting {spec.name} at {time.strftime('%Y-%m-%d %H:%M:%S')} ===\n"
    state.log_file.write(banner)
    try:
        state.proc = subprocess.Popen(
            spec.command,
            cwd=spec.cwd,
            env=env,
            stdout=state.log_file,
            stderr=subprocess.STDOUT,
            stdin=subprocess.DEVNULL,
        )
        state.started_at = time.monotonic()
        logger.info(f"started {spec.name} (pid={state.proc.pid}): {' '.join(spec.command)}")
    except Exception as e:
        logger.error(f"failed to start {spec.name}: {e}")
        state.proc = None
        state.failure_count += 1


def _schedule_restart(state: ServiceState) -> None:
    backoff = min(BACKOFF_BASE_SECONDS * (2 ** state.failure_count), BACKOFF_MAX_SECONDS)
    state.next_restart_at = time.monotonic() + backoff
    logger.warning(f"{state.spec.name} exited — restarting in {backoff:.0f}s (failure #{state.failure_count})")


def _terminate(state: ServiceState) -> None:
    if not state.proc:
        return
    try:
        state.proc.terminate()
        try:
            state.proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            logger.warning(f"{state.spec.name} did not exit in time, killing")
            state.proc.kill()
            state.proc.wait(timeout=5)
    except Exception as e:
        logger.error(f"error stopping {state.spec.name}: {e}")
    finally:
        if state.log_file:
            state.log_file.close()
            state.log_file = None


def _execute_command(state: ServiceState, action: str) -> None:
    """Operator-initiated action from the dashboard. Supports:
      - "restart" (original): terminate if running, the main loop
        respawns it on the next tick.
      - "start": spawn the service if it's not already running. Used
        after a deliberate stop.
      - "stop": graceful terminate and mark `stopped=True` so the main
        loop does NOT auto-restart it. Re-enable with "start".
    Anything else is ignored with a warning."""
    if action == "restart":
        # clear any deliberate-stop so the next main-loop tick is allowed
        # to respawn it
        state.stopped = False
        if state.proc is not None:
            _terminate(state)
            state.proc = None
        state.failure_count = 0
        state.next_restart_at = 0.0
        return
    if action == "start":
        state.stopped = False
        state.failure_count = 0
        state.next_restart_at = 0.0
        if state.proc is None:
            _spawn(state, _CURRENT_LOG_DIR or LOG_DIR)
        return
    if action == "stop":
        if state.proc is not None:
            _terminate(state)
            state.proc = None
        state.stopped = True
        return
    logger.warning(f"ignoring unknown remote command {action!r} for {state.spec.name}")


def _execute_systemd_action(service_name: str, action: str) -> None:
    """Control allowlisted RoboPark systemd units or Windows scheduled tasks."""
    if action not in {"start", "stop", "restart"}:
        logger.warning("rejected unsupported system service action %r %r", service_name, action)
        return
    aliases = {
        "runtime": "robopark-robot-runtime-*.service",
        "mesh": "robopark-robot-runtime-*.service",
        "vision": "robopark-robot-runtime-*.service",
        "preview": "robopark-robot-runtime-*.service",
        "conversation": "robopark-robot-conversation-*.service",
        "motor": "robopark-robot-motor-*.service",
        "screen": "robopark-robot-screen-*.service",
    }
    if sys.platform == "win32":
        prefixes = {
            "runtime": "RoboPark-robot-runtime-",
            "mesh": "RoboPark-robot-runtime-",
            "vision": "RoboPark-robot-runtime-",
            "preview": "RoboPark-robot-runtime-",
            "conversation": "RoboPark-robot-conversation-",
            "motor": "RoboPark-robot-motor-",
            "screen": "RoboPark-robot-screen-",
        }
        prefix = prefixes.get(service_name)
        if not prefix:
            logger.warning("rejected non-RoboPark Windows service %r", service_name)
            return
        try:
            listed = subprocess.run(
                ["schtasks", "/Query", "/FO", "CSV", "/NH"],
                capture_output=True, text=True, timeout=10, check=False,
            )
            tasks = []
            for row in csv.reader(listed.stdout.splitlines()):
                task = row[0].lstrip("\\") if row else ""
                if task.lower().startswith(prefix.lower()):
                    tasks.append(task)
            for task in tasks:
                if action in {"stop", "restart"}:
                    subprocess.run(["schtasks", "/End", "/TN", task], capture_output=True, timeout=20, check=False)
                if action in {"start", "restart"}:
                    result = subprocess.run(["schtasks", "/Run", "/TN", task], capture_output=True, text=True, timeout=20, check=False)
                    if result.returncode:
                        logger.error("schtasks %s %s failed: %s", action, task, (result.stderr or result.stdout).strip())
            if not tasks:
                logger.warning("no installed Windows task matches service %r", service_name)
        except Exception as exc:
            logger.error("Windows service action failed: %s", exc)
        return
    if sys.platform != "linux":
        logger.warning("service actions are unsupported on %s", sys.platform)
        return
    pattern = aliases.get(service_name, service_name)
    if not re.fullmatch(r"robopark-[a-z0-9*-]+\.service", pattern):
        logger.warning("rejected non-RoboPark service %r", service_name)
        return
    try:
        listed = subprocess.run(
            ["systemctl", "list-unit-files", pattern, "--no-legend", "--no-pager"],
            capture_output=True, text=True, timeout=5, check=False,
        )
        units = []
        for line in listed.stdout.splitlines():
            unit = line.strip().split()[0] if line.strip() else ""
            if re.fullmatch(r"robopark-[a-z0-9-]+\.service", unit):
                units.append(unit)
        if not units:
            logger.warning("no installed unit matches service %r", service_name)
            return
        for unit in units:
            result = subprocess.run(
                ["systemctl", action, unit], capture_output=True, text=True, timeout=20, check=False,
            )
            if result.returncode:
                logger.error("systemctl %s %s failed: %s", action, unit, (result.stderr or result.stdout).strip())
            else:
                logger.info("systemctl %s %s completed", action, unit)
    except Exception as exc:
        logger.error("system service action failed: %s", exc)


# ── Operator shell (C1 gap-fill) ──
# Allows the dashboard to read service logs and run a small set of
# pre-approved diagnostic commands on the robot. Results are POSTed back
# to the scheduler (POST /api/devices/{id}/supervisor-output) which
# caches them per (device, command-id) so the dashboard can poll the
# result without holding a long-lived connection to the robot.
#
# SECURITY: the allowlist below is the only thing that gets executed.
# Operators cannot pass arbitrary commands; the parser rejects anything
# not on this list with a 400-equivalent ("command not allowed").

import secrets as _secrets  # noqa: E402

# A list of allow-listed command specs. Each entry has a fixed argv;
# placeholders in <angle brackets> get substituted from the request.
# A request that doesn't match any entry is rejected.
SHELL_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", "<dir>"]},
    {"name": "systemctl-status", "argv": ["systemctl", "status", "<name>", "--no-pager", "-n", "30"]},
    {"name": "journalctl",    "argv": ["journalctl", "-u", "<name>", "-n", "200", "--no-pager"]},
    {"name": "tail",          "argv": ["tail", "-n", "<n>", "<path>"]},
]

# Cap output size per command to avoid an operator accidentally filling
# the supervisor output buffer with 100MB of `ps aux`.
SHELL_OUTPUT_MAX_BYTES = 64 * 1024
JOURNAL_OUTPUT_MAX_BYTES = 512 * 1024
SHELL_RUN_TIMEOUT_SECONDS = 8.0

VOICE_ENGINE_COMMANDS = {
    "voice-engine-status",
    "voice-engine-switch",
    "voice-engine-restart",
    "voice-engine-stop",
    "voice-engine-trigger",
    "voice-engine-dispose",
    "voice-engine-recover",
    "voice-engine-audio-status",
    "voice-engine-audio-release",
    "voice-engine-audio-recover",
    "conversation-journal-backfill",
}


def _systemctl(action: str, unit: str) -> tuple[bool, str]:
    try:
        proc = subprocess.run(
            ["systemctl", action, unit],
            capture_output=True,
            text=True,
            timeout=20,
            check=False,
        )
        detail = (proc.stderr or proc.stdout or "").strip()
        return proc.returncode == 0, detail
    except (FileNotFoundError, subprocess.TimeoutExpired, OSError) as exc:
        return False, f"{type(exc).__name__}: {exc}"


def _user_systemctl(action: str, unit: str) -> tuple[bool, str]:
    """Control a kiosk user unit, including from a root-owned supervisor."""
    try:
        identities: list[tuple[int, Optional[str]]] = [(os.getuid(), None)]
        if os.getuid() == 0:
            import pwd

            for service_path in Path("/home").glob(f"*/.config/systemd/user/{unit}"):
                username = service_path.parts[2]
                identity = (pwd.getpwnam(username).pw_uid, username)
                if identity not in identities:
                    identities.append(identity)
        details = []
        for uid, username in identities:
            env = {
                **os.environ,
                "XDG_RUNTIME_DIR": f"/run/user/{uid}",
                "DBUS_SESSION_BUS_ADDRESS": f"unix:path=/run/user/{uid}/bus",
            }
            argv = ["systemctl", "--user", action, unit]
            if username:
                argv = ["runuser", "-u", username, "--", *argv]
            proc = subprocess.run(
                argv,
                capture_output=True,
                text=True,
                timeout=20,
                check=False,
                env=env,
            )
            detail = (proc.stderr or proc.stdout or "").strip()
            if proc.returncode == 0:
                return True, detail
            if detail:
                details.append(detail)
        return False, "; ".join(details)
    except (AttributeError, FileNotFoundError, subprocess.TimeoutExpired, OSError) as exc:
        return False, f"{type(exc).__name__}: {exc}"


def _conversation_endpoint(port: int, method: str = "GET") -> tuple[bool, dict]:
    request = urllib.request.Request(
        f"http://127.0.0.1:{port}/",
        method=method,
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(request, timeout=3) as response:
            payload = json.loads(response.read().decode("utf-8") or "{}")
            return 200 <= response.status < 300, payload
    except Exception as exc:
        return False, {"error": f"{type(exc).__name__}: {exc}"}


def _audio_device_owners() -> list[dict]:
    """Return processes with an open ALSA device, using procfs as ground truth."""
    owners: list[dict] = []
    proc_root = Path("/proc")
    if not proc_root.exists():
        return owners
    for process_dir in proc_root.iterdir():
        if not process_dir.name.isdigit():
            continue
        devices: set[str] = set()
        try:
            for descriptor in (process_dir / "fd").iterdir():
                try:
                    target = os.readlink(descriptor)
                except OSError:
                    continue
                if target.startswith("/dev/snd/"):
                    devices.add(target)
        except (FileNotFoundError, PermissionError, ProcessLookupError):
            continue
        if not devices:
            continue
        try:
            command = (process_dir / "cmdline").read_bytes().replace(b"\0", b" ").decode("utf-8", "replace").strip()
        except (FileNotFoundError, PermissionError, ProcessLookupError, OSError):
            command = ""
        try:
            process_name = (process_dir / "comm").read_text(encoding="utf-8", errors="replace").strip()
        except (FileNotFoundError, PermissionError, ProcessLookupError, OSError):
            process_name = "unknown"
        try:
            cgroup = (process_dir / "cgroup").read_text(encoding="utf-8", errors="replace")
            service = next(
                (part for part in cgroup.replace("\n", "/").split("/") if part.endswith(".service")),
                None,
            )
        except (FileNotFoundError, PermissionError, ProcessLookupError, OSError):
            service = None
        owners.append({
            "pid": int(process_dir.name),
            "process": process_name,
            "command": command[:512],
            "service": service,
            "devices": sorted(devices),
        })
    return sorted(owners, key=lambda item: item["pid"])


def _audio_owner_groups(owners: list[dict]) -> list[str]:
    """Collapse parent/child media processes belonging to one service owner."""
    groups = {
        str(owner.get("service") or f"pid:{owner.get('pid', 'unknown')}")
        for owner in owners
    }
    return sorted(groups)


def _wait_for_conversation_endpoint(port: int, timeout: float = 30.0) -> tuple[bool, dict]:
    deadline = time.monotonic() + timeout
    last: dict = {"error": "conversation endpoint did not become ready"}
    while time.monotonic() < deadline:
        ready, payload = _conversation_endpoint(port)
        if ready:
            return True, payload
        last = payload
        time.sleep(0.5)
    return False, last


def _conversation_management_api(port: int, path: str, method: str = "GET",
                                 payload: Optional[dict] = None) -> tuple[bool, dict]:
    """Call the robot-local supervisor without touching media or systemd."""
    request = urllib.request.Request(
        f"http://127.0.0.1:{port}{path}",
        data=(json.dumps(payload).encode("utf-8") if payload is not None else None),
        headers={"Content-Type": "application/json"},
        method=method,
    )
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            body = response.read().decode("utf-8")
        return True, json.loads(body) if body else {}
    except urllib.error.HTTPError as exc:
        try:
            detail = json.loads(exc.read().decode("utf-8") or "{}")
        except Exception:
            detail = {"error": str(exc)}
        return False, detail
    except Exception as exc:
        return False, {"error": str(exc)}


def _voice_engine_command(name: str, params: dict) -> dict:
    """Manage one robot's mutually exclusive RoboVoice/ElevenLabs owner."""
    import re

    robot = str(params.get("robot", "")).strip()
    engine = str(params.get("engine", "elevenlabs")).strip().lower()
    try:
        port = int(params.get("port", 5060))
    except (TypeError, ValueError):
        return {"ok": False, "error": "invalid motion port"}
    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,63}", robot):
        return {"ok": False, "error": "invalid robot name"}
    if engine not in {"robovoice", "elevenlabs"}:
        return {"ok": False, "error": "engine must be robovoice or elevenlabs"}
    if not 1024 <= port <= 65535:
        return {"ok": False, "error": "motion port must be between 1024 and 65535"}

    conversation_unit = f"robopark-robot-conversation-{robot}.service"
    robovoice_unit = f"robopark-robot-runtime-{robot}.service"
    kiosk_unit = "robopark-kiosk.service"

    if name == "conversation-journal-backfill":
        units = [conversation_unit, robovoice_unit]
        chunks = []
        for unit in units:
            try:
                proc = subprocess.run(
                    ["journalctl", "-u", unit, "--since", "3 days ago", "--output", "short-iso", "--no-pager"],
                    capture_output=True,
                    text=True,
                    timeout=SHELL_RUN_TIMEOUT_SECONDS,
                    check=False,
                )
                if proc.stdout:
                    chunks.append(f"# unit={unit}\n{proc.stdout}")
            except (FileNotFoundError, subprocess.TimeoutExpired, OSError) as exc:
                chunks.append(f"# unit={unit} error={type(exc).__name__}: {exc}")
        output = "\n".join(chunks)
        return {
            "ok": True,
            "exit_code": 0,
            "stdout": output[:JOURNAL_OUTPUT_MAX_BYTES],
            "stderr": "",
            "truncated": len(output) > JOURNAL_OUTPUT_MAX_BYTES,
        }

    if name == "voice-engine-audio-status":
        owners = _audio_device_owners()
        return {
            "ok": True,
            "exit_code": 0,
            "stdout": json.dumps({
                "audio_owners": owners,
                "audio_owner_groups": _audio_owner_groups(owners),
                "owner_count": len(owners),
            }),
            "stderr": "",
        }

    if name in {"voice-engine-audio-release", "voice-engine-audio-recover"}:
        if not bool(params.get("force")):
            return {
                "ok": False, "exit_code": 1, "stdout": "",
                "stderr": "audio release is destructive and requires force=true",
            }
        ok, health = _conversation_management_api(port, "/health")
        if ok and health.get("session_active"):
            return {
                "ok": False, "exit_code": 1, "stdout": json.dumps(health),
                "stderr": "active sessions cannot release their media lease",
            }
        # The management plane never kills arbitrary ALSA owners. Return exact
        # ownership so the operator can resolve the owning service explicitly.
        return {
            "ok": bool(ok and health.get("media_lease_available")),
            "exit_code": 0 if ok and health.get("media_lease_available") else 1,
            "stdout": json.dumps(health),
            "stderr": "foreign media owner must be stopped explicitly" if ok else health.get("error", "endpoint unavailable"),
        }

    if name == "voice-engine-switch":
        return {"ok": False, "error": "engine changes must be staged through voice configuration"}
    elif name == "voice-engine-restart":
        if engine == "elevenlabs":
            ok, payload = _conversation_management_api(port, "/restart-when-idle", "POST", {})
            detail = payload.get("error")
        else:
            return {"ok": False, "error": "RoboVoice restart requires an explicit engine transition"}
        if not ok:
            return {"ok": False, "error": detail or f"could not restart {engine} engine"}
    elif name == "voice-engine-stop":
        if engine == "elevenlabs":
            ok, payload = _conversation_management_api(port, "/stop", "POST", {})
            detail = payload.get("error")
        else:
            return {"ok": False, "error": "RoboVoice stop requires an explicit engine transition"}
        if not ok:
            return {"ok": False, "error": detail or f"could not stop {engine} engine"}
    elif name == "voice-engine-trigger":
        if engine != "elevenlabs":
            return {"ok": False, "error": "RoboVoice sessions are triggered through LiveKit"}
        ok, payload = _conversation_management_api(port, "/trigger", "POST", {"reason": "dashboard"})
        return {"ok": ok, "exit_code": 0 if ok else 1, "stdout": json.dumps(payload), "stderr": ""}
    elif name in {"voice-engine-dispose", "voice-engine-recover"}:
        if engine != "elevenlabs":
            return {"ok": False, "error": "session disposal is only available for ElevenLabs"}
        path = "/restart?force=true" if name == "voice-engine-recover" else "/stop"
        if name == "voice-engine-recover" and not bool(params.get("force")):
            path = "/restart-when-idle"
        ok, payload = _conversation_management_api(port, path, "POST", {})
        return {"ok": ok, "exit_code": 0 if ok else 1, "stdout": json.dumps(payload), "stderr": payload.get("error", "")}

    health_ready, health = _conversation_management_api(port, "/health")
    state_ready, state = _conversation_management_api(port, "/state")
    config_ready, configuration = _conversation_management_api(port, "/configuration")
    conversation_active, _ = _systemctl("is-active", conversation_unit)
    robovoice_runtime_active, _ = _systemctl("is-active", robovoice_unit)
    robovoice_kiosk_active, _ = _user_systemctl("is-active", kiosk_unit)
    robovoice_active = robovoice_runtime_active or robovoice_kiosk_active
    endpoint_ready, endpoint = (state_ready, state)
    observed = "conflict" if conversation_active and robovoice_active else (
        "elevenlabs" if conversation_active else "robovoice" if robovoice_active else "none"
    )
    audio_owners = _audio_device_owners()
    status = {
        "desired_engine": engine,
        "observed_engine": observed,
        "conversation_service_active": conversation_active,
        "robovoice_service_active": robovoice_active,
        "robovoice_runtime_active": robovoice_runtime_active,
        "robovoice_kiosk_active": robovoice_kiosk_active,
        "endpoint_ready": endpoint_ready,
        "endpoint": endpoint if endpoint_ready else None,
        "endpoint_error": None if endpoint_ready else endpoint.get("error"),
        "health": health if health_ready else None,
        "configuration": configuration if config_ready else None,
        "audio_owners": audio_owners,
        "audio_owner_groups": _audio_owner_groups(audio_owners),
    }
    audio_conflict = bool((health.get("media_lease") or {}).get("conflicts")) if health_ready else len(status["audio_owner_groups"]) > 1
    status["audio_conflict"] = audio_conflict
    healthy = observed == engine and not audio_conflict and (engine != "elevenlabs" or endpoint_ready)
    return {
        "ok": healthy,
        "exit_code": 0 if healthy else 1,
        "stdout": json.dumps(status),
        "stderr": "" if healthy else (
            "multiple independent processes own ALSA devices"
            if audio_conflict else "desired and observed voice engines do not match"
        ),
    }


def _run_shell_command(name: str, params: dict) -> dict:
    """Resolve `name` against the allowlist, substitute params, and run.

    Returns a dict {ok, exit_code, stdout, stderr, error}. The caller
    POSTs this to the scheduler as the supervisor-output payload."""
    if name in VOICE_ENGINE_COMMANDS:
        return _voice_engine_command(name, params)

    spec = None
    for s in SHELL_ALLOWLIST:
        if s["name"] == name:
            spec = s
            break
    if spec is None:
        return {"ok": False, "error": f"command {name!r} is not in the allowlist"}
    argv = []
    for piece in spec["argv"]:
        if piece.startswith("<") and piece.endswith(">"):
            key = piece[1:-1]
            val = params.get(key)
            if val is None or not isinstance(val, str) or not val.strip():
                return {"ok": False, "error": f"missing required parameter {key!r} for command {name!r}"}
            # Reject anything that smells like a shell metachar — keep it
            # strictly to filenames / unit names. Service names are owned
            # by supervisor.json on this robot, not by the network.
            if any(c in val for c in ("\x00", "\n", "\r")):
                return {"ok": False, "error": "invalid characters in parameter"}
            argv.append(val)
        else:
            argv.append(piece)
    try:
        proc = subprocess.run(
            argv,
            capture_output=True,
            text=True,
            timeout=SHELL_RUN_TIMEOUT_SECONDS,
            check=False,
        )
        out = (proc.stdout or "")[:SHELL_OUTPUT_MAX_BYTES]
        err = (proc.stderr or "")[:SHELL_OUTPUT_MAX_BYTES]
        return {
            "ok": True,
            "exit_code": proc.returncode,
            "stdout": out,
            "stderr": err,
            "truncated": len(proc.stdout or "") > SHELL_OUTPUT_MAX_BYTES or len(proc.stderr or "") > SHELL_OUTPUT_MAX_BYTES,
        }
    except subprocess.TimeoutExpired:
        return {"ok": False, "error": f"command timed out after {SHELL_RUN_TIMEOUT_SECONDS}s"}
    except FileNotFoundError as e:
        return {"ok": False, "error": f"command not found: {e}"}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def _tail_log_file(path: Path, lines: int) -> dict:
    """Return the last `lines` lines of a log file as a string.

    Resolves `path` against LOG_DIR if it's relative; rejects anything
    that tries to escape (../) for safety, even though the operator
    can already read anything on the robot via the allowlist."""
    try:
        lines = max(1, min(int(lines), 2000))
    except (TypeError, ValueError):
        lines = 200
    if not path.is_absolute():
        path = (LOG_DIR / path).resolve()
    # Belt-and-suspenders: don't let `..` traversal escape the log dir
    # (the dashboard only ever sends a service name, but defend anyway).
    try:
        path.relative_to(LOG_DIR.resolve())
    except ValueError:
        return {"ok": False, "error": "log path is outside the supervisor log directory"}
    if not path.exists():
        return {"ok": False, "error": f"no such log file: {path}"}
    try:
        # Read the tail efficiently: seek to ~32KB from the end and split.
        size = path.stat().st_size
        chunk = min(size, 64 * 1024)
        with path.open("rb") as f:
            if size > chunk:
                f.seek(size - chunk)
            data = f.read().decode("utf-8", errors="replace")
        all_lines = data.splitlines()
        tail = all_lines[-lines:]
        return {"ok": True, "path": str(path), "lines": len(tail), "total_lines": len(all_lines), "content": "\n".join(tail)}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def _post_supervisor_output(identity, kind: str, service: Optional[str], payload: dict, request_id: Optional[str] = None) -> None:
    """Send the result of a tail_logs or shell_run back to the scheduler.

    Uses the same httpx call style as the status report itself; no
    retry, no queue — if the scheduler is down we drop the result.
    The dashboard times out and reports a clear error in that case."""
    device_id, scheduler_url, token = identity
    try:
        import httpx
    except ImportError:
        return
    url = f"{scheduler_url.rstrip('/')}/api/devices/{device_id}/supervisor-output"
    body = {"kind": kind, "service": service, "payload": payload, "request_id": request_id}
    try:
        httpx.post(url, json=body, headers=_scheduler_headers(token), timeout=5.0)
    except Exception as e:
        logger.debug(f"supervisor-output POST failed: {e}")


def _handle_shell_command(identity, cmd: dict) -> None:
    """Process a single shell/log request returned by the scheduler.

    `cmd` is {id, kind: "shell_run"|"tail_logs"|"speaker_test", service,
    params}. Runs the request, posts the result back, and returns."""
    kind = cmd.get("kind", "shell_run")
    service = cmd.get("service")
    request_id = cmd.get("id")
    params = cmd.get("params") or {}
    if kind == "tail_logs":
        # params: {lines: N, path?: override path}
        path_param = params.get("path")
        if path_param:
            log_path = Path(path_param)
        elif service:
            log_path = (_CURRENT_LOG_DIR or LOG_DIR) / f"{service}.log"
        else:
            _post_supervisor_output(identity, kind, service, {"ok": False, "error": "tail_logs requires service or path"}, request_id)
            return
        result = _tail_log_file(log_path, int(params.get("lines", 200)))
    elif kind == "shell_run":
        result = _run_shell_command(params.get("name", ""), params)
    elif kind == "speaker_test":
        result = _speaker_roundtrip_test(params)
    elif kind == "motor_discover":
        result = _discover_motor_registry(params)
    elif kind == "motor_sequence":
        result = _run_motor_sequence(params)
    else:
        result = {"ok": False, "error": f"unknown shell command kind: {kind!r}"}
    _post_supervisor_output(identity, kind, service, result, request_id)


def _discover_motor_registry(params: dict) -> dict:
    """Read the robot-local motor registry without touching any GPIO output."""
    import httpx
    import re

    base = str(params.get("motor_server_url") or "http://127.0.0.1:8001").rstrip("/")
    if not (base.startswith("http://127.0.0.1:") or base.startswith("http://localhost:")):
        return {"ok": False, "error": "motor server must be robot-local"}
    try:
        with httpx.Client(timeout=5.0, headers=_motor_headers()) as client:
            response = client.get(f"{base}/list-motors")
            response.raise_for_status()
            raw_motors = response.json().get("motors", [])
        registry, used_ids, used_gpios = [], set(), set()
        for index, raw in enumerate(raw_motors[:32]):
            gpio = int(raw.get("gpio", -1))
            if gpio < 2 or gpio > 27 or gpio in used_gpios:
                continue
            name = str(raw.get("name") or f"Relay {index + 1}").strip()[:60]
            base_id = re.sub(r"[^a-z0-9_-]+", "-", name.lower()).strip("-") or f"relay-{index + 1}"
            motor_id, suffix = base_id[:32], 2
            while motor_id in used_ids:
                tail = f"-{suffix}"
                motor_id = f"{base_id[:32-len(tail)]}{tail}"
                suffix += 1
            used_ids.add(motor_id)
            used_gpios.add(gpio)
            registry.append({
                "id": motor_id,
                "name": name,
                "gpio": gpio,
                "active_high": bool(raw.get("active_high", True)),
                "max_duration_ms": 3000,
            })
        return {"ok": True, "registry": registry, "count": len(registry), "motor_server_url": base}
    except Exception as exc:
        return {"ok": False, "error": f"motor registry discovery failed: {type(exc).__name__}: {exc}"}


def _run_motor_sequence(params: dict) -> dict:
    """Run one validated scheduler sequence against the robot-local motor API."""
    import httpx
    base = str(params.get("motor_server_url") or "http://127.0.0.1:8001").rstrip("/")
    if not (base.startswith("http://127.0.0.1:") or base.startswith("http://localhost:")):
        return {"ok": False, "error": "motor server must be robot-local"}
    registry = {str(item.get("id")): item for item in (params.get("registry") or [])}
    steps = params.get("steps") or []
    started = time.monotonic()
    completed = []
    gpio_log = []
    try:
        with httpx.Client(timeout=8.0, headers=_motor_headers()) as client:
            if params.get("stop_all"):
                response = client.post(f"{base}/stop-motors", json={})
                response.raise_for_status()
                return {"ok": True, "stopped": True, "sequence_id": "emergency-stop"}
            existing = client.get(f"{base}/list-motors").json().get("motors", [])
            existing_names = {str(item.get("name")) for item in existing}
            for motor_id, motor in registry.items():
                body = {"name": motor_id, "gpio": int(motor["gpio"]), "active_high": bool(motor.get("active_high", True))}
                response = (client.put(f"{base}/update-motor/{motor_id}", json=body)
                            if motor_id in existing_names else client.post(f"{base}/add-motor", json=body))
                response.raise_for_status()
            if params.get("test_all"):
                duration_ms = max(50, min(1000, int(params.get("duration_ms", 300))))
                pause_ms = max(0, min(2000, int(params.get("pause_ms", 200))))
                response = client.post(f"{base}/test", json={
                    "seconds_on": duration_ms / 1000.0,
                    "seconds_pause": pause_ms / 1000.0,
                    "pins": [int(motor["gpio"]) for motor in registry.values()],
                })
                response.raise_for_status()
                deadline = time.monotonic() + len(registry) * ((duration_ms + pause_ms) / 1000.0) + 3.0
                last_test_action = None
                while time.monotonic() < deadline:
                    status = client.get(f"{base}/status").json()
                    action = str(status.get("last_action") or "")
                    if action.startswith("test:gpio:") and action != last_test_action:
                        try:
                            observed_gpio = int(action.rsplit(":", 1)[-1])
                        except ValueError:
                            observed_gpio = -1
                        if observed_gpio >= 2:
                            gpio_log.append({
                                "gpio": observed_gpio,
                                "status": "active_observed",
                                "elapsed_ms": int((time.monotonic() - started) * 1000),
                            })
                        last_test_action = action
                    if status.get("status") == "idle":
                        if status.get("error"):
                            raise RuntimeError(f"registered GPIO test failed: {status['error']}")
                        completed = [{"motor_id": motor_id, "gpio": int(motor["gpio"])}
                                     for motor_id, motor in registry.items()]
                        break
                    time.sleep(0.05)
                else:
                    raise TimeoutError("registered GPIO test did not return to idle")
            for index, step in enumerate(steps):
                motor_id = str(step.get("motor_id"))
                motor = registry.get(motor_id)
                if not motor:
                    raise ValueError(f"step {index + 1} references unknown motor {motor_id}")
                delay_ms = max(0, min(30000, int(step.get("delay_ms", 0))))
                duration_ms = max(50, min(int(motor.get("max_duration_ms", 3000)), int(step.get("duration_ms", 500))))
                if delay_ms:
                    time.sleep(delay_ms / 1000.0)
                response = client.post(f"{base}/trigger-motor", json={"motor_name": motor_id, "seconds": duration_ms / 1000.0})
                response.raise_for_status()
                deadline = time.monotonic() + duration_ms / 1000.0 + 2.0
                while time.monotonic() < deadline:
                    status_response = client.get(f"{base}/status")
                    status_response.raise_for_status()
                    status = status_response.json()
                    if status.get("status") == "idle":
                        if status.get("error"):
                            raise RuntimeError(f"motor {motor_id} failed: {status['error']}")
                        break
                    time.sleep(0.05)
                else:
                    raise TimeoutError(f"motor {motor_id} did not return to idle")
                completed.append({"motor_id": motor_id, "gpio": int(motor["gpio"]), "duration_ms": duration_ms})
    except Exception as exc:
        try:
            httpx.post(f"{base}/stop-motors", json={}, headers=_motor_headers(), timeout=3.0)
        except Exception:
            pass
        return {"ok": False, "error": f"{type(exc).__name__}: {exc}", "completed_steps": completed,
                "gpio_log": gpio_log,
                "duration_ms": int((time.monotonic() - started) * 1000), "session_id": params.get("session_id")}
    return {"ok": True, "pass": True, "sequence_id": params.get("sequence_id"), "completed_steps": completed,
            "gpio_log": gpio_log,
            "duration_ms": int((time.monotonic() - started) * 1000), "motor_server_url": base,
            "session_id": params.get("session_id")}


# ── Speaker roundtrip test (C2 gap-fill) ──
# Plays a short tone through the robot's configured output device while
# simultaneously recording from the configured input device. Returns
# {played, recorded_rms, peak_db, duration_ms, output_device,
# input_device, error}.
#
# Used by the dashboard's "Test speaker" button. Helps catch the
# classic failure modes: speaker muted, wrong output device, mic
# pointing the wrong way, USB audio device unplugged. The result is
# a quick PASS/FAIL plus a peak-dB readout that the operator can
# read at a glance.

# Default test tone parameters — overridden by per-request params
SPEAKER_TEST_FREQUENCY_HZ = 1000.0
SPEAKER_TEST_DURATION_S = 0.6
SPEAKER_TEST_SAMPLE_RATE = 48000
SPEAKER_TEST_AMPLITUDE = 0.6   # 0..1; conservative so it doesn't clip
SPEAKER_TEST_OUTPUT_DEVICE = None  # None = use ROBOPARK_AUDIO_OUTPUT / default
SPEAKER_TEST_INPUT_DEVICE = None
# Peak dB threshold: anything below this is reported as "no signal
# detected" (PASS means the robot heard the tone back; FAIL means
# the mic didn't pick it up). The default (-30 dB) is conservative
# for a quiet indoor environment; can be overridden per request.
SPEAKER_TEST_PEAK_DB_THRESHOLD = -30.0


def _resolve_audio_device(pa, selected: str, kind: str) -> Optional[int]:
    """Resolve a free-form device name/index to a PyAudio device index.

    kind is "input" or "output". None / "default" picks the WASAPI
    default device (per the same logic preview_agent.py uses for the
    mic). If the name doesn't match any device, returns None and lets
    PyAudio pick its global default — that may still work, but the
    result is recorded so the operator can see it."""
    if selected is None:
        selected = "default"
    s = str(selected).strip()
    if s == "" or s.lower() == "default":
        try:
            info = (pa.get_default_input_device_info() if kind == "input"
                    else pa.get_default_output_device_info())
            idx = info.get("index")
            if idx is not None and idx >= 0:
                return int(idx)
        except Exception:
            pass
        return None
    if s.isdigit():
        return int(s)
    needle = s.lower()
    want_channels = 1 if kind == "input" else 0  # input: must have >0
    for i in range(pa.get_device_count()):
        info = pa.get_device_info_by_index(i)
        if kind == "output" and info.get("maxOutputChannels", 0) <= 0:
            continue
        if kind == "input" and info.get("maxInputChannels", 0) <= 0:
            continue
        if needle in str(info.get("name", "")).lower():
            return i
    return None


def _audio_rate_candidates(info: dict, requested_rate: int) -> list[int]:
    """Return practical PCM rates with the device native rate first."""
    rates = []
    for value in (
        info.get("defaultSampleRate"),
        requested_rate,
        48000,
        44100,
        32000,
        24000,
        16000,
    ):
        try:
            rate = int(round(float(value)))
        except (TypeError, ValueError):
            continue
        if rate > 0 and rate not in rates:
            rates.append(rate)
    return rates


def _resample_pcm16_mono(raw: bytes, source_rate: int, target_rate: int) -> bytes:
    """Linearly resample little-endian mono PCM16 without optional DSP deps."""
    if source_rate == target_rate or not raw:
        return raw
    source_count = len(raw) // 2
    target_count = max(1, int(round(source_count * target_rate / source_rate)))
    source = struct.unpack(f"<{source_count}h", raw[:source_count * 2])
    if source_count == 1:
        return struct.pack("<h", source[0]) * target_count
    scale = (source_count - 1) / max(1, target_count - 1)
    result = bytearray(target_count * 2)
    for index in range(target_count):
        position = index * scale
        left = int(position)
        right = min(left + 1, source_count - 1)
        fraction = position - left
        value = int(round(source[left] + (source[right] - source[left]) * fraction))
        struct.pack_into("<h", result, index * 2, value)
    return bytes(result)


def _open_pcm_stream(pa, pyaudio_module, kind: str, device_index: Optional[int],
                     requested_rate: int):
    """Open a PCM16 stream using the first format the device accepts."""
    try:
        if device_index is None:
            info = (pa.get_default_output_device_info() if kind == "output"
                    else pa.get_default_input_device_info())
            device_index = int(info.get("index"))
        else:
            info = pa.get_device_info_by_index(device_index)
    except Exception as exc:
        raise OSError(f"selected {kind} device is unavailable: {exc}") from exc

    channel_key = "maxOutputChannels" if kind == "output" else "maxInputChannels"
    max_channels = int(info.get(channel_key, 0) or 0)
    if max_channels < 1:
        raise OSError(f"{info.get('name', device_index)} has no {kind} channels")
    channels = [2, 1] if kind == "output" and max_channels >= 2 else [1]
    errors = []
    for rate in _audio_rate_candidates(info, requested_rate):
        for channel_count in channels:
            kwargs = {
                "format": pyaudio_module.paInt16,
                "channels": channel_count,
                "rate": rate,
                kind: True,
                f"{kind}_device_index": device_index,
            }
            if kind == "input":
                kwargs["frames_per_buffer"] = 1024
            try:
                stream = pa.open(**kwargs)
                return stream, rate, channel_count, info
            except Exception as exc:
                errors.append(f"{rate}Hz/{channel_count}ch: {exc}")
    attempts = "; ".join(errors[-4:])
    raise OSError(
        f"{info.get('name', device_index)} supports no usable PCM16 {kind} format"
        + (f" ({attempts})" if attempts else "")
    )


def _linux_aplay_pcm16(raw: bytes, selected_output: str, sample_rate: int,
                       channels: int = 1) -> tuple[bool, str]:
    """Play PCM through the same ALSA plughw path proven during Pi setup.

    PortAudio and RoboVision enumerate devices in different index spaces. The
    inventory name contains the stable ALSA hw tuple, so use it directly and
    let ALSA's plug layer adapt the cached TTS sample rate/channel count.
    """
    import re

    match = re.search(r"hw:(\d+),(\d+)", str(selected_output), re.IGNORECASE)
    if not match:
        return False, "selected output has no ALSA hw address"
    alsa_device = f"plughw:{match.group(1)},{match.group(2)}"
    command = [
        "aplay", "-q", "-D", alsa_device, "-t", "raw", "-f", "S16_LE",
        "-r", str(sample_rate), "-c", str(channels),
    ]
    last_error = "aplay failed"
    # ALSA can retain an exclusive USB handle briefly after the LiveKit
    # playback stream closes. Retry for a bounded period instead of declaring
    # a valid device unsupported on the first EBUSY response.
    try:
        from media_lock import media_lock
        with media_lock("speaker", timeout=8.0):
            for attempt in range(5):
                if attempt:
                    time.sleep(0.5)
                try:
                    completed = subprocess.run(
                        command,
                        input=raw,
                        stdout=subprocess.DEVNULL,
                        stderr=subprocess.PIPE,
                        timeout=max(5.0, len(raw) / max(1, sample_rate * channels * 2) + 3.0),
                        check=False,
                    )
                except Exception as exc:
                    last_error = f"{type(exc).__name__}: {exc}"
                    continue
                if completed.returncode == 0:
                    return True, alsa_device
                last_error = completed.stderr.decode("utf-8", errors="replace").strip() or f"aplay exited {completed.returncode}"
    except TimeoutError as exc:
        return False, str(exc)
    return False, f"{alsa_device}: {last_error}"


def _linux_mic_groundtruth(selected_input: str, duration: float = 4.0) -> dict:
    """Capture the fleet USB mic through the exact onsite-proven ALSA path."""
    import re

    match = re.search(r"hw:(\d+),(\d+)", str(selected_input), re.IGNORECASE)
    if not match:
        return {"ok": False, "error": "selected input has no ALSA hw address"}
    alsa_device = f"plughw:{match.group(1)},{match.group(2)}"
    seconds = max(1, min(10, int(round(duration))))
    command = [
        "arecord", "-q", "-D", alsa_device, "-t", "raw", "-f", "S16_LE",
        "-r", "48000", "-c", "1", "-d", str(seconds),
    ]
    started = time.monotonic()
    try:
        from media_lock import media_lock
        with media_lock("microphone", timeout=8.0):
            completed = subprocess.run(
                command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                timeout=seconds + 4.0,
                check=False,
            )
    except Exception as exc:
        return {
            "ok": False,
            "mode": "mic_groundtruth",
            "error": f"{type(exc).__name__}: {exc}",
            "input_device": alsa_device,
            "duration_ms": int((time.monotonic() - started) * 1000),
        }
    raw = completed.stdout or b""
    stderr = completed.stderr.decode("utf-8", errors="replace").strip()
    if completed.returncode != 0 or len(raw) < 2:
        return {
            "ok": False,
            "mode": "mic_groundtruth",
            "error": stderr or f"arecord exited {completed.returncode}",
            "exit_code": completed.returncode,
            "bytes": len(raw),
            "input_device": alsa_device,
            "command": " ".join(command),
            "duration_ms": int((time.monotonic() - started) * 1000),
        }
    sample_count = len(raw) // 2
    peak = 0
    square_sum = 0
    nonzero = 0
    for (value,) in struct.iter_unpack("<h", raw[:sample_count * 2]):
        magnitude = abs(value)
        peak = max(peak, magnitude)
        square_sum += value * value
        if value:
            nonzero += 1
    rms = (square_sum / sample_count) ** 0.5
    peak_db = round(20.0 * math.log10(max(1, peak) / 32767.0), 1)
    rms_db = round(20.0 * math.log10(max(1.0, rms) / 32767.0), 1)
    return {
        "ok": True,
        "pass": nonzero > 0 and peak > 0,
        "mode": "mic_groundtruth",
        "exit_code": completed.returncode,
        "bytes": len(raw),
        "samples": sample_count,
        "nonzero_samples": nonzero,
        "nonzero_percent": round(nonzero * 100.0 / sample_count, 3),
        "recorded_peak": peak,
        "recorded_peak_db": peak_db,
        "recorded_rms": round(rms, 2),
        "recorded_rms_db": rms_db,
        "sample_rate": 48000,
        "channels": 1,
        "input_device": alsa_device,
        "command": " ".join(command),
        "stderr": stderr,
        "duration_ms": int((time.monotonic() - started) * 1000),
        "diagnostic": "ground truth captured through the onsite-proven ALSA arecord path",
    }


def _speaker_roundtrip_test(params: dict) -> dict:
    """Play a test tone + record the mic simultaneously; return metrics.

    params may include {frequency, duration, amplitude, output, input,
    threshold_db}. Anything missing falls back to the module constants."""
    mode = str(params.get("mode", "tone")).lower()
    playback_only = bool(params.get("playback_only", mode == "tts"))
    freq = float(params.get("frequency", SPEAKER_TEST_FREQUENCY_HZ))
    dur = float(params.get("duration", SPEAKER_TEST_DURATION_S))
    amp = float(params.get("amplitude", SPEAKER_TEST_AMPLITUDE))
    threshold_db = float(params.get("threshold_db", SPEAKER_TEST_PEAK_DB_THRESHOLD))
    # RoboVision inventory indices belong to sounddevice and may not match
    # PyAudio's index space. Prefer the reported hardware name for lookup.
    out_name = params.get("output_name") or params.get("output", os.environ.get("ROBOPARK_AUDIO_OUTPUT") or SPEAKER_TEST_OUTPUT_DEVICE)
    in_name = params.get("input_name") or params.get("input", os.environ.get("ROBOPARK_AUDIO_INPUT") or SPEAKER_TEST_INPUT_DEVICE)
    source_rate = int(params.get("sample_rate", SPEAKER_TEST_SAMPLE_RATE))
    if mode == "mic_groundtruth":
        if not sys.platform.startswith("linux"):
            return {"ok": False, "error": "mic_groundtruth requires Linux ALSA"}
        return _linux_mic_groundtruth(str(in_name), float(params.get("duration", 4.0)))
    if mode not in ("tone", "tts"):
        return {"ok": False, "error": f"unsupported speaker test mode: {mode}"}
    try:
        import pyaudio
    except ImportError as e:
        return {"ok": False, "error": f"pyaudio not installed on this robot: {e}"}
    if mode == "tone" and not (50.0 <= freq <= 8000.0):
        return {"ok": False, "error": f"frequency {freq}Hz out of allowed range (50..8000)"}
    if not (0.1 <= dur <= 3.0):
        return {"ok": False, "error": f"duration {dur}s out of allowed range (0.1..3.0)"}
    if not (0.05 <= amp <= 1.0):
        return {"ok": False, "error": f"amplitude {amp} out of allowed range (0.05..1.0)"}
    mono_pcm = None
    if mode == "tts":
        try:
            mono_pcm = base64.b64decode(params.get("audio_pcm_base64") or "", validate=True)
        except Exception as e:
            return {"ok": False, "error": f"invalid cached TTS audio: {e}"}
        if len(mono_pcm) < 480 or len(mono_pcm) % 2:
            return {"ok": False, "error": "cached TTS audio is empty or malformed"}
        dur = len(mono_pcm) / 2 / source_rate
        if dur > 10.0:
            return {"ok": False, "error": "cached TTS audio exceeds the 10 second test limit"}
    n_samples = int(source_rate * dur)
    if n_samples < 480:
        return {"ok": False, "error": "duration too short"}

    # Cached character voice is output-only. On Linux, bypass PortAudio's
    # unrelated index space and use the exact ALSA plughw endpoint selected by
    # RoboVision. This is the same path used by the successful onsite test.
    if playback_only and mono_pcm is not None and sys.platform.startswith("linux"):
        t0 = time.monotonic()
        played, detail = _linux_aplay_pcm16(mono_pcm, str(out_name), source_rate)
        duration_ms = int((time.monotonic() - t0) * 1000)
        if not played:
            return {
                "ok": False,
                "error": f"ALSA playback failed: {detail}",
                "duration_ms": duration_ms,
                "output_device": str(out_name),
                "input_device": "not opened (speaker-only test)",
                "duration": dur,
            }
        return {
            "ok": True,
            "pass": True,
            "played": True,
            "playback_only": True,
            "mode": mode,
            "text": params.get("text"),
            "cache_hit": bool(params.get("cache_hit")),
            "tts_provider": params.get("tts_provider"),
            "tts_voice": params.get("tts_voice"),
            "duration": dur,
            "duration_ms": duration_ms,
            "sample_rate": source_rate,
            "output_sample_rate": source_rate,
            "output_channels": 1,
            "output_device": detail,
            "input_device": "not opened (speaker-only test)",
            "diagnostic": "cached voice played through the selected ALSA plughw endpoint",
        }

    pa = pyaudio.PyAudio()
    out_idx = _resolve_audio_device(pa, out_name, "output")
    in_idx = None if playback_only else _resolve_audio_device(pa, in_name, "input")
    def _dev_name(idx):
        if idx is None: return "(default)"
        try: return pa.get_device_info_by_index(idx).get("name", str(idx))
        except Exception: return str(idx)
    out_label = _dev_name(out_idx)
    in_label = "not opened (speaker-only test)" if playback_only else _dev_name(in_idx)

    recorded_peak = 0
    recorded_rms = 0.0
    err = None
    output_rate = None
    input_rate = None
    output_channels = None
    input_channels = None
    t0 = time.monotonic()
    out_stream = None
    in_stream = None
    try:
        out_stream, output_rate, output_channels, out_info = _open_pcm_stream(
            pa, pyaudio, "output", out_idx, source_rate
        )
        out_label = str(out_info.get("name", out_label))
        if not playback_only:
            in_stream, input_rate, input_channels, in_info = _open_pcm_stream(
                pa, pyaudio, "input", in_idx, source_rate
            )
            in_label = str(in_info.get("name", in_label))

        if mono_pcm is not None:
            output_mono = _resample_pcm16_mono(mono_pcm, source_rate, output_rate)
        else:
            output_samples = int(output_rate * dur)
            frames = bytearray(output_samples * 2)
            peak_value = int(32767 * amp)
            for n in range(output_samples):
                env = min(1.0, n / (output_rate * 0.004),
                          (output_samples - n) / (output_rate * 0.008))
                value = int(peak_value * env * math.sin(2 * math.pi * freq * n / output_rate))
                struct.pack_into("<h", frames, n * 2, value)
            output_mono = bytes(frames)
        if output_channels == 2:
            frames = bytearray(len(output_mono) * 2)
            for pos in range(0, len(output_mono), 2):
                frames[pos * 2:pos * 2 + 4] = output_mono[pos:pos + 2] * 2
            raw = bytes(frames)
        else:
            raw = output_mono

        chunk = 1024
        if in_stream is not None:
            in_stream.start_stream()
        chunk_bytes = chunk * 2 * output_channels
        pos = 0
        peak_acc = 0
        rms_acc = 0.0
        n_samp = 0
        deadline = time.monotonic() + dur + 1.5
        while pos < len(raw) and time.monotonic() < deadline:
            end = min(pos + chunk_bytes, len(raw))
            out_stream.write(raw[pos:end])
            pos = end
            try:
                if in_stream is None:
                    continue
                avail = in_stream.get_read_available()
                if avail and avail > 0:
                    data = in_stream.read(avail, exception_on_overflow=False)
                    for s in range(0, len(data) - 1, 2):
                        v = int.from_bytes(data[s:s+2], "little", signed=True)
                        a = abs(v)
                        if a > peak_acc: peak_acc = a
                        rms_acc += v * v
                        n_samp += 1
            except Exception:
                pass
        out_stream.stop_stream(); out_stream.close(); out_stream = None
        if in_stream is not None:
            try:
                tail_frames = min(int(input_rate * 0.35), max(0, in_stream.get_read_available()))
                tail = in_stream.read(tail_frames, exception_on_overflow=False) if tail_frames else b""
                for s in range(0, len(tail) - 1, 2):
                    v = int.from_bytes(tail[s:s+2], "little", signed=True)
                    a = abs(v)
                    if a > peak_acc: peak_acc = a
                    rms_acc += v * v
                    n_samp += 1
            except Exception:
                pass
        if in_stream is not None:
            in_stream.stop_stream(); in_stream.close(); in_stream = None
        recorded_peak = peak_acc
        recorded_rms = (rms_acc / max(1, n_samp)) ** 0.5
    except Exception as e:
        err = f"{type(e).__name__}: {e}"
    finally:
        for stream in (out_stream, in_stream):
            if stream is not None:
                try: stream.stop_stream()
                except Exception: pass
                try: stream.close()
                except Exception: pass
        try: pa.terminate()
        except Exception: pass
    duration_ms = int((time.monotonic() - t0) * 1000)
    if err:
        return {"ok": False, "error": err, "duration_ms": duration_ms,
                "output_device": out_label, "input_device": in_label,
                "frequency": freq, "duration": dur}
    if playback_only:
        return {
            "ok": True,
            "pass": True,
            "played": True,
            "playback_only": True,
            "mode": mode,
            "text": params.get("text"),
            "cache_hit": bool(params.get("cache_hit")),
            "tts_provider": params.get("tts_provider"),
            "tts_voice": params.get("tts_voice"),
            "duration": dur,
            "duration_ms": duration_ms,
            "sample_rate": source_rate,
            "output_sample_rate": output_rate,
            "output_channels": output_channels,
            "output_device": out_label,
            "input_device": in_label,
            "diagnostic": "speaker playback completed; microphone was intentionally not opened",
        }
    if recorded_peak > 0:
        peak_db = 20.0 * math.log10(recorded_peak / 32767.0)
    else:
        peak_db = -120.0
    if recorded_rms > 0:
        rms_db = 20.0 * math.log10(recorded_rms / 32767.0)
    else:
        rms_db = -120.0
    pass_ = peak_db >= threshold_db
    return {
        "ok": True,
        "pass": pass_,
        "mode": mode,
        "text": params.get("text"),
        "cache_hit": bool(params.get("cache_hit")),
        "tts_provider": params.get("tts_provider"),
        "tts_voice": params.get("tts_voice"),
        "frequency": freq,
        "duration": dur,
        "duration_ms": duration_ms,
        "sample_rate": source_rate,
        "output_sample_rate": output_rate,
        "input_sample_rate": input_rate,
        "output_channels": output_channels,
        "input_channels": input_channels,
        "amplitude": amp,
        "output_device": out_label,
        "input_device": in_label,
        "recorded_peak": int(recorded_peak),
        "recorded_peak_db": round(peak_db, 1),
        "recorded_rms": round(recorded_rms, 1),
        "recorded_rms_db": round(rms_db, 1),
        "threshold_db": threshold_db,
        "diagnostic": ("speaker + mic round-trip OK" if pass_ else
                       "no signal detected — check that the speaker is on, the mic isn't muted, and the right output/input devices are selected"),
    }


def run(config_path: Path, commands_only: bool = False) -> None:
    global _CURRENT_LOG_DIR
    if commands_only:
        # Unified robot-runtime already owns preview, vision, audio and motors.
        # Do not require a legacy supervisor config or launch duplicate owners.
        services, log_dir = [], LOG_DIR
    else:
        services, log_dir = _load_config(config_path)
    _CURRENT_LOG_DIR = log_dir
    enabled = [s for s in services if s.enabled]
    if not enabled and not commands_only:
        raise SystemExit(f"No enabled services in {config_path} — nothing to supervise.")

    states = [ServiceState(spec=s) for s in enabled]
    for state in states:
        _spawn(state, log_dir)

    shutdown = {"flag": False}

    def _on_signal(signum, frame):
        logger.info(f"received signal {signum}, shutting down all services…")
        shutdown["flag"] = True

    signal.signal(signal.SIGINT, _on_signal)
    signal.signal(signal.SIGTERM, _on_signal)

    if commands_only:
        logger.info("command-only supervisor active; waiting for scheduler requests")
    else:
        logger.info(f"supervising {len(states)} service(s): {', '.join(s.spec.name for s in states)}")

    last_status_report_at = 0.0
    try:
        while not shutdown["flag"]:
            now = time.monotonic()
            for state in states:
                if state.proc is None:
                    if not state.stopped and now >= state.next_restart_at:
                        _spawn(state, log_dir)
                    continue
                ret = state.proc.poll()
                if ret is None:
                    continue  # still running
                # Exited. A long uptime before exit resets the backoff —
                # otherwise a service that's crash-looping keeps backing off.
                uptime = now - state.started_at
                state.last_exit_code = ret
                if uptime >= STABLE_UPTIME_SECONDS:
                    state.failure_count = 0
                else:
                    state.failure_count += 1
                if state.log_file:
                    state.log_file.write(f"=== supervisor: {state.spec.name} exited with code {ret} (uptime {uptime:.0f}s) ===\n")
                    state.log_file.close()
                    state.log_file = None
                state.proc = None
                _schedule_restart(state)
            if now - last_status_report_at >= STATUS_REPORT_INTERVAL_SECONDS:
                last_status_report_at = now
                identity = _load_robopark_identity()
                if identity:
                    commands = _report_status(states, identity)
                    by_name = {s.spec.name: s for s in states}
                    for cmd in commands:
                        kind = cmd.get("kind", "supervisor_action")
                        target = by_name.get(cmd.get("service_name"))
                        if kind == "supervisor_action" and target:
                            _execute_command(target, cmd.get("action", "restart"))
                        elif kind == "supervisor_action":
                            _execute_systemd_action(cmd.get("service_name", ""), cmd.get("action", "restart"))
                        elif kind in ("shell_run", "tail_logs", "speaker_test", "motor_discover", "motor_sequence"):
                            # run on a background thread so we don't block
                            # the 2s poll loop on a slow command
                            import threading
                            threading.Thread(target=_handle_shell_command, args=(identity, cmd), daemon=True).start()
                        else:
                            logger.warning(f"unknown remote command kind: {kind!r}")
                else:
                    logger.debug("not enrolled yet (no ~/.robopark/preview_agent.json + device_token) -- skipping status report")
            time.sleep(POLL_INTERVAL_SECONDS)
    finally:
        for state in states:
            _terminate(state)
        logger.info("all services stopped, exiting")


def main() -> None:
    # stdout/stderr are fully buffered (not line-buffered) once redirected to
    # a file or pipe -- which is exactly how Task Scheduler/systemd run this.
    # Without this, restart/crash log lines can sit unflushed for minutes.
    sys.stdout.reconfigure(line_buffering=True)
    sys.stderr.reconfigure(line_buffering=True)
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG_FILE, help="path to supervisor.json")
    parser.add_argument("--commands-only", action="store_true",
                        help="consume remote commands without launching duplicate robot services")
    args = parser.parse_args()
    run(args.config, commands_only=args.commands_only)


if __name__ == "__main__":
    main()