voice2 1.0.0 — full-duplex interruptible voice engine for local AI

.gitignore

@@ -0,0 +1,10 @@
+__pycache__/
+*.pyc
+*.egg-info/
+dist/
+build/
+.venv/
+venv/
+voice_engine.jsonl
+*.onnx
+*.onnx.json

.pytest_cache/.gitignore

@@ -0,0 +1,2 @@
+# Created by pytest automatically.
+*

.pytest_cache/CACHEDIR.TAG

@@ -0,0 +1,4 @@
+Signature: 8a477f597d28d172789f06886806bc55
+# This file is a cache directory tag created by pytest.
+# For information about cache directory tags, see:
+#	https://bford.info/cachedir/spec.html

.pytest_cache/README.md

@@ -0,0 +1,8 @@
+# pytest cache directory #
+
+This directory contains data from the pytest's cache plugin,
+which provides the `--lf` and `--ff` options, as well as the `cache` fixture.
+
+**Do not** commit this to version control.
+
+See [the docs](https://docs.pytest.org/en/stable/how-to/cache.html) for more information.

.pytest_cache/v/cache/nodeids

@@ -0,0 +1,16 @@
+[
+  "voice2/tests/test_interrupt_controller.py::test_clear",
+  "voice2/tests/test_interrupt_controller.py::test_debounce_allows_after_window",
+  "voice2/tests/test_interrupt_controller.py::test_debounce_blocks_second",
+  "voice2/tests/test_interrupt_controller.py::test_trigger_sets_event",
+  "voice2/tests/test_ring_buffer.py::test_append_and_get",
+  "voice2/tests/test_ring_buffer.py::test_capacity_trimming",
+  "voice2/tests/test_ring_buffer.py::test_clear",
+  "voice2/tests/test_ring_buffer.py::test_preroll_extraction",
+  "voice2/tests/test_state_controller.py::test_floor_blocks_agent_speaking",
+  "voice2/tests/test_state_controller.py::test_invalid_transition_rejected",
+  "voice2/tests/test_state_controller.py::test_snapshot",
+  "voice2/tests/test_state_controller.py::test_stopped_blocks_all",
+  "voice2/tests/test_state_controller.py::test_turn_counter",
+  "voice2/tests/test_state_controller.py::test_valid_transition"
+]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Rhet Wike
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,94 @@
+---
+license: mit
+language:
+- en
+tags:
+- voice
+- speech
+- full-duplex
+- barge-in
+- vad
+- asr
+- tts
+- piper
+- whisper
+- silero
+- local-ai
+- voice-assistant
+---
+
+# voice2
+
+A full-duplex, interruptible voice engine for local AI, **built and run daily as the voice of a fully local companion** before being extracted for release. Plain Python threads, CPU-only defaults, no cloud, no API keys. Talk to your model — and talk over it.
+
+voice2 turns any `callable(text) -> str` into a hands-free voice conversation: mic → Silero VAD → faster-whisper ASR → your model → Piper TTS. Speak over the reply (or tap spacebar) and playback stops mid-chunk in ~100–200 ms, exactly like interrupting a person.
+
+Code is mirrored on GitHub: https://github.com/AIIT-GLITCH/voice2
+
+## Engine details
+
+- **What it is:** a turn-taking state machine, not a demo loop. Explicit states (`IDLE → LISTENING → THINKING → SPEAKING → INTERRUPTING`), a validated transition table, and a `FloorOwner` (USER / AGENT / NONE) that arbitrates who may speak. The agent can never talk over you.
+- **Barge-in:** a fast energy-gated VAD watches the mic *only while the engine is speaking*; a debounced central `InterruptController` also accepts spacebar and programmatic triggers.
+- **Stale-turn suppression:** every utterance gets a `turn_id`; replies to an abandoned turn are discarded at every stage (think, TTS, playback).
+- **Invariants, enforced:** a background checker audits rules like *SPEAKING ⇒ floor == AGENT* and *interrupted ⇒ no new TTS*, with forced repair plus a structural gate in the playback hot path.
+- **Observability:** every event is a JSON line with per-turn latency marks (`asr_ms`, `think_ms`, `interrupt_stop_ms`, `total_turn_ms`).
+- **Degrades gracefully:** no mic → text mode; TTS missing → silent replies, still logs; keyboard hook fails → engine keeps running.
+
+## Backends
+
+| Stage | Default | Swap point |
+|---|---|---|
+| VAD (quality gate) | Silero VAD via `torch.hub` | `ListenWorker` |
+| ASR | faster-whisper `small.en`, int8, CPU | `backends/asr.py` (Protocol) |
+| LLM | any `callable(text) -> str` | `backends/llm.py` |
+| TTS | Piper CLI, any `.onnx` voice | `backends/tts.py` (Protocol) |
+
+## Limitations — stated honestly
+
+- English-first defaults: ASR ships as `small.en`. Other Whisper models load with one config line, but nothing else was tested.
+- The LLM callable is synchronous: TTS starts after the full reply returns (Piper then streams sentence-by-sentence). No token-level streaming yet.
+- Barge-in is energy-based with an absolute RMS floor of 0.06 — tuned on open speakers in a quiet room. Headsets and noisy rooms need recalibration. There is no echo cancellation.
+- Keyboard interrupt uses POSIX `termios` — Linux/macOS terminals only.
+- Unit tests cover the control plane (state transitions, floor rules, interrupt debounce, ring buffer). Audio I/O paths were validated by months of daily use, not by CI.
+
+## How to run
+
+```bash
+git clone https://github.com/AIIT-GLITCH/voice2
+cd voice2
+pip install -r requirements.txt
+# put a Piper voice at ~/.local/share/piper-voices/ (or set VOICE2_PIPER_MODEL)
+python -m voice2.main          # echo backend — proves the loop, no LLM needed
+python examples/http_llm.py    # wire any local HTTP LLM
+```
+
+```python
+from voice2 import VoiceEngine, VoiceConfig
+
+def ask(text: str) -> str:
+    return my_model.reply(text)   # any callable(text) -> str
+
+engine = VoiceEngine(VoiceConfig(), ask)
+engine.load_models()
+engine.start()   # talk naturally; speak over it to interrupt
+```
+
+## Provenance
+
+voice2 was written as the voice front-end for **Buddy**, a fully local AI companion running on a single RTX 3090 in Council Hill, Oklahoma, and carried his daily conversations for months before release. The design bias throughout: the user always wins the floor, and a companion you can't interrupt isn't a companion. Released alongside [Tessera-1B](https://huggingface.co/AIIT-Threshold/Tessera-1B) as part of AIIT-THRESHOLD's open stack.
+
+## License
+
+MIT © 2026 Rhet Dillard Wike, AIIT-THRESHOLD, Oklahoma.
+
+## Citation
+
+```bibtex
+@software{wike2026voice2,
+  author = {Wike, Rhet Dillard},
+  title  = {voice2: a full-duplex, interruptible voice engine for local AI},
+  year   = {2026},
+  url    = {https://github.com/AIIT-GLITCH/voice2},
+  note   = {AIIT-THRESHOLD}
+}
+```

examples/http_llm.py

@@ -0,0 +1,48 @@
+"""
+Wire the voice engine to any local LLM behind an HTTP API.
+
+Works with anything that takes text and returns text — llama.cpp server,
+Ollama, vLLM, or your own FastAPI wrapper. Edit `ask()` to match your API.
+
+Usage:
+    LLM_URL=http://localhost:8000/chat python examples/http_llm.py
+"""
+import os
+import signal
+
+import httpx
+
+from voice2 import VoiceEngine, VoiceConfig
+
+LLM_URL = os.environ.get("LLM_URL", "http://localhost:11434/api/generate")
+MODEL = os.environ.get("LLM_MODEL", "llama3.2")
+
+
+def ask(text: str) -> str:
+    """Ollama-style example. Adapt the payload/response to your server."""
+    try:
+        r = httpx.post(
+            LLM_URL,
+            json={"model": MODEL, "prompt": text, "stream": False},
+            timeout=120.0,
+        )
+        r.raise_for_status()
+        return r.json().get("response", "").strip() or "..."
+    except Exception as e:
+        return f"Backend error: {e}"
+
+
+def main() -> None:
+    engine = VoiceEngine(VoiceConfig(), ask)
+    engine.load_models()
+    engine.start()
+    print("Online. Talk naturally. Space = interrupt, Ctrl+C = quit.")
+    try:
+        signal.pause()
+    except KeyboardInterrupt:
+        pass
+    engine.stop()
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -0,0 +1,31 @@
+[build-system]
+requires = ["setuptools>=64"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "voice2"
+version = "1.0.0"
+description = "Full-duplex, interruptible voice engine for local AI - barge-in, floor management, stale-turn suppression"
+readme = "README.md"
+license = { text = "MIT" }
+authors = [{ name = "Rhet Wike" }]
+requires-python = ">=3.10"
+dependencies = [
+    "numpy",
+    "sounddevice",
+    "faster-whisper",
+    "torch",
+]
+keywords = ["voice", "speech", "tts", "asr", "llm", "barge-in", "full-duplex", "piper", "whisper", "local-ai"]
+classifiers = [
+    "License :: OSI Approved :: MIT License",
+    "Operating System :: POSIX :: Linux",
+    "Programming Language :: Python :: 3",
+    "Topic :: Multimedia :: Sound/Audio :: Speech",
+]
+
+[project.urls]
+Homepage = "https://github.com/AIIT-GLITCH/voice2"
+
+[tool.setuptools.packages.find]
+include = ["voice2*"]

requirements.txt

@@ -0,0 +1,5 @@
+numpy
+sounddevice
+faster-whisper
+torch          # Silero VAD via torch.hub (CPU build is fine)
+httpx          # examples/http_llm.py only

voice2/__init__.py

@@ -0,0 +1,6 @@
+"""voice2 — production-grade stateful interruptible voice engine."""
+from .engine import VoiceEngine
+from .config import VoiceConfig
+from .enums import EngineState, FloorOwner, InterruptSource
+
+__all__ = ["VoiceEngine", "VoiceConfig", "EngineState", "FloorOwner", "InterruptSource"]

voice2/audio_broadcaster.py

@@ -0,0 +1,44 @@
+"""AudioBroadcaster — fan-out mic frames to N consumers. No frame theft."""
+import queue
+import threading
+from collections import defaultdict
+from typing import Dict
+import numpy as np
+from . import logging_util as log
+
+
+class AudioBroadcaster:
+    def __init__(self, shutdown: threading.Event) -> None:
+        self._shutdown = shutdown
+        self._lock = threading.Lock()
+        self._consumers: Dict[str, queue.Queue] = {}
+        self._dropped: Dict[str, int] = defaultdict(int)
+
+    def subscribe(self, name: str, maxsize: int = 500) -> queue.Queue:
+        q: queue.Queue = queue.Queue(maxsize=maxsize)
+        with self._lock:
+            self._consumers[name] = q
+        log.event("broadcaster", "subscribed", meta={"name": name, "maxsize": maxsize})
+        return q
+
+    def unsubscribe(self, name: str) -> None:
+        with self._lock:
+            self._consumers.pop(name, None)
+
+    def publish(self, frame: np.ndarray) -> None:
+        with self._lock:
+            consumers = list(self._consumers.items())
+        for name, q in consumers:
+            try:
+                q.put_nowait(frame)
+            except queue.Full:
+                # Drop oldest, push newest — never block the publisher
+                try:
+                    q.get_nowait()
+                    q.put_nowait(frame)
+                    self._dropped[name] += 1
+                except Exception:
+                    pass
+
+    def dropped_count(self, name: str) -> int:
+        return self._dropped.get(name, 0)

voice2/backends/asr.py

@@ -0,0 +1,22 @@
+"""ASR backend wrapper. Swap internals without touching workers."""
+from typing import Protocol
+import numpy as np
+
+
+class ASRBackend(Protocol):
+    def transcribe(self, audio: np.ndarray) -> str: ...
+
+
+class FasterWhisperASR:
+    def __init__(self, model_size: str = "small.en", device: str = "cpu",
+                 compute_type: str = "int8") -> None:
+        from faster_whisper import WhisperModel
+        self._model = WhisperModel(model_size, device=device,
+                                   compute_type=compute_type)
+
+    def transcribe(self, audio: np.ndarray) -> str:
+        audio_f32 = audio.astype("float32")
+        segments, _ = self._model.transcribe(audio_f32, language="en",
+                                              beam_size=1, vad_filter=False,
+                                              condition_on_previous_text=False)
+        return " ".join(s.text.strip() for s in segments).strip()

voice2/backends/llm.py

@@ -0,0 +1,15 @@
+"""LLM backend wrapper. Wire any callable(text) -> str here."""
+from typing import Callable, Protocol
+
+
+class LLMBackend(Protocol):
+    def reply(self, text: str) -> str: ...
+
+
+class CallableLLM:
+    """Wraps any callable(text) -> str as an LLM backend."""
+    def __init__(self, fn: Callable[[str], str]) -> None:
+        self._fn = fn
+
+    def reply(self, text: str) -> str:
+        return self._fn(text)

voice2/backends/tts.py

@@ -0,0 +1,53 @@
+"""TTS backend — yields PCM chunks. Swap without touching PlaybackWorker."""
+from typing import Iterable, Protocol
+import io
+import os
+import subprocess
+import numpy as np
+
+
+class TTSBackend(Protocol):
+    def synthesize(self, text: str) -> Iterable[np.ndarray]: ...
+
+
+class PiperTTS:
+    """Yields one PCM chunk per sentence. Starts playback before full synthesis."""
+
+    def __init__(self, model_path: str, sample_rate: int = 22050) -> None:
+        self._model = model_path
+        self._sr = sample_rate
+        self._length_scale = os.environ.get("VOICE2_TTS_LENGTH_SCALE")
+        self._noise_scale = os.environ.get("VOICE2_TTS_NOISE_SCALE")
+        self._noise_w = os.environ.get("VOICE2_TTS_NOISE_W")
+        self._volume = os.environ.get("VOICE2_TTS_VOLUME")
+
+    def synthesize(self, text: str) -> Iterable[np.ndarray]:
+        # Piper reads text from stdin, outputs raw PCM on stdout
+        cmd = ["piper", "--model", self._model, "--output-raw"]
+        if self._length_scale:
+            cmd += ["--length-scale", self._length_scale]
+        if self._noise_scale:
+            cmd += ["--noise-scale", self._noise_scale]
+        if self._noise_w:
+            cmd += ["--noise-w-scale", self._noise_w]
+        if self._volume:
+            cmd += ["--volume", self._volume]
+        proc = subprocess.Popen(
+            cmd,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+        )
+        assert proc.stdin and proc.stdout
+        proc.stdin.write(text.encode())
+        proc.stdin.close()
+
+        chunk_bytes = int(self._sr * 0.020) * 2  # 20ms of int16
+        while True:
+            raw = proc.stdout.read(chunk_bytes)
+            if not raw:
+                break
+            pcm = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0
+            yield pcm
+
+        proc.wait()

voice2/config.py

@@ -0,0 +1,102 @@
+"""Voice engine configuration. Dataclasses only. No logic here."""
+import os
+from dataclasses import dataclass, field
+from typing import Optional
+
+
+@dataclass
+class AudioConfig:
+    sample_rate: int = 16000
+    channels: int = 1
+    dtype: str = "float32"
+    block_ms: int = 32
+    input_device: int | None = None
+    output_device: int | None = None
+
+    @property
+    def blocksize(self) -> int:
+        return int(self.sample_rate * self.block_ms / 1000)
+
+
+@dataclass
+class VADConfig:
+    """Normal utterance capture — patient."""
+    threshold: float = 0.5
+    pre_roll_ms: int = 400
+    min_utterance_ms: int = 300
+    min_voiced_ms: int = 320
+    min_voice_prob: float = 0.45
+    min_voiced_ratio: float = 0.08
+    end_silence_ms: int = 5000   # patient: give the speaker room to think mid-sentence
+    max_utterance_sec: float = 90.0
+
+
+@dataclass
+class InterruptVADConfig:
+    """Aggressive barge-in detection during SPEAKING. Fast and cheap."""
+    threshold: float = 0.4
+    consecutive_frames_required: int = 4   # ~128ms at 32ms blocks
+    energy_multiplier: float = 2.0         # must exceed N× playback bleed baseline
+    refractory_ms: int = 500               # don't re-trigger within this window
+    check_interval_ms: int = 32            # match block_ms
+
+
+@dataclass
+class ASRConfig:
+    model_size: str = "small.en"
+    device: str = "cpu"
+    compute_type: str = "int8"
+    language: str = "en"
+    beam_size: int = 1
+    vad_filter: bool = False
+    condition_on_previous_text: bool = False
+
+
+@dataclass
+class TTSConfig:
+    model_path: str = os.environ.get(
+        "VOICE2_PIPER_MODEL",
+        os.path.expanduser("~/.local/share/piper-voices/en_US-ryan-high.onnx"),
+    )
+    sample_rate: int = 22050
+    chunk_ms: int = 20          # playback chunk size — 20ms = fast interrupt response
+
+
+@dataclass
+class PlaybackConfig:
+    chunk_ms: int = 20          # must match TTSConfig.chunk_ms
+    discard_on_interrupt: bool = True
+
+
+@dataclass
+class InterruptConfig:
+    debounce_ms: int = 200
+    keyboard_key: str = " "     # spacebar
+
+
+@dataclass
+class RingBufferConfig:
+    seconds: float = 2.0
+
+
+@dataclass
+class QueueConfig:
+    """Sizes for all internal queues. Tune for latency vs. drop behavior."""
+    broadcast_listen_maxsize: int = 500
+    broadcast_interrupt_maxsize: int = 200
+    transcript_maxsize: int = 50
+    playback_maxsize: int = 200
+
+
+@dataclass
+class VoiceConfig:
+    audio: AudioConfig = field(default_factory=AudioConfig)
+    vad: VADConfig = field(default_factory=VADConfig)
+    interrupt_vad: InterruptVADConfig = field(default_factory=InterruptVADConfig)
+    asr: ASRConfig = field(default_factory=ASRConfig)
+    tts: TTSConfig = field(default_factory=TTSConfig)
+    playback: PlaybackConfig = field(default_factory=PlaybackConfig)
+    interrupt: InterruptConfig = field(default_factory=InterruptConfig)
+    ring: RingBufferConfig = field(default_factory=RingBufferConfig)
+    queues: QueueConfig = field(default_factory=QueueConfig)
+    log_file: str | None = "voice_engine.jsonl"

voice2/engine.py

@@ -0,0 +1,299 @@
+"""VoiceEngine — orchestration only. Wires all workers, manages lifecycle."""
+import os
+import queue
+import subprocess
+import sys
+import threading
+from typing import Callable
+
+import sounddevice as sd
+import numpy as np
+
+# Optional companion visualizer: point VOICE2_EDGE_GLOW at a script that tails
+# the engine's JSONL log (e.g. a screen-edge glow while speaking). Off by default.
+EDGE_GLOW_PATH = os.path.expanduser(os.environ.get('VOICE2_EDGE_GLOW', ''))
+
+from .config import VoiceConfig
+from .enums import EngineState, TransitionReason
+from .shared_state import SharedState
+from .state_controller import StateController
+from .floor_manager import FloorManager
+from .interrupt_controller import InterruptController
+from .audio_broadcaster import AudioBroadcaster
+from .ring_buffer import RingAudioBuffer
+from .workers.listen import ListenWorker
+from .workers.interrupt_detector import InterruptDetectorWorker
+from .workers.think import ThinkWorker
+from .workers.playback import PlaybackWorker
+from .workers.keyboard import KeyboardWorker
+from .backends.asr import FasterWhisperASR
+from .backends.tts import PiperTTS
+from .backends.llm import CallableLLM
+from .invariants import InvariantChecker
+from .tones import UICues
+from . import logging_util as log
+
+
+class VoiceEngine:
+    def __init__(self, config: VoiceConfig, ask_fn: Callable[[str], str]) -> None:
+        self.cfg = config
+        self._ask_fn = ask_fn
+        self._started = False
+
+        # Control plane
+        self.shared = SharedState()
+        self.ctrl = StateController(self.shared)
+        self.floor = FloorManager(self.shared, self.ctrl)
+        self.interrupt = InterruptController(
+            self.shared, self.ctrl, self.floor,
+            debounce_ms=config.interrupt.debounce_ms,
+        )
+
+        # Data plane
+        self.broadcaster = AudioBroadcaster(self.shared.shutdown)
+        self.ring = RingAudioBuffer(
+            config.audio.sample_rate, config.audio.channels,
+            config.ring.seconds,
+        )
+        self._transcript_q: queue.Queue = queue.Queue(maxsize=50)
+        self._threads: list[threading.Thread] = []
+        self._stream: sd.InputStream | None = None
+
+        # UI cues
+        self.cues = UICues(
+            sample_rate=config.tts.sample_rate,
+            device=config.audio.output_device,
+        )
+
+        # Backends (lazy loaded)
+        self._asr = None
+        self._tts = None
+        self._llm = None
+
+        # Workers (built after load_models)
+        self._listen_worker = None
+        self._interrupt_det = None
+        self._think_worker = None
+        self._playback_worker = None
+        self._keyboard_worker = None
+
+    # ── Lifecycle ──
+
+    def load_models(self) -> None:
+        log.event("engine", "load_models_start")
+        try:
+            self._asr = FasterWhisperASR(
+                self.cfg.asr.model_size,
+                self.cfg.asr.device,
+                self.cfg.asr.compute_type,
+            )
+        except Exception as e:
+            log.event("engine", "asr_load_failed", error=str(e))
+            self.shared.asr_available = False
+
+        try:
+            self._tts = PiperTTS(self.cfg.tts.model_path, self.cfg.tts.sample_rate)
+        except Exception as e:
+            log.event("engine", "tts_load_failed", error=str(e))
+            self.shared.voice_out_available = False
+
+        self._llm = CallableLLM(self._ask_fn)
+        log.event("engine", "load_models_done")
+
+    def start(self) -> None:
+        if self._started:
+            return
+
+        log.init(self.cfg.log_file)
+        log.event("engine", "starting")
+
+        if self._asr is None:
+            self.load_models()
+
+        # Test audio I/O
+        self._test_audio_in()
+        self._test_audio_out()
+
+        # Build workers
+        qcfg = self.cfg.queues
+        listen_q = self.broadcaster.subscribe("listen", maxsize=qcfg.broadcast_listen_maxsize)
+        interrupt_q = self.broadcaster.subscribe("interrupt_det", maxsize=qcfg.broadcast_interrupt_maxsize)
+
+        self._playback_worker = PlaybackWorker(
+            self.shared, self.ctrl, self.floor, self.interrupt,
+            sample_rate=self.cfg.tts.sample_rate,
+            chunk_ms=self.cfg.playback.chunk_ms,
+            output_device=self.cfg.audio.output_device,
+            cues=self.cues,
+        )
+        self._think_worker = ThinkWorker(
+            self._transcript_q, self.shared, self.ctrl, self.floor,
+            self.interrupt, self._llm, self._playback_worker, self._tts,
+            cues=self.cues,
+        )
+        self._listen_worker = ListenWorker(
+            listen_q, self.shared, self.ctrl, self.floor, self.interrupt,
+            self.ring, self.cfg.vad, self.cfg.audio.sample_rate,
+            self._transcript_q, self._asr, cues=self.cues,
+        )
+        self._interrupt_det = InterruptDetectorWorker(
+            interrupt_q, self.shared, self.ctrl, self.interrupt,
+            self.cfg.interrupt_vad,
+        )
+        self._keyboard_worker = KeyboardWorker(
+            self.shared, self.interrupt, self.cfg.interrupt.keyboard_key,
+        )
+
+        # Launch workers (Thread subclasses — call .start() directly)
+        for worker in (self._listen_worker, self._interrupt_det,
+                       self._think_worker, self._playback_worker):
+            worker.start()
+            self._threads.append(worker)
+
+        # Keyboard in isolated try — failure here doesn't kill engine
+        try:
+            self._keyboard_worker.start()
+            self._threads.append(self._keyboard_worker)
+        except Exception as e:
+            log.event("engine", "keyboard_start_failed", error=str(e))
+            self.shared.keyboard_interrupt_available = False
+
+        # Start mic stream
+        self._start_mic()
+
+        # Invariant monitor
+        self._invariants = InvariantChecker(self.shared, self.ctrl)
+        self._spawn("invariants", lambda: self._invariants.run_loop(interval_sec=2.0))
+
+        # Engine starts in IDLE — no transition needed, just record it
+        self._started = True
+        log.event("engine", "online", state=self.ctrl.get_state().name)
+
+        # Launch optional edge-glow companion (VOICE2_EDGE_GLOW).
+        # Runs under system python so it can use GUI bindings not in this venv.
+        self._edge_glow = None
+        if (self.cfg.log_file and EDGE_GLOW_PATH
+                and os.path.exists(EDGE_GLOW_PATH) and os.environ.get('DISPLAY')):
+            try:
+                self._edge_glow = subprocess.Popen(
+                    ['/usr/bin/python3', EDGE_GLOW_PATH,
+                     '--log', self.cfg.log_file,
+                     '--rate', str(self.cfg.tts.sample_rate)],
+                    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                )
+                log.event("engine", "edge_glow_started", pid=self._edge_glow.pid)
+            except Exception as e:
+                log.event("engine", "edge_glow_start_failed", error=str(e))
+
+    def stop(self) -> None:
+        if not self._started:
+            return
+        log.event("engine", "stopping")
+        self.shared.shutdown.set()
+        self.shared.interrupted.set()  # unblock any active playback loop immediately
+        # Shut down cue stream first (no concurrent sd.play calls after this)
+        if self.cues:
+            try:
+                self.cues.close()
+            except Exception:
+                pass
+        if self._stream:
+            try:
+                self._stream.abort()
+                self._stream.close()
+            except Exception:
+                pass
+        for t in self._threads:
+            t.join(timeout=2.0)
+        if getattr(self, "_edge_glow", None):
+            try:
+                self._edge_glow.terminate()
+                self._edge_glow.wait(timeout=2.0)
+            except Exception:
+                pass
+        self._started = False
+        log.event("engine", "stopped")
+
+    def submit_text(self, text: str) -> None:
+        """Fallback: inject text directly as if ASR produced it."""
+        from .logging_util import LatencyTrace
+        turn_id = self.ctrl.start_new_turn()
+        trace = LatencyTrace(turn_id)
+        trace.mark("turn_start")
+        self._transcript_q.put_nowait((text, turn_id, trace))
+
+    def status(self) -> dict:
+        return {
+            **self.ctrl.snapshot(),
+            "started": self._started,
+            "interrupted": self.shared.interrupted.is_set(),
+            "speaking": self.shared.speaking.is_set(),
+            "thinking": self.shared.thinking.is_set(),
+            "listening": self.shared.listening.is_set(),
+            "capabilities": {
+                "voice_in_available": self.shared.voice_in_available,
+                "voice_out_available": self.shared.voice_out_available,
+                "asr_available": self.shared.asr_available,
+                "llm_available": self.shared.llm_available,
+                "keyboard_interrupt_available": self.shared.keyboard_interrupt_available,
+                "vad_interrupt_available": self.shared.vad_interrupt_available,
+            },
+            "worker_health": {
+                t.name: t.is_alive()
+                for t in self._threads
+            },
+        }
+
+    def join(self) -> None:
+        for t in self._threads:
+            t.join()
+
+    # ── Internal ──
+
+    def _spawn(self, name: str, target) -> None:
+        t = threading.Thread(target=target, name=f"voice-{name}", daemon=True)
+        t.start()
+        self._threads.append(t)
+
+    def _start_mic(self) -> None:
+        cfg = self.cfg.audio
+
+        def _callback(indata, frames, time_info, status):
+            frame = indata[:, 0].copy()
+            self.ring.append(frame)
+            self.broadcaster.publish(frame)
+
+        try:
+            self._stream = sd.InputStream(
+                samplerate=cfg.sample_rate,
+                channels=cfg.channels,
+                dtype=cfg.dtype,
+                blocksize=cfg.blocksize,
+                device=cfg.input_device,
+                callback=_callback,
+            )
+            self._stream.start()
+            log.event("engine", "mic_started")
+        except Exception as e:
+            log.event("engine", "mic_failed", error=str(e))
+            self.shared.voice_in_available = False
+
+    def _test_audio_in(self) -> None:
+        try:
+            sd.check_input_settings(
+                device=self.cfg.audio.input_device,
+                samplerate=self.cfg.audio.sample_rate,
+            )
+        except Exception as e:
+            log.event("engine", "audio_in_unavailable", error=str(e))
+            self.shared.voice_in_available = False
+
+    def _test_audio_out(self) -> None:
+        try:
+            sd.check_output_settings(
+                device=self.cfg.audio.output_device,
+                samplerate=self.cfg.tts.sample_rate,
+            )
+        except Exception as e:
+            log.event("engine", "audio_out_unavailable", error=str(e))
+            self.shared.voice_out_available = False

voice2/enums.py

@@ -0,0 +1,43 @@
+"""All enums for the voice engine. Single source of truth."""
+from enum import Enum, auto
+
+
+class EngineState(Enum):
+    IDLE = auto()
+    LISTENING = auto()
+    THINKING = auto()
+    SPEAKING = auto()
+    INTERRUPTING = auto()
+    FALLBACK_TEXT = auto()
+    ERROR = auto()
+    STOPPED = auto()
+
+
+class FloorOwner(Enum):
+    NONE = auto()
+    USER = auto()
+    AGENT = auto()
+
+
+class TransitionReason(Enum):
+    STARTUP = auto()
+    SPEECH_DETECTED = auto()
+    SPEECH_ENDED = auto()
+    ASR_COMPLETE = auto()
+    THINK_START = auto()
+    THINK_COMPLETE = auto()
+    PLAYBACK_START = auto()
+    PLAYBACK_COMPLETE = auto()
+    INTERRUPT = auto()
+    INTERRUPT_RESOLVED = auto()
+    ERROR_RECOVERY = auto()
+    CAPABILITY_FAILURE = auto()
+    SHUTDOWN = auto()
+    FALLBACK = auto()
+
+
+class InterruptSource(Enum):
+    KEYBOARD = auto()
+    VAD = auto()
+    PROGRAMMATIC = auto()
+    PARTIAL_ASR = auto()  # hook for future use

voice2/floor_manager.py

@@ -0,0 +1,68 @@
+"""FloorManager — policy object for turn-taking arbitration."""
+import threading
+from .enums import FloorOwner, EngineState, TransitionReason, InterruptSource
+from .shared_state import SharedState
+from .state_controller import StateController
+from . import logging_util as log
+
+
+class FloorManager:
+    def __init__(self, shared: SharedState, ctrl: StateController) -> None:
+        self._shared = shared
+        self._ctrl = ctrl
+        self._lock = threading.RLock()  # reentrant — request_agent_floor calls can_agent_speak
+
+    def can_agent_speak(self, turn_id: int = 0) -> bool:
+        """Agent may speak only if floor is NONE or AGENT and state allows.
+        If turn_id provided, also verifies the turn is still current."""
+        with self._lock:
+            state = self._ctrl.get_state()
+            floor = self._ctrl.get_floor_owner()
+            if state in (EngineState.STOPPED, EngineState.ERROR):
+                return False
+            if floor == FloorOwner.USER:
+                return False
+            if turn_id and turn_id != self._shared.current_turn_id:
+                return False
+            return True
+
+    def request_user_floor(self, reason: str = "", turn_id: int = 0) -> bool:
+        with self._lock:
+            self._ctrl.set_floor(FloorOwner.USER, reason=reason, turn_id=turn_id)
+            return True
+
+    def request_agent_floor(self, reason: str = "", turn_id: int = 0) -> bool:
+        with self._lock:
+            if not self.can_agent_speak():
+                log.event("floor", "agent_floor_denied",
+                          state=self._ctrl.get_state().name,
+                          floor_owner=self._ctrl.get_floor_owner().name,
+                          turn_id=turn_id, reason=reason)
+                return False
+            self._ctrl.set_floor(FloorOwner.AGENT, reason=reason, turn_id=turn_id)
+            return True
+
+    def release_floor(self, reason: str = "", turn_id: int = 0) -> None:
+        with self._lock:
+            self._ctrl.set_floor(FloorOwner.NONE, reason=reason, turn_id=turn_id)
+
+    def should_commit_user_audio(self) -> bool:
+        """Is ASR result from user currently valid to act on?"""
+        floor = self._ctrl.get_floor_owner()
+        state = self._ctrl.get_state()
+        if state in (EngineState.STOPPED, EngineState.ERROR):
+            return False
+        # User speech is always commitable if user has or can take floor
+        return floor in (FloorOwner.USER, FloorOwner.NONE)
+
+    def handle_interrupt(self, source: InterruptSource, reason: str = "") -> None:
+        """User is interrupting. Floor goes back to USER immediately."""
+        with self._lock:
+            self._shared.speaking.clear()
+            self._ctrl.set_floor(FloorOwner.USER,
+                                  reason=f"interrupt:{source.name}:{reason}")
+            log.event("floor", "interrupt_floor_taken",
+                      state=self._ctrl.get_state().name,
+                      floor_owner=FloorOwner.USER.name,
+                      turn_id=self._shared.current_turn_id,
+                      source=source.name)

voice2/interrupt_controller.py

@@ -0,0 +1,83 @@
+"""InterruptController — central interrupt trigger with debounce."""
+import threading
+import time
+from .enums import InterruptSource, EngineState, TransitionReason
+from .shared_state import SharedState
+from .state_controller import StateController
+from .floor_manager import FloorManager
+from .logging_util import LatencyTrace
+from . import logging_util as log
+
+
+class InterruptController:
+    def __init__(
+        self,
+        shared: SharedState,
+        ctrl: StateController,
+        floor: FloorManager,
+        debounce_ms: int = 200,
+    ) -> None:
+        self._shared = shared
+        self._ctrl = ctrl
+        self._floor = floor
+        self._debounce_sec = debounce_ms / 1000.0
+        self._lock = threading.Lock()
+        self._last_trigger_ts: float = 0.0
+        self._last_source: InterruptSource | None = None
+        self._trace: LatencyTrace | None = None
+
+    def set_trace(self, trace: LatencyTrace) -> None:
+        self._trace = trace
+
+    def trigger(self, source: InterruptSource, reason: str = "", **meta) -> bool:
+        with self._lock:
+            now = time.monotonic()
+            # Debounce — first trigger wins within the window
+            since = now - self._last_trigger_ts
+            if since < self._debounce_sec:
+                log.event("interrupt", "debounced",
+                          state=self._ctrl.get_state().name,
+                          floor_owner=self._ctrl.get_floor_owner().name,
+                          turn_id=self._shared.current_turn_id,
+                          source=source.name, since_ms=int(since * 1000))
+                return False
+
+            self._last_trigger_ts = now
+            self._last_source = source
+
+        # Mark timing
+        if self._trace:
+            self._trace.mark("interrupt_detected")
+
+        log.event("interrupt", "triggered",
+                  state=self._ctrl.get_state().name,
+                  floor_owner=self._ctrl.get_floor_owner().name,
+                  turn_id=self._shared.current_turn_id,
+                  source=source.name, reason=reason, **meta)
+
+        # 1. Set interrupted event — playback loop polls this
+        self._shared.interrupted.set()
+
+        # 2. Floor immediately back to user
+        self._floor.handle_interrupt(source=source, reason=reason)
+
+        # 3. State → INTERRUPTING
+        self._ctrl.transition(
+            EngineState.INTERRUPTING,
+            TransitionReason.INTERRUPT,
+            source=source.name,
+            reason_text=reason,
+        )
+
+        return True
+
+    def clear(self) -> None:
+        """Called after interrupt is fully resolved and we're back to listening."""
+        self._shared.interrupted.clear()
+        log.event("interrupt", "cleared",
+                  state=self._ctrl.get_state().name,
+                  floor_owner=self._ctrl.get_floor_owner().name,
+                  turn_id=self._shared.current_turn_id)
+
+    def is_interrupted(self) -> bool:
+        return self._shared.interrupted.is_set()

voice2/invariants.py

@@ -0,0 +1,81 @@
+"""
+System invariants — called periodically or on every transition.
+Violations are logged critical and trigger forced repair.
+
+INVARIANT 1: SPEAKING => FloorOwner == AGENT
+INVARIANT 2: interrupted.is_set() => no new TTS begins
+INVARIANT 3: Only ListenWorker advances committed turn_id
+INVARIANT 4: Response plays only if turn_id matches + floor + no interrupt + not shutdown
+INVARIANT 5: No direct state mutation outside StateController
+INVARIANT 6: No subscriber failure may stall mic capture
+"""
+
+import threading
+from .enums import EngineState, FloorOwner
+from .shared_state import SharedState
+from .state_controller import StateController
+from . import logging_util as log
+
+
+class InvariantChecker:
+    def __init__(self, shared: SharedState, ctrl: StateController) -> None:
+        self._shared = shared
+        self._ctrl = ctrl
+        self._lock = threading.Lock()
+
+    def check_all(self) -> list[str]:
+        """Run all invariant checks. Returns list of violation descriptions."""
+        violations = []
+        violations += self._check_speaking_floor()
+        violations += self._check_interrupt_blocks_playback()
+        for v in violations:
+            log.event("invariant", "VIOLATION",
+                      state=self._ctrl.get_state().name,
+                      floor_owner=self._ctrl.get_floor_owner().name,
+                      turn_id=self._shared.current_turn_id,
+                      violation=v)
+        return violations
+
+    def _check_speaking_floor(self) -> list[str]:
+        state = self._ctrl.get_state()
+        floor = self._ctrl.get_floor_owner()
+        if state == EngineState.SPEAKING and floor != FloorOwner.AGENT:
+            msg = f"INV1: SPEAKING but floor={floor.name}, expected AGENT"
+            # Force repair
+            self._ctrl.set_floor(FloorOwner.AGENT, reason="invariant_repair")
+            return [msg]
+        return []
+
+    def _check_interrupt_blocks_playback(self) -> list[str]:
+        # Invariant 2 is enforced structurally in PlaybackWorker._speak()
+        # Here we just audit and log if something slipped through
+        if (self._shared.interrupted.is_set()
+                and self._ctrl.get_state() == EngineState.SPEAKING):
+            return ["INV2: speaking while interrupted flag is set"]
+        return []
+
+    def playback_is_allowed(self, turn_id: int) -> tuple[bool, str]:
+        """
+        Invariant 4: A response may only play if:
+          - turn_id == current_turn_id
+          - floor manager grants agent floor
+          - interrupt flag is clear
+          - not shutting down
+        Returns (allowed, reason).
+        """
+        if self._shared.shutdown.is_set():
+            return False, "shutdown"
+        if self._shared.interrupted.is_set():
+            return False, "interrupted"
+        if turn_id != self._shared.current_turn_id:
+            return False, f"stale_turn:{turn_id}!={self._shared.current_turn_id}"
+        if self._ctrl.get_floor_owner() == FloorOwner.USER:
+            return False, "user_owns_floor"
+        return True, "ok"
+
+    def run_loop(self, interval_sec: float = 2.0) -> None:
+        """Background invariant monitor. Run in daemon thread."""
+        import time
+        while not self._shared.shutdown.is_set():
+            time.sleep(interval_sec)
+            self.check_all()

voice2/logging_util.py

@@ -0,0 +1,75 @@
+"""Structured JSON-line event logger for the voice engine."""
+import json
+import threading
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+
+_log_lock = threading.Lock()
+_log_file: str | None = None
+
+
+def init(path: str | None) -> None:
+    global _log_file
+    _log_file = path
+
+
+def event(subsystem: str, ev: str, **meta: Any) -> None:
+    """All metadata is keyword-only. No positional args beyond subsystem + ev."""
+    turn_id = meta.get("turn_id", 0)
+    state = meta.get("state", "")
+    record = {
+        "ts": time.monotonic(),
+        "wall_ts": datetime.now(timezone.utc).isoformat(),
+        "subsystem": subsystem,
+        "event": ev,
+        **meta,
+    }
+    line = json.dumps(record, default=str)
+    with _log_lock:
+        print(f"[{subsystem}] {ev}" + (f" | turn={turn_id}" if turn_id else "")
+              + (f" | state={state}" if state else ""))
+        if _log_file:
+            try:
+                with open(_log_file, "a") as f:
+                    f.write(line + "\n")
+            except Exception:
+                pass
+
+
+class LatencyTrace:
+    """Per-turn monotonic timing. Thread-safe reads, single-writer assumed."""
+
+    FIELDS = [
+        "turn_start", "speech_detect_start", "speech_detect_end",
+        "asr_start", "asr_end", "think_start", "first_token", "think_end",
+        "tts_start", "tts_first_sample", "playback_start",
+        "interrupt_detected", "interrupt_executed",
+        "playback_end", "turn_end",
+    ]
+
+    def __init__(self, turn_id: int) -> None:
+        self.turn_id = turn_id
+        self._marks: dict[str, float] = {}
+
+    def mark(self, name: str) -> None:
+        self._marks[name] = time.monotonic()
+
+    def duration(self, a: str, b: str) -> float | None:
+        if a in self._marks and b in self._marks:
+            return round(self._marks[b] - self._marks[a], 4)
+        return None
+
+    def to_dict(self) -> dict:
+        out: dict[str, Any] = {"turn_id": self.turn_id, "marks": {}}
+        for k, v in self._marks.items():
+            out["marks"][k] = round(v, 4)
+        # Key durations
+        out["durations"] = {
+            "asr_ms": int((self.duration("asr_start", "asr_end") or 0) * 1000),
+            "think_ms": int((self.duration("think_start", "think_end") or 0) * 1000),
+            "interrupt_stop_ms": int((self.duration("interrupt_detected", "interrupt_executed") or 0) * 1000),
+            "total_turn_ms": int((self.duration("turn_start", "turn_end") or 0) * 1000),
+        }
+        return out

voice2/main.py

@@ -0,0 +1,71 @@
+"""
+Minimal bootstrap — run the engine with an echo backend, no LLM required.
+
+Usage:
+    python -m voice2.main
+
+Wire your own model by replacing `backend` with any callable(text) -> str.
+See examples/ for an HTTP LLM backend.
+"""
+import os
+import signal
+
+from voice2 import VoiceEngine, VoiceConfig
+from voice2.config import VADConfig, InterruptVADConfig, InterruptConfig
+
+
+def backend(text: str) -> str:
+    """Echo backend — proves the full loop (mic -> ASR -> reply -> TTS)."""
+    return f"You said: {text}"
+
+
+def main() -> None:
+    cfg = VoiceConfig(
+        vad=VADConfig(end_silence_ms=5000, max_utterance_sec=90.0),
+        interrupt_vad=InterruptVADConfig(
+            consecutive_frames_required=4,
+            energy_multiplier=2.0,
+        ),
+        interrupt=InterruptConfig(debounce_ms=200, keyboard_key=" "),
+        log_file=os.path.expanduser("~/voice_engine.jsonl"),
+    )
+
+    engine = VoiceEngine(cfg, backend)
+
+    print("=" * 60)
+    print("  VOICE ENGINE v2 — Full-Duplex Interruptible")
+    print("  Space = interrupt | Ctrl+C = quit")
+    print("=" * 60)
+
+    print("[boot] Loading models...")
+    engine.load_models()
+
+    print("[boot] Starting engine...")
+    engine.start()
+
+    print(f"[boot] Online. Status: {engine.status()}")
+    print("[boot] Listening. Talk naturally.\n")
+
+    # Text fallback if voice_in failed
+    if not engine.shared.voice_in_available:
+        print("[fallback] Mic unavailable — TEXT MODE. Type messages, Enter to send.")
+        while not engine.shared.shutdown.is_set():
+            try:
+                text = input("You: ").strip()
+                if text:
+                    engine.submit_text(text)
+            except (EOFError, KeyboardInterrupt):
+                break
+    else:
+        try:
+            signal.pause()
+        except KeyboardInterrupt:
+            pass
+
+    print("\n[shutdown] Stopping engine...")
+    engine.stop()
+    print("[shutdown] Done.")
+
+
+if __name__ == "__main__":
+    main()

voice2/ring_buffer.py

@@ -0,0 +1,44 @@
+"""RingAudioBuffer — thread-safe rolling mic history for pre-roll."""
+import threading
+from collections import deque
+import numpy as np
+
+
+class RingAudioBuffer:
+    def __init__(self, sample_rate: int, channels: int, seconds: float,
+                 dtype: str = "float32") -> None:
+        self._sr = sample_rate
+        self._ch = channels
+        self._dtype = dtype
+        self._capacity_samples = int(sample_rate * seconds)
+        self._buf: deque[np.ndarray] = deque()
+        self._size = 0  # samples stored
+        self._lock = threading.Lock()
+
+    def append(self, frame: np.ndarray) -> None:
+        with self._lock:
+            self._buf.append(frame)
+            self._size += len(frame)
+            # Trim oldest frames to stay within capacity
+            while self._size > self._capacity_samples and self._buf:
+                oldest = self._buf.popleft()
+                self._size -= len(oldest)
+
+    def get_last(self, seconds: float) -> np.ndarray:
+        """Return up to `seconds` of most recent audio."""
+        n = int(self._sr * seconds)
+        with self._lock:
+            if not self._buf:
+                return np.zeros(0, dtype=self._dtype)
+            frames = list(self._buf)
+        combined = np.concatenate(frames)
+        return combined[-n:] if len(combined) > n else combined
+
+    def clear(self) -> None:
+        with self._lock:
+            self._buf.clear()
+            self._size = 0
+
+    def size_seconds(self) -> float:
+        with self._lock:
+            return self._size / self._sr

voice2/shared_state.py

@@ -0,0 +1,27 @@
+"""SharedState — synchronized primitives only. No business logic."""
+import threading
+from dataclasses import dataclass, field
+
+
+@dataclass
+class SharedState:
+    # Lifecycle
+    shutdown: threading.Event = field(default_factory=threading.Event)
+
+    # Floor signals
+    interrupted: threading.Event = field(default_factory=threading.Event)
+    speaking: threading.Event = field(default_factory=threading.Event)
+    thinking: threading.Event = field(default_factory=threading.Event)
+    listening: threading.Event = field(default_factory=threading.Event)
+    mic_gated: threading.Event = field(default_factory=threading.Event)
+
+    # Capability flags — set once at startup, read-only after that
+    voice_in_available: bool = True
+    voice_out_available: bool = True
+    asr_available: bool = True
+    llm_available: bool = True
+    keyboard_interrupt_available: bool = True
+    vad_interrupt_available: bool = True
+
+    # Turn tracking — only StateController writes this
+    current_turn_id: int = 0

voice2/state_controller.py

@@ -0,0 +1,186 @@
+"""StateController — owns all state transitions. Single lock. Atomic only."""
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+from .enums import EngineState, FloorOwner, TransitionReason
+from .shared_state import SharedState
+from . import logging_util as log
+
+
+@dataclass(frozen=True)
+class StateSnapshot:
+    """Immutable atomic snapshot. Callers compare against this, never hold lock."""
+    state: EngineState
+    floor_owner: FloorOwner
+    turn_id: int
+    ts: float
+
+# Transition validation table: (from_state, to_state) -> allowed
+_VALID = {
+    (EngineState.IDLE, EngineState.LISTENING),
+    (EngineState.IDLE, EngineState.THINKING),   # submit_text() path: no LISTENING step
+    (EngineState.IDLE, EngineState.FALLBACK_TEXT),
+    (EngineState.IDLE, EngineState.ERROR),
+    (EngineState.IDLE, EngineState.STOPPED),
+    (EngineState.LISTENING, EngineState.THINKING),
+    (EngineState.LISTENING, EngineState.IDLE),
+    (EngineState.LISTENING, EngineState.INTERRUPTING),
+    (EngineState.LISTENING, EngineState.ERROR),
+    (EngineState.LISTENING, EngineState.STOPPED),
+    (EngineState.THINKING, EngineState.SPEAKING),
+    (EngineState.THINKING, EngineState.IDLE),
+    (EngineState.THINKING, EngineState.INTERRUPTING),
+    (EngineState.THINKING, EngineState.ERROR),
+    (EngineState.THINKING, EngineState.STOPPED),
+    (EngineState.SPEAKING, EngineState.IDLE),
+    (EngineState.SPEAKING, EngineState.LISTENING),
+    (EngineState.SPEAKING, EngineState.INTERRUPTING),
+    (EngineState.SPEAKING, EngineState.ERROR),
+    (EngineState.SPEAKING, EngineState.STOPPED),
+    (EngineState.INTERRUPTING, EngineState.LISTENING),
+    (EngineState.INTERRUPTING, EngineState.IDLE),
+    (EngineState.INTERRUPTING, EngineState.ERROR),
+    (EngineState.INTERRUPTING, EngineState.STOPPED),
+    (EngineState.ERROR, EngineState.IDLE),
+    (EngineState.ERROR, EngineState.STOPPED),
+    (EngineState.FALLBACK_TEXT, EngineState.THINKING),
+    (EngineState.FALLBACK_TEXT, EngineState.STOPPED),
+}
+
+# Floor ownership rules: cannot enter SPEAKING if floor is USER
+_FLOOR_BLOCKS: dict[EngineState, set[FloorOwner]] = {
+    EngineState.SPEAKING: {FloorOwner.USER},
+}
+
+
+@dataclass
+class TransitionRecord:
+    ts: float
+    from_state: EngineState
+    to_state: EngineState
+    reason: TransitionReason
+    floor_owner: FloorOwner
+    turn_id: int
+    meta: dict = field(default_factory=dict)
+
+
+class StateController:
+    def __init__(self, shared: SharedState) -> None:
+        self._shared = shared
+        self._lock = threading.RLock()
+        self._state = EngineState.IDLE
+        self._floor = FloorOwner.NONE
+        self._history: list[TransitionRecord] = []
+
+    # ── Reads ──
+
+    def get_state(self) -> EngineState:
+        with self._lock:
+            return self._state
+
+    def get_floor_owner(self) -> FloorOwner:
+        with self._lock:
+            return self._floor
+
+    def get_snapshot(self) -> StateSnapshot:
+        """Atomic frozen snapshot — callers may read freely without holding lock."""
+        with self._lock:
+            return StateSnapshot(
+                state=self._state,
+                floor_owner=self._floor,
+                turn_id=self._shared.current_turn_id,
+                ts=time.monotonic(),
+            )
+
+    def snapshot(self) -> dict[str, Any]:
+        s = self.get_snapshot()
+        return {
+            "state": s.state.name,
+            "floor_owner": s.floor_owner.name,
+            "turn_id": s.turn_id,
+            "history_len": len(self._history),
+        }
+
+    # ── Turn counter ──
+
+    def start_new_turn(self) -> int:
+        with self._lock:
+            self._shared.current_turn_id += 1
+            log.event("state", "new_turn",
+                      state=self._state.name, floor_owner=self._floor.name,
+                      turn_id=self._shared.current_turn_id)
+            return self._shared.current_turn_id
+
+    def current_turn(self) -> int:
+        return self._shared.current_turn_id
+
+    # ── State transition ──
+
+    def transition(
+        self,
+        new_state: EngineState,
+        reason: TransitionReason,
+        **meta: Any,
+    ) -> bool:
+        with self._lock:
+            if self._state == EngineState.STOPPED and new_state != EngineState.STOPPED:
+                log.event("state", "transition_rejected",
+                          from_state=self._state.name, to_state=new_state.name,
+                          floor_owner=self._floor.name,
+                          turn_id=self._shared.current_turn_id,
+                          reason="STOPPED_no_exit")
+                return False
+
+            pair = (self._state, new_state)
+            if pair not in _VALID:
+                log.event("state", "transition_rejected",
+                          from_state=self._state.name, to_state=new_state.name,
+                          floor_owner=self._floor.name,
+                          turn_id=self._shared.current_turn_id,
+                          reason="invalid_transition")
+                return False
+
+            blocked = _FLOOR_BLOCKS.get(new_state, set())
+            if self._floor in blocked:
+                log.event("state", "transition_rejected",
+                          from_state=self._state.name, to_state=new_state.name,
+                          floor_owner=self._floor.name,
+                          turn_id=self._shared.current_turn_id,
+                          reason="floor_blocked")
+                return False
+
+            rec = TransitionRecord(
+                ts=time.monotonic(),
+                from_state=self._state,
+                to_state=new_state,
+                reason=reason,
+                floor_owner=self._floor,
+                turn_id=self._shared.current_turn_id,
+                meta=meta,
+            )
+            self._history.append(rec)
+            self._state = new_state
+
+            _RESERVED = {"turn_id", "state", "floor_owner", "reason",
+                         "from_state", "to_state", "old_state", "new_state"}
+            safe_meta = {k: v for k, v in meta.items() if k not in _RESERVED}
+            log.event("state", "transition",
+                      from_state=self._history[-1].from_state.name,
+                      state=new_state.name, floor_owner=self._floor.name,
+                      turn_id=self._shared.current_turn_id,
+                      reason=reason.name, **safe_meta)
+            return True
+
+    # ── Floor control ──
+
+    def set_floor(self, owner: FloorOwner, reason: str = "", **meta: Any) -> None:
+        with self._lock:
+            _RESERVED = {"turn_id", "state", "floor_owner", "reason"}
+            safe_meta = {k: v for k, v in meta.items() if k not in _RESERVED}
+            self._floor = owner
+            log.event("floor", "floor_set",
+                      state=self._state.name, floor_owner=owner.name,
+                      turn_id=self._shared.current_turn_id,
+                      reason=reason, **safe_meta)

voice2/tests/test_interrupt_controller.py

@@ -0,0 +1,53 @@
+"""Unit tests — InterruptController debounce and trigger."""
+import time
+from ..shared_state import SharedState
+from ..state_controller import StateController
+from ..floor_manager import FloorManager
+from ..interrupt_controller import InterruptController
+from ..enums import InterruptSource, EngineState, TransitionReason
+
+
+def make_stack():
+    shared = SharedState()
+    ctrl = StateController(shared)
+    floor = FloorManager(shared, ctrl)
+    interrupt = InterruptController(shared, ctrl, floor, debounce_ms=200)
+    return shared, ctrl, floor, interrupt
+
+
+def test_trigger_sets_event():
+    shared, ctrl, floor, interrupt = make_stack()
+    ctrl.transition(EngineState.LISTENING, TransitionReason.SPEECH_DETECTED)
+    ctrl.transition(EngineState.THINKING, TransitionReason.ASR_COMPLETE)
+    # Need SPEAKING state for transition to INTERRUPTING
+    # go through full path
+    result = interrupt.trigger(InterruptSource.KEYBOARD, "test")
+    assert shared.interrupted.is_set()
+
+
+def test_debounce_blocks_second():
+    shared, ctrl, floor, interrupt = make_stack()
+    ctrl.transition(EngineState.LISTENING, TransitionReason.SPEECH_DETECTED)
+    interrupt.trigger(InterruptSource.KEYBOARD, "first")
+    result = interrupt.trigger(InterruptSource.KEYBOARD, "second")
+    assert not result  # debounced
+
+
+def test_debounce_allows_after_window():
+    shared, ctrl, floor, interrupt = make_stack()
+    interrupt._debounce_sec = 0.05
+    ctrl.transition(EngineState.LISTENING, TransitionReason.SPEECH_DETECTED)
+    interrupt.trigger(InterruptSource.KEYBOARD, "first")
+    time.sleep(0.1)
+    shared.interrupted.clear()  # simulate resolution
+    # Now trigger again — should work
+    interrupt._last_trigger_ts = 0.0
+    result = interrupt.trigger(InterruptSource.KEYBOARD, "second")
+    assert result
+
+
+def test_clear():
+    shared, ctrl, floor, interrupt = make_stack()
+    shared.interrupted.set()
+    interrupt.clear()
+    assert not shared.interrupted.is_set()

voice2/tests/test_ring_buffer.py

@@ -0,0 +1,36 @@
+"""Unit tests — RingAudioBuffer."""
+import numpy as np
+from ..ring_buffer import RingAudioBuffer
+
+
+def test_append_and_get():
+    buf = RingAudioBuffer(sample_rate=16000, channels=1, seconds=2.0)
+    frame = np.ones(1600, dtype="float32")
+    buf.append(frame)
+    out = buf.get_last(0.1)
+    assert len(out) == 1600
+
+
+def test_capacity_trimming():
+    buf = RingAudioBuffer(sample_rate=16000, channels=1, seconds=1.0)
+    for _ in range(20):
+        buf.append(np.ones(1600, dtype="float32"))
+    # Should not exceed capacity
+    assert buf.size_seconds() <= 1.1
+
+
+def test_preroll_extraction():
+    buf = RingAudioBuffer(sample_rate=16000, channels=1, seconds=2.0)
+    buf.append(np.zeros(8000, dtype="float32"))
+    buf.append(np.ones(8000, dtype="float32"))
+    out = buf.get_last(0.5)
+    assert len(out) == 8000
+    # Last 0.5s should be the ones
+    assert np.all(out == 1.0)
+
+
+def test_clear():
+    buf = RingAudioBuffer(sample_rate=16000, channels=1, seconds=2.0)
+    buf.append(np.ones(1600, dtype="float32"))
+    buf.clear()
+    assert buf.size_seconds() == 0.0

voice2/tests/test_state_controller.py

@@ -0,0 +1,58 @@
+"""Unit tests — StateController."""
+import pytest
+from ..shared_state import SharedState
+from ..state_controller import StateController
+from ..enums import EngineState, FloorOwner, TransitionReason
+
+
+def make_ctrl():
+    return StateController(SharedState())
+
+
+def test_valid_transition():
+    ctrl = make_ctrl()
+    assert ctrl.transition(EngineState.LISTENING, TransitionReason.SPEECH_DETECTED)
+    assert ctrl.get_state() == EngineState.LISTENING
+
+
+def test_invalid_transition_rejected():
+    ctrl = make_ctrl()
+    # Can't go IDLE -> SPEAKING directly
+    result = ctrl.transition(EngineState.SPEAKING, TransitionReason.PLAYBACK_START)
+    assert not result
+    assert ctrl.get_state() == EngineState.IDLE
+
+
+def test_stopped_blocks_all():
+    ctrl = make_ctrl()
+    ctrl.transition(EngineState.LISTENING, TransitionReason.SPEECH_DETECTED)
+    ctrl.transition(EngineState.STOPPED, TransitionReason.SHUTDOWN)
+    result = ctrl.transition(EngineState.IDLE, TransitionReason.STARTUP)
+    assert not result
+
+
+def test_floor_blocks_agent_speaking():
+    shared = SharedState()
+    ctrl = StateController(shared)
+    # Set floor to USER then try to go SPEAKING
+    ctrl.set_floor(FloorOwner.USER, "test")
+    ctrl.transition(EngineState.LISTENING, TransitionReason.SPEECH_DETECTED)
+    result = ctrl.transition(EngineState.THINKING, TransitionReason.ASR_COMPLETE)
+    # THINKING is allowed regardless of floor
+    assert result
+    result = ctrl.transition(EngineState.SPEAKING, TransitionReason.PLAYBACK_START)
+    assert not result  # USER has floor — blocked
+
+
+def test_turn_counter():
+    ctrl = make_ctrl()
+    t1 = ctrl.start_new_turn()
+    t2 = ctrl.start_new_turn()
+    assert t2 == t1 + 1
+
+
+def test_snapshot():
+    ctrl = make_ctrl()
+    snap = ctrl.snapshot()
+    assert snap["state"] == "IDLE"
+    assert snap["floor_owner"] == "NONE"

voice2/tones.py

@@ -0,0 +1,148 @@
+"""Soft UI tones — dedicated persistent OutputStream, no sd.play() calls."""
+import queue
+import threading
+import numpy as np
+import sounddevice as sd
+
+
+def _tone(
+    freq: float,
+    duration_ms: int,
+    volume: float,
+    sample_rate: int,
+    fade_ms: int = 30,
+) -> np.ndarray:
+    """Sine wave with fade-in and fade-out to avoid clicks."""
+    n = int(sample_rate * duration_ms / 1000)
+    t = np.linspace(0, duration_ms / 1000, n, endpoint=False)
+    wave = np.sin(2 * np.pi * freq * t).astype(np.float32) * volume
+
+    fade = int(sample_rate * fade_ms / 1000)
+    fade = min(fade, n // 2)
+    ramp = np.linspace(0, 1, fade)
+    wave[:fade] *= ramp
+    wave[-fade:] *= ramp[::-1]
+    return wave
+
+
+def _chord(freqs, duration_ms, volume, sample_rate, fade_ms=30) -> np.ndarray:
+    wave = sum(_tone(f, duration_ms, volume / len(freqs), sample_rate, fade_ms) for f in freqs)
+    return wave.astype(np.float32)
+
+
+def _sequence(tones, gap_ms, sample_rate) -> np.ndarray:
+    gap = np.zeros(int(sample_rate * gap_ms / 1000), dtype=np.float32)
+    parts = []
+    for i, t in enumerate(tones):
+        parts.append(t)
+        if i < len(tones) - 1:
+            parts.append(gap)
+    return np.concatenate(parts)
+
+
+class UICues:
+    """
+    Play soft tones for voice engine state transitions.
+
+    Uses a dedicated persistent sd.OutputStream so tone playback never
+    conflicts with the TTS OutputStream in PlaybackWorker.  All public
+    methods are non-blocking: they enqueue the pre-rendered numpy array
+    and return immediately.
+    """
+
+    def __init__(self, sample_rate: int = 22050, device=None, volume: float = 0.18):
+        self._sr = sample_rate
+        self._dev = device
+        self._vol = volume
+        self._q: queue.Queue = queue.Queue(maxsize=16)
+        self._stream: sd.OutputStream | None = None
+        self._thread: threading.Thread | None = None
+        self._open()
+
+    # ── Lifecycle ──
+
+    def _open(self) -> None:
+        """Open dedicated cue stream and start drain thread."""
+        try:
+            self._stream = sd.OutputStream(
+                samplerate=self._sr,
+                channels=1,
+                dtype="float32",
+                device=self._dev,
+                blocksize=512,
+            )
+            self._stream.start()
+        except Exception as e:
+            self._stream = None
+            return  # cues silently disabled
+
+        self._thread = threading.Thread(target=self._drain, daemon=True, name="voice-cues")
+        self._thread.start()
+
+    def _drain(self) -> None:
+        """Pull tone arrays from queue and write to the open stream."""
+        while True:
+            try:
+                samples = self._q.get(timeout=0.5)
+            except queue.Empty:
+                if self._stream is None:
+                    break
+                continue
+            if samples is None:  # sentinel
+                break
+            if self._stream is None:
+                continue
+            try:
+                self._stream.write(samples)
+            except Exception:
+                pass
+
+    def close(self) -> None:
+        """Graceful shutdown — call from engine.stop()."""
+        self._q.put(None)  # sentinel to drain thread (blocking — guaranteed delivery)
+        if self._stream:
+            try:
+                self._stream.stop()
+                self._stream.close()
+            except Exception:
+                pass
+            self._stream = None
+
+    def _play(self, samples: np.ndarray) -> None:
+        """Non-blocking enqueue. Drops silently if queue full."""
+        if self._stream is None:
+            return
+        try:
+            self._q.put_nowait(samples)
+        except queue.Full:
+            pass
+
+    # ── Cues ──
+
+    def listening(self) -> None:
+        """Soft rising two-note: ready to hear you."""
+        tones = [
+            _tone(660, 80, self._vol, self._sr),
+            _tone(880, 100, self._vol, self._sr),
+        ]
+        self._play(_sequence(tones, 20, self._sr))
+
+    def thinking(self) -> None:
+        """Single soft mid-note ping: processing."""
+        self._play(_tone(528, 90, self._vol * 0.7, self._sr))
+
+    def speaking(self) -> None:
+        """Gentle chord: about to speak."""
+        self._play(_chord([528, 660], 100, self._vol * 0.8, self._sr))
+
+    def interrupted(self) -> None:
+        """Soft descending note: cut off."""
+        tones = [
+            _tone(660, 70, self._vol * 0.7, self._sr),
+            _tone(440, 90, self._vol * 0.6, self._sr),
+        ]
+        self._play(_sequence(tones, 15, self._sr))
+
+    def error(self) -> None:
+        """Low soft thud: something wrong."""
+        self._play(_tone(220, 150, self._vol * 0.5, self._sr))

voice2/turn_context.py

@@ -0,0 +1,54 @@
+"""Turn-level types. Per-turn state is separate from global engine state."""
+import time
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Iterable
+
+import numpy as np
+
+from .enums import InterruptSource
+
+if TYPE_CHECKING:
+    from .logging_util import LatencyTrace
+
+
+@dataclass
+class TurnContext:
+    turn_id: int
+    created_ts: float = field(default_factory=time.monotonic)
+    transcript: str | None = None
+    audio: np.ndarray | None = None
+    generation_id: str | None = None
+    cancelled: bool = False
+    stale: bool = False
+    latency: "LatencyTrace | None" = None
+
+    def mark_stale(self) -> None:
+        self.stale = True
+
+    def mark_cancelled(self) -> None:
+        self.cancelled = True
+
+    def is_live(self) -> bool:
+        return not self.stale and not self.cancelled
+
+
+@dataclass
+class PlaybackItem:
+    turn_id: int
+    text: str
+    created_ts: float = field(default_factory=time.monotonic)
+    # Pre-synthesized audio iterator. If None, PlaybackWorker synthesizes from text.
+    audio_iter: "Iterable[np.ndarray] | None" = None
+
+    def is_stale(self, current_turn_id: int) -> bool:
+        return self.turn_id != current_turn_id
+
+
+@dataclass
+class InterruptRecord:
+    source: InterruptSource
+    ts: float
+    state: str
+    floor_owner: str
+    turn_id: int
+    reason: str

voice2/workers/interrupt_detector.py

@@ -0,0 +1,85 @@
+"""InterruptDetectorWorker — fast VAD/energy barge-in during SPEAKING only."""
+import queue
+import time
+import threading
+import numpy as np
+
+from ..enums import EngineState, InterruptSource
+from ..shared_state import SharedState
+from ..state_controller import StateController
+from ..interrupt_controller import InterruptController
+from ..config import InterruptVADConfig
+from .. import logging_util as log
+
+
+class InterruptDetectorWorker(threading.Thread):
+    def __init__(
+        self,
+        audio_queue: queue.Queue,
+        shared: SharedState,
+        ctrl: StateController,
+        interrupt: InterruptController,
+        cfg: InterruptVADConfig,
+    ) -> None:
+        super().__init__(name="voice-interrupt_det", daemon=True)
+        self._q = audio_queue
+        self._shared = shared
+        self._ctrl = ctrl
+        self._interrupt = interrupt
+        self._cfg = cfg
+
+    def run(self) -> None:
+        """Runs in daemon thread. Only hot during SPEAKING."""
+        baseline_rms = 0.001
+        consecutive = 0
+        last_refractory = 0.0
+
+        while not self._shared.shutdown.is_set():
+            # Only active during SPEAKING — idle otherwise
+            if self._ctrl.get_state() != EngineState.SPEAKING:
+                # Drain queue so it doesn't fill up while inactive
+                try:
+                    self._q.get(timeout=0.05)
+                except queue.Empty:
+                    pass
+                consecutive = 0
+                baseline_rms = 0.001
+                continue
+
+            try:
+                frame = self._q.get(timeout=0.05)
+            except queue.Empty:
+                continue
+
+            # Compute RMS energy
+            rms = float(np.sqrt(np.mean(frame.astype(np.float32) ** 2)))
+
+            # Slowly update baseline from speaker bleed (slower = more stable floor)
+            baseline_rms = 0.995 * baseline_rms + 0.005 * rms
+
+            # Threshold: multiplier × baseline, but NEVER below 0.06 absolute floor
+            # 0.06 blocks ambient noise and most speaker bleed; a normal speaking voice clears it
+            threshold = max(baseline_rms * self._cfg.energy_multiplier, 0.06)
+
+            if rms > threshold:
+                consecutive += 1
+            else:
+                consecutive = 0
+
+            if consecutive >= self._cfg.consecutive_frames_required:
+                now = time.monotonic()
+                refractory_sec = self._cfg.refractory_ms / 1000.0
+                if (now - last_refractory) > refractory_sec:
+                    last_refractory = now
+                    consecutive = 0
+                    log.event("interrupt_detector", "vad_triggered",
+                              state=self._ctrl.get_state().name,
+                              floor_owner=self._ctrl.get_floor_owner().name,
+                              turn_id=self._shared.current_turn_id,
+                              rms=round(rms, 4),
+                              baseline=round(baseline_rms, 4))
+                    self._interrupt.trigger(
+                        InterruptSource.VAD,
+                        reason="energy_threshold_exceeded",
+                        rms=round(rms, 4),
+                    )

voice2/workers/keyboard.py

@@ -0,0 +1,53 @@
+"""KeyboardWorker — spacebar interrupt. Isolated so failure doesn't kill engine."""
+import threading
+import sys
+import termios
+import tty
+
+from ..enums import InterruptSource
+from ..shared_state import SharedState
+from ..interrupt_controller import InterruptController
+from .. import logging_util as log
+
+
+def _read_char() -> str:
+    fd = sys.stdin.fileno()
+    old = termios.tcgetattr(fd)
+    try:
+        tty.setraw(fd)
+        return sys.stdin.read(1)
+    finally:
+        termios.tcsetattr(fd, termios.TCSADRAIN, old)
+
+
+class KeyboardWorker(threading.Thread):
+    def __init__(
+        self,
+        shared: SharedState,
+        interrupt: InterruptController,
+        key: str = " ",
+    ) -> None:
+        super().__init__(name="voice-keyboard", daemon=True)
+        self._shared = shared
+        self._interrupt = interrupt
+        self._key = key
+
+    def run(self) -> None:
+        log.event("keyboard", "listener_started", meta={"key": repr(self._key)})
+        while not self._shared.shutdown.is_set():
+            try:
+                ch = _read_char()
+                if ch == self._key:
+                    log.event("keyboard", "spacebar_pressed")
+                    self._interrupt.trigger(
+                        InterruptSource.KEYBOARD,
+                        reason="spacebar_pressed",
+                    )
+                elif ch in ("\x03", "\x04"):  # Ctrl+C or Ctrl+D
+                    self._shared.shutdown.set()
+                    break
+            except Exception as e:
+                log.event("keyboard", "listener_error", meta={"error": str(e)})
+                # Don't kill the engine over keyboard issues
+                break
+        log.event("keyboard", "listener_stopped")

voice2/workers/listen.py

@@ -0,0 +1,302 @@
+"""ListenWorker — patient VAD utterance capture with pre-roll."""
+import queue
+import re
+import threading
+import time
+import numpy as np
+
+from ..enums import EngineState, FloorOwner, TransitionReason
+from ..shared_state import SharedState
+from ..state_controller import StateController
+from ..floor_manager import FloorManager
+from ..interrupt_controller import InterruptController
+from ..ring_buffer import RingAudioBuffer
+from ..logging_util import LatencyTrace
+from ..config import VADConfig
+from .. import logging_util as log
+
+
+class ListenWorker(threading.Thread):
+    def __init__(
+        self,
+        audio_queue: queue.Queue,
+        shared: SharedState,
+        ctrl: StateController,
+        floor: FloorManager,
+        interrupt: InterruptController,
+        ring: RingAudioBuffer,
+        vad_cfg: VADConfig,
+        sample_rate: int,
+        transcript_queue: queue.Queue,
+        asr_backend,
+        cues=None,
+    ) -> None:
+        super().__init__(name="voice-listen", daemon=True)
+        self._cues = cues
+        self._q = audio_queue
+        self._shared = shared
+        self._ctrl = ctrl
+        self._floor = floor
+        self._interrupt = interrupt
+        self._ring = ring
+        self._cfg = vad_cfg
+        self._sr = sample_rate
+        self._out_q = transcript_queue
+        self._asr = asr_backend
+        self._silence_threshold = 0.02   # onset: need intentional voice, not ambient noise
+        self._collect_silence_threshold = 0.015  # during collection: slightly more lenient
+        self._vad_model = None
+        self._torch = None
+
+    def run(self) -> None:
+        try:
+            import torch
+            model, utils = torch.hub.load(
+                repo_or_dir="snakers4/silero-vad",
+                model="silero_vad",
+                force_reload=False,
+                trust_repo=True,
+            )
+            (get_speech_timestamps, _, read_audio, *_) = utils
+            self._vad_model = model
+            self._torch = torch
+        except Exception as e:
+            log.event("listen", "vad_load_error", error=str(e))
+            self._vad_model = None
+            self._torch = None
+
+        while not self._shared.shutdown.is_set():
+            try:
+                frame = self._q.get(timeout=0.1)
+            except queue.Empty:
+                continue
+
+            # Always feed ring buffer
+            self._ring.append(frame)
+
+            # Only capture if engine is in a state that allows it
+            state = self._ctrl.get_state()
+            if state not in (EngineState.IDLE, EngineState.LISTENING,
+                             EngineState.INTERRUPTING):
+                continue
+
+            # Simple energy gate to detect speech onset
+            rms = float(np.sqrt(np.mean(frame.astype(np.float32) ** 2)))
+            if rms < self._silence_threshold:
+                continue
+
+            # Speech detected — start an utterance
+            turn_id = self._ctrl.start_new_turn()
+            trace = LatencyTrace(turn_id)
+            trace.mark("turn_start")
+            trace.mark("speech_detect_start")
+
+            self._floor.request_user_floor(reason="speech_onset", turn_id=turn_id)
+            # Only transition if not already LISTENING
+            if self._ctrl.get_state() != EngineState.LISTENING:
+                self._ctrl.transition(EngineState.LISTENING, TransitionReason.SPEECH_DETECTED,
+                                      turn_id=turn_id)
+            self._interrupt.clear()
+
+            log.event("listen", "utterance_start",
+                      state=EngineState.LISTENING.name,
+                      floor_owner=FloorOwner.USER.name,
+                      turn_id=turn_id)
+            if self._cues:
+                self._cues.listening()
+
+            # Collect utterance until silence
+            utterance = self._collect_utterance(turn_id, trace)
+
+            if utterance is None or len(utterance) == 0:
+                self._ctrl.transition(EngineState.IDLE, TransitionReason.SPEECH_ENDED,
+                                      turn_id=turn_id)
+                continue
+
+            quality = self._speech_quality(utterance)
+            if not quality["ok"]:
+                log.event("listen", "utterance_discarded",
+                          turn_id=turn_id,
+                          reason=quality["reason"],
+                          duration_ms=quality["duration_ms"],
+                          voiced_ms=quality["voiced_ms"],
+                          max_prob=round(quality["max_prob"], 3),
+                          voiced_ratio=round(quality["voiced_ratio"], 3))
+                self._floor.release_floor(reason="audio_quality_gate", turn_id=turn_id)
+                self._ctrl.transition(EngineState.IDLE, TransitionReason.SPEECH_ENDED,
+                                      turn_id=turn_id)
+                continue
+
+            trace.mark("speech_detect_end")
+            trace.mark("asr_start")
+            log.event("listen", "asr_start", turn_id=turn_id,
+                      duration_sec=round(len(utterance) / self._sr, 2))
+
+            try:
+                text = self._clean_transcript(self._asr.transcribe(utterance))
+                trace.mark("asr_end")
+                log.event("listen", "asr_complete", turn_id=turn_id,
+                          text_preview=text[:80],
+                          asr_ms=int((trace.duration("asr_start", "asr_end") or 0) * 1000))
+
+                if self._is_transcript_noise(text, quality):
+                    log.event("listen", "asr_discarded", turn_id=turn_id,
+                              reason="noise_transcript",
+                              text_preview=text[:80])
+                    self._floor.release_floor(reason="noise_transcript", turn_id=turn_id)
+                    self._ctrl.transition(EngineState.IDLE, TransitionReason.SPEECH_ENDED)
+                elif text and self._floor.should_commit_user_audio():
+                    # Release floor so agent can respond
+                    self._floor.release_floor(reason="asr_committed", turn_id=turn_id)
+                    self._out_q.put_nowait((text, turn_id, trace))
+                else:
+                    log.event("listen", "asr_discarded", turn_id=turn_id,
+                              reason="empty_or_floor_denied")
+                    self._floor.release_floor(reason="asr_discarded", turn_id=turn_id)
+                    self._ctrl.transition(EngineState.IDLE, TransitionReason.SPEECH_ENDED)
+
+            except Exception as e:
+                log.event("listen", "asr_error", turn_id=turn_id, error=str(e))
+                self._floor.release_floor(reason="asr_error", turn_id=turn_id)
+                self._ctrl.transition(EngineState.IDLE, TransitionReason.SPEECH_ENDED)
+
+    def _collect_utterance(self, turn_id: int, trace: LatencyTrace) -> np.ndarray | None:
+        """Collect frames until end_silence_ms of quiet. Returns full audio."""
+        frames = []
+        # Pre-roll from ring buffer
+        pre_roll_sec = self._cfg.pre_roll_ms / 1000.0
+        pre = self._ring.get_last(pre_roll_sec)
+        if len(pre) > 0:
+            frames.append(pre)
+
+        silence_samples = int(self._sr * self._cfg.end_silence_ms / 1000.0)
+        max_samples = int(self._sr * self._cfg.max_utterance_sec)
+        silent_count = 0
+        total_samples = 0
+
+        deadline = time.monotonic() + self._cfg.max_utterance_sec + 2
+
+        while not self._shared.shutdown.is_set():
+            if time.monotonic() > deadline:
+                break
+            try:
+                frame = self._q.get(timeout=0.1)
+            except queue.Empty:
+                continue
+
+            self._ring.append(frame)
+            frames.append(frame)
+            total_samples += len(frame)
+
+            rms = float(np.sqrt(np.mean(frame.astype(np.float32) ** 2)))
+            if rms < self._collect_silence_threshold:
+                silent_count += len(frame)
+                if silent_count >= silence_samples:
+                    break
+            else:
+                silent_count = 0
+
+            if total_samples >= max_samples:
+                break
+
+        if not frames:
+            return None
+
+        audio = np.concatenate(frames)
+        min_samples = int(self._sr * self._cfg.min_utterance_ms / 1000.0)
+        if len(audio) < min_samples:
+            return None
+        return audio
+
+    def _speech_quality(self, audio: np.ndarray) -> dict:
+        duration_ms = int(len(audio) / self._sr * 1000)
+        result = {
+            "ok": True,
+            "reason": "ok",
+            "duration_ms": duration_ms,
+            "voiced_ms": 0,
+            "max_prob": 0.0,
+            "voiced_ratio": 0.0,
+        }
+        if duration_ms < self._cfg.min_utterance_ms:
+            result.update(ok=False, reason="too_short")
+            return result
+
+        if self._vad_model is None or self._torch is None:
+            # Better to allow text through the post-ASR noise gate than dead-stop
+            # voice if Silero failed to load.
+            return result
+
+        frame_len = 512
+        probs = []
+        audio_f32 = audio.astype(np.float32)
+        for start in range(0, max(len(audio_f32) - frame_len + 1, 0), frame_len):
+            frame = audio_f32[start:start + frame_len]
+            if frame.shape[0] != frame_len:
+                continue
+            if float(np.sqrt(np.mean(frame ** 2))) < 0.003:
+                probs.append(0.0)
+                continue
+            try:
+                with self._torch.no_grad():
+                    prob = float(self._vad_model(self._torch.from_numpy(frame), self._sr).item())
+            except Exception as e:
+                log.event("listen", "quality_vad_error", error=str(e))
+                return result
+            probs.append(prob)
+
+        if not probs:
+            result.update(ok=False, reason="no_frames")
+            return result
+
+        voiced = [p for p in probs if p >= self._cfg.min_voice_prob]
+        voiced_ms = int(len(voiced) * frame_len / self._sr * 1000)
+        voiced_ratio = len(voiced) / max(len(probs), 1)
+        max_prob = max(probs)
+        result.update(
+            voiced_ms=voiced_ms,
+            max_prob=max_prob,
+            voiced_ratio=voiced_ratio,
+        )
+        if max_prob < self._cfg.min_voice_prob:
+            result.update(ok=False, reason="no_speech_probability")
+        elif voiced_ms < self._cfg.min_voiced_ms:
+            result.update(ok=False, reason="too_little_voiced_audio")
+        elif voiced_ratio < self._cfg.min_voiced_ratio:
+            result.update(ok=False, reason="voiced_ratio_low")
+        return result
+
+    @staticmethod
+    def _clean_transcript(text: str) -> str:
+        return re.sub(r"\s+", " ", (text or "").strip())
+
+    @staticmethod
+    def _is_transcript_noise(text: str, quality: dict) -> bool:
+        norm = re.sub(r"[^a-z0-9' ]+", "", (text or "").lower()).strip()
+        noise = {
+            "",
+            ".",
+            "...",
+            "uh",
+            "um",
+            "umm",
+            "hmm",
+            "hm",
+            "mm",
+            "ah",
+            "oh",
+            "you",
+            "thank you",
+            "thanks",
+            "thanks for watching",
+            "bye",
+            "bye bye",
+            "subtitles by the amaraorg community",
+        }
+        if norm in noise:
+            return True
+        if quality.get("duration_ms", 0) < 1100:
+            words = [w for w in norm.split() if w]
+            if len(words) <= 1 and len(norm) <= 5:
+                return True
+        return len(norm) <= 1

voice2/workers/playback.py

@@ -0,0 +1,185 @@
+"""PlaybackWorker — chunk-by-chunk TTS playback with hard interrupt guarantee."""
+import queue
+import socket
+import threading
+import time
+import numpy as np
+import sounddevice as sd
+
+# Fan out played TTS samples to localhost UDP so companion visualizers
+# (e.g. edge_glow) can derive a sample-accurate envelope without guessing
+# through PulseAudio monitor routing. Fire-and-forget; listener optional.
+_TAP_HOST = "127.0.0.1"
+_TAP_PORT = 47121
+
+from ..enums import EngineState, TransitionReason, FloorOwner
+from ..shared_state import SharedState
+from ..state_controller import StateController
+from ..floor_manager import FloorManager
+from ..interrupt_controller import InterruptController
+from ..logging_util import LatencyTrace
+from .. import logging_util as log
+
+
+class PlaybackWorker(threading.Thread):
+    def __init__(
+        self,
+        shared: SharedState,
+        ctrl: StateController,
+        floor: FloorManager,
+        interrupt: InterruptController,
+        sample_rate: int = 22050,
+        chunk_ms: int = 20,
+        output_device: int | None = None,
+        cues=None,
+    ) -> None:
+        super().__init__(name="voice-playback", daemon=True)
+        self._cues = cues
+        self._shared = shared
+        self._ctrl = ctrl
+        self._floor = floor
+        self._interrupt = interrupt
+        self._sr = sample_rate
+        self._chunk_samples = int(sample_rate * chunk_ms / 1000)
+        self._device = output_device
+        self._queue: queue.Queue = queue.Queue(maxsize=200)
+        self._stream: sd.OutputStream | None = None
+        self._lock = threading.Lock()
+        try:
+            self._tap_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            self._tap_sock.setblocking(False)
+        except Exception:
+            self._tap_sock = None
+
+    def submit(self, text: str, turn_id: int, tts_backend, trace: LatencyTrace) -> None:
+        """Non-blocking. Puts work on internal queue."""
+        self._queue.put_nowait((text, turn_id, tts_backend, trace))
+
+    def run(self) -> None:
+        """Worker loop. Run in its own daemon thread."""
+        log.event("playback", "worker_started")
+        try:
+            while not self._shared.shutdown.is_set():
+                try:
+                    item = self._queue.get(timeout=0.1)
+                except queue.Empty:
+                    continue
+                text, turn_id, tts_backend, trace = item
+                self._speak(text, turn_id, tts_backend, trace)
+        except Exception as e:
+            log.event("playback", "worker_crashed", error=str(e))
+            raise
+
+    def _speak(self, text: str, turn_id: int, tts_backend, trace: LatencyTrace) -> None:
+        log.event("playback", "speak_entry", turn_id=turn_id,
+                  current_turn=self._ctrl.current_turn(),
+                  floor=self._ctrl.get_floor_owner().name,
+                  state=self._ctrl.get_state().name)
+        # Stale generation guard — if turn moved on, suppress
+        if turn_id != self._ctrl.current_turn():
+            log.event("playback", "stale_suppressed", turn_id=turn_id,
+                      current_turn=self._ctrl.current_turn())
+            return
+
+        if not self._floor.request_agent_floor(reason="playback_start", turn_id=turn_id):
+            log.event("playback", "floor_denied", turn_id=turn_id)
+            return
+
+        ok = self._ctrl.transition(EngineState.SPEAKING, TransitionReason.PLAYBACK_START,
+                                   turn_id=turn_id)
+        if not ok:
+            self._floor.release_floor(reason="transition_failed", turn_id=turn_id)
+            return
+
+        self._shared.speaking.set()
+        if self._cues:
+            self._cues.speaking()
+        trace.mark("tts_start")
+        trace.mark("playback_start")
+        log.event("playback", "start",
+                  state=EngineState.SPEAKING.name,
+                  floor_owner=FloorOwner.AGENT.name,
+                  turn_id=turn_id)
+
+        try:
+            stream = sd.OutputStream(
+                samplerate=self._sr,
+                channels=1,
+                dtype="float32",
+                device=self._device,
+                blocksize=self._chunk_samples,
+            )
+            stream.start()
+            first = True
+
+            for chunk in tts_backend.synthesize(text):
+                # Split into target chunk sizes for tight interrupt polling
+                for subchunk in self._split(chunk):
+                    if self._shared.shutdown.is_set():
+                        stream.abort()
+                        return
+                    if self._interrupt.is_interrupted():
+                        if self._cues:
+                            self._cues.interrupted()
+                        trace.mark("interrupt_executed")
+                        log.event("playback", "interrupt_executed",
+                                  turn_id=turn_id,
+                                  queued_chunks_discarded=self._queue.qsize())
+                        stream.abort()
+                        # Drain pending TTS for this turn
+                        self._drain(turn_id)
+                        return
+                    if first:
+                        trace.mark("tts_first_sample")
+                        first = False
+                    stream.write(subchunk)
+                    if self._tap_sock is not None:
+                        try:
+                            self._tap_sock.sendto(
+                                subchunk.tobytes(), (_TAP_HOST, _TAP_PORT)
+                            )
+                        except Exception:
+                            pass
+
+            stream.stop()
+            stream.close()
+            trace.mark("playback_end")
+            log.event("playback", "complete", turn_id=turn_id,
+                      duration_ms=int((trace.duration("playback_start", "playback_end") or 0) * 1000))
+
+        except Exception as e:
+            log.event("playback", "error", turn_id=turn_id, error=str(e))
+        finally:
+            self._shared.speaking.clear()
+            self._floor.release_floor(reason="playback_finished", turn_id=turn_id)
+            state = self._ctrl.get_state()
+            if state == EngineState.SPEAKING:
+                self._ctrl.transition(EngineState.LISTENING,
+                                      TransitionReason.PLAYBACK_COMPLETE,
+                                      turn_id=turn_id)
+
+    def _split(self, chunk: np.ndarray):
+        """Yield chunk_samples-sized sub-arrays for tight interrupt polling."""
+        for i in range(0, len(chunk), self._chunk_samples):
+            yield chunk[i:i + self._chunk_samples].astype(np.float32)
+
+    def _drain(self, turn_id: int) -> None:
+        """Discard queued TTS items for this turn only; preserve items for other turns."""
+        keep = []
+        drained = 0
+        while not self._queue.empty():
+            try:
+                item = self._queue.get_nowait()
+            except queue.Empty:
+                break
+            if len(item) >= 2 and item[1] == turn_id:
+                drained += 1
+            else:
+                keep.append(item)
+        for item in keep:
+            try:
+                self._queue.put_nowait(item)
+            except queue.Full:
+                pass  # queue filled while we were draining — item lost
+        if drained:
+            log.event("playback", "drained", turn_id=turn_id, count=drained)

voice2/workers/think.py

@@ -0,0 +1,96 @@
+"""ThinkWorker — consumes ASR transcripts, calls LLM, submits to playback."""
+import queue
+import threading
+
+from ..enums import EngineState, TransitionReason
+from ..shared_state import SharedState
+from ..state_controller import StateController
+from ..floor_manager import FloorManager
+from ..interrupt_controller import InterruptController
+from ..logging_util import LatencyTrace
+from .. import logging_util as log
+
+
+class ThinkWorker(threading.Thread):
+    def __init__(
+        self,
+        transcript_queue: queue.Queue,
+        shared: SharedState,
+        ctrl: StateController,
+        floor: FloorManager,
+        interrupt: InterruptController,
+        llm_backend,
+        playback_worker,
+        tts_backend,
+        cues=None,
+    ) -> None:
+        super().__init__(name="voice-think", daemon=True)
+        self._cues = cues
+        self._in_q = transcript_queue
+        self._shared = shared
+        self._ctrl = ctrl
+        self._floor = floor
+        self._interrupt = interrupt
+        self._llm = llm_backend
+        self._playback = playback_worker
+        self._tts = tts_backend
+
+    def run(self) -> None:
+        while not self._shared.shutdown.is_set():
+            try:
+                text, turn_id, trace = self._in_q.get(timeout=0.1)
+            except queue.Empty:
+                continue
+
+            # Stale generation guard — newer turn already started
+            if turn_id != self._ctrl.current_turn():
+                log.event("think", "stale_discarded", turn_id=turn_id,
+                          current=self._ctrl.current_turn())
+                continue
+
+            self._ctrl.transition(EngineState.THINKING, TransitionReason.THINK_START,
+                                   turn_id=turn_id)
+            self._shared.thinking.set()
+            if self._cues:
+                self._cues.thinking()
+            trace.mark("think_start")
+            log.event("think", "start",
+                      state=EngineState.THINKING.name,
+                      floor_owner=self._ctrl.get_floor_owner().name,
+                      turn_id=turn_id, input_preview=text[:80])
+
+            try:
+                reply = self._llm.reply(text)
+                trace.mark("first_token")
+                trace.mark("think_end")
+                self._shared.thinking.clear()
+
+                log.event("think", "complete", turn_id=turn_id,
+                          think_ms=int((trace.duration("think_start", "think_end") or 0) * 1000),
+                          reply_preview=reply[:80])
+
+                # Stale check again — could have been interrupted while thinking
+                if turn_id != self._ctrl.current_turn():
+                    log.event("think", "post_think_stale_suppressed", turn_id=turn_id)
+                    continue
+
+                if not self._floor.can_agent_speak(turn_id):
+                    log.event("think", "floor_blocked_post_think", turn_id=turn_id)
+                    self._ctrl.transition(EngineState.IDLE, TransitionReason.THINK_COMPLETE,
+                                          turn_id=turn_id)
+                    continue
+
+                if reply:
+                    # Hand off to PlaybackWorker — it owns the SPEAKING transition
+                    self._interrupt.set_trace(trace)
+                    self._playback.submit(reply, turn_id, self._tts, trace)
+                    log.event("think", "submitted_to_playback", turn_id=turn_id)
+                else:
+                    self._ctrl.transition(EngineState.IDLE, TransitionReason.THINK_COMPLETE,
+                                          turn_id=turn_id)
+
+            except Exception as e:
+                self._shared.thinking.clear()
+                log.event("think", "error", turn_id=turn_id, error=str(e))
+                self._ctrl.transition(EngineState.IDLE, TransitionReason.ERROR_RECOVERY,
+                                      turn_id=turn_id)