from __future__ import annotations import asyncio import logging import os from contextlib import asynccontextmanager, suppress from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Literal from uuid import uuid4 from fastapi import FastAPI, Header, HTTPException, WebSocket, WebSocketDisconnect from fastapi.responses import FileResponse, RedirectResponse, JSONResponse from pydantic import BaseModel, Field from device_gateway.auth import DeviceCredentialStore from device_gateway.collectors.system import SystemCollector from device_gateway.collectors.host_state import host_state from device_gateway.protocol import ServerSnapshot from device_gateway.integrations.pi import PiSessionEvent, PiSessionRegistry from device_gateway.state import TargetStore, normalize_target, split_target from device_gateway.voice import VoiceManager, configured_stt from device_gateway.speech import SpeechManager, SpeechPreferences, configured_tts, HaihubSummary, MODES from device_gateway.control.settings import quick_catalog from device_gateway.control.models import ControlError, Input, Project, Settings from device_gateway.control.projects import ProjectRegistry from device_gateway.control.pi import PiAdapter from device_gateway.control.codex import CodexAdapter from device_gateway.control.approvals import ApprovalRegistry from device_gateway.control.pi_approval import PiApprovalServer from device_gateway.control.user_input import UserInputs from device_gateway.control.operations import Operations from device_gateway.control.navigator import Navigator,CAPABILITIES logger = logging.getLogger("uvicorn.error.device_gateway") system_collector = SystemCollector(hostname=os.environ.get("GATEWAY_SERVER_NAME")) snapshot_interval_seconds = float(os.environ.get("GATEWAY_SNAPSHOT_INTERVAL", "2")) # Last derived host state, used only for threshold hysteresis. host_states: dict[str, str] = {} credential_store = DeviceCredentialStore( os.environ.get( "DEVICE_CREDENTIALS_FILE", "~/.config/esp32-device-gateway/device.json", ) ) target_store = TargetStore( os.environ.get( "GATEWAY_STATE_FILE", "~/.config/esp32-device-gateway/targets.json", ) ) admin_page_path = os.path.join(os.path.dirname(__file__), "static", "admin.html") pi_sessions = PiSessionRegistry() project_registry = ProjectRegistry(os.environ.get("GATEWAY_PROJECTS_FILE", "~/.config/esp32-device-gateway/projects.json")) approvals = ApprovalRegistry() user_inputs=UserInputs() operations=Operations(os.environ.get("GATEWAY_OPERATIONS_FILE","~/.config/esp32-device-gateway/operations.sqlite")) adapters = {"pi": PiAdapter(project_registry, pi_sessions, approvals=approvals,user_inputs=user_inputs)} if os.environ.get("GATEWAY_CODEX_ENABLED", "0") == "1": adapters["codex"] = CodexAdapter(project_registry, approvals=approvals,user_inputs=user_inputs) def utc_now() -> datetime: return datetime.now(timezone.utc) def gateway_timezone_offset_minutes() -> int: configured = os.environ.get("GATEWAY_TIMEZONE_OFFSET_MINUTES") if configured is not None: try: value = int(configured) except ValueError: value = 0 logger.warning("invalid GATEWAY_TIMEZONE_OFFSET_MINUTES=%r; using UTC", configured) return max(-720, min(840, value)) offset = datetime.now().astimezone().utcoffset() return int(offset.total_seconds() // 60) if offset is not None else 0 @dataclass class Device: device_id: str board: str firmware: str websocket: WebSocket | None connection_id: str = field(default_factory=lambda: uuid4().hex) online: bool = True last_seen: datetime = field(default_factory=utc_now) last_event: dict[str, Any] | None = None settings_lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False) settings_task: asyncio.Task | None = field(default=None, repr=False) settings_id: str | None = None settings_generation: int = 0 last_agent_state: str = "" capabilities: set[str] = field(default_factory=set) send_lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False) def public(self) -> dict[str, Any]: return { "device_id": self.device_id, "board": self.board, "firmware": self.firmware, "online": self.online, "capabilities": sorted(self.capabilities), "last_seen": self.last_seen.isoformat(), "last_event": self.last_event, } class ApprovalDecision(BaseModel): decision: str = Field(min_length=1,max_length=40) class DisplayMessage(BaseModel): title: str = Field(min_length=1, max_length=80) body: str = Field(min_length=1, max_length=500) class TargetSelection(BaseModel): target: str = Field(pattern=r"^(server|(?:agent:(?:pi|codex)|pi):[A-Za-z0-9._:-]{1,160})$") class NotificationTest(BaseModel): kind: Literal["done", "failed"] devices: dict[str, Device] = {} async def send_to_device(device: Device, payload: dict[str, Any]) -> None: if not device.online or device.websocket is None: raise RuntimeError("device is offline") async with device.send_lock: await asyncio.wait_for(device.websocket.send_json(payload), 3) def current_device(device): return devices.get(device.device_id) is device and device.online and device.websocket is not None voices = VoiceManager(configured_stt(), lambda did: target_store.get(did), lambda agent: adapter_for(agent), send_to_device,current_device,lambda:bool(approvals.list_pending()),generation_for=lambda did:target_store.generation(did), context_prepare=lambda d,e:navigator.voice_prepare(d,e),context_valid=lambda d,c:navigator.voice_valid(d,c),context_submit=lambda d,c,t:navigator.voice_submit(d,c,t)) def speech_blocked(device): voice = voices.requests.get(device.connection_id) return bool(approvals.list_pending()) or bool(voice and voice.state in ('RECORDING', 'TRANSCRIBING', 'AWAITING_CONFIRMATION', 'SENDING')) speech_preferences = SpeechPreferences(os.environ.get('GATEWAY_SPEECH_FILE', '~/.config/esp32-device-gateway/speech.json')) speeches = SpeechManager(configured_tts(), HaihubSummary(), speech_preferences, lambda did: target_store.get(did), send_to_device, current_device, speech_blocked) def on_agent_terminal(agent, session_id, state, final_text, *, name='', latest_activity=''): for device in list(devices.values()): speeches.terminal(device, agent, session_id, state, final_text, name, latest_activity) def approvals_changed(changed): if approvals.list_pending(): for device in list(devices.values()): speeches.cancel(device) voices.cancel(device) device.settings_id = None changed.set() async def device_settings(device, event): if event['type'] == 'settings_cancel': if device.settings_id == event.get('request_id'): device.settings_id = None return async with device.settings_lock: await _device_settings(device, event) async def _device_settings(device, event): rid = event.get('request_id') if not isinstance(rid, str) or not 1 <= len(rid) <= 64: return if event['type'] == 'settings_cancel': if device.settings_id == rid: device.settings_id = None return target = target_store.get(device.device_id) generation=target_store.generation(device.device_id) parsed = split_target(target) if not parsed: await send_to_device(device, {'type':'settings_error','request_id':rid,'message':'NO AGENT SELECTED'}) return if approvals.list_pending() or user_inputs.pending() or not current_device(device): return if event['type'] == 'settings_get': device.settings_id = rid;device.settings_generation=generation elif device.settings_id != rid or device.settings_generation!=generation or event.get('target') != target: return def valid(): return current_device(device) and device.settings_id == rid and target_store.get(device.device_id) == target and target_store.generation(device.device_id)==generation and not approvals.list_pending() try: agent, sid = parsed adapter = adapter_for(agent) session = await adapter.get_session(sid) if not valid(): return if event['type'] == 'settings_apply': # Consume identity; a duplicate apply cannot mutate twice. settings = await adapter.update_settings(sid, Settings.model_validate(event.get('settings', {})), still_valid=valid) else: settings = await adapter.get_settings(sid) if event.get("include_permissions") and agent=="codex":settings["permissions"]=await adapter.permissions(sid) if not valid(): return recent = await adapter.list_sessions(project_registry.get(session.project_id)) settings = quick_catalog(settings, [s.model for s in sorted(recent, key=lambda s:s.updated_at, reverse=True) if s.model]) if valid(): await send_to_device(device, dict(type='settings_updated' if event['type']=='settings_apply' else 'settings_state', request_id=rid,target=target,target_generation=generation,name=session.name,**settings)) if event['type']=='settings_apply': device.settings_id = None except Exception as exc: if valid(): await send_to_device(device, {'type':'settings_error','request_id':rid, 'message':exc.detail if isinstance(exc, ControlError) else 'UPDATE FAILED'}) async def broadcast_server_snapshot(payload: dict[str, Any]) -> None: for device in list(devices.values()): if not device.online or device.websocket is None: continue try: await send_to_device(device, payload) except Exception: speeches.cancel(device, notify=False, forget=True) voices.cancel(device, notify=False, forget=True) device.settings_id = None if device.websocket: with suppress(Exception): await device.websocket.close(code=1011) device.online = False device.websocket = None device.last_seen = utc_now() logger.warning("device %s disconnected while pushing server_snapshot", device.device_id) async def monitor_system() -> None: while True: snapshot = system_collector.collect() await broadcast_server_snapshot(device_snapshot_payload(snapshot)) await push_agent_targets() await asyncio.sleep(snapshot_interval_seconds) def device_snapshot_payload(snapshot: ServerSnapshot | None = None) -> dict[str, Any]: """Existing server snapshot plus derived glance fields. No new message type, no extra frame and no device-driven refresh: the same 2 s frame simply carries a few more small fields. Account identity, plan and credits are never included. """ snapshot = snapshot or system_collector.latest or system_collector.collect() payload = snapshot.model_dump() adapter = adapters.get("codex") payload["host_state"] = host_state(snapshot.cpu_percent, snapshot.memory_percent, valid=True, gateway_online=True, previous=host_states.get("last")) host_states["last"] = payload["host_state"] quota = getattr(adapter, "quota", None) payload.update(quota.wire() if quota is not None else {"codex_quota_valid": False, "codex_remaining_percent": 0, "codex_resets_at": None, "codex_window_minutes": None}) return payload async def synchronize_approvals(): # One delivery worker owns presentations. New clients are picked up by the # same worker, avoiding resolve/reconnect races and duplicate notifications. for device in list(devices.values()): connection = device.connection_id if not device.online or not device.websocket: approvals.presentations.pop(connection, None) continue try: current_id = approvals.presentations.get(connection) current = approvals.get(current_id) if current_id and (current is None or current.state != "PENDING"): await send_to_device(device, {"type": "approval_cancelled", "approval_id": current_id, "state": current.state if current else "CANCELLED"}) approvals.presentations.pop(connection, None) if connection not in approvals.presentations: pending = approvals.list_pending() if pending: approval = pending[0] speeches.cancel(device) voices.cancel(device) device.settings_id = None approvals.presentations[connection] = approval.approval_id await asyncio.wait_for(send_to_device(device, approvals.wire(approval)), 3) except Exception: approvals.presentations.pop(connection, None) speeches.cancel(device, notify=False, forget=True) voices.cancel(device, notify=False, forget=True) device.settings_id = None device.online = False if device.websocket: with suppress(Exception): await device.websocket.close(code=1011) device.websocket = None for device in list(devices.values()): connection=device.connection_id if not current_device(device):user_inputs.presentations.pop(connection,None);continue if 'native-interaction' not in device.capabilities:continue previous=user_inputs.presentations.get(connection) pending=user_inputs.pending() if not approvals.list_pending() else [] entry=pending[0] if pending else None presentation=(entry['request_id'],entry['questions'][entry['index']]['id']) if entry else None if previous!=presentation: if previous: voices.cancel(device) await send_to_device(device,dict(type='user_input_cancelled',request_id=previous[0],question_id=previous[1])) user_inputs.presentations.pop(connection,None) if entry: voices.cancel(device);device.settings_id=None user_inputs.presentations[connection]=presentation await send_to_device(device,user_inputs.wire(entry)) # Session status is an overlay over real lifecycle state; resolution never # overwrites a concurrent DONE/FAILED event with a guessed RUNNING state. for session in list(pi_sessions.sessions.values()): await push_selected_pi_session(session.session_id) async def approval_delivery(changed): while True: await changed.wait() changed.clear() await synchronize_approvals() def invalidate_target(device): speeches.cancel(device);voices.cancel(device);device.settings_id=None;device.last_agent_state='' # Lambdas keep test/admin replacement stores authoritative. class DynamicNavigator: def __init__(self):self.instance=None;self.identity=None def __getattr__(self,name): identity=(id(project_registry),id(adapters),id(target_store),id(operations),id(user_inputs)) if self.identity!=identity: self.instance=Navigator(project_registry,adapters,target_store,operations,user_inputs,send_to_device,current_device,agent_target_payload,invalidate_target);self.identity=identity return getattr(self.instance,name) navigator=DynamicNavigator() @asynccontextmanager async def lifespan(_: FastAPI): changed = asyncio.Event() approvals.on_change = lambda: approvals_changed(changed) user_inputs.on_change=lambda:changed.set() if "codex" in adapters: adapters["codex"].on_activity = queue_codex_activity adapters["codex"].on_terminal = on_agent_terminal approval_server = PiApprovalServer(approvals, project_registry) # Synthetic Pi approval endpoint is intentionally not started in 0.6b. delivery_task = asyncio.create_task(approval_delivery(changed), name="approval-delivery") monitor_task = asyncio.create_task(monitor_system(), name="system-monitor") try: yield finally: approvals.close() for entry in user_inputs.pending():user_inputs.cancel(entry) await speeches.close() await voices.close() await approval_server.close() await asyncio.gather(*(adapter.close() for adapter in adapters.values())) await synchronize_approvals() approvals.on_change = None user_inputs.on_change=None for task in (monitor_task, delivery_task): task.cancel() with suppress(asyncio.CancelledError): await task app = FastAPI(title="Device Lab Gateway", version="0.5.0", lifespan=lifespan) @app.exception_handler(ControlError) async def control_error_handler(request, exc: ControlError): return JSONResponse(status_code=exc.status, content={"detail": exc.detail}) def adapter_for(agent: str): if agent not in adapters: raise ControlError("agent adapter not enabled", 404) return adapters[agent] @app.get("/api/approvals") async def list_approvals(): return [approval.model_dump(mode="json") for approval in approvals.list_pending()] @app.post("/api/approvals/{approval_id}/decision") async def decide_approval(approval_id: str, value: ApprovalDecision): if not approvals.resolve(approval_id, value.decision): raise HTTPException(status_code=409, detail="approval is missing or already resolved") return {"accepted": True, "approval_id": approval_id, "state": approvals.get(approval_id).state} @app.get("/api/projects") async def list_projects(): return project_registry.list() @app.post("/api/projects", status_code=201) async def add_project(project: Project): return project_registry.add(project) @app.delete("/api/projects/{project_id}", status_code=204) async def delete_project(project_id: str): project_registry.delete(project_id) @app.get("/api/agents") async def list_agents(): return [{"agent": name, "capabilities": ["list_sessions", "resume_session", "send_input", "get_settings", "list_models", "update_settings"]} for name in adapters] @app.get("/api/projects/{project_id}/sessions") async def project_sessions(project_id: str): project = project_registry.get(project_id) sessions, errors = [], {} results = await asyncio.gather(*(adapter.list_sessions(project) for adapter in adapters.values()), return_exceptions=True) for name, result in zip(adapters, results): if isinstance(result, Exception): errors[name] = result.detail if isinstance(result, ControlError) else "agent catalog unavailable" else: sessions.extend(result) return {"sessions": sessions, "errors": errors} @app.get("/api/agents/{agent}/sessions/{session_id}") async def get_agent_session(agent: str, session_id: str): return await adapter_for(agent).get_session(session_id) @app.post("/api/agents/{agent}/sessions/{session_id}/resume") async def resume_agent_session(agent: str, session_id: str): resumed = await adapter_for(agent).resume_session(session_id) return {"accepted": True, "agent": agent, "session_id": session_id, "resumed": resumed} @app.post("/api/agents/{agent}/sessions/{session_id}/input", status_code=202) async def agent_input(agent: str, session_id: str, value: Input): return await adapter_for(agent).send_input(session_id, value.text) @app.get("/api/agents/{agent}/sessions/{session_id}/settings") async def get_settings(agent: str, session_id: str): return await adapter_for(agent).get_settings(session_id) @app.post("/api/agents/{agent}/sessions/{session_id}/settings") async def update_settings(agent: str, session_id: str, value: Settings): return await adapter_for(agent).update_settings(session_id, value) @app.get("/health") async def health() -> dict[str, Any]: return {"status": "ok", "control_version": "0.5.0", "online_devices": sum(d.online for d in devices.values())} @app.get("/", include_in_schema=False) async def root() -> RedirectResponse: return RedirectResponse("/admin", status_code=307) @app.get("/admin", include_in_schema=False) async def admin_page() -> FileResponse: return FileResponse(admin_page_path, media_type="text/html") @app.get("/api/admin.js", include_in_schema=False) async def admin_script() -> FileResponse: return FileResponse(os.path.join(os.path.dirname(admin_page_path), "admin.js"), media_type="text/javascript") @app.get("/server") async def server_status() -> dict[str, Any]: snapshot = system_collector.latest or system_collector.collect() return snapshot.model_dump() @app.get("/devices") async def list_devices() -> list[dict[str, Any]]: result = [] for device in sorted(devices.values(), key=lambda d: d.device_id): item = device.public() item["target"] = target_store.get(device.device_id) item["speech_mode"] = speeches.preferences.get(device.device_id, item["target"]) result.append(item) return result @app.get("/api/status") async def admin_status() -> dict[str, Any]: snapshot = system_collector.latest or system_collector.collect() sessions = pi_sessions.public() for session in sessions: pending = approvals.list_pending("pi", session["session_id"]) if pending: session["state"] = "WAITING_APPROVAL" session["waiting_reason"] = pending[0].title + ": " + pending[0].detail return { "server": snapshot.model_dump(), "devices": await list_devices(), "pi_sessions": sessions, } def pi_target_payload( session_id: str, notification: Literal["done", "failed"] | None = None ) -> dict[str, Any]: session = pi_sessions.get(session_id) if session is None: return { "type": "agent_state", "agent": "pi", "target": f"agent:pi:{session_id}", "session_id": session_id, "name": session_id, "cwd": "", "state": "OFFLINE", "elapsed_seconds": 0, "waiting_reason": None, } payload = { "type": "agent_state", "agent": "pi", "target": f"agent:pi:{session_id}", **session.public(include_history=True), } pending = approvals.list_pending("pi", session_id) if pending: payload["state"] = "WAITING_APPROVAL" payload["waiting_reason"] = pending[0].title + ": " + pending[0].detail if notification is not None: payload["notification"] = notification return payload async def push_selected_pi_session( session_id: str, notification: Literal["done", "failed"] | None = None ) -> None: target = f"agent:pi:{session_id}" payload = await agent_target_payload(target) if notification is not None: payload["notification"] = notification for device in list(devices.values()): if target_store.get(device.device_id) != target: continue generation = target_store.generation(device.device_id) # Re-verify after the catalog await; a terminal notification must never # reach a device that already switched to another session/generation. if target_store.get(device.device_id) != target or target_store.generation(device.device_id) != generation: continue try: await send_to_device(device, dict(payload, target_generation=generation)) except Exception: speeches.cancel(device, notify=False, forget=True) voices.cancel(device, notify=False, forget=True) device.settings_id = None if device.websocket: with suppress(Exception): await device.websocket.close(code=1011) device.online = False device.websocket = None device.last_seen = utc_now() async def push_selected_pi_activity(event: PiSessionEvent) -> None: session = pi_sessions.get(event.session_id) if session is None or not session.latest_activity: return payload = { "type": "agent_activity", "agent": "pi", "target": f"agent:pi:{event.session_id}", "session_id": event.session_id, "activity_type": session.activity_type, "activity": session.latest_activity, "updated_at": ( session.activity_updated_at.isoformat() if session.activity_updated_at is not None else None ), } for device in list(devices.values()): if target_store.get(device.device_id) != f"agent:pi:{event.session_id}": continue try: await send_to_device(device, payload) except Exception: speeches.cancel(device, notify=False, forget=True) voices.cancel(device, notify=False, forget=True) device.settings_id = None if device.websocket: with suppress(Exception): await device.websocket.close(code=1011) device.online = False device.websocket = None device.last_seen = utc_now() async def agent_target_payload(target): parsed = split_target(target) if not parsed: return {'type':'target_state', 'target':'server'} agent, sid = parsed try: session = await adapter_for(agent).get_session(sid) if agent == 'codex' and hasattr(adapter_for(agent), 'observe_session'): await adapter_for(agent).observe_session(sid) payload = {'type':'agent_state','target':normalize_target(target), **session.model_dump()} payload['recent_text'] = payload.get('recent_text','').encode('utf-8')[-4000:].decode('utf-8', errors='ignore') if agent == 'pi' and pi_sessions.get(sid) and session.live: overlay=pi_target_payload(sid) # A restarted pager may only know a heartbeat. Preserve native # persisted assistant text already recovered by the adapter. for field in ('recent_text','final_text','latest_activity'): if not overlay.get(field) and payload.get(field):overlay[field]=payload[field] payload.update(overlay) return payload except ControlError: if agent == 'pi' and pi_sessions.get(sid): return pi_target_payload(sid) return {'type':'agent_state','target':normalize_target(target),'agent':agent,'session_id':sid, 'name':sid,'state':'OFFLINE','elapsed_seconds':0,'model':None,'effort':None} async def push_agent_targets(agent=None, sid=None, activity=False): cache = {} for device in list(devices.values()): target = target_store.get(device.device_id) generation=target_store.generation(device.device_id) parsed = split_target(target) if not current_device(device) or not parsed or (agent and parsed != (agent, sid)): continue try: if target not in cache: cache[target] = await agent_target_payload(target) payload = cache[target] # Verify again after the catalog await; old targets cannot overwrite new ones. if target_store.get(device.device_id) != target or target_store.generation(device.device_id)!=generation or not current_device(device): continue if activity: await send_to_device(device, dict(type='agent_activity',target=target,target_generation=generation,agent=parsed[0],session_id=parsed[1], activity_type=payload.get('activity_type'),activity=payload.get('latest_activity',''), recent_text=payload.get('recent_text',''))) else: outgoing = dict(payload,target_generation=generation) if parsed[0] == 'codex' and device.last_agent_state in ('RUNNING','WAITING_APPROVAL') and payload['state'] in ('DONE','FAILED'): outgoing['notification'] = payload['state'].lower() device.last_agent_state = payload['state'] await send_to_device(device, outgoing) except Exception: logger.warning('Agent target delivery failed') _codex_push_tasks = {} def queue_codex_activity(sid, method): # Coalesce rapid deltas; reader never awaits an RPC made on its own socket. async def deliver(): try: await asyncio.sleep(.15) await push_agent_targets('codex', sid) await push_agent_targets('codex', sid, activity=True) finally: _codex_push_tasks.pop(sid, None) if sid not in _codex_push_tasks: _codex_push_tasks[sid] = voices.spawn(deliver()) @app.post("/integrations/pi/events", include_in_schema=False) async def pi_event( event: PiSessionEvent, x_device_lab_pi_token: str = Header(default=""), ) -> dict[str, Any]: if not credential_store.authenticate_pi(x_device_lab_pi_token): raise HTTPException(status_code=401, detail="invalid Pi integration token") session = pi_sessions.apply(event) if event.event in ("session_shutdown", "agent_end"): approvals.cancel_session("pi", event.session_id) user_inputs.cancel_session("pi",event.session_id) if event.event == "agent_end" and session.pending_notification: on_agent_terminal("pi", session.session_id, session.state, session.final_text, name=session.name, latest_activity=session.latest_activity) if event.event == "activity": await push_selected_pi_activity(event) else: await push_selected_pi_session(session.session_id, session.pending_notification) return {"accepted": True, "session": session.public()} @app.post("/devices/{device_id}/target") async def select_target(device_id: str, selection: TargetSelection) -> dict[str, Any]: if device_id not in devices: raise HTTPException(status_code=404, detail="device not found") target = normalize_target(selection.target) parsed = split_target(target) if parsed: agent, sid = parsed if not (agent == 'pi' and pi_sessions.get(sid)): await adapter_for(agent).get_session(sid) device = devices[device_id] if target_store.get(device_id) != target: speeches.cancel(device) voices.cancel(device) device.settings_id = None device.last_agent_state = '' value=target_store.set(device_id,target) generation=value["generation"] delivered = False if current_device(device): try: payload = await agent_target_payload(target) if target_store.generation(device_id)==generation: await send_to_device(device,dict(payload,target_generation=generation)) device.last_agent_state = payload.get("state", "") await speeches.settings(device) delivered = True except Exception: speeches.cancel(device, notify=False, forget=True) voices.cancel(device, notify=False, forget=True) device.settings_id = None if device.websocket: with suppress(Exception): await device.websocket.close(code=1011) device.online = False device.websocket = None return {"device_id": device_id, "target": target, "delivered": delivered} @app.post("/devices/{device_id}/message") async def send_message(device_id: str, message: DisplayMessage) -> dict[str, Any]: device = devices.get(device_id) if device is None: raise HTTPException(status_code=404, detail="device not found") if not device.online or device.websocket is None: raise HTTPException(status_code=409, detail="device is offline") payload = {"type": "display_message", "title": message.title, "body": message.body} try: await send_to_device(device, payload) except Exception as exc: device.online = False device.websocket = None raise HTTPException(status_code=409, detail="device disconnected") from exc logger.info("sent display_message to device %s", device_id) return {"sent": True, "device_id": device_id, "message": payload} @app.post("/devices/{device_id}/test-notification") async def test_notification(device_id: str, request: NotificationTest) -> dict[str, Any]: device = devices.get(device_id) if device is None: raise HTTPException(status_code=404, detail="device not found") if not device.online or device.websocket is None: raise HTTPException(status_code=409, detail="device is offline") try: await send_to_device(device, {"type": "notification_test", "kind": request.kind}) except Exception as exc: device.online = False device.websocket = None raise HTTPException(status_code=409, detail="device disconnected") from exc logger.info("sent %s notification test to device %s", request.kind, device_id) return {"sent": True, "device_id": device_id, "kind": request.kind} class SpeechSelection(BaseModel): target: str = Field(min_length=1, max_length=190) mode: Literal['OFF', 'MANUAL', 'AUTO_REPLY', 'AUTO_SUMMARY'] class SpeechTest(BaseModel): target: str = Field(min_length=1, max_length=190) model_config = {'extra': 'forbid'} @app.post('/api/devices/{device_id}/speech') async def set_speech_mode(device_id: str, value: SpeechSelection): device = devices.get(device_id) if not device: raise HTTPException(404, 'Device not found') if target_store.get(device_id) != normalize_target(value.target): raise HTTPException(409, 'Target changed; refresh settings') speeches.cancel(device) speeches.preferences.set(device_id, value.target, value.mode) await speeches.settings(device) return {'mode': value.mode, 'target': normalize_target(value.target)} @app.post('/api/devices/{device_id}/speech-test') async def test_speech(device_id: str, value: SpeechTest): device = devices.get(device_id) if not device: raise HTTPException(404, 'Device not found') if target_store.get(device_id) != normalize_target(value.target): raise HTTPException(409, 'Target changed; refresh settings') if not speeches.start(device, 'screen', '语音播放测试成功。'): raise HTTPException(409, 'Speech unavailable, device busy or firmware lacks speech_playback') return {'accepted': True} @app.websocket("/device/ws") async def device_websocket(websocket: WebSocket) -> None: await websocket.accept() device: Device | None = None try: hello = await websocket.receive_json() if not isinstance(hello, dict) or hello.get("type") != "hello": await websocket.close(code=1008, reason="first message must be hello") return required = ("device_id", "device_token", "board", "firmware") if any(not isinstance(hello.get(key), str) or not hello[key] for key in required): await websocket.close(code=1008, reason="invalid hello") return device_id = hello["device_id"] if not credential_store.authenticate(device_id, hello["device_token"]): logger.warning("rejected unauthorized device %s", device_id) await websocket.close(code=1008, reason="unauthorized device") return previous = devices.get(device_id) if previous is not None and previous.websocket is not None: speeches.cancel(previous, notify=False, forget=True) voices.cancel(previous, notify=False, forget=True) await previous.websocket.close(code=1012, reason="replaced by a new connection") device = Device( device_id=device_id, board=hello["board"], firmware=hello["firmware"], websocket=websocket, capabilities={v for v in hello.get("capabilities", []) if isinstance(v, str)} if isinstance(hello.get("capabilities"), list) else set(), ) devices[device_id] = device await send_to_device(device, { "type": "hello_ack", "server_time_ms": int(utc_now().timestamp() * 1000), "timezone_offset_minutes": gateway_timezone_offset_minutes(), "capabilities":sorted(CAPABILITIES & device.capabilities), "target_generation":target_store.generation(device_id), }) snapshot = system_collector.latest or system_collector.collect() await send_to_device(device, device_snapshot_payload(snapshot)) target = target_store.get(device_id) payload = await agent_target_payload(target) await send_to_device(device,dict(payload,**({"target_generation":target_store.generation(device_id)} if "target-select" in device.capabilities else {}))) device.last_agent_state = payload.get("state", "") await speeches.settings(device) approvals.changed() logger.info( "device %s online board=%s firmware=%s", device.device_id, device.board, device.firmware ) while True: event = await websocket.receive_json() if not isinstance(event, dict) or not isinstance(event.get("type"), str): logger.warning("device %s sent invalid event", device.device_id) continue device.last_seen = utc_now() # Never expose audio/transcript payloads through Admin or logs. device.last_event = {"type": event["type"]} if event["type"].startswith(("voice_","settings_","speech_","user_input_","session_","creation_","history_")) else event if event["type"].startswith("speech_"): await speeches.handle(device, event) elif event["type"].startswith("voice_"): if event["type"] == "voice_start": speeches.cancel(device) await voices.handle(device, event) elif event["type"] in ("settings_get", "settings_apply", "settings_cancel"): if event["type"] == "settings_get": speeches.cancel(device) if event["type"] == "settings_cancel": await device_settings(device, event) else: if device.settings_task is None or device.settings_task.done(): device.settings_task = voices.spawn(device_settings(device, event)) else: await send_to_device(device, {"type":"settings_error", "request_id":event.get("request_id"), "message":"Settings request busy"}) elif event['type'] in ('catalog_get','target_select','target_get','creation_get','session_create','session_operation_get','history_get','user_input_response'): if event['type']=='user_input_response':await navigator.handle(device,event) elif device.settings_task is None or device.settings_task.done():device.settings_task=voices.spawn(navigator.handle(device,event)) else:await send_to_device(device,dict(type='navigator_error' if event['type']!='history_get' else 'history_error',request_id=event.get('request_id'),message='Busy')) elif event["type"] == "approval_response": if devices.get(device.device_id) is device and device.online: decision=event.get("decision") approval=approvals.get(event.get("approval_id")) # Legacy binary firmware cannot spell native cancel. Never # turn its Reject into acceptance or a nonexistent decline. if ('native-interaction' not in device.capabilities and decision=='reject' and approval and 'decline' not in approval.choices and 'cancel' in approval.choices): decision='cancel' approvals.resolve(event.get("approval_id"), decision, device_connection=device.connection_id) elif event["type"] == "button_event": logger.info( "device %s button clicked button=%s action=%s", device.device_id, event.get("button", "unknown"), event.get("action", "unknown"), ) elif event["type"] == "heartbeat": logger.debug("device %s heartbeat uptime=%s", device.device_id, event.get("uptime")) else: logger.info("device %s event type=%s", device.device_id, event["type"]) except WebSocketDisconnect: pass except Exception: logger.exception("device websocket failed") finally: if device is not None: speeches.cancel(device, notify=False, forget=True) voices.cancel(device, notify=False, forget=True) device.settings_id = None approvals.presentations.pop(device.connection_id, None) user_inputs.presentations.pop(device.connection_id,None) if device is not None and devices.get(device.device_id) is device: speeches.cancel(device, notify=False, forget=True) voices.cancel(device, notify=False, forget=True) device.settings_id = None if device.websocket: with suppress(Exception): await device.websocket.close(code=1011) device.online = False device.websocket = None device.last_seen = utc_now() logger.info("device %s offline", device.device_id)