Keep listening while an utterance is being transcribed

stt.finish() used to run inline in the frame loop, deafening the
assistant for up to ~4s per utterance while the server returned the
transcript (and whatever was said meanwhile was later drained). The
finish now runs on a background thread per utterance; the segmenter
keeps consuming mic frames, so speech during the wait becomes its own
utterance and is transcribed too. Transcripts are delivered oldest-
first between frames, deferred while an utterance is in progress, and
the command/follow-up window no longer times out on transcription
latency. Speaking stays half-duplex (echo protection).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

bolt_assistant/assistant.py

@@ -12,6 +12,11 @@ Speaking is half-duplex: the mic is ignored while Bolt talks and for a
 moment after, so his own voice through the speakers can't wake him (his
 replies say "Bolt" often enough).
 
+Transcribing is not: waiting for the server's transcript happens on a
+background thread, so the mic keeps being segmented meanwhile and whatever
+is said next becomes its own utterance and is sent off too. Finished
+transcripts are delivered oldest-first between mic frames.
+
 Everything that touches hardware or the network is injected, so the flow
 is tested with fakes; __main__ wires in the real mic, server and speaker.
 """
@@ -22,11 +27,12 @@ import platform
 import queue
 import threading
 import time
+from collections import deque
 from typing import Callable, Optional
 
 from . import audio, config, server, speech_text
 from .listener import Segmenter
-from .stt import SttError, Transcription
+from .stt import Transcription
 from .tools import Tools
 from .wake import find_wake
 
@@ -57,6 +63,17 @@ def annotate(text: str, commands: str) -> str:
             f"{can} Never make up a command's output.]")
 
 
+class _PendingTranscript:
+    """One utterance already sent off, its transcript still on the way."""
+
+    __slots__ = ("thread", "text", "error")
+
+    def __init__(self):
+        self.thread: Optional[threading.Thread] = None
+        self.text = ""
+        self.error: Optional[Exception] = None
+
+
 class Assistant:
     def __init__(
         self,
@@ -89,6 +106,7 @@ class Assistant:
         self._deadline = 0.0
         self._follow_ups = 0
         self._stt = None
+        self._pending: "deque[_PendingTranscript]" = deque()
         self._mute_until = 0.0
         self._voice = ""  # a voice the server picked (speak_as) sticks for later replies
         self._tts_failed = False
@@ -109,8 +127,9 @@ class Assistant:
                 frame = self.frames.get(timeout=0.2)
             except queue.Empty:
                 self._check_deadline()
-                continue
+            else:
                 self.process(frame)
+            self.handle_transcripts()
 
     def listening_banner(self) -> str:
         return 'Listening for "Hey Bolt" (or "Thunderbolt", "Hey Assistant"). Ctrl+C to quit.'
@@ -131,18 +150,53 @@ class Assistant:
                 self._stt = None
             elif event.kind == "end":
                 stt, self._stt = self._stt, None
+                if stt is not None:
+                    self._start_finish(stt, event.audio)
+
+    def _start_finish(self, stt, utterance: bytes) -> None:
+        """finish() waits on the server; done here, that wait would deafen
+        the loop. Run it on a thread and keep segmenting the mic meanwhile."""
+        pending = _PendingTranscript()
+
+        def work() -> None:
             try:
-                    text = stt.finish(event.audio) if stt is not None else ""
-                except SttError as exc:
-                    self.log(f"! transcription failed: {exc}")
+                pending.text = stt.finish(utterance)
+            except Exception as exc:  # SttError, or the socket misbehaving
+                pending.error = exc
+
+        pending.thread = threading.Thread(target=work, name="stt-finish", daemon=True)
+        self._pending.append(pending)
+        pending.thread.start()
+
+    def handle_transcripts(self, wait: bool = False) -> None:
+        """Deliver finished transcripts, oldest first. One still in flight —
+        or an utterance still being spoken — ends the pass rather than
+        blocking, unless *wait* (tests, shutdown). The head stays queued
+        while heard() runs so _check_deadline knows speech already happened."""
+        while self._pending:
+            if self.segmenter.in_speech and not wait:
+                return
+            head = self._pending[0]
+            if wait:
+                head.thread.join()
+            if head.thread.is_alive():
+                return
+            if head.error is not None:
+                self._pending.popleft()
+                self.log(f"! transcription failed: {head.error}")
                 if self.state != "idle":
                     self._chime(audio.ERROR_CHIME)
                 continue
-                self.heard(text)
+            try:
+                self.heard(head.text)
+            finally:
+                self._pending.popleft()
 
     def _check_deadline(self) -> None:
+        # A pending transcript means something *was* said in time — it's
+        # just still at the server; the window mustn't close on that latency.
         if self.state in ("command", "follow_up") and self.clock() > self._deadline \
-                and not self.segmenter.in_speech:
+                and not self.segmenter.in_speech and not self._pending:
             if self.verbose:
                 self.log("  (stopped waiting)")
             self.state = "idle"

tests/test_assistant.py

@@ -111,19 +111,22 @@ def test_long_speech_is_cut_off_at_the_cap():
 # ── the flow ────────────────────────────────────────────────────────────────
 
 class FakeTranscription:
-    """Each utterance returns the next scripted transcript."""
+    """Each utterance takes the next scripted transcript. Bound at creation
+    (the "start" event, on the main thread) because finish() now runs on a
+    worker thread per utterance and two may be in flight at once."""
 
     script = []
 
     def __init__(self):
         self.frames = 0
         self.cancelled = False
+        self.transcript = FakeTranscription.script.pop(0)
 
     def add(self, frames):
         self.frames += len(frames)
 
     def finish(self, audio):
-        return FakeTranscription.script.pop(0)
+        return self.transcript
 
     def cancel(self):
         self.cancelled = True
@@ -141,7 +144,7 @@ def utterance():
     return [frame(LOUD)] * 8 + [frame(QUIET)] * 12
 
 
-def make(replies, commands="read", streaming=False, **kw):
+def make(replies, commands="read", streaming=False, transcription=FakeTranscription, **kw):
     asked, played, logs = [], [], []
 
     def converse(message, on_command, on_say=None):
@@ -152,7 +155,7 @@ def make(replies, commands="read", streaming=False, **kw):
         return server.Reply(reply)
 
     clock = Clock()
-    a = Assistant(transcription=FakeTranscription, converse=converse,
+    a = Assistant(transcription=transcription, converse=converse,
                   converse_stream=None if not streaming else kw.pop("stream"),
                   tts=lambda text, voice: [text.encode()], play=lambda chunks: played.extend(chunks),
                   tools=Tools(mode=commands, log=logs.append), log=logs.append, clock=clock, **kw)
@@ -163,6 +166,7 @@ def say(a, transcript):
     FakeTranscription.script.append(transcript)
     for f in utterance():
         a.process(f)
+    a.handle_transcripts(wait=True)  # what run() does between frames
 
 
 def spoken(played):
@@ -225,6 +229,27 @@ def test_mic_audio_during_speech_is_discarded():
     assert not a.segmenter.in_speech
 
 
+def test_listening_continues_while_a_transcript_is_in_flight():
+    import threading
+
+    release = threading.Event()
+
+    class Slow(FakeTranscription):
+        def finish(self, audio):
+            release.wait(timeout=5)
+            return self.transcript
+
+    a, asked, played, logs, _ = make(["One.", "Two."], transcription=Slow)
+    FakeTranscription.script += ["hey bolt first thing", "hey bolt second thing"]
+    for f in utterance() + utterance():  # the second said while the first transcribes
+        a.process(f)
+    assert asked == []  # both transcripts still at the "server"
+    release.set()
+    a.handle_transcripts(wait=True)
+    assert len(asked) == 2  # nothing said during the wait was lost
+    assert asked[0].startswith("first thing") and asked[1].startswith("second thing")
+
+
 def test_streamed_sentences_are_spoken_as_they_arrive():
     def stream(message, on_say, on_command):
         on_say("First.")