#!/usr/bin/env python3
"""Direct motion/voice-triggered ElevenLabs conversation runtime for a robot."""
from __future__ import annotations
import argparse
import audioop
import hmac
import json
import math
import os
import queue
import re
import signal
import shutil
import subprocess
import threading
import time
import urllib.parse
import urllib.request
import uuid
from collections import deque
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Callable
import pyaudio
from elevenlabs.client import ElevenLabs
from elevenlabs.conversational_ai.conversation import AudioInterface, ClientTools, Conversation
from supervisor_store import SupervisorStore
SDK_RATE = 16000
SAMPLE_WIDTH = 2
TRIGGER_CHUNK = 2048
OUTPUT_QUEUE_CHUNKS = 256
SESSION_STOP_GRACE_SECONDS = 3.0
SESSION_FORCE_CLOSE_SECONDS = 2.0
SPEAKER_ECHO_TAIL_SECONDS = 0.7
PENDING_MOTION_TTL_SECONDS = 15.0
def log(message: str) -> None:
print(f"[conversation] {message}", flush=True)
def pcm_rms(data: bytes) -> int:
if not data:
return 0
count = len(data) // SAMPLE_WIDTH
if count == 0:
return 0
total = 0
for index in range(0, count * SAMPLE_WIDTH, SAMPLE_WIDTH):
sample = int.from_bytes(data[index:index + SAMPLE_WIDTH], "little", signed=True)
total += sample * sample
return int(math.sqrt(total / count))
def find_device(pa: pyaudio.PyAudio, selector: str, direction: str) -> tuple[int, dict]:
required_channels = "maxInputChannels" if direction == "input" else "maxOutputChannels"
candidates: list[tuple[int, dict]] = []
for index in range(pa.get_device_count()):
info = pa.get_device_info_by_index(index)
if int(info.get(required_channels, 0)) > 0:
candidates.append((index, info))
if not candidates:
raise RuntimeError(f"no audio {direction} devices found")
value = str(selector).strip()
if value.casefold() in {"default", "system default"}:
try:
info = (
pa.get_default_input_device_info()
if direction == "input"
else pa.get_default_output_device_info()
)
return int(info["index"]), info
except Exception as exc:
raise RuntimeError(f"default audio {direction} device is unavailable: {exc}") from exc
if value.isdigit():
wanted = int(value)
for index, info in candidates:
if index == wanted:
return index, info
raise RuntimeError(f"audio {direction} device index {wanted} is unavailable")
normalized = value.casefold()
exact = [(index, info) for index, info in candidates if str(info["name"]).casefold() == normalized]
partial = [(index, info) for index, info in candidates if normalized in str(info["name"]).casefold()]
matches = exact or partial
if not matches:
available = ", ".join(f"{index}:{info['name']}" for index, info in candidates)
raise RuntimeError(f"audio {direction} device '{selector}' not found; available: {available}")
return matches[0]
def candidate_rates(info: dict) -> list[int]:
values = [int(float(info.get("defaultSampleRate", 0))), 48000, 44100, 32000, 16000]
return list(dict.fromkeys(rate for rate in values if rate > 0))
def stable_audio_label(selector: str) -> str:
"""Remove the boot-order ALSA suffix while preserving product identity."""
return re.sub(r"\s*\((?:plug)?hw:\d+,\d+\)\s*$", "", str(selector), flags=re.I).strip()
def resolve_alsa_capture(selector: str) -> str | None:
"""Resolve a logical USB microphone label to its current ALSA card."""
if os.name != "posix" or not shutil.which("arecord"):
return None
label = stable_audio_label(selector)
parts = [part.strip() for part in label.split(":") if part.strip() not in {"", "-"}]
try:
output = subprocess.run(
["arecord", "-l"], capture_output=True, text=True, timeout=3, check=False
).stdout
except Exception:
return None
candidates = []
for line in output.splitlines():
match = re.match(r"^card\s+(\d+):.*device\s+(\d+):", line.strip(), re.I)
if not match:
continue
folded = line.casefold()
if parts and not all(part.casefold() in folded for part in parts):
continue
# Exact case is a useful discriminator for the fleet's two generic
# USB products ("Usb Audio Device" mic vs "USB Audio Device" output).
case_matches = sum(part in line for part in parts)
candidates.append((case_matches, f"plughw:{match.group(1)},{match.group(2)}"))
if not candidates:
return None
candidates.sort(reverse=True)
return candidates[0][1]
def audio_inventory() -> dict:
pa = pyaudio.PyAudio()
try:
inputs = []
outputs = []
for index in range(pa.get_device_count()):
info = pa.get_device_info_by_index(index)
row = {
"index": index,
"name": str(info.get("name", "unknown")),
"rate": int(float(info.get("defaultSampleRate", 0))),
}
if int(info.get("maxInputChannels", 0)) > 0:
inputs.append({**row, "channels": int(info["maxInputChannels"])})
if int(info.get("maxOutputChannels", 0)) > 0:
outputs.append({**row, "channels": int(info["maxOutputChannels"])})
return {"inputs": inputs, "outputs": outputs}
finally:
pa.terminate()
def media_lease_status() -> dict:
"""Report ALSA owners without opening or terminating any media device."""
owners: dict[int, dict] = {}
if os.name == "posix" and Path("/proc").exists():
for process_dir in Path("/proc").glob("[0-9]*"):
try:
pid = int(process_dir.name)
devices = sorted({
os.readlink(fd)
for fd in (process_dir / "fd").iterdir()
if os.readlink(fd).startswith("/dev/snd/")
})
if not devices:
continue
command = (process_dir / "cmdline").read_bytes().replace(b"\0", b" ").decode(
"utf-8", errors="replace"
).strip()
owners[pid] = {"pid": pid, "command": command, "devices": devices}
except (FileNotFoundError, PermissionError, ProcessLookupError, OSError, ValueError):
continue
rows = list(owners.values())
foreign = [row for row in rows if row["pid"] != os.getpid()]
return {
"authoritative_pid": os.getpid(),
"available": not foreign,
"owners": rows,
"conflicts": foreign,
}
class NativeAudioInterface(AudioInterface):
"""PyAudio interface with deterministic devices and native-rate conversion."""
def __init__(self, input_selector: str, output_selector: str) -> None:
self.input_selector = input_selector
self.output_selector = output_selector
self.pa: pyaudio.PyAudio | None = None
self.in_stream = None
self.input_process: subprocess.Popen | None = None
self.input_backend = "pyaudio"
self.input_name = input_selector
self.output_name = output_selector
self.out_stream = None
self.input_callback: Callable[[bytes], None] | None = None
self.output_queue: queue.Queue[bytes] = queue.Queue(maxsize=OUTPUT_QUEUE_CHUNKS)
self.stop_event = threading.Event()
self.input_thread: threading.Thread | None = None
self.output_thread: threading.Thread | None = None
self.input_frames_per_buffer = 320
self.input_rate = SDK_RATE
self.output_rate = SDK_RATE
self.output_channels = 1
self.input_rate_state = None
self.output_rate_state = None
self.lock = threading.Lock()
self.started = threading.Event()
self.output_chunks = 0
self.output_bytes = 0
self.input_chunks = 0
self.forwarded_input_chunks = 0
self.input_peak = 0
self.suppressed_input_chunks = 0
self.input_read_errors = 0
self.last_input_error: str | None = None
self.output_write_errors = 0
self.last_output_error: str | None = None
self.fatal_error = threading.Event()
self.fatal_error_message: str | None = None
self.suppress_input_until = 0.0
self.echo_lock = threading.Lock()
self.echo_gate_announced = False
def _suppress_input_for(self, seconds: float) -> None:
with self.echo_lock:
self.suppress_input_until = max(
self.suppress_input_until,
time.monotonic() + max(0.0, seconds),
)
def _input_is_suppressed(self) -> bool:
with self.echo_lock:
return time.monotonic() < self.suppress_input_until
def _open_input(self, index: int, info: dict, alsa_device: str | None = None):
alsa_match = re.search(r"\((?:plug)?hw:(\d+),(\d+)\)", str(info.get("name", "")), re.I)
if alsa_device is None and alsa_match:
alsa_device = f"plughw:{alsa_match.group(1)},{alsa_match.group(2)}"
if os.name == "posix" and alsa_device and shutil.which("arecord"):
device = alsa_device
command = [
"arecord", "-q", "-D", device, "-t", "raw", "-f", "S16_LE",
"-r", "48000", "-c", "1", "--period-size", "960",
]
process = subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
bufsize=0,
)
time.sleep(0.15)
if process.poll() is not None:
raise RuntimeError(f"microphone arecord failed on {device} (exit {process.returncode})")
self.input_process = process
self.input_backend = f"arecord {device}"
# This is the same onsite-proven path used by mic ground-truth
# tests. Several fleet USB interfaces expose near-silent samples
# when ALSA is asked to capture at 16 kHz directly. Capture at
# 48 kHz and resample to the ElevenLabs 16 kHz SDK format below.
self.input_rate = 48000
self.input_frames_per_buffer = int(self.input_rate / 50)
return None
errors = []
for rate in candidate_rates(info):
try:
frames_per_buffer = max(256, int(rate / 50))
stream = self.pa.open(
format=pyaudio.paInt16,
channels=1,
rate=rate,
input=True,
input_device_index=index,
frames_per_buffer=frames_per_buffer,
start=True,
)
self.input_rate = rate
self.input_frames_per_buffer = frames_per_buffer
self.input_backend = "pyaudio"
return stream
except Exception as exc:
errors.append(f"{rate}Hz: {exc}")
raise RuntimeError(f"microphone open failed ({'; '.join(errors)})")
def _open_output(self, index: int, info: dict):
errors = []
max_channels = max(1, int(info.get("maxOutputChannels", 1)))
for channels in ([2, 1] if max_channels >= 2 else [1]):
for rate in candidate_rates(info):
try:
stream = self.pa.open(
format=pyaudio.paInt16,
channels=channels,
rate=rate,
output=True,
output_device_index=index,
frames_per_buffer=max(256, int(rate / 16)),
start=True,
)
self.output_rate = rate
self.output_channels = channels
return stream
except Exception as exc:
errors.append(f"{rate}Hz/{channels}ch: {exc}")
raise RuntimeError(f"speaker open failed ({'; '.join(errors)})")
def start(self, input_callback: Callable[[bytes], None]):
with self.lock:
if self.started.is_set():
return
self.input_callback = input_callback
self.stop_event.clear()
self.pa = pyaudio.PyAudio()
input_label = stable_audio_label(self.input_selector)
resolved_capture = resolve_alsa_capture(input_label)
if resolved_capture:
input_index = -1
input_info = {"name": input_label}
log(f"microphone identity resolved: {input_label} -> {resolved_capture}")
else:
input_index, input_info = find_device(self.pa, input_label, "input")
output_index, output_info = find_device(
self.pa, stable_audio_label(self.output_selector), "output"
)
self.input_name = str(input_info["name"])
self.output_name = str(output_info["name"])
try:
self.in_stream = self._open_input(input_index, input_info, resolved_capture)
self.out_stream = self._open_output(output_index, output_info)
except Exception:
self._close_streams()
raise
self.output_thread = threading.Thread(
target=self._output_loop, name="robot-speaker", daemon=True
)
self.input_thread = threading.Thread(
target=self._input_loop, name="robot-microphone", daemon=True
)
self.output_thread.start()
self.input_thread.start()
self.started.set()
log(
f"audio active: mic {input_info['name']} at {self.input_rate}Hz; "
f"capture {self.input_backend}; speaker {output_info['name']} "
f"at {self.output_rate}Hz/{self.output_channels}ch"
)
def status(self) -> dict:
return {
"active": self.started.is_set() and not self.stop_event.is_set(),
"input": self.input_name,
"input_backend": self.input_backend,
"input_rate": self.input_rate,
"input_chunks": self.input_chunks,
"forwarded_input_chunks": self.forwarded_input_chunks,
"suppressed_input_chunks": self.suppressed_input_chunks,
"input_peak": self.input_peak,
"input_read_errors": self.input_read_errors,
"last_input_error": self.last_input_error,
"output": self.output_name,
"output_rate": self.output_rate,
"output_channels": self.output_channels,
"output_chunks": self.output_chunks,
"output_bytes": self.output_bytes,
"output_write_errors": self.output_write_errors,
"last_output_error": self.last_output_error,
"fatal_error": self.fatal_error_message,
"echo_gate_active": self._input_is_suppressed(),
}
def _input_loop(self) -> None:
read_errors = 0
while not self.stop_event.is_set():
try:
if self.input_process is not None:
data = self.input_process.stdout.read(self.input_frames_per_buffer * SAMPLE_WIDTH)
if not data and self.input_process.poll() is not None:
if self.stop_event.is_set():
break
raise RuntimeError(f"arecord exited with code {self.input_process.returncode}")
else:
data = self.in_stream.read(
self.input_frames_per_buffer,
exception_on_overflow=False,
)
read_errors = 0
except Exception as exc:
read_errors += 1
self.input_read_errors += 1
self.last_input_error = str(exc)
if read_errors <= 3 or read_errors % 50 == 0:
log(f"microphone read failed ({read_errors}): {exc}")
if read_errors >= 3:
self.fatal_error_message = f"microphone capture failed repeatedly: {exc}"
self.fatal_error.set()
break
time.sleep(0.02)
continue
if not data:
continue
self.input_chunks += 1
try:
self.input_peak = max(self.input_peak, audioop.max(data, SAMPLE_WIDTH))
except Exception:
pass
if self._input_is_suppressed() or not self.output_queue.empty():
self.suppressed_input_chunks += 1
continue
try:
converted, self.input_rate_state = audioop.ratecv(
data, SAMPLE_WIDTH, 1, self.input_rate, SDK_RATE, self.input_rate_state
)
if converted and self.input_callback:
self.input_callback(converted)
self.forwarded_input_chunks += 1
except Exception as exc:
log(f"microphone callback failed: {exc}")
def output(self, audio: bytes):
if self.stop_event.is_set() or not audio:
return
# ElevenLabs delivers mono PCM16 at SDK_RATE. Gate capture for the
# queued playback duration plus room/speaker decay so the agent cannot
# transcribe its own response and recursively answer itself.
self._suppress_input_for((len(audio) / (SDK_RATE * SAMPLE_WIDTH)) + SPEAKER_ECHO_TAIL_SECONDS)
if not self.echo_gate_announced:
self.echo_gate_announced = True
log("speaker echo gate active; microphone STT paused during playback")
try:
self.output_queue.put_nowait(audio)
except queue.Full:
# Bound latency and memory if the hardware stalls. Fresh speech is
# more useful than replaying stale buffered audio after recovery.
try:
self.output_queue.get_nowait()
except queue.Empty:
pass
try:
self.output_queue.put_nowait(audio)
except queue.Full:
pass
def _output_loop(self) -> None:
write_errors = 0
while not self.stop_event.is_set():
try:
data = self.output_queue.get(timeout=0.2)
except queue.Empty:
continue
try:
self._suppress_input_for(
(len(data) / (SDK_RATE * SAMPLE_WIDTH)) + SPEAKER_ECHO_TAIL_SECONDS
)
converted, self.output_rate_state = audioop.ratecv(
data, SAMPLE_WIDTH, 1, SDK_RATE, self.output_rate, self.output_rate_state
)
if self.output_channels == 2:
converted = audioop.tostereo(converted, SAMPLE_WIDTH, 1, 1)
self.out_stream.write(converted, exception_on_underflow=False)
write_errors = 0
self._suppress_input_for(SPEAKER_ECHO_TAIL_SECONDS)
self.output_chunks += 1
self.output_bytes += len(data)
if self.output_chunks == 1:
log("speaker received first agent audio frame")
except Exception as exc:
write_errors += 1
self.output_write_errors += 1
self.last_output_error = str(exc)
log(f"speaker write failed: {exc}")
if write_errors >= 3:
self.fatal_error_message = f"speaker playback failed repeatedly: {exc}"
self.fatal_error.set()
break
def interrupt(self):
while True:
try:
self.output_queue.get_nowait()
except queue.Empty:
break
self.output_rate_state = None
self._suppress_input_for(SPEAKER_ECHO_TAIL_SECONDS)
log("agent playback interrupted")
def _close_streams(self) -> None:
if self.input_process is not None:
try:
self.input_process.terminate()
self.input_process.wait(timeout=1)
except Exception:
try:
self.input_process.kill()
except Exception:
pass
self.input_process = None
for stream in (self.in_stream, self.out_stream):
if stream is not None:
try:
stream.stop_stream()
except Exception:
pass
try:
stream.close()
except Exception:
pass
self.in_stream = None
self.out_stream = None
if self.pa is not None:
try:
self.pa.terminate()
except Exception:
pass
self.pa = None
def stop(self):
with self.lock:
if self.stop_event.is_set() and not self.started.is_set():
return
self.stop_event.set()
if self.input_process is not None:
try:
self.input_process.terminate()
except Exception:
pass
if self.input_thread and self.input_thread is not threading.current_thread():
self.input_thread.join(timeout=2)
if self.output_thread and self.output_thread is not threading.current_thread():
self.output_thread.join(timeout=2)
self._close_streams()
self.started.clear()
self.input_callback = None
while True:
try:
self.output_queue.get_nowait()
except queue.Empty:
break
class VoiceTrigger:
def __init__(self, selector: str, threshold: int, trigger: Callable[[str], None]) -> None:
self.selector = selector
self.threshold = threshold
self.trigger = trigger
self.stop_event = threading.Event()
self.thread: threading.Thread | None = None
self.stream = None
self.pa: pyaudio.PyAudio | None = None
def start(self) -> None:
if self.thread and self.thread.is_alive():
return
self.stop_event.clear()
self.thread = threading.Thread(target=self._run, name="voice-trigger", daemon=True)
self.thread.start()
def stop(self) -> None:
self.stop_event.set()
if self.stream is not None:
try:
self.stream.stop_stream()
self.stream.close()
except Exception:
pass
self.stream = None
if self.thread and self.thread is not threading.current_thread():
self.thread.join(timeout=2)
if self.pa is not None:
try:
self.pa.terminate()
except Exception:
pass
self.pa = None
def _run(self) -> None:
try:
self.pa = pyaudio.PyAudio()
index, info = find_device(self.pa, self.selector, "input")
last_error = None
for rate in candidate_rates(info):
try:
self.stream = self.pa.open(
format=pyaudio.paInt16,
channels=1,
rate=rate,
input=True,
input_device_index=index,
frames_per_buffer=TRIGGER_CHUNK,
start=True,
)
log(f"voice trigger armed on {info['name']} at {rate}Hz")
break
except Exception as exc:
last_error = exc
if self.stream is None:
raise RuntimeError(f"could not open trigger microphone: {last_error}")
consecutive = 0
warmup = 3
while not self.stop_event.is_set():
data = self.stream.read(TRIGGER_CHUNK, exception_on_overflow=False)
if warmup:
warmup -= 1
continue
level = pcm_rms(data)
consecutive = consecutive + 1 if level >= self.threshold else 0
if consecutive >= 3:
log(f"voice trigger detected (RMS {level})")
self.trigger("voice")
return
except Exception as exc:
if not self.stop_event.is_set():
log(f"voice trigger unavailable: {exc}")
class MotorRegistry:
def __init__(self, motors: dict[str, int], active_high: bool, default_pulse_ms: int) -> None:
self.motors = motors
self.active_high = active_high
self.default_pulse_ms = default_pulse_ms
self.devices = {}
self.lock = threading.Lock()
def names(self) -> list[str]:
return sorted(self.motors)
def _device(self, name: str):
normalized = str(name).strip().lower().replace("_", "-")
if normalized not in self.motors:
raise ValueError(
f"unknown motor '{name}'; registered motors: {', '.join(self.names()) or 'none'}"
)
if normalized not in self.devices:
from gpiozero import OutputDevice
self.devices[normalized] = OutputDevice(
self.motors[normalized],
active_high=self.active_high,
initial_value=False,
)
return normalized, self.devices[normalized]
def execute(self, parameters: dict) -> dict:
name = parameters.get("name") or parameters.get("motor")
action = str(parameters.get("action") or "pulse").strip().lower()
requested_ms = parameters.get("duration_ms", self.default_pulse_ms)
try:
duration_ms = int(requested_ms)
except (TypeError, ValueError):
duration_ms = self.default_pulse_ms
duration_ms = max(50, min(duration_ms, 3000))
with self.lock:
normalized, device = self._device(name)
pin = self.motors[normalized]
if action == "off":
device.off()
log(f"motor {normalized} GPIO{pin}: OFF")
elif action in {"pulse", "on", "activate", "move"}:
log(f"motor {normalized} GPIO{pin}: ON for {duration_ms}ms")
device.on()
time.sleep(duration_ms / 1000)
device.off()
log(f"motor {normalized} GPIO{pin}: OFF")
else:
raise ValueError("action must be pulse, on, activate, move, or off")
return {
"ok": True,
"motor": normalized,
"gpio": pin,
"action": action,
"duration_ms": 0 if action == "off" else duration_ms,
}
def close(self) -> None:
for device in self.devices.values():
try:
device.off()
device.close()
except Exception:
pass
self.devices.clear()
class SchedulerReporter:
"""Best-effort central oversight for robot-local ElevenLabs calls."""
def __init__(self, config: dict, store: SupervisorStore) -> None:
self.base_url = str(config.get("scheduler_url") or "").rstrip("/")
self.device_id = str(config.get("scheduler_device_id") or "").strip()
self.device_token = str(config.get("scheduler_device_token") or "").strip()
if not (self.base_url and self.device_id and self.device_token):
try:
config_dir = Path.home() / ".robopark"
preview = json.loads((config_dir / "preview_agent.json").read_text(encoding="utf-8"))
self.base_url = str(preview.get("scheduler_url") or self.base_url).rstrip("/")
self.device_id = str(preview.get("device_id") or self.device_id).strip()
token_path = config_dir / "device_token"
self.device_token = (
token_path.read_text(encoding="utf-8").strip()
if token_path.exists()
else str(preview.get("device_token") or self.device_token).strip()
)
except Exception:
pass
self.agent_id = str(config.get("agent_id") or "").strip()
self.branch_id = str(config.get("branch_id") or "").strip() or None
self.session_id: str | None = None
self.event_token: str | None = None
self.sequence = 0
self.lock = threading.Lock()
self.store = store
self.command_handler: Callable[[dict], dict] | None = None
self.state_provider: Callable[[], dict] | None = None
self.stop_event = threading.Event()
self.jobs: queue.Queue[tuple[str, dict, dict] | None] = queue.Queue(maxsize=256)
self.worker: threading.Thread | None = None
@property
def enabled(self) -> bool:
return bool(self.base_url and self.device_id and self.device_token)
def _post(self, path: str, payload: dict, headers: dict | None = None) -> dict:
request = urllib.request.Request(
f"{self.base_url}{path}",
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json", **(headers or {})},
method="POST",
)
with urllib.request.urlopen(request, timeout=4) as response:
body = response.read().decode("utf-8")
return json.loads(body) if body else {}
def _get(self, path: str, headers: dict | None = None) -> dict:
request = urllib.request.Request(
f"{self.base_url}{path}", headers=headers or {}, method="GET"
)
with urllib.request.urlopen(request, timeout=4) as response:
body = response.read().decode("utf-8")
return json.loads(body) if body else {}
def _sync_durable_state(self) -> None:
if not self.enabled:
return
headers = {"Authorization": f"Bearer {self.device_token}"}
if self.state_provider is not None:
self._post(
f"/api/devices/{urllib.parse.quote(self.device_id, safe='')}/voice-runtime-state",
self.state_provider(), headers,
)
events = self.store.events_after(0, 250)
if events:
result = self._post(
f"/api/devices/{urllib.parse.quote(self.device_id, safe='')}/telemetry/events",
{"events": events},
headers,
)
acknowledged = int(result.get("acknowledged_sequence") or 0)
if acknowledged:
self.store.acknowledge(acknowledged)
commands = self._get(
f"/api/devices/{urllib.parse.quote(self.device_id, safe='')}/commands/pending",
headers,
).get("commands") or []
for command in commands:
command_id = str(command.get("command_id") or "")
if not command_id:
continue
previous = self.store.command_result(command_id)
if previous:
self._post(
f"/api/devices/{urllib.parse.quote(self.device_id, safe='')}/commands/{command_id}/ack",
{"status": previous["status"], "result": previous["result"]},
headers,
)
continue
self._post(
f"/api/devices/{urllib.parse.quote(self.device_id, safe='')}/commands/{command_id}/ack",
{"status": "acknowledged", "result": {}}, headers,
)
self._post(
f"/api/devices/{urllib.parse.quote(self.device_id, safe='')}/commands/{command_id}/ack",
{"status": "running", "result": {}}, headers,
)
try:
if self.command_handler is None:
raise RuntimeError("local command handler is unavailable")
result = self.command_handler(command)
status = "succeeded" if result.get("ok") else "failed"
except Exception as exc:
result = {"ok": False, "error": str(exc)}
status = "failed"
self.store.complete_command(command_id, status, result)
self._post(
f"/api/devices/{urllib.parse.quote(self.device_id, safe='')}/commands/{command_id}/ack",
{"status": status, "result": result}, headers,
)
def _run(self) -> None:
while not self.stop_event.is_set():
try:
job = self.jobs.get(timeout=2)
except queue.Empty:
job = ()
if job is None:
return
if job:
path, payload, headers = job
try:
self._post(path, payload, headers)
except Exception as exc:
log(f"central oversight write failed; event remains local: {exc}")
try:
self._sync_durable_state()
except Exception:
# Gateway connectivity never changes local voice readiness.
pass
def start_worker(self) -> None:
if self.worker is None or not self.worker.is_alive():
self.worker = threading.Thread(target=self._run, name="management-sync", daemon=True)
self.worker.start()
def _enqueue(self, path: str, payload: dict, headers: dict) -> None:
if not self.enabled:
return
self.start_worker()
try:
self.jobs.put_nowait((path, payload, headers))
except queue.Full:
log("central oversight queue full; dropping telemetry event")
def start(self, reason: str) -> None:
if not self.enabled:
return
with self.lock:
self.session_id = None
self.event_token = None
self.sequence = 0
try:
result = self._post(
f"/api/devices/{urllib.parse.quote(self.device_id, safe='')}/direct-voice/sessions",
{"agent_id": self.agent_id, "branch_id": self.branch_id, "trigger_reason": reason},
{"Authorization": f"Bearer {self.device_token}"},
)
with self.lock:
self.session_id = str(result.get("session_id") or "") or None
self.event_token = str(result.get("event_token") or "") or None
if self.session_id:
log(f"central oversight session registered: {self.session_id}")
except Exception as exc:
log(f"central oversight unavailable; continuing locally: {exc}")
def event(self, stage: str, status: str, message: str, details: dict | None = None) -> None:
pipeline_stage = {
"session_starting": "voice_worker",
"provider_connected": "voice_worker",
"user_transcript": "stt_listening",
"agent_response": "llm_response",
"session_error": "voice_worker",
"session_disposed": "session_ended",
}.get(stage)
if pipeline_stage is None:
return
pipeline_status = "running" if status == "active" else status
with self.lock:
session_id, token = self.session_id, self.event_token
if not session_id or not token:
return
self._enqueue(
f"/api/sessions/{urllib.parse.quote(session_id, safe='')}/pipeline-events",
{"stage": pipeline_stage, "status": pipeline_status, "message": message, "source": "robot_hardware", "details": details or {}},
{"X-RoboPark-Session-Token": token},
)
def turn(self, role: str, text: str) -> None:
# Transcript delivery is exclusively cursor-based through the durable
# spool. Posting it here as well produced duplicate historical turns.
return
def end(self, reason: str, duration: float, error: str | None) -> None:
with self.lock:
session_id = self.session_id
self.session_id = None
self.event_token = None
if not session_id or not self.enabled:
return
self._enqueue(
f"/api/devices/{urllib.parse.quote(self.device_id, safe='')}/direct-voice/sessions/{urllib.parse.quote(session_id, safe='')}/end",
{"reason": reason, "duration_seconds": duration, "error": error},
{"Authorization": f"Bearer {self.device_token}"},
)
class Runtime:
def __init__(self, config: dict) -> None:
self.config = config
# Session disposal is local and deterministic. Recycling the systemd
# worker after every timeout caused port races and stale ALSA owners.
self.config["recycle_after_session"] = False
self.shutdown = threading.Event()
self.trigger_event = threading.Event()
self.trigger_reason = "unknown"
self.state_lock = threading.Lock()
self.active = False
self.last_session_end = 0.0
self.session_started_at = 0.0
self.sessions_started = 0
self.sessions_completed = 0
self.triggers_accepted = 0
self.triggers_rejected = 0
self.pending_motion_at = 0.0
self.last_error: str | None = None
self.telemetry_last_error: str | None = None
self.vision_routed = False
self.vision_last_error: str | None = None
self.vision_process: subprocess.Popen | None = None
self.vision_process_started = False
self.vision_restarts = 0
self.recycle_required = False
self.force_stop_event = threading.Event()
self.paused = False
self.apply_when_idle_requested = False
self.restart_when_idle_requested = False
self.stop_when_idle_requested = False
self.local_session_id: str | None = None
self.lifecycle_events: deque[dict] = deque(maxlen=80)
self.store = SupervisorStore(
str(config["config_path"]),
str(config.get("scheduler_device_id") or config.get("robot_name") or "robot"),
)
self.paused = bool(self.store.get_state("paused", False))
self._initialize_configuration_state()
self.scheduler_reporter = SchedulerReporter(config, self.store)
self.scheduler_reporter.command_handler = self.handle_management_command
self.scheduler_reporter.state_provider = self.reconciliation_state
self.last_session: dict | None = None
self.last_trigger_reason: str | None = None
self.audio_devices = audio_inventory()
self.current_audio: NativeAudioInterface | None = None
self.voice_trigger = VoiceTrigger(
config["audio_input"], int(config["voice_threshold"]), self.trigger
)
self.httpd: ThreadingHTTPServer | None = None
self.http_thread: threading.Thread | None = None
self.motor_registry = MotorRegistry(
config.get("motors") or {},
bool(config.get("motor_active_high")),
int(config.get("motor_pulse_ms") or 300),
)
self._event("runtime_ready", "ElevenLabs runtime initialized")
self.scheduler_reporter.start_worker()
def _initialize_configuration_state(self) -> None:
applied = self.store.get_state("applied_configuration")
if not applied:
applied = self._configuration_snapshot(
int(self.config.get("applied_revision") or self.config.get("desired_revision") or 1)
)
self.store.set_state("applied_configuration", applied)
self.store.set_state("last_known_good_configuration", applied)
self.config["applied_revision"] = int(applied.get("applied_revision") or 1)
self.config["desired_revision"] = int(
self.store.get_state("desired_revision", self.config["applied_revision"])
)
def _configuration_snapshot(self, revision: int | None = None) -> dict:
applied_revision = int(revision or self.config.get("applied_revision") or 1)
desired_revision = int(self.config.get("desired_revision") or applied_revision)
return {
"voice_engine_default": str(self.config.get("voice_engine_default") or "elevenlabs"),
"robovoice_enabled": bool(self.config.get("robovoice_enabled", True)),
"allow_session_override": bool(self.config.get("allow_session_override", True)),
"apply_changes_when_idle": bool(self.config.get("apply_changes_when_idle", True)),
"desired_revision": desired_revision,
"applied_revision": applied_revision,
"character_id": str(self.config.get("character_id") or self.config.get("robot_name") or ""),
"elevenlabs": {
"agent_id": str(self.config.get("agent_id") or ""),
"branch_id": str(self.config.get("branch_id") or "") or None,
},
"robovoice": {"server_id": self.config.get("robovoice_server_id")},
"audio": {
"input_identity": str(self.config.get("audio_input") or ""),
"output_identity": str(self.config.get("audio_output") or ""),
},
}
def configuration(self) -> dict:
applied = self.store.get_state("applied_configuration", self._configuration_snapshot())
staged = self.store.get_state("staged_configuration")
desired_revision = int(self.store.get_state("desired_revision", applied.get("applied_revision", 1)))
applied["desired_revision"] = desired_revision
return {
**applied,
"desired_revision": desired_revision,
"applied_revision": int(applied.get("applied_revision") or 1),
"application_state": (
"waiting_for_idle" if staged and self.active else
"staged" if staged else "applied"
),
"staged_configuration": staged,
"last_known_good_revision": int(
(self.store.get_state("last_known_good_configuration") or applied).get("applied_revision") or 1
),
}
@staticmethod
def _validate_staged_configuration(payload: dict) -> dict:
forbidden = {"api_key", "token", "secret", "password", "device_token"}
def inspect(value, path="configuration"):
if isinstance(value, dict):
for key, child in value.items():
if any(part in str(key).casefold() for part in forbidden):
raise ValueError(f"credentials are not accepted in {path}")
inspect(child, f"{path}.{key}")
elif isinstance(value, list):
for child in value:
inspect(child, path)
inspect(payload)
engine = str(payload.get("voice_engine_default") or "elevenlabs").casefold()
if engine not in {"elevenlabs", "robovoice"}:
raise ValueError("voice_engine_default must be elevenlabs or robovoice")
elevenlabs = payload.get("elevenlabs") or {}
if engine == "elevenlabs" and not str(elevenlabs.get("agent_id") or "").strip():
raise ValueError("elevenlabs.agent_id is required")
audio = payload.get("audio") or {}
if not str(audio.get("input_identity") or "").strip():
raise ValueError("audio.input_identity is required")
if not str(audio.get("output_identity") or "").strip():
raise ValueError("audio.output_identity is required")
return payload
def stage_configuration(self, payload: dict) -> dict:
current = self.configuration()
merged = {
key: value for key, value in current.items()
if key not in {"application_state", "staged_configuration", "last_known_good_revision"}
}
for key in (
"voice_engine_default", "robovoice_enabled", "allow_session_override",
"apply_changes_when_idle", "character_id", "elevenlabs", "robovoice", "audio",
):
if key in payload:
merged[key] = payload[key]
revision = max(
int(payload.get("desired_revision") or 0),
int(current["desired_revision"]) + 1,
)
merged["desired_revision"] = revision
merged["applied_revision"] = int(current["applied_revision"])
self._validate_staged_configuration(merged)
self.store.set_state("desired_revision", revision)
self.store.set_state("staged_configuration", merged)
self.config["desired_revision"] = revision
self._event("configuration_staged", f"Configuration revision {revision} staged", "pending")
return self.configuration()
def apply_staged_configuration(self) -> dict:
staged = self.store.get_state("staged_configuration")
if not staged:
return {"ok": True, "configuration": self.configuration(), "changed": False}
if self.active:
self.apply_when_idle_requested = True
return {"ok": True, "waiting_for_idle": True, "configuration": self.configuration()}
previous = self._configuration_snapshot(int(self.config.get("applied_revision") or 1))
try:
self._validate_staged_configuration(staged)
audio = staged["audio"]
elevenlabs = staged.get("elevenlabs") or {}
if staged["voice_engine_default"] == "elevenlabs":
pa = pyaudio.PyAudio()
try:
find_device(pa, audio["input_identity"], "input")
find_device(pa, audio["output_identity"], "output")
finally:
pa.terminate()
self.config.update({
"voice_engine_default": staged["voice_engine_default"],
"robovoice_enabled": bool(staged.get("robovoice_enabled", True)),
"allow_session_override": bool(staged.get("allow_session_override", True)),
"apply_changes_when_idle": bool(staged.get("apply_changes_when_idle", True)),
"character_id": staged.get("character_id"),
"agent_id": elevenlabs.get("agent_id") or self.config.get("agent_id"),
"branch_id": elevenlabs.get("branch_id"),
"audio_input": audio["input_identity"],
"audio_output": audio["output_identity"],
"desired_revision": int(staged["desired_revision"]),
"applied_revision": int(staged["desired_revision"]),
})
Path(self.config["config_path"]).write_text(
json.dumps(self.config, indent=2) + "\n", encoding="utf-8"
)
applied = self._configuration_snapshot(self.config["applied_revision"])
applied["desired_revision"] = self.config["applied_revision"]
self.store.set_state("last_known_good_configuration", previous)
self.store.set_state("applied_configuration", applied)
self.store.set_state("desired_revision", self.config["applied_revision"])
self.store.set_state("staged_configuration", None)
self.audio_devices = audio_inventory()
self.restart_when_idle_requested = True
self._event("configuration_applied", f"Configuration revision {self.config['applied_revision']} applied")
return {"ok": True, "changed": True, "configuration": self.configuration()}
except Exception as exc:
self.store.set_state("applied_configuration", previous)
self._event("configuration_rollback", f"Staged configuration rejected: {exc}", "failed")
return {"ok": False, "error": str(exc), "rolled_back": True, "configuration": self.configuration()}
def _event(self, stage: str, message: str, status: str = "ok", **details) -> None:
event = {
"timestamp": time.time(),
"stage": stage,
"status": status,
"message": message,
**details,
}
self.lifecycle_events.append(event)
self._spool_event(f"lifecycle.{stage}", event)
self.scheduler_reporter.event(stage, status, message, details)
def _spool_event(self, event_type: str, payload: dict) -> None:
try:
self.store.append_event(event_type, payload, self.local_session_id)
except Exception as exc:
# Observability storage can degrade, but it cannot break media or
# provider callbacks. Health exposes the failure for management.
self.telemetry_last_error = str(exc)
log(f"telemetry spool write failed; voice continues locally: {exc}")
def status(self) -> dict:
with self.state_lock:
cooldown_remaining = max(
0.0,
float(self.config["cooldown"]) - (time.monotonic() - self.last_session_end),
) if self.last_session_end else 0.0
return {
"engine": "elevenlabs",
"effective_engine": self.configuration()["voice_engine_default"],
"engine_source": "robot",
"fallback_available": bool(self.config.get("robovoice_enabled", True)),
"robot_name": self.config.get("robot_name"),
"agent_id": self.config.get("agent_id"),
"branch_id": self.config.get("branch_id"),
"central_oversight": {
"enabled": self.scheduler_reporter.enabled,
"session_id": self.scheduler_reporter.session_id,
"queued_writes": self.scheduler_reporter.jobs.qsize(),
"telemetry": self.store.stats(),
},
"configuration": self.configuration(),
"paused": self.paused,
"active": self.active,
"queued": self.trigger_event.is_set(),
"state": (
"active" if self.active else
("queued" if self.trigger_event.is_set() else
("cooldown" if cooldown_remaining else "armed"))
),
"cooldown_remaining_ms": int(cooldown_remaining * 1000),
"sessions_started": self.sessions_started,
"sessions_completed": self.sessions_completed,
"triggers_accepted": self.triggers_accepted,
"triggers_rejected": self.triggers_rejected,
"pending_motion": bool(
self.pending_motion_at
and time.monotonic() - self.pending_motion_at <= PENDING_MOTION_TTL_SECONDS
),
"last_error": self.last_error,
"telemetry_last_error": self.telemetry_last_error,
"last_trigger_reason": self.last_trigger_reason,
"session_started_at": self.session_started_at,
"session_age_seconds": (
round(time.monotonic() - self.session_started_at, 1)
if self.active and self.session_started_at else 0
),
"last_session": self.last_session,
"lifecycle_events": list(self.lifecycle_events),
"vision_routed": self.vision_routed,
"vision_last_error": self.vision_last_error,
"vision_managed": bool(self.config.get("manage_vision")),
"vision_process_running": bool(
self.vision_process is not None and self.vision_process.poll() is None
),
"vision_restarts": self.vision_restarts,
"configured_audio": {
"input": self.config["audio_input"],
"output": self.config["audio_output"],
},
"active_audio": self.current_audio.status() if self.current_audio else None,
"audio_devices": self.audio_devices,
}
def reconciliation_state(self) -> dict:
configuration = self.configuration()
return {
"applied_revision": int(configuration["applied_revision"]),
"desired_revision": int(configuration["desired_revision"]),
"effective_engine": "elevenlabs",
"state": "active" if self.active else "paused" if self.paused else "armed",
"session": ({
"session_id": self.local_session_id,
"trigger_reason": self.last_trigger_reason,
} if self.active else None),
"health": {
"service_running": True,
"endpoint_ready": self.http_thread is None or self.http_thread.is_alive(),
"session_active": self.active,
"telemetry_connected": self.scheduler_reporter.enabled,
},
}
def trigger(self, reason: str) -> dict:
with self.state_lock:
if self.shutdown.is_set():
self.triggers_rejected += 1
self._event("trigger_rejected", "Runtime is shutting down", "failed", reason=reason)
return {"accepted": False, "reason": "shutting_down"}
if self.paused:
self.triggers_rejected += 1
return {"accepted": False, "reason": "stopped"}
if self.active:
if reason == "motion":
self.pending_motion_at = time.monotonic()
self.triggers_rejected += 1
self._event("trigger_rejected", "Session active; motion retained", "pending", reason=reason)
return {"accepted": False, "reason": "active", "pending": reason == "motion"}
cooldown_remaining = (
float(self.config["cooldown"]) - (time.monotonic() - self.last_session_end)
if self.last_session_end else 0.0
)
if cooldown_remaining > 0:
if reason == "motion":
self.pending_motion_at = time.monotonic()
self.triggers_rejected += 1
self._event("trigger_rejected", "Cooldown active; motion retained", "pending", reason=reason)
return {
"accepted": False,
"reason": "cooldown",
"pending": reason == "motion",
"retry_after_ms": int(cooldown_remaining * 1000),
}
if self.trigger_event.is_set():
self.triggers_rejected += 1
self._event("trigger_rejected", "A trigger is already queued", "pending", reason=reason)
return {"accepted": False, "reason": "already_queued"}
self.trigger_reason = reason
self.last_trigger_reason = reason
self.trigger_event.set()
self.triggers_accepted += 1
self._event("trigger_accepted", f"{reason} trigger accepted", "ok", reason=reason)
return {"accepted": True, "reason": reason}
def handle_management_command(self, command: dict) -> dict:
operation = str(command.get("operation") or "")
payload = command.get("payload") or {}
force = bool(command.get("force"))
if operation == "voice.trigger":
self.paused = False
self.store.set_state("paused", False)
return {"ok": True, **self.trigger(str(payload.get("reason") or "gateway"))}
if operation == "voice.configuration.stage":
return {"ok": True, "configuration": self.stage_configuration(payload)}
if operation == "voice.configuration.apply_when_idle":
self.apply_when_idle_requested = True
return {"ok": True, "waiting_for_idle": self.active}
if operation == "voice.restart_when_idle":
self.restart_when_idle_requested = True
return {"ok": True, "waiting_for_idle": self.active}
if operation == "voice.stop_when_idle":
self.stop_when_idle_requested = True
return {"ok": True, "waiting_for_idle": self.active}
if operation == "voice.configuration.rollback":
previous = self.store.get_state("last_known_good_configuration")
if not previous:
return {"ok": False, "error": "no last-known-good configuration exists"}
previous["desired_revision"] = int(self.configuration()["desired_revision"]) + 1
self.store.set_state("staged_configuration", previous)
self.store.set_state("desired_revision", previous["desired_revision"])
self.apply_when_idle_requested = True
return {"ok": True, "waiting_for_idle": self.active}
if operation == "voice.media.release_stale":
if not force:
return {"ok": False, "error": "force authorization is required"}
lease = media_lease_status()
if self.active:
return {"ok": False, "error": "active session owns the media lease", "media_lease": lease}
if lease["conflicts"]:
return {
"ok": False,
"error": "foreign live process owns ALSA; stop that named service explicitly",
"media_lease": lease,
}
return {"ok": True, "released": True, "media_lease": lease}
if operation in {"voice.restart", "voice.stop", "voice.configuration.apply"}:
if not force:
return {"ok": False, "error": "force authorization is required"}
if operation == "voice.restart":
self.restart_when_idle_requested = True
elif operation == "voice.stop":
self.stop_when_idle_requested = True
else:
self.apply_when_idle_requested = True
if self.active:
self.force_stop_event.set()
return {"ok": True, "disposing": self.active}
return {"ok": False, "error": f"unsupported operation: {operation}"}
def start_motion_server(self) -> None:
runtime = self
class Handler(BaseHTTPRequestHandler):
def _remote(self) -> bool:
return self.client_address[0] not in {"127.0.0.1", "::1"}
def _authorized(self) -> bool:
if not self._remote():
return True
supplied = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query).get("token", [""])[0]
expected = str(runtime.config.get("tailscale_status_token") or "")
return bool(expected) and hmac.compare_digest(supplied, expected)
def _reply(self, status: int, payload: dict) -> None:
body = json.dumps(payload).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _body(self) -> dict:
length = int(self.headers.get("Content-Length", "0") or 0)
return json.loads(self.rfile.read(length) or b"{}")
def _force(self) -> bool:
value = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query).get("force", [""])[0]
return value.casefold() in {"1", "true", "yes"}
def _html(self) -> None:
body = """RoboPark audioRoboPark active devices
loading...
""".encode()
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
if not self._authorized():
self._reply(403, {"ok": False, "error": "invalid or missing device-status token"})
return
parsed = urllib.parse.urlparse(self.path)
path = parsed.path
if path == "/" and "text/html" in self.headers.get("Accept", ""):
self._html()
return
state = runtime.status()
if path == "/health":
configured = state["configured_audio"]
input_names = {row["name"] for row in runtime.audio_devices["inputs"]}
output_names = {row["name"] for row in runtime.audio_devices["outputs"]}
lease = media_lease_status()
self._reply(200, {
"ok": True,
"voice_ready": (
configured["input"] in input_names
and configured["output"] in output_names
and lease["available"]
and not runtime.paused
),
"service_running": True,
"endpoint_ready": True,
"audio_input_resolved": configured["input"] in input_names,
"audio_output_resolved": configured["output"] in output_names,
"media_lease_available": lease["available"],
"provider_reachable": runtime.last_error is None,
"configuration_valid": runtime.store.get_state("staged_configuration") is None,
"trigger_armed": not runtime.paused,
"session_active": runtime.active,
"telemetry_connected": runtime.scheduler_reporter.enabled,
"telemetry_mode": "connected" if runtime.scheduler_reporter.enabled else "buffering_locally",
"media_lease": lease,
})
return
if path == "/configuration":
self._reply(200, {"ok": True, **runtime.configuration()})
return
if path == "/sessions/current":
self._reply(200, {
"ok": True,
"active": runtime.active,
"session": ({
"session_id": runtime.local_session_id,
"started_at": runtime.session_started_at,
"trigger_reason": runtime.last_trigger_reason,
"engine": "elevenlabs",
"agent_id": runtime.config.get("agent_id"),
} if runtime.active else None),
"last_session": runtime.last_session,
})
return
if path == "/events":
query = urllib.parse.parse_qs(parsed.query)
after = int(query.get("after", ["0"])[0] or 0)
limit = int(query.get("limit", ["250"])[0] or 250)
events = runtime.store.events_after(after, limit)
self._reply(200, {
"ok": True,
"events": events,
"next_cursor": events[-1]["sequence"] if events else after,
"spool": runtime.store.stats(),
})
return
if path not in {"/", "/status", "/state"}:
self._reply(404, {"ok": False, "error": "unknown management endpoint"})
return
self._reply(200, {
"ok": True,
"armed": True,
"motors": runtime.motor_registry.motors,
**state,
})
def do_POST(self):
parsed_path = urllib.parse.urlparse(self.path)
path = parsed_path.path
if parsed_path.path == "/devices" and self._authorized():
length = int(self.headers.get("Content-Length", "0") or 0)
try:
payload = json.loads(self.rfile.read(length) or b"{}")
selected_input = str(payload.get("input") or "").strip()
selected_output = str(payload.get("output") or "").strip()
valid_inputs = {row["name"] for row in runtime.audio_devices["inputs"]}
valid_outputs = {row["name"] for row in runtime.audio_devices["outputs"]}
if selected_input not in valid_inputs or selected_output not in valid_outputs:
raise ValueError("selection must use devices currently registered by PyAudio")
runtime.config["audio_input"] = selected_input
runtime.config["audio_output"] = selected_output
Path(runtime.config["config_path"]).write_text(
json.dumps(runtime.config, indent=2) + "\n", encoding="utf-8"
)
self._reply(202, {"ok": True, "input": selected_input, "output": selected_output, "restarting": True})
threading.Timer(0.2, runtime.shutdown.set).start()
except Exception as exc:
self._reply(400, {"ok": False, "error": str(exc)})
return
if self._remote():
self._reply(403, {"ok": False, "error": "tailnet status endpoint is read-only"})
return
if path == "/trigger":
if not runtime.config["motion_enabled"]:
self._reply(409, {"ok": False, "error": "motion trigger is disabled"})
return
body = self._body()
runtime.paused = False
runtime.store.set_state("paused", False)
admission = runtime.trigger(str(body.get("reason") or "management_test"))
self._reply(202 if admission["accepted"] else 409, {"ok": admission["accepted"], **admission})
return
if path == "/configuration/stage":
try:
self._reply(202, {"ok": True, **runtime.stage_configuration(self._body())})
except Exception as exc:
self._reply(400, {"ok": False, "error": str(exc)})
return
if path == "/configuration/apply-when-idle":
runtime.apply_when_idle_requested = True
result = runtime.apply_staged_configuration() if not runtime.active else {
"ok": True, "waiting_for_idle": True, "configuration": runtime.configuration()
}
self._reply(202, result)
return
if path == "/configuration/apply":
if not self._force():
self._reply(409, {"ok": False, "error": "force=true is required"})
return
runtime.apply_when_idle_requested = True
if runtime.active:
runtime.force_stop_event.set()
self._reply(202, {"ok": True, "disposing": runtime.active, "configuration": runtime.configuration()})
return
if path == "/restart-when-idle":
runtime.restart_when_idle_requested = True
runtime._event("restart_requested", "Restart scheduled for idle", "pending")
self._reply(202, {"ok": True, "waiting_for_idle": runtime.active})
return
if path == "/restart":
if not self._force():
self._reply(409, {"ok": False, "error": "force=true is required"})
return
runtime.restart_when_idle_requested = True
if runtime.active:
runtime.force_stop_event.set()
self._reply(202, {"ok": True, "disposing": runtime.active})
return
if path == "/stop":
runtime.stop_when_idle_requested = True
if self._force() and runtime.active:
runtime.force_stop_event.set()
elif runtime.active:
self._reply(202, {"ok": True, "waiting_for_idle": True})
return
runtime.paused = True
runtime.store.set_state("paused", True)
runtime.trigger_event.clear()
runtime.voice_trigger.stop()
runtime._event("runtime_stopped", "Voice triggers paused", "pending")
self._reply(202, {"ok": True, "paused": True})
return
if path == "/events/ack":
body = self._body()
cursor = runtime.store.acknowledge(int(body.get("sequence") or 0))
self._reply(200, {"ok": True, "acknowledged_sequence": cursor})
return
if path in {"/session/stop", "/recover"}:
if not self._force():
runtime.stop_when_idle_requested = True
self._reply(202, {"ok": True, "waiting_for_idle": runtime.active})
return
runtime.force_stop_event.set()
if runtime.current_audio is not None:
runtime.current_audio.stop()
runtime.pending_motion_at = 0.0
runtime.trigger_event.clear()
stage = "recovery_requested" if path == "/recover" else "dispose_requested"
runtime._event(stage, "Audio released and session disposal requested", "pending")
self._reply(202, {
"ok": True,
"disposing": runtime.active,
"audio_released": True,
})
return
if path.startswith("/motors/"):
name = path.split("/", 2)[2]
length = int(self.headers.get("Content-Length", "0") or 0)
try:
body = json.loads(self.rfile.read(length) or b"{}")
body["name"] = name
self._reply(200, runtime.motor_registry.execute(body))
except Exception as exc:
self._reply(400, {"ok": False, "error": str(exc)})
return
if not runtime.config["motion_enabled"]:
self._reply(404, {"ok": False, "error": "motion trigger is disabled"})
return
admission = runtime.trigger("motion")
self._reply(202, {
"ok": True,
"queued": admission["accepted"],
**admission,
})
def log_message(self, format, *args):
return
ThreadingHTTPServer.allow_reuse_address = True
ThreadingHTTPServer.daemon_threads = True
self.httpd = ThreadingHTTPServer(
(self.config.get("status_bind_host") or self.config["motion_host"], int(self.config["motion_port"])), Handler
)
self.http_thread = threading.Thread(
target=self.httpd.serve_forever,
name="motion-webhook",
daemon=True,
)
self.http_thread.start()
log(f"robot control endpoint armed at http://{self.config['motion_host']}:{self.config['motion_port']}/")
if self.config.get("tailscale_status_ip"):
log(
f"tailnet device selector: http://{self.config['tailscale_status_ip']}:"
f"{self.config['motion_port']}/?token={self.config['tailscale_status_token']}"
)
if self.config.get("vision_url") and self.config["motion_enabled"]:
threading.Thread(
target=self._maintain_vision_route,
name="robovision-router",
daemon=True,
).start()
def _maintain_vision_route(self) -> None:
target = f"http://127.0.0.1:{self.config['motion_port']}/"
payload = json.dumps({"url": target}).encode()
arm_payload = json.dumps({"active": True}).encode()
last_result = None
while not self.shutdown.is_set():
try:
for endpoint, body in (
("/api/motion/webhook", payload),
("/api/motion/toggle", arm_payload),
):
request = urllib.request.Request(
f"{self.config['vision_url']}{endpoint}",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=3) as response:
if response.status >= 300:
raise RuntimeError(f"{endpoint} returned HTTP {response.status}")
with self.state_lock:
self.vision_routed = True
self.vision_last_error = None
if last_result is not True:
log(f"RoboVision motion armed and routed to {target}")
last_result = True
# Reassert frequently enough to recover from a replaced camera
# process without waiting for an operator or a full service restart.
delay = 5
except Exception as exc:
with self.state_lock:
self.vision_routed = False
self.vision_last_error = str(exc)
if last_result is not False:
log(f"RoboVision webhook configuration failed; retrying: {exc}")
last_result = False
if self.config.get("manage_vision"):
self._ensure_vision_process()
delay = 2
self.shutdown.wait(delay)
def _ensure_vision_process(self) -> None:
process = self.vision_process
if process is not None and process.poll() is None:
return
if process is not None:
log(f"managed RoboVision exited with code {process.returncode}; restarting")
script = self.config.get("vision_script")
python = self.config.get("vision_python")
if not script or not python:
with self.state_lock:
self.vision_last_error = "managed RoboVision runtime path is missing"
return
target = f"http://127.0.0.1:{self.config['motion_port']}/"
env = os.environ.copy()
env["PYTHONUNBUFFERED"] = "1"
env["ROBOPARK_CAMERA_DEVICE"] = str(self.config.get("camera_device") or "auto")
try:
self.vision_process = subprocess.Popen(
[
str(python),
str(script),
"--port", str(self.config["vision_port"]),
"--motion-webhook-url", target,
"--motion-active",
],
env=env,
)
self.vision_process_started = True
self.vision_restarts += 1
log(
f"managed RoboVision started (pid {self.vision_process.pid}) on "
f"{self.config['vision_url']} with camera {env['ROBOPARK_CAMERA_DEVICE']}"
)
except Exception as exc:
with self.state_lock:
self.vision_last_error = str(exc)
log(f"managed RoboVision start failed; retrying: {exc}")
def _stop_vision_process(self) -> None:
process = self.vision_process
if not self.vision_process_started or process is None or process.poll() is not None:
return
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
log("forcing managed RoboVision process closed")
process.kill()
process.wait(timeout=2)
log("managed RoboVision stopped")
def run_session(self, reason: str) -> None:
with self.state_lock:
self.active = True
self.session_started_at = time.monotonic()
self.sessions_started += 1
session_number = self.sessions_started
self.last_error = None
self.force_stop_event.clear()
self.local_session_id = f"session_{uuid.uuid4().hex}"
self.scheduler_reporter.start(reason)
if self.scheduler_reporter.session_id:
self.local_session_id = self.scheduler_reporter.session_id
self._event("session_starting", f"Starting ElevenLabs session #{session_number}", "active", reason=reason)
self.voice_trigger.stop()
audio = NativeAudioInterface(self.config["audio_input"], self.config["audio_output"])
self.current_audio = audio
last_activity = [time.monotonic()]
agent_busy_until = [0.0]
ended = threading.Event()
conversation = None
client_tools = None
def on_user(text: str) -> None:
last_activity[0] = time.monotonic()
self._event("user_transcript", text or "No speech decoded", "ok" if text.strip(" .") else "failed")
self.scheduler_reporter.turn("user", text)
self._spool_event(
"transcript.final",
{"role": "user", "text": text, "is_final": True, "engine": "elevenlabs"},
)
log(f"user: {text}")
def on_agent(text: str) -> None:
now = time.monotonic()
last_activity[0] = now
agent_busy_until[0] = now + max(1.5, len(text.split()) / 2.2)
self._event("agent_response", text or "Empty agent response", "ok" if text.strip() else "failed")
self.scheduler_reporter.turn("assistant", text)
self._spool_event(
"transcript.final",
{"role": "assistant", "text": text, "is_final": True, "engine": "elevenlabs"},
)
log(f"agent: {text}")
def on_end() -> None:
ended.set()
started = time.monotonic()
log(f"starting ElevenLabs session #{session_number} ({reason})")
try:
client = ElevenLabs(api_key=self.config.get("api_key"))
client_tools = ClientTools()
if self.motor_registry.names():
client_tools.register("robotMotor", self.motor_registry.execute)
log(f"agent tool robotMotor registered for: {', '.join(self.motor_registry.names())}")
conversation = Conversation(
client=client,
agent_id=self.config["agent_id"],
requires_auth=bool(self.config["requires_auth"]),
audio_interface=audio,
callback_user_transcript=on_user,
callback_agent_response=on_agent,
callback_end_session=on_end,
client_tools=client_tools,
)
branch_id = str(self.config.get("branch_id") or "").strip()
if branch_id:
if self.config.get("requires_auth"):
query = urllib.parse.urlencode({
"agent_id": self.config["agent_id"],
"branch_id": branch_id,
})
request = urllib.request.Request(
f"https://api.elevenlabs.io/v1/convai/conversation/get-signed-url?{query}",
headers={
"Accept": "application/json",
"xi-api-key": str(self.config.get("api_key") or ""),
},
)
with urllib.request.urlopen(request, timeout=15) as response:
signed_url = json.loads(response.read().decode("utf-8"))["signed_url"]
conversation._get_signed_url = lambda: signed_url
else:
original_get_wss_url = conversation._get_wss_url
def branch_wss_url():
parsed = urllib.parse.urlparse(original_get_wss_url())
params = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
params.append(("branch_id", branch_id))
return urllib.parse.urlunparse(
parsed._replace(query=urllib.parse.urlencode(params))
)
conversation._get_wss_url = branch_wss_url
log(f"pinned ElevenLabs branch {branch_id}")
conversation.start_session()
self._event("provider_connected", "ElevenLabs websocket connected", "active")
if self.motor_registry.names():
deadline = time.monotonic() + 10
while getattr(conversation, "_ws", None) is None and time.monotonic() < deadline:
time.sleep(0.1)
if getattr(conversation, "_ws", None) is not None:
conversation.send_contextual_update(
"MANDATORY ROBOT CONTROL POLICY: When the user asks to move, turn, "
"activate, test, or switch a physical part, call robotMotor before "
"responding. Do not merely describe or claim movement. "
"This robot exposes the robotMotor client tool. "
f"Registered motor names: {', '.join(self.motor_registry.names())}. "
"Only use these exact names. Use action pulse by default. Activations "
"are automatically time-bounded, and only confirm movement after the "
"tool returns ok=true."
)
log("registered motor names sent to agent context")
while not self.shutdown.is_set() and not self.force_stop_event.is_set() and not ended.wait(0.25):
now = time.monotonic()
if audio.fatal_error.is_set():
self.recycle_required = True
raise RuntimeError(audio.fatal_error_message or "fatal audio device failure")
if audio.started.is_set() and now - started >= 8 and audio.input_chunks == 0:
self.recycle_required = True
raise RuntimeError("microphone capture produced no PCM within 8 seconds")
thread = getattr(conversation, "_thread", None)
if thread is not None and not thread.is_alive():
break
if self.config["max_session"] > 0 and now - started >= self.config["max_session"]:
log("maximum session timeout reached")
break
if (
self.config["idle_timeout"] > 0
and now - last_activity[0] >= self.config["idle_timeout"]
and now >= agent_busy_until[0]
and audio.started.is_set()
):
log("conversation idle timeout reached")
break
except Exception as exc:
with self.state_lock:
self.last_error = str(exc)
log(f"conversation failed: {exc}")
self._event("session_error", str(exc), "failed")
finally:
ws = getattr(conversation, "_ws", None) if conversation is not None else None
thread = getattr(conversation, "_thread", None) if conversation is not None else None
if conversation is not None:
try:
conversation.end_session()
except Exception as exc:
log(f"conversation end request failed: {exc}")
audio.stop()
self.current_audio = None
if thread is not None and thread is not threading.current_thread():
thread.join(timeout=SESSION_STOP_GRACE_SECONDS)
if thread.is_alive() and ws is not None:
log("forcing stale ElevenLabs websocket closed")
try:
ws.close()
except Exception:
pass
thread.join(timeout=SESSION_FORCE_CLOSE_SECONDS)
if thread.is_alive():
log("ElevenLabs worker did not exit before disposal deadline")
self.recycle_required = True
elif conversation is None and client_tools is not None:
try:
client_tools.stop()
except Exception:
pass
with self.state_lock:
self.active = False
self.last_session_end = time.monotonic()
self.session_started_at = 0.0
self.sessions_completed += 1
elapsed = time.monotonic() - started
self.last_session = {
"number": session_number,
"reason": reason,
"duration_seconds": round(elapsed, 1),
"ended_at": time.time(),
"error": self.last_error,
"audio": audio.status(),
}
self._event(
"session_disposed",
f"Session #{session_number} disposed and media released",
"failed" if self.last_error else "ok",
duration_seconds=round(elapsed, 1),
)
self.scheduler_reporter.end(
"error" if self.last_error else ("operator_stop" if self.force_stop_event.is_set() else "completed"),
elapsed,
self.last_error,
)
self.local_session_id = None
log(
f"session #{session_number} disposed in {elapsed:.1f}s; "
f"speaker frames={audio.output_chunks}, bytes={audio.output_bytes}; "
f"mic chunks={audio.input_chunks}, forwarded={audio.forwarded_input_chunks}, "
f"peak={audio.input_peak}, echo-suppressed={audio.suppressed_input_chunks}; "
"triggers will re-arm after cooldown"
)
def _process_idle_requests(self) -> None:
if self.active:
return
if self.apply_when_idle_requested:
result = self.apply_staged_configuration()
if result.get("ok") and not result.get("waiting_for_idle"):
self.apply_when_idle_requested = False
if self.stop_when_idle_requested:
self.stop_when_idle_requested = False
self.paused = True
self.store.set_state("paused", True)
self.trigger_event.clear()
self.voice_trigger.stop()
self._event("runtime_stopped", "Voice triggers paused after session disposal", "pending")
if self.restart_when_idle_requested:
self.restart_when_idle_requested = False
self._event("restart_begin", "Clean worker restart beginning", "pending")
self.recycle_required = True
self.shutdown.set()
self.trigger_event.set()
def run(self) -> None:
if self.config.get("always_on") or self.config["motion_enabled"] or self.motor_registry.names():
self.start_motion_server()
if self.config["voice_trigger_enabled"] and not self.paused:
self.voice_trigger.start()
if self.config.get("always_on"):
self.trigger_reason = "always-on"
self.trigger_event.set()
log(f"ready: robot={self.config['robot_name']} agent={self.config['agent_id']}")
while not self.shutdown.is_set():
self._process_idle_requests()
if self.shutdown.is_set():
break
if self.http_thread is not None and not self.http_thread.is_alive():
log("robot control endpoint stopped unexpectedly; recycling runtime")
self.recycle_required = True
break
if not self.trigger_event.wait(0.5):
continue
if self.shutdown.is_set():
break
self.trigger_event.clear()
reason = self.trigger_reason
self.run_session(reason)
self._process_idle_requests()
if self.recycle_required:
log("recycling runtime for clean media state")
break
while not self.shutdown.is_set() and (
time.monotonic() - self.last_session_end < self.config["cooldown"]
):
time.sleep(0.2)
with self.state_lock:
pending_age = (
time.monotonic() - self.pending_motion_at
if self.pending_motion_at else float("inf")
)
if pending_age <= PENDING_MOTION_TTL_SECONDS and not self.shutdown.is_set():
self.pending_motion_at = 0.0
self.trigger_reason = "motion-continuation"
self.trigger_event.set()
self.triggers_accepted += 1
log("recent motion promoted to a continuation session")
elif self.pending_motion_at:
self.pending_motion_at = 0.0
if self.config.get("always_on") and not self.shutdown.is_set():
self.trigger_reason = "always-on-reconnect"
self.trigger_event.set()
log("always-on session queued for reconnect")
if self.config["voice_trigger_enabled"] and not self.shutdown.is_set():
self.voice_trigger.start()
self.voice_trigger.stop()
self.motor_registry.close()
if self.httpd is not None:
self.httpd.shutdown()
self.httpd.server_close()
self._stop_vision_process()
self.scheduler_reporter.stop_event.set()
try:
self.scheduler_reporter.jobs.put_nowait(None)
except queue.Full:
pass
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--config", required=True)
args = parser.parse_args()
config = json.loads(Path(args.config).read_text(encoding="utf-8"))
config["config_path"] = str(Path(args.config).resolve())
if not config.get("always_on") and not config.get("motion_enabled") and not config.get("voice_trigger_enabled"):
raise SystemExit("at least one trigger must be enabled")
runtime = Runtime(config)
def stop(_signum, _frame):
log("shutdown requested")
runtime.shutdown.set()
runtime.trigger_event.set()
signal.signal(signal.SIGTERM, stop)
signal.signal(signal.SIGINT, stop)
runtime.run()
if __name__ == "__main__":
main()