|
|
@@ -0,0 +1,475 @@
|
|
|
+"""Assistant controller: turns the STT event stream into mode and command state.
|
|
|
+
|
|
|
+Phase 2 scope: no LLM, no tools, no control socket. The service consumes events,
|
|
|
+maintains the mode machine, assembles command text, and exposes an API that the
|
|
|
+Phase 3 control socket will call.
|
|
|
+"""
|
|
|
+
|
|
|
+import logging
|
|
|
+import threading
|
|
|
+import time
|
|
|
+from typing import Callable
|
|
|
+
|
|
|
+from .. import events
|
|
|
+from ..config import Config
|
|
|
+from .executor import ExecutionResult
|
|
|
+from .modes import Mode, ModeMachine
|
|
|
+
|
|
|
+logger = logging.getLogger(__name__)
|
|
|
+
|
|
|
+MODE_EVENT = "mode"
|
|
|
+COMMAND_TEXT_EVENT = "command_text"
|
|
|
+ERROR_EVENT = "error"
|
|
|
+TOOL_CALL_EVENT = "tool_call"
|
|
|
+TOOL_RESULT_EVENT = "tool_result"
|
|
|
+CHAT_MESSAGE_EVENT = "chat_message"
|
|
|
+HEARTBEAT_EVENT = "heartbeat"
|
|
|
+
|
|
|
+CAPTURING = "capturing"
|
|
|
+READY = "ready"
|
|
|
+EXECUTING = "executing"
|
|
|
+RELAYED_TYPES = (events.TRANSCRIPT, events.SPEECH_START, events.SPEECH_STOP)
|
|
|
+
|
|
|
+
|
|
|
+class AssistantService:
|
|
|
+ """Maintains assistant mode, command text, and readiness from STT events."""
|
|
|
+
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ cfg: Config,
|
|
|
+ subscriber,
|
|
|
+ emit: Callable[..., None] | None = None,
|
|
|
+ time_fn: Callable[[], float] = time.time,
|
|
|
+ executor=None,
|
|
|
+ audit=None,
|
|
|
+ relay: Callable[[dict], None] | None = None,
|
|
|
+ chat_fn: Callable[[str], str] | None = None,
|
|
|
+ chat_reset: Callable[[], None] | None = None,
|
|
|
+ ):
|
|
|
+ """Create the service.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ cfg: Configuration (uses `[assistant]` and `[events]`).
|
|
|
+ subscriber: An object exposing `events()` (and `close()`).
|
|
|
+ emit: Sink for assistant events (`mode`, `command_text`, `error`, `tool_*`).
|
|
|
+ time_fn: Clock (injectable for tests).
|
|
|
+ executor: Object with `run(command_text, emit) -> ExecutionResult`.
|
|
|
+ audit: Object with `record(**fields)`; called once per execution.
|
|
|
+ relay: Sink for relayed STT events (`transcript`, `speech_*`).
|
|
|
+ chat_fn: Callable sending one chat turn and returning the reply.
|
|
|
+ chat_reset: Callable clearing the chat session on exit.
|
|
|
+ """
|
|
|
+ self.cfg = cfg
|
|
|
+ self._subscriber = subscriber
|
|
|
+ self._emit = emit if emit is not None else (lambda *a, **k: None)
|
|
|
+ self._relay = relay if relay is not None else (lambda event: None)
|
|
|
+ self._now = time_fn
|
|
|
+ self.executor = executor
|
|
|
+ self._audit = audit
|
|
|
+ self.chat_fn = chat_fn
|
|
|
+ self.chat_reset = chat_reset
|
|
|
+ self.modes = ModeMachine()
|
|
|
+
|
|
|
+ # Speech/segment pipeline state (derived from the event stream)
|
|
|
+ self.speech_open = False
|
|
|
+ self.awaited_seg_id: int | None = None
|
|
|
+
|
|
|
+ # Command buffer
|
|
|
+ self.arm_ts: float | None = None
|
|
|
+ self.command_text = ""
|
|
|
+ self.command_edited = False
|
|
|
+
|
|
|
+ # Chat session (volatile)
|
|
|
+ self.chat_history: list[str] = []
|
|
|
+ self._pending_chat: list[str] = []
|
|
|
+
|
|
|
+ # Health
|
|
|
+ self._last_heartbeat_ts: float | None = None
|
|
|
+ self._worker_stall_s = 0.0
|
|
|
+ self._queue_depth = 0
|
|
|
+ self._device_ok = True
|
|
|
+ self._last_heartbeat_emit = 0.0
|
|
|
+ self._last_activity_ts = 0.0
|
|
|
+ self._last_chat_activity = 0.0
|
|
|
+
|
|
|
+ self._last_emitted: tuple = ()
|
|
|
+ self._stop = threading.Event()
|
|
|
+ self._reader: threading.Thread | None = None
|
|
|
+ self._ticker: threading.Thread | None = None
|
|
|
+ self._exec_thread: threading.Thread | None = None
|
|
|
+
|
|
|
+ # ------------------------------------------------------------------ lifecycle
|
|
|
+
|
|
|
+ def start(self) -> None:
|
|
|
+ """Start the event reader and the timeout ticker."""
|
|
|
+ self._reader = threading.Thread(
|
|
|
+ target=self._read_loop, name="assistant-events", daemon=True
|
|
|
+ )
|
|
|
+ self._ticker = threading.Thread(
|
|
|
+ target=self._tick_loop, name="assistant-tick", daemon=True
|
|
|
+ )
|
|
|
+ self._reader.start()
|
|
|
+ self._ticker.start()
|
|
|
+
|
|
|
+ def stop(self) -> None:
|
|
|
+ """Stop the reader/ticker and close the subscription."""
|
|
|
+ self._stop.set()
|
|
|
+ try:
|
|
|
+ self._subscriber.close()
|
|
|
+ except Exception: # noqa: BLE001 - best-effort shutdown
|
|
|
+ pass
|
|
|
+ for thread in (self._reader, self._ticker):
|
|
|
+ if thread is not None:
|
|
|
+ thread.join(timeout=1)
|
|
|
+
|
|
|
+ def _read_loop(self) -> None:
|
|
|
+ try:
|
|
|
+ for event in self._subscriber.events():
|
|
|
+ if self._stop.is_set():
|
|
|
+ break
|
|
|
+ self.handle_event(event)
|
|
|
+ except Exception as e: # noqa: BLE001 - keep the service alive
|
|
|
+ logger.error("Assistant event loop error: %s", e)
|
|
|
+
|
|
|
+ def _tick_loop(self) -> None:
|
|
|
+ while not self._stop.is_set():
|
|
|
+ self.tick()
|
|
|
+ self._stop.wait(0.1)
|
|
|
+
|
|
|
+ # ------------------------------------------------------------------ queries
|
|
|
+
|
|
|
+ def stt_ok(self) -> bool:
|
|
|
+ """Health of the STT side, from heartbeat recency and worker stall."""
|
|
|
+ if self._last_heartbeat_ts is None:
|
|
|
+ return True
|
|
|
+ now = self._now()
|
|
|
+ if now - self._last_heartbeat_ts > self.cfg.assistant.stall_timeout_ms / 1000.0:
|
|
|
+ return False
|
|
|
+ return self._worker_stall_s <= self.cfg.assistant.worker_stall_ms / 1000.0
|
|
|
+
|
|
|
+ def ready(self) -> bool:
|
|
|
+ """True when no speech is open, every finalized segment is done, and STT is healthy."""
|
|
|
+ return (
|
|
|
+ not self.speech_open
|
|
|
+ and self.awaited_seg_id is None
|
|
|
+ and self.stt_ok()
|
|
|
+ )
|
|
|
+
|
|
|
+ def turn_ended(self) -> bool:
|
|
|
+ """True when the pipeline is ready and the room has been quiet for `turn_gap_s`."""
|
|
|
+ if not self.ready():
|
|
|
+ return False
|
|
|
+ return (self._now() - self._last_activity_ts) >= self.cfg.assistant.turn_gap_s
|
|
|
+
|
|
|
+ def command_phase(self) -> str:
|
|
|
+ if self.modes.mode is Mode.EXECUTING:
|
|
|
+ return EXECUTING
|
|
|
+ if self.modes.mode is Mode.COMMAND_ARMED:
|
|
|
+ return READY if self.ready() else CAPTURING
|
|
|
+ return "idle"
|
|
|
+
|
|
|
+ def status(self) -> dict:
|
|
|
+ """Snapshot for the control socket (`status`)."""
|
|
|
+ return {
|
|
|
+ "mode": self.modes.mode.value,
|
|
|
+ "ready": self.ready(),
|
|
|
+ "stt_ok": self.stt_ok(),
|
|
|
+ "phase": self.command_phase(),
|
|
|
+ "command_text": self.command_text,
|
|
|
+ "speech_open": self.speech_open,
|
|
|
+ "awaited_seg_id": self.awaited_seg_id,
|
|
|
+ "chat_turns": len(self.chat_history),
|
|
|
+ }
|
|
|
+
|
|
|
+ # ------------------------------------------------------------------ actions
|
|
|
+
|
|
|
+ def set_sinks(self, emit: Callable[..., None], relay: Callable[[dict], None]) -> None:
|
|
|
+ """Replace the event sinks (used by the runner once the control server exists)."""
|
|
|
+ self._emit = emit
|
|
|
+ self._relay = relay
|
|
|
+
|
|
|
+ def edit(self, text: str) -> bool:
|
|
|
+ """Replace the command buffer text (refused unless command-armed)."""
|
|
|
+ if self.modes.mode is not Mode.COMMAND_ARMED:
|
|
|
+ return False
|
|
|
+ self.command_text = text
|
|
|
+ self.command_edited = True
|
|
|
+ self._emit_command_text(force=True)
|
|
|
+ return True
|
|
|
+
|
|
|
+ def arm(self) -> bool:
|
|
|
+ """Enter command mode (refused unless idle)."""
|
|
|
+ if not self.modes.arm():
|
|
|
+ return False
|
|
|
+ self.arm_ts = self._now()
|
|
|
+ self.command_text = ""
|
|
|
+ self.command_edited = False
|
|
|
+ self._emit_mode()
|
|
|
+ self._emit_command_text(force=True)
|
|
|
+ return True
|
|
|
+
|
|
|
+ def cancel(self) -> bool:
|
|
|
+ """Leave command mode, discarding the buffer."""
|
|
|
+ if not self.modes.cancel():
|
|
|
+ return False
|
|
|
+ self._clear_command()
|
|
|
+ self._emit_mode()
|
|
|
+ self._emit_command_text(force=True)
|
|
|
+ return True
|
|
|
+
|
|
|
+ def execute(self) -> str | None:
|
|
|
+ """Return the command text and enter `executing`, or None if not allowed.
|
|
|
+
|
|
|
+ When an executor is configured, execution runs on a worker thread so the
|
|
|
+ caller (control socket) is never blocked.
|
|
|
+ """
|
|
|
+ if self.modes.mode is not Mode.COMMAND_ARMED or not self.ready():
|
|
|
+ return None
|
|
|
+ text = self.command_text
|
|
|
+ edited = self.command_edited
|
|
|
+ if not self.modes.execute():
|
|
|
+ return None
|
|
|
+ self._emit_mode()
|
|
|
+ self._emit_command_text(force=True)
|
|
|
+ if self.executor is not None:
|
|
|
+ self._exec_thread = threading.Thread(
|
|
|
+ target=self._run_execution,
|
|
|
+ args=(text, edited),
|
|
|
+ name="assistant-exec",
|
|
|
+ daemon=True,
|
|
|
+ )
|
|
|
+ self._exec_thread.start()
|
|
|
+ return text
|
|
|
+
|
|
|
+ def _run_execution(self, text: str, edited: bool) -> None:
|
|
|
+ start = self._now()
|
|
|
+ result = ExecutionResult(ok=False, result="no executor configured")
|
|
|
+ executor = self.executor
|
|
|
+ if executor is not None:
|
|
|
+ try:
|
|
|
+ result = executor.run(text, self._emit)
|
|
|
+ except Exception as e: # noqa: BLE001 - an executor crash must not kill the service
|
|
|
+ logger.error("Executor error: %s", e)
|
|
|
+ result = ExecutionResult(ok=False, result=f"executor error: {e}")
|
|
|
+
|
|
|
+ latency_ms = int((self._now() - start) * 1000)
|
|
|
+ self._emit(
|
|
|
+ TOOL_RESULT_EVENT,
|
|
|
+ tool=result.tool,
|
|
|
+ ok=result.ok,
|
|
|
+ result=result.result,
|
|
|
+ latency_ms=latency_ms,
|
|
|
+ )
|
|
|
+ if self._audit is not None:
|
|
|
+ self._audit.record(
|
|
|
+ mode="command",
|
|
|
+ text=text,
|
|
|
+ edited=edited,
|
|
|
+ tool=result.tool,
|
|
|
+ args=result.args,
|
|
|
+ side_effect=result.side_effect,
|
|
|
+ requires_confirmation=result.requires_confirmation,
|
|
|
+ confirmed=result.confirmed,
|
|
|
+ ok=result.ok,
|
|
|
+ result=result.result,
|
|
|
+ latency_ms=latency_ms,
|
|
|
+ )
|
|
|
+ self.finish()
|
|
|
+
|
|
|
+ def finish(self) -> bool:
|
|
|
+ """Leave `executing` and discard the buffer."""
|
|
|
+ if not self.modes.finish():
|
|
|
+ return False
|
|
|
+ self._clear_command()
|
|
|
+ self._emit_mode()
|
|
|
+ self._emit_command_text(force=True)
|
|
|
+ return True
|
|
|
+
|
|
|
+ def toggle_chat(self) -> bool:
|
|
|
+ """Enter chat mode from idle, or leave chat mode."""
|
|
|
+ if self.modes.mode is Mode.CHAT:
|
|
|
+ return self._exit_chat("toggle")
|
|
|
+ if not self.modes.enter_chat():
|
|
|
+ return False
|
|
|
+ self.chat_history = []
|
|
|
+ self._pending_chat = []
|
|
|
+ self._last_chat_activity = self._now()
|
|
|
+ self._emit_mode()
|
|
|
+ return True
|
|
|
+
|
|
|
+ # ------------------------------------------------------------------ events
|
|
|
+
|
|
|
+ def handle_event(self, event: dict) -> None:
|
|
|
+ """Consume one STT event."""
|
|
|
+ etype = event.get("type")
|
|
|
+ now = self._now()
|
|
|
+
|
|
|
+ if etype in RELAYED_TYPES:
|
|
|
+ self._relay(event)
|
|
|
+
|
|
|
+ if etype == events.STATUS:
|
|
|
+ self._on_status(now)
|
|
|
+ elif etype == events.SPEECH_START:
|
|
|
+ self.speech_open = True
|
|
|
+ self._last_activity_ts = now
|
|
|
+ elif etype == events.SPEECH_STOP:
|
|
|
+ self.speech_open = False
|
|
|
+ self.awaited_seg_id = event.get("seg_id")
|
|
|
+ self._last_activity_ts = now
|
|
|
+ elif etype == events.SEGMENT_DONE:
|
|
|
+ if event.get("seg_id") == self.awaited_seg_id:
|
|
|
+ self.awaited_seg_id = None
|
|
|
+ self._last_activity_ts = now
|
|
|
+ elif etype == events.TRANSCRIPT:
|
|
|
+ self._on_transcript(event, now)
|
|
|
+ elif etype == events.HEARTBEAT:
|
|
|
+ self._last_heartbeat_ts = now
|
|
|
+ self._worker_stall_s = float(event.get("worker_stall_s", 0.0) or 0.0)
|
|
|
+ self._queue_depth = int(event.get("queue_depth", 0) or 0)
|
|
|
+ self._device_ok = bool(event.get("device_ok", True))
|
|
|
+
|
|
|
+ self._check_timeouts(now)
|
|
|
+ self._emit_command_text()
|
|
|
+
|
|
|
+ def tick(self) -> None:
|
|
|
+ """Evaluate timeouts, dispatch pending chat, and emit a heartbeat."""
|
|
|
+ now = self._now()
|
|
|
+ self._check_timeouts(now)
|
|
|
+ if (
|
|
|
+ self.modes.mode is Mode.CHAT
|
|
|
+ and self._pending_chat
|
|
|
+ and self.chat_fn is not None
|
|
|
+ and self.turn_ended()
|
|
|
+ ):
|
|
|
+ self._dispatch_chat()
|
|
|
+ self._emit_heartbeat(now)
|
|
|
+ self._emit_command_text()
|
|
|
+
|
|
|
+ def _emit_heartbeat(self, now: float) -> None:
|
|
|
+ if now - self._last_heartbeat_emit < 1.0:
|
|
|
+ return
|
|
|
+ self._last_heartbeat_emit = now
|
|
|
+ self._emit(
|
|
|
+ HEARTBEAT_EVENT,
|
|
|
+ mode=self.modes.mode.value,
|
|
|
+ stt_ok=self.stt_ok(),
|
|
|
+ in_flight=self.modes.mode is Mode.EXECUTING,
|
|
|
+ worker_stall_s=round(self._worker_stall_s, 3),
|
|
|
+ queue_depth=self._queue_depth,
|
|
|
+ device_ok=self._device_ok,
|
|
|
+ )
|
|
|
+
|
|
|
+ def _on_status(self, now: float) -> None:
|
|
|
+ # Reconnect rule: the replayed ring is informational. Never rebuild speech
|
|
|
+ # state from it; reset so a stale speech_start cannot pin speech_open.
|
|
|
+ self.speech_open = False
|
|
|
+ self.awaited_seg_id = None
|
|
|
+ self._last_activity_ts = now
|
|
|
+
|
|
|
+ def _on_transcript(self, event: dict, now: float) -> None:
|
|
|
+ text = (event.get("text") or "").strip()
|
|
|
+ if not text:
|
|
|
+ return
|
|
|
+ self._last_activity_ts = now
|
|
|
+
|
|
|
+ if self.modes.mode is Mode.COMMAND_ARMED:
|
|
|
+ ts = float(event.get("ts", now))
|
|
|
+ # Ignore transcripts from before arming (e.g. ring replay on attach).
|
|
|
+ if self.arm_ts is None or ts >= self.arm_ts:
|
|
|
+ self.command_text = f"{self.command_text} {text}".strip()
|
|
|
+ elif self.modes.mode is Mode.CHAT:
|
|
|
+ self.chat_history.append(text)
|
|
|
+ self._pending_chat.append(text)
|
|
|
+ self._last_chat_activity = now
|
|
|
+ if self._is_exit_phrase(text):
|
|
|
+ self._exit_chat("exit_phrase")
|
|
|
+ self._emit_command_text()
|
|
|
+
|
|
|
+ def _is_exit_phrase(self, text: str) -> bool:
|
|
|
+ normalized = text.lower().strip().rstrip(".!?,")
|
|
|
+ return any(phrase in normalized for phrase in self.cfg.assistant.chat_exit_phrases)
|
|
|
+
|
|
|
+ def _check_timeouts(self, now: float) -> None:
|
|
|
+ if self.modes.mode is Mode.COMMAND_ARMED:
|
|
|
+ if (
|
|
|
+ self.arm_ts is not None
|
|
|
+ and now - self.arm_ts > self.cfg.assistant.command_arm_timeout_s
|
|
|
+ ):
|
|
|
+ self._disarm("arm_timeout")
|
|
|
+ elif not self.stt_ok():
|
|
|
+ self._disarm("stt_stall")
|
|
|
+ elif self.modes.mode is Mode.CHAT:
|
|
|
+ if now - self._last_chat_activity > self.cfg.assistant.chat_idle_timeout_s:
|
|
|
+ self._exit_chat("idle_timeout")
|
|
|
+
|
|
|
+ # ------------------------------------------------------------------ helpers
|
|
|
+
|
|
|
+ def _disarm(self, reason: str) -> None:
|
|
|
+ if not self.modes.cancel():
|
|
|
+ return
|
|
|
+ self._clear_command()
|
|
|
+ self._emit_mode()
|
|
|
+ self._emit_command_text(force=True)
|
|
|
+ self._emit(ERROR_EVENT, code=reason, message=f"command {reason}")
|
|
|
+
|
|
|
+ def _exit_chat(self, reason: str) -> bool:
|
|
|
+ if not self.modes.exit_chat():
|
|
|
+ return False
|
|
|
+ self.chat_history = []
|
|
|
+ self._pending_chat = []
|
|
|
+ if self.chat_reset is not None:
|
|
|
+ try:
|
|
|
+ self.chat_reset()
|
|
|
+ except Exception as e: # noqa: BLE001 - resetting chat must not break the mode change
|
|
|
+ logger.warning("chat reset failed: %s", e)
|
|
|
+ self._emit_mode()
|
|
|
+ if reason != "toggle":
|
|
|
+ self._emit(ERROR_EVENT, code=reason, message=f"chat {reason}")
|
|
|
+ return True
|
|
|
+
|
|
|
+ def _dispatch_chat(self) -> None:
|
|
|
+ text = " ".join(self._pending_chat).strip()
|
|
|
+ self._pending_chat = []
|
|
|
+ if not text:
|
|
|
+ return
|
|
|
+ self._emit(CHAT_MESSAGE_EVENT, role="user", text=text)
|
|
|
+ thread = threading.Thread(
|
|
|
+ target=self._run_chat, args=(text,), name="assistant-chat", daemon=True
|
|
|
+ )
|
|
|
+ thread.start()
|
|
|
+
|
|
|
+ def _run_chat(self, text: str) -> None:
|
|
|
+ chat_fn = self.chat_fn
|
|
|
+ if chat_fn is None:
|
|
|
+ return
|
|
|
+ try:
|
|
|
+ reply = chat_fn(text)
|
|
|
+ except Exception as e: # noqa: BLE001 - an LLM failure must not kill the service
|
|
|
+ logger.error("Chat error: %s", e)
|
|
|
+ self._emit(ERROR_EVENT, code="llm_error", message=str(e))
|
|
|
+ return
|
|
|
+ if reply:
|
|
|
+ self._emit(CHAT_MESSAGE_EVENT, role="assistant", text=reply)
|
|
|
+
|
|
|
+ def _clear_command(self) -> None:
|
|
|
+ self.arm_ts = None
|
|
|
+ self.command_text = ""
|
|
|
+ self.command_edited = False
|
|
|
+
|
|
|
+ def _emit_mode(self) -> None:
|
|
|
+ self._emit(MODE_EVENT, mode=self.modes.mode.value)
|
|
|
+
|
|
|
+ def _emit_command_text(self, force: bool = False) -> None:
|
|
|
+ if self.modes.mode not in (Mode.COMMAND_ARMED, Mode.EXECUTING):
|
|
|
+ return
|
|
|
+ phase = self.command_phase()
|
|
|
+ key = (self.command_text, self.command_edited, phase)
|
|
|
+ if not force and key == self._last_emitted:
|
|
|
+ return
|
|
|
+ self._last_emitted = key
|
|
|
+ self._emit(
|
|
|
+ COMMAND_TEXT_EVENT,
|
|
|
+ text=self.command_text,
|
|
|
+ edited=self.command_edited,
|
|
|
+ phase=phase,
|
|
|
+ )
|