Bug fixes

README.md

@@ -145,8 +145,8 @@ ring_size    = 200                          # Recent events replayed to a newly
 control_socket        = ""                  # Default: $XDG_RUNTIME_DIR/stt-assistant.sock
 stall_timeout_ms      = 3000                # No heartbeat for this long -> STT considered offline
 worker_stall_ms       = 15000               # Heartbeats arriving but a transcription runs this long -> degraded
-command_arm_timeout_s = 10                  # Disarm a command if no turn is captured in this time
-chat_idle_timeout_s   = 300                 # Leave chat mode after this much silence
+command_arm_timeout_s = 10.0                # Disarm a command if no turn is captured in this time
+chat_idle_timeout_s   = 300.0               # Leave chat mode after this much silence
 turn_gap_s            = 1.5                 # Quiet gap that ends a turn
 require_viewer        = true                # Refuse to execute commands with no TUI attached
 chat_exit_phrases     = ["end conversation", "that's all", "stop listening"]
@@ -160,7 +160,7 @@ enabled              = true
 endpoint             = "http://localhost:9931/v1/chat/completions"
 temperature          = 0.7
 max_tokens           = 1024                 # reasoning models spend tokens before the answer; keep generous
-timeout_s            = 120
+timeout_s            = 120.0
 cache_prompt         = true                 # reuse the KV prefix (append-only history keeps it warm)
 max_prompt_tokens    = 64000                # ~half of the 131072 context
 trim_block_turns     = 4                    # trim history in blocks, not per-turn, to limit re-prefill

config.example.toml

@@ -53,8 +53,8 @@ ring_size    = 200
 control_socket        = ""
 stall_timeout_ms      = 3000
 worker_stall_ms       = 15000
-command_arm_timeout_s = 10
-chat_idle_timeout_s   = 300
+command_arm_timeout_s = 10.0
+chat_idle_timeout_s   = 300.0
 turn_gap_s            = 1.5
 require_viewer        = true
 chat_exit_phrases     = ["end conversation", "that's all", "stop listening"]
@@ -68,7 +68,7 @@ enabled               = true
 endpoint              = "http://localhost:9931/v1/chat/completions"
 temperature           = 0.7
 max_tokens            = 1024
-timeout_s             = 120
+timeout_s             = 120.0
 cache_prompt          = true
 max_prompt_tokens     = 64000
 trim_block_turns      = 4

src/stt_daemon/assistant/audit.py

@@ -36,7 +36,7 @@ class AuditLog:
         try:
             self.directory.mkdir(parents=True, exist_ok=True)
             record = {"ts": datetime.now().isoformat(), **fields}
-            with open(self.path_for_today(), "a") as f:
+            with open(self.path_for_today(), "a", encoding="utf-8") as f:
                 json.dump(record, f, ensure_ascii=False)
                 f.write("\n")
         except Exception as e:  # noqa: BLE001 - auditing is best effort

src/stt_daemon/assistant/executor.py

@@ -80,21 +80,18 @@ class LlmExecutor:
             )
 
             for call in response.tool_calls:
-                tool = self.registry.get(call.name)
                 args, error = self.registry.validate(call.name, call.arguments)
+                tool = self.registry.get(call.name)
+                if error is not None or tool is None:
+                    message = error or f"unknown tool: {call.name}"
+                    emit("error", code="invalid_tool_call", message=message)
+                    return ExecutionResult(ok=False, result=message, tool=call.name)
                 emit(
                     "tool_call",
                     tool=call.name,
                     args=args or {},
-                    requires_confirmation=bool(tool and tool.requires_confirmation),
+                    requires_confirmation=tool.requires_confirmation,
                 )
-                if error is not None:
-                    emit("error", code="invalid_tool_call", message=error)
-                    return ExecutionResult(ok=False, result=error, tool=call.name)
-                if tool is None:
-                    message = f"unknown tool: {call.name}"
-                    emit("error", code="invalid_tool_call", message=message)
-                    return ExecutionResult(ok=False, result=message, tool=call.name)
                 if tool.requires_confirmation:
                     message = f"{call.name} requires confirmation"
                     emit("error", code="confirmation_required", message=message)

src/stt_daemon/assistant/service.py

@@ -94,6 +94,9 @@ class AssistantService:
 
         self._last_emitted: tuple = ()
         self._stop = threading.Event()
+        # Guards mode transitions, the command buffer, and the chat buffers
+        # against the reader, ticker, and control-socket threads.
+        self._lock = threading.RLock()
         self._reader: threading.Thread | None = None
         self._ticker: threading.Thread | None = None
         self._exec_thread: threading.Thread | None = None
@@ -102,6 +105,8 @@ class AssistantService:
 
     def start(self) -> None:
         """Start the event reader and the timeout ticker."""
+        if self._reader is not None or self._ticker is not None:
+            return
         self._reader = threading.Thread(
             target=self._read_loop, name="assistant-events", daemon=True
         )
@@ -190,29 +195,32 @@ class AssistantService:
 
     def edit(self, text: str) -> bool:
         """Replace the command buffer text (refused unless command-armed)."""
-        if self.modes.mode is not Mode.COMMAND_ARMED:
-            return False
-        self.command_text = text
-        self.command_edited = True
+        with self._lock:
+            if self.modes.mode is not Mode.COMMAND_ARMED:
+                return False
+            self.command_text = text
+            self.command_edited = True
         self._emit_command_text(force=True)
         return True
 
     def arm(self) -> bool:
         """Enter command mode (refused unless idle)."""
-        if not self.modes.arm():
-            return False
-        self.arm_ts = self._now()
-        self.command_text = ""
-        self.command_edited = False
+        with self._lock:
+            if not self.modes.arm():
+                return False
+            self.arm_ts = self._now()
+            self.command_text = ""
+            self.command_edited = False
         self._emit_mode()
         self._emit_command_text(force=True)
         return True
 
     def cancel(self) -> bool:
         """Leave command mode, discarding the buffer."""
-        if not self.modes.cancel():
-            return False
-        self._clear_command()
+        with self._lock:
+            if not self.modes.cancel():
+                return False
+            self._clear_command()
         self._emit_mode()
         self._emit_command_text(force=True)
         return True
@@ -223,12 +231,15 @@ class AssistantService:
         When an executor is configured, execution runs on a worker thread so the
         caller (control socket) is never blocked.
         """
-        if self.modes.mode is not Mode.COMMAND_ARMED or not self.ready():
-            return None
-        text = self.command_text
-        edited = self.command_edited
-        if not self.modes.execute():
-            return None
+        # The readiness check and the mode transition must be atomic: the
+        # reader thread mutates speech_open/awaited_seg_id concurrently.
+        with self._lock:
+            if self.modes.mode is not Mode.COMMAND_ARMED or not self.ready():
+                return None
+            text = self.command_text
+            edited = self.command_edited
+            if not self.modes.execute():
+                return None
         self._emit_mode()
         self._emit_command_text(force=True)
         if self.executor is not None:
@@ -278,9 +289,10 @@ class AssistantService:
 
     def finish(self) -> bool:
         """Leave `executing` and discard the buffer."""
-        if not self.modes.finish():
-            return False
-        self._clear_command()
+        with self._lock:
+            if not self.modes.finish():
+                return False
+            self._clear_command()
         self._emit_mode()
         self._emit_command_text(force=True)
         return True
@@ -289,11 +301,12 @@ class AssistantService:
         """Enter chat mode from idle, or leave chat mode."""
         if self.modes.mode is Mode.CHAT:
             return self._exit_chat("toggle")
-        if not self.modes.enter_chat():
-            return False
-        self.chat_history = []
-        self._pending_chat = []
-        self._last_chat_activity = self._now()
+        with self._lock:
+            if not self.modes.enter_chat():
+                return False
+            self.chat_history = []
+            self._pending_chat = []
+            self._last_chat_activity = self._now()
         self._emit_mode()
         return True
 
@@ -376,11 +389,13 @@ class AssistantService:
             ts = float(event.get("ts", now))
             # Ignore transcripts from before arming (e.g. ring replay on attach).
             if self.arm_ts is None or ts >= self.arm_ts:
-                self.command_text = f"{self.command_text} {text}".strip()
+                with self._lock:
+                    self.command_text = f"{self.command_text} {text}".strip()
         elif self.modes.mode is Mode.CHAT:
-            self.chat_history.append(text)
-            self._pending_chat.append(text)
-            self._last_chat_activity = now
+            with self._lock:
+                self.chat_history.append(text)
+                self._pending_chat.append(text)
+                self._last_chat_activity = now
             if self._is_exit_phrase(text):
                 self._exit_chat("exit_phrase")
         self._emit_command_text()
@@ -408,18 +423,20 @@ class AssistantService:
     # ------------------------------------------------------------------ helpers
 
     def _disarm(self, reason: str) -> None:
-        if not self.modes.cancel():
-            return
-        self._clear_command()
+        with self._lock:
+            if not self.modes.cancel():
+                return
+            self._clear_command()
         self._emit_mode()
         self._emit_command_text(force=True)
         self._emit(ERROR_EVENT, code=reason, message=f"command {reason}")
 
     def _exit_chat(self, reason: str) -> bool:
-        if not self.modes.exit_chat():
-            return False
-        self.chat_history = []
-        self._pending_chat = []
+        with self._lock:
+            if not self.modes.exit_chat():
+                return False
+            self.chat_history = []
+            self._pending_chat = []
         if self.chat_reset is not None:
             try:
                 self.chat_reset()
@@ -431,8 +448,9 @@ class AssistantService:
         return True
 
     def _dispatch_chat(self) -> None:
-        text = " ".join(self._pending_chat).strip()
-        self._pending_chat = []
+        with self._lock:
+            text = " ".join(self._pending_chat).strip()
+            self._pending_chat = []
         if not text:
             return
         self._emit(CHAT_MESSAGE_EVENT, role="user", text=text)

src/stt_daemon/audio.py

@@ -116,26 +116,25 @@ class ParecCapture(AudioCapture):
 
     def frames(self) -> Iterator[bytes]:
         """Yield frames from parec stdout."""
-        if not self._proc or not self._proc.stdout:
+        # Hold a local reference: close() may clear self._proc mid-stream.
+        proc = self._proc
+        if proc is None or proc.stdout is None:
             raise RuntimeError("parec process not running")
 
-        try:
-            while True:
-                frame = self._proc.stdout.read(self._frame_bytes)
-                if not frame:
-                    raise RuntimeError("parec EOF (stream ended unexpectedly)")
-                if len(frame) < self._frame_bytes:
-                    raise RuntimeError(
-                        f"parec short read: {len(frame)} < {self._frame_bytes}"
-                    )
-
-                # Apply gain if needed
-                if self.cfg.gain != 1.0:
-                    frame = self._apply_gain(frame)
-
-                yield frame
-        except KeyboardInterrupt:
-            pass
+        while True:
+            frame = proc.stdout.read(self._frame_bytes)
+            if not frame:
+                raise RuntimeError("parec EOF (stream ended unexpectedly)")
+            if len(frame) < self._frame_bytes:
+                raise RuntimeError(
+                    f"parec short read: {len(frame)} < {self._frame_bytes}"
+                )
+
+            # Apply gain if needed
+            if self.cfg.gain != 1.0:
+                frame = self._apply_gain(frame)
+
+            yield frame
 
     def _apply_gain(self, frame: bytes) -> bytes:
         """Apply gain to frame."""
@@ -149,14 +148,16 @@ class ParecCapture(AudioCapture):
         return self._proc is not None and self._proc.poll() is None
 
     def close(self) -> None:
-        """Terminate parec process."""
-        if self._proc:
-            self._proc.terminate()
+        """Terminate parec process. Idempotent: safe to call more than once."""
+        proc = self._proc
+        self._proc = None
+        if proc is not None:
+            proc.terminate()
             try:
-                self._proc.wait(timeout=2)
+                proc.wait(timeout=2)
             except subprocess.TimeoutExpired:
-                self._proc.kill()
-                self._proc.wait()
+                proc.kill()
+                proc.wait()
 
 
 class SoundDeviceCapture(AudioCapture):
@@ -182,6 +183,7 @@ class SoundDeviceCapture(AudioCapture):
         self.cfg = cfg
         self._queue = None
         self._stream = None
+        self._read_timeout_s = 5.0
         self._device = self._resolve_device(cfg.device)
 
         self._frame_samples = round(cfg.sample_rate * cfg.frame_ms / 1000)
@@ -225,21 +227,29 @@ class SoundDeviceCapture(AudioCapture):
         return self._frame_bytes
 
     def frames(self) -> Iterator[bytes]:
-        """Yield frames from queue."""
+        """Yield frames from queue.
+
+        Raises RuntimeError when the stream stalls (no frames for
+        `_read_timeout_s`), matching ParecCapture's EOF behavior so the
+        daemon's reader thread logs the failure instead of silently exiting.
+        """
         if not self._queue:
             raise RuntimeError("sounddevice stream not initialized")
 
-        try:
-            while True:
-                frame = self._queue.get(timeout=5)
+        while True:
+            try:
+                frame = self._queue.get(timeout=self._read_timeout_s)
+            except queue.Empty:
+                raise RuntimeError(
+                    "sounddevice stream stalled: no frames for "
+                    f"{self._read_timeout_s:.0f}s (device lost?)"
+                ) from None
 
-                # Apply gain if needed
-                if self.cfg.gain != 1.0:
-                    frame = self._apply_gain(frame)
+            # Apply gain if needed
+            if self.cfg.gain != 1.0:
+                frame = self._apply_gain(frame)
 
-                yield frame
-        except Exception:
-            pass
+            yield frame
 
     def _apply_gain(self, frame: bytes) -> bytes:
         """Apply gain to frame."""
@@ -253,10 +263,12 @@ class SoundDeviceCapture(AudioCapture):
         return self._stream is not None and bool(self._stream.active)
 
     def close(self) -> None:
-        """Close the stream."""
-        if self._stream:
-            self._stream.stop()
-            self._stream.close()
+        """Close the stream. Idempotent: safe to call more than once."""
+        stream = self._stream
+        self._stream = None
+        if stream is not None:
+            stream.stop()
+            stream.close()
 
 
 def open_capture(cfg: AudioConfig) -> AudioCapture:

src/stt_daemon/bus.py

@@ -115,8 +115,7 @@ class EventPublisher:
     def _bind(self) -> None:
         path = Path(self.path)
         path.parent.mkdir(parents=True, exist_ok=True)
-        if path.exists():
-            path.unlink()
+        path.unlink(missing_ok=True)
         server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
         server.bind(str(path))
         os.chmod(str(path), 0o600)

src/stt_daemon/cli.py

@@ -261,8 +261,9 @@ def _handle_run(args) -> int:
         daemon = Daemon(cfg, log_level=log_level)
 
         def handle_signal(signum, frame):
-            daemon.stop()
-            sys.exit(0)
+            # Only request the stop here: joining threads or raising SystemExit
+            # inside a signal handler is re-entrant on a second SIGTERM.
+            daemon.request_stop()
 
         signal.signal(signal.SIGINT, handle_signal)
         signal.signal(signal.SIGTERM, handle_signal)

src/stt_daemon/config.py

@@ -1,9 +1,25 @@
 """Configuration management for STT daemon."""
 
 from dataclasses import dataclass, field
+import logging
 import tomllib
 from pathlib import Path
 
+logger = logging.getLogger(__name__)
+
+KNOWN_SECTIONS = {
+    "audio",
+    "vad",
+    "whisper",
+    "output",
+    "sink",
+    "daemon",
+    "events",
+    "assistant",
+    "audit",
+    "llm",
+}
+
 
 @dataclass
 class AudioConfig:
@@ -159,6 +175,12 @@ def load_config(path: str | None = None) -> Config:
         with open(config_path, "rb") as f:
             data = tomllib.load(f)
 
+        unknown = set(data) - KNOWN_SECTIONS
+        if unknown:
+            logger.warning(
+                "Ignoring unknown config sections: %s", ", ".join(sorted(unknown))
+            )
+
         # Audio section
         if "audio" in data:
             audio_data = data["audio"]

src/stt_daemon/ctl.py

@@ -88,6 +88,10 @@ def main() -> int:
                     continue
                 print(json.dumps(reply, ensure_ascii=False))
     finally:
+        try:
+            sock.shutdown(socket.SHUT_RDWR)
+        except OSError:
+            pass
         sock.close()
     return 0
 

src/stt_daemon/daemon.py

@@ -93,6 +93,10 @@ class Daemon:
         finally:
             self.stop()
 
+    def request_stop(self) -> None:
+        """Ask the run loop to stop. Signal-safe: cleanup happens in stop()."""
+        self._stop_event.set()
+
     def _build_transcriber(self) -> Transcriber:
         return Transcriber(
             model=self.cfg.whisper.model,

src/stt_daemon/output.py

@@ -72,17 +72,20 @@ class TranscriptSink:
         if not result.text:
             return None
 
+        # One timestamp for the whole record: stdout, JSONL name/ts, and the
+        # WAV path must agree even when emit() straddles midnight.
+        now = datetime.now()
+
         audio_path = None
         if self.save_audio and audio is not None:
             try:
-                audio_path = self._save_audio(audio, reason)
+                audio_path = self._save_audio(audio, reason, now)
             except Exception as e:
                 # Best-effort side output: a failed save must not drop the transcript.
                 logger.warning("Failed to save audio for this utterance: %s", e)
 
         # Stdout line: [HH:MM:SS] (lang) text  [rms=0.041 end_of_turn]
         if self.stdout:
-            now = datetime.now()
             timestamp = now.strftime("%H:%M:%S")
             line = f"[{timestamp}] ({result.language}) {result.text}  [rms={rms:.3f} {reason}]"
             print(line, flush=True)
@@ -90,11 +93,11 @@ class TranscriptSink:
         # JSONL file per day
         if self.jsonl_dir:
             self.jsonl_dir.mkdir(parents=True, exist_ok=True)
-            today = datetime.now().strftime("%Y-%m-%d")
+            today = now.strftime("%Y-%m-%d")
             jsonl_path = self.jsonl_dir / f"{today}.jsonl"
 
             record = {
-                "ts": datetime.now().isoformat(),
+                "ts": now.isoformat(),
                 "lang": result.language,
                 "lang_prob": result.language_probability,
                 "duration_s": duration_s,
@@ -105,14 +108,13 @@ class TranscriptSink:
             if audio_path is not None:
                 record["audio"] = str(audio_path)
 
-            with open(jsonl_path, "a") as f:
+            with open(jsonl_path, "a", encoding="utf-8") as f:
                 json.dump(record, f, ensure_ascii=False)
                 f.write("\n")
 
         return str(audio_path) if audio_path is not None else None
 
-    def _save_audio(self, audio: np.ndarray, reason: str) -> Path:
-        now = datetime.now()
+    def _save_audio(self, audio: np.ndarray, reason: str, now: datetime) -> Path:
         day_dir = self.audio_dir / now.strftime("%Y-%m-%d")
         day_dir.mkdir(parents=True, exist_ok=True)
         name = f"{now.strftime('%H%M%S')}-{now.strftime('%f')[:3]}-{reason}.wav"

src/stt_daemon/state.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+import tempfile
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
 
@@ -60,7 +61,7 @@ class StateStore:
             if not self.path.exists():
                 return None
 
-            with open(self.path, "r") as f:
+            with open(self.path, "r", encoding="utf-8") as f:
                 data = json.load(f)
 
             # Reconstruct Loopback objects
@@ -86,7 +87,8 @@ class StateStore:
     def save(self, state: AudioState) -> None:
         """Save state to file.
 
-        Creates parent directories if needed.
+        Creates parent directories if needed. Writes via a temp file and
+        atomic rename so a crash mid-write cannot corrupt the state file.
 
         Args:
             state: AudioState to persist.
@@ -94,8 +96,19 @@ class StateStore:
         self.path.parent.mkdir(parents=True, exist_ok=True)
 
         data = asdict(state)
-        with open(self.path, "w") as f:
-            json.dump(data, f, indent=2)
+        fd, tmp_path = tempfile.mkstemp(
+            dir=self.path.parent, prefix=".audio-", suffix=".tmp"
+        )
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
+                json.dump(data, f, indent=2)
+            os.replace(tmp_path, self.path)
+        except BaseException:
+            try:
+                os.unlink(tmp_path)
+            except OSError:
+                pass
+            raise
 
     def clear(self) -> None:
         """Remove the state file if it exists."""

src/stt_daemon/tui/app.py

@@ -40,6 +40,7 @@ class AssistantTUI(App):
         self.path = path
         self.state = ViewState()
         self.client = ControlClient(path)
+        self.connect_error: str | None = None
         self._events: queue.Queue = queue.Queue()
         self._seen_transcripts = 0
         self._seen_lower = 0
@@ -63,11 +64,17 @@ class AssistantTUI(App):
             self.client.connect()
             self.client.subscribe(viewer=True)
         except OSError as e:
-            self.exit(message=f"cannot connect to {self.path}: {e}")
+            self.connect_error = f"cannot connect to {self.path}: {e}"
+            self.exit()
             return
-        self.client.start(lambda event: self._events.put(event))
+        self.client.start(
+            lambda event: self._events.put(event), on_state=self._on_conn_state
+        )
         self.set_interval(0.1, self._drain)
 
+    def _on_conn_state(self, connected: bool) -> None:
+        self.state.connected = connected
+
     def _drain(self) -> None:
         while True:
             try:
@@ -80,10 +87,12 @@ class AssistantTUI(App):
     def _render(self) -> None:
         state = self.state
         health = "ok" if state.health_ok() else "DOWN"
+        conn = "ok" if state.connected else "LOST"
         speaking = "●" if state.speech_open else "○"
         status = (
             f"[{state.mode_style()} bold]{state.mode}[/]  "
-            f"stt:{health}  speech:{speaking}  worker_stall:{state.worker_stall_s:.1f}s  "
+            f"stt:{health}  conn:{conn}  speech:{speaking}  "
+            f"worker_stall:{state.worker_stall_s:.1f}s  "
             f"backlog:{state.queue_depth}  phase:{state.phase}"
         )
         self.query_one("#status", Static).update(status)
@@ -187,7 +196,11 @@ def main() -> int:
     if args.dump is not None:
         return _dump(path, args.dump)
 
-    AssistantTUI(path).run()
+    app = AssistantTUI(path)
+    app.run()
+    if app.connect_error:
+        print(f"error: {app.connect_error}", file=sys.stderr)
+        return 1
     return 0
 
 

src/stt_daemon/tui/client.py

@@ -9,6 +9,8 @@ from typing import Callable
 logger = logging.getLogger(__name__)
 
 READ_CHUNK = 65536
+RECONNECT_MIN_S = 0.2
+RECONNECT_MAX_S = 5.0
 
 
 class ControlClient:
@@ -19,11 +21,16 @@ class ControlClient:
         self._sock: socket.socket | None = None
         self._stop = threading.Event()
         self._thread: threading.Thread | None = None
+        self._viewer = True
 
     def connect(self) -> None:
         """Connect to the control socket (raises OSError on failure)."""
         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-        sock.connect(self.path)
+        try:
+            sock.connect(self.path)
+        except OSError:
+            sock.close()
+            raise
         self._sock = sock
 
     def send(self, type_: str, **fields) -> None:
@@ -35,27 +42,52 @@ class ControlClient:
 
     def subscribe(self, viewer: bool = True) -> None:
         """Subscribe as a viewer to receive broadcasts."""
+        self._viewer = viewer
         self.send("subscribe", viewer=viewer)
 
-    def start(self, on_event: Callable[[dict], None]) -> None:
-        """Start the background reader, invoking `on_event` per message."""
+    def start(
+        self,
+        on_event: Callable[[dict], None],
+        on_state: Callable[[bool], None] | None = None,
+    ) -> None:
+        """Start the background reader, invoking `on_event` per message.
+
+        When the connection drops, the reader reconnects with backoff until
+        `close()`; `on_state` is called with False on drop and True on reconnect.
+        """
         self._thread = threading.Thread(
-            target=self._read_loop, args=(on_event,), name="tui-client", daemon=True
+            target=self._read_loop,
+            args=(on_event, on_state),
+            name="tui-client",
+            daemon=True,
         )
         self._thread.start()
 
-    def _read_loop(self, on_event: Callable[[dict], None]) -> None:
-        sock = self._sock
-        if sock is None:
-            return
+    def _read_loop(self, on_event, on_state) -> None:
+        while not self._stop.is_set():
+            sock = self._sock
+            if sock is None:
+                return
+            self._pump(sock, on_event)
+            if self._stop.is_set():
+                return
+            if on_state is not None:
+                on_state(False)
+            self._drop_socket()
+            if not self._reconnect():
+                return
+            if on_state is not None:
+                on_state(True)
+
+    def _pump(self, sock: socket.socket, on_event) -> None:
         buffer = bytearray()
         while not self._stop.is_set():
             try:
                 chunk = sock.recv(READ_CHUNK)
             except OSError:
-                break
+                return
             if not chunk:
-                break
+                return
             buffer.extend(chunk)
             while True:
                 newline = buffer.find(b"\n")
@@ -70,17 +102,37 @@ class ControlClient:
                 except (UnicodeDecodeError, json.JSONDecodeError) as e:
                     logger.debug("dropping malformed event: %s", e)
 
-    def close(self) -> None:
-        """Stop the reader and close the socket."""
-        self._stop.set()
+    def _drop_socket(self) -> None:
         sock = self._sock
         self._sock = None
         if sock is not None:
             try:
-                sock.shutdown(socket.SHUT_RDWR)
+                sock.close()
             except OSError:
                 pass
+
+    def _reconnect(self) -> bool:
+        delay = RECONNECT_MIN_S
+        while not self._stop.is_set():
             try:
-                sock.close()
+                self.connect()
+                self.subscribe(viewer=self._viewer)
+                return True
+            except OSError:
+                self._stop.wait(delay)
+                delay = min(delay * 2, RECONNECT_MAX_S)
+        return False
+
+    def close(self) -> None:
+        """Stop the reader and close the socket."""
+        self._stop.set()
+        sock = self._sock
+        if sock is not None:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
             except OSError:
                 pass
+        self._drop_socket()
+        thread = self._thread
+        if thread is not None and thread is not threading.current_thread():
+            thread.join(timeout=1)

src/stt_daemon/tui/state.py

@@ -23,8 +23,9 @@ class ViewState:
     """Everything the TUI renders."""
 
     mode: str = "idle"
-    stt_ok: bool = True
+    stt_ok: bool = False
     device_ok: bool = True
+    connected: bool = True
     speech_open: bool = False
     worker_stall_s: float = 0.0
     queue_depth: int = 0

tests/test_assistant_service.py

@@ -1,6 +1,8 @@
 """Tests for the assistant service (Phase 2) using synthetic event streams."""
 
 import itertools
+import threading
+import time
 
 from stt_daemon import events
 from stt_daemon.assistant.service import AssistantService
@@ -282,3 +284,47 @@ def test_cancel_clears_command_buffer():
     assert service.modes.mode.value == "idle"
     assert service.command_text == ""
     assert service.arm_ts is None
+
+
+def test_pending_chat_is_not_lost_across_threads():
+    """Transcripts appended on the reader thread while the ticker dispatches
+    must all reach the model: the join-and-clear in _dispatch_chat is atomic."""
+    cfg = Config()
+    cfg.assistant.turn_gap_s = 0.01
+    dispatched: list[str] = []
+
+    def emit(type_, **fields):
+        if type_ == "chat_message" and fields.get("role") == "user":
+            dispatched.append(fields["text"])
+
+    service = AssistantService(cfg, FakeSubscriber(), emit=emit, chat_fn=lambda text: "ok")
+    service.toggle_chat()
+
+    total = 200
+    producer = threading.Thread(
+        target=lambda: [
+            service.handle_event(
+                {
+                    "v": 1,
+                    "type": events.TRANSCRIPT,
+                    "seq": i,
+                    "ts": time.time(),
+                    "text": f"w{i}",
+                }
+            )
+            for i in range(total)
+        ],
+        daemon=True,
+    )
+    producer.start()
+
+    deadline = time.time() + 10.0
+    while time.time() < deadline:
+        service.tick()
+        words = sum(len(text.split()) for text in dispatched)
+        if not producer.is_alive() and words == total:
+            break
+        time.sleep(0.002)
+    producer.join(timeout=5)
+
+    assert sum(len(text.split()) for text in dispatched) == total

tests/test_audio.py

@@ -0,0 +1,116 @@
+"""Tests for audio capture backends using fakes (no hardware, no subprocesses)."""
+
+import queue
+
+import pytest
+
+from stt_daemon.audio import ParecCapture, SoundDeviceCapture
+from stt_daemon.config import AudioConfig
+
+
+class FakeStdout:
+    def __init__(self, chunks):
+        self._chunks = list(chunks)
+
+    def read(self, _n):
+        return self._chunks.pop(0) if self._chunks else b""
+
+
+class FakeProc:
+    def __init__(self, chunks=()):
+        self.stdout = FakeStdout(chunks)
+        self.terminated = 0
+        self.killed = 0
+        self.waits = 0
+
+    def terminate(self):
+        self.terminated += 1
+
+    def wait(self, timeout=None):
+        self.waits += 1
+        return 0
+
+    def kill(self):
+        self.killed += 1
+
+    def poll(self):
+        return None
+
+
+class FakeStream:
+    def __init__(self):
+        self.stopped = 0
+        self.closed = 0
+        self.active = True
+
+    def stop(self):
+        self.stopped += 1
+        self.active = False
+
+    def close(self):
+        self.closed += 1
+
+
+def _parec(chunks=()):
+    cap = ParecCapture.__new__(ParecCapture)
+    cap.cfg = AudioConfig()
+    cap._frame_bytes = 4
+    cap._proc = FakeProc(chunks)
+    return cap
+
+
+def _sounddevice():
+    cap = SoundDeviceCapture.__new__(SoundDeviceCapture)
+    cap.cfg = AudioConfig()
+    cap._queue = queue.Queue()
+    cap._read_timeout_s = 0.05
+    cap._stream = FakeStream()
+    return cap
+
+
+def test_parec_frames_yield_until_eof():
+    cap = _parec(chunks=[b"abcd", b"efgh", b""])
+    gen = cap.frames()
+    assert next(gen) == b"abcd"
+    assert next(gen) == b"efgh"
+    with pytest.raises(RuntimeError, match="parec EOF"):
+        next(gen)
+
+
+def test_parec_short_read_raises():
+    cap = _parec(chunks=[b"ab"])
+    with pytest.raises(RuntimeError, match="short read"):
+        next(cap.frames())
+
+
+def test_parec_close_is_idempotent():
+    cap = _parec()
+    proc = cap._proc
+    cap.close()
+    cap.close()
+    assert proc.terminated == 1
+    assert proc.waits == 1
+    assert cap.healthy() is False
+
+
+def test_sounddevice_frames_yield_queued_frames():
+    cap = _sounddevice()
+    cap._queue.put(b"frame1")
+    gen = cap.frames()
+    assert next(gen) == b"frame1"
+
+
+def test_sounddevice_stall_raises_instead_of_silent_exit():
+    cap = _sounddevice()
+    with pytest.raises(RuntimeError, match="stalled"):
+        next(cap.frames())
+
+
+def test_sounddevice_close_is_idempotent():
+    cap = _sounddevice()
+    stream = cap._stream
+    cap.close()
+    cap.close()
+    assert stream.stopped == 1
+    assert stream.closed == 1
+    assert cap.healthy() is False

tests/test_config.py

@@ -246,3 +246,20 @@ backlog_warn_at = 3
 
     assert cfg.daemon.queue_size == 20
     assert cfg.daemon.backlog_warn_at == 3
+
+
+def test_unknown_sections_warn_and_are_ignored(caplog):
+    """A typo'd section like [assitant] must warn, not silently use defaults."""
+    config_toml = """
+[assitant]
+require_viewer = false
+"""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        config_path = Path(tmpdir) / "config.toml"
+        config_path.write_text(config_toml)
+
+        with caplog.at_level("WARNING", logger="stt_daemon.config"):
+            cfg = load_config(str(config_path))
+
+    assert "assitant" in caplog.text
+    assert cfg.assistant.require_viewer is True

tests/test_daemon_events.py

@@ -353,3 +353,28 @@ def test_inbound_data_on_subscriber_socket_is_ignored(sock_path):
     raw.close()
     subscriber.close()
     daemon.stop()
+
+
+def test_request_stop_shuts_down_run_loop(tmp_path):
+    """request_stop() (the signal-handler path) must unwind run() cleanly."""
+    sock_path = str(tmp_path / "events.sock")
+    cfg = _base_config(sock_path)
+    capture = FakeCapture([_frame(0)] * 5, stall_after=1)
+    sink = FakeSink()
+    daemon = Daemon(
+        cfg,
+        log_level=logging.WARNING,
+        capture_factory=lambda: capture,
+        vad_factory=lambda _aggr: FakeVad([]),
+        transcriber_factory=lambda: FakeTranscriber(),
+        sink_factory=lambda: sink,
+    )
+
+    thread = threading.Thread(target=daemon.run, daemon=True)
+    thread.start()
+    time.sleep(0.2)
+    daemon.request_stop()
+    thread.join(timeout=5)
+
+    assert not thread.is_alive()
+    assert capture._closed.is_set()

tests/test_executor.py

@@ -62,6 +62,7 @@ def test_unknown_tool_is_rejected():
 
     assert result.ok is False
     assert any(e["type"] == "error" and e["code"] == "invalid_tool_call" for e in emitted)
+    assert not any(e["type"] == "tool_call" for e in emitted)
 
 
 def test_malformed_arguments_are_rejected():
@@ -74,6 +75,7 @@ def test_malformed_arguments_are_rejected():
 
     assert result.ok is False
     assert any(e["type"] == "error" and e["code"] == "invalid_tool_call" for e in emitted)
+    assert not any(e["type"] == "tool_call" for e in emitted)
 
 
 def test_tool_loop_is_capped():

tests/test_output.py

@@ -73,6 +73,42 @@ def test_saves_wav_when_enabled(tmp_path):
     assert record["audio"] == str(wavs[0])
 
 
+def test_single_timestamp_per_emit(tmp_path, monkeypatch):
+    """All artifacts of one emit() share one timestamp, even across midnight."""
+    import stt_daemon.output as output
+    from datetime import datetime as real_datetime
+
+    calls = []
+    t1 = real_datetime(2026, 9, 20, 23, 59, 59)
+    t2 = real_datetime(2026, 9, 21, 0, 0, 1)
+
+    class FakeDatetime:
+        @staticmethod
+        def now():
+            calls.append(1)
+            return t1 if len(calls) == 1 else t2
+
+    monkeypatch.setattr(output, "datetime", FakeDatetime)
+
+    logs = tmp_path / "logs"
+    sink = TranscriptSink(
+        stdout=False,
+        jsonl_dir=str(logs),
+        save_audio=True,
+        audio_dir=str(tmp_path / "audio"),
+        sample_rate=16000,
+    )
+    sink.emit(_result(), 1.0, rms=0.1, reason="pause", audio=np.zeros(160, dtype=np.float32))
+
+    assert len(calls) == 1
+    jsonl = next(logs.glob("*.jsonl"))
+    assert jsonl.name == "2026-09-20.jsonl"
+    record = json.loads(jsonl.read_text().splitlines()[0])
+    assert record["ts"].startswith("2026-09-20")
+    wav = next((tmp_path / "audio").rglob("*.wav"))
+    assert wav.parent.name == "2026-09-20"
+
+
 def test_no_wav_when_disabled(tmp_path):
     sink = TranscriptSink(
         stdout=False,

tests/test_tui_app.py

@@ -3,6 +3,7 @@
 import asyncio
 import os
 import shutil
+import sys
 import tempfile
 
 import pytest
@@ -114,3 +115,35 @@ def test_tui_mounts_subscribes_and_ingests_broadcasts(server):
 
     asyncio.run(scenario())
     assert app.state.mode == "command_armed"
+
+
+def test_tui_records_connect_failure(tmp_path):
+    pytest.importorskip("textual")
+    from stt_daemon.tui.app import AssistantTUI
+
+    app = AssistantTUI(str(tmp_path / "missing.sock"))
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+
+    asyncio.run(scenario())
+    assert app.connect_error is not None
+    assert "cannot connect" in app.connect_error
+
+
+def test_main_exit_code_reflects_connect_failure(monkeypatch, tmp_path, capsys):
+    pytest.importorskip("textual")
+    from stt_daemon.tui import app as tui_app
+
+    monkeypatch.setattr(
+        tui_app.AssistantTUI,
+        "run",
+        lambda self: setattr(self, "connect_error", "boom"),
+    )
+    monkeypatch.setattr(sys, "argv", ["stt-tui", "--socket", str(tmp_path / "x.sock")])
+    assert tui_app.main() == 1
+    assert "boom" in capsys.readouterr().err
+
+    monkeypatch.setattr(tui_app.AssistantTUI, "run", lambda self: None)
+    assert tui_app.main() == 0

tests/test_tui_client.py

@@ -97,3 +97,31 @@ def test_client_receives_relayed_transcript(server):
 
     assert _wait(lambda: any(e["type"] == "transcript" and e["text"] == "hello" for e in received))
     client.close()
+
+
+def test_client_reconnects_after_server_restart(server):
+    path, service, control = server
+    received: list[dict] = []
+    states: list[bool] = []
+
+    client = ControlClient(path)
+    client.connect()
+    client.subscribe(viewer=True)
+    client.start(received.append, on_state=states.append)
+    assert _wait(lambda: any(e["type"] == "status" for e in received))
+
+    control.close()
+    assert _wait(lambda: False in states)
+
+    cfg = Config()
+    cfg.assistant.control_socket = path
+    control2 = ControlServer(cfg, service, path=path)
+    service.set_sinks(emit=control2.broadcast, relay=control2.relay)
+    control2.start()
+    try:
+        assert _wait(lambda: states[-1] is True if states else False)
+        received.clear()
+        assert _wait(lambda: any(e["type"] == "status" for e in received))
+    finally:
+        client.close()
+        control2.close()

tests/test_tui_state.py

@@ -10,6 +10,14 @@ def test_mode_event_updates_mode_and_style():
     assert state.mode_style() == "yellow"
 
 
+def test_initial_health_is_not_optimistic():
+    """Before the first heartbeat arrives, health must read as unknown/down."""
+    state = ViewState()
+    assert state.health_ok() is False
+    state.ingest({"type": "heartbeat", "stt_ok": True, "device_ok": True})
+    assert state.health_ok() is True
+
+
 def test_heartbeat_self_corrects_missed_mode():
     state = ViewState()
     state.ingest(