Let a button press cut the current reply (barge-in)

Every press bumps a turn counter and kills the running aplay; queued
audio from an older turn is dropped, and reply() now runs on its own
thread and returns once its turn is stale, so the button loop is never
blocked. A cut-off exchange is left out of history so user and
assistant messages still alternate.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 65 insertions, 34 deletions

diff --git a/assistant.py b/assistant.py
index 635fb81..7724b73 100644
--- a/assistant.py
+++ b/assistant.py
@@ -260,35 +260,51 @@ def ask(messages, calls, tools=None):
                 yield delta["content"]
 
 
+turn = 0          # bumped by every button press: older replies and audio are stale
+player = None     # the aplay speaking now, killed by a new press
+
+
 def speaker(q):
+    global player
     from kokoro_onnx import Kokoro
     k = Kokoro(CFG["kokoro"]["model"], CFG["kokoro"]["voices"])
     # Each voice is a weighted blend of Kokoro style vectors.
     voices = {l: sum(w * k.get_voice_style(n) for n, w in v["blend"].items())
               for l, v in CFG["voice"].items()}
-    play = None
+    play, playing = None, None            # aplay and the turn it speaks for
     while True:
-        item = q.get()
-        if item is None:                  # end of reply
-            if play:
-                play.stdin.close()
-                play.wait()
-                play = None
+        t, item = q.get()
+        if t != turn:                     # from a reply the user cut off
             continue
-        if isinstance(item, float):       # seconds of silence
-            if play:
-                play.stdin.write(bytes(int(sr * item) * 2))
-            continue
-        text, lang = item
-        audio, sr = k.create(text, voice=voices[lang], lang=KLANG[lang])
-        if play is None:
-            play = subprocess.Popen(["aplay", "-q", "-r", str(sr), "-f", "S16_LE", "-c", "1"],
-                                    stdin=subprocess.PIPE)
-            # A new stream wakes the suspended sink and the first ~0.2 s is
-            # lost: lead with silence so it eats that, not the first word.
-            play.stdin.write(bytes(int(sr * 0.3) * 2))
-        # The pipe buffer lets synthesis of the next sentence overlap playback.
-        play.stdin.write((audio.clip(-1, 1) * 32767).astype("<i2").tobytes())
+        if play and playing != t:         # never reuse a cut-off stream
+            play.kill()
+            play = None
+        try:
+            if item is None:              # end of reply
+                if play:
+                    play.stdin.close()
+                    play.wait()
+                    play = None
+                continue
+            if isinstance(item, float):   # seconds of silence
+                if play:
+                    play.stdin.write(bytes(int(sr * item) * 2))
+                continue
+            text, lang = item
+            audio, sr = k.create(text, voice=voices[lang], lang=KLANG[lang])
+            if t != turn:                 # cut off while synthesising
+                continue
+            if play is None:
+                play = subprocess.Popen(["aplay", "-q", "-r", str(sr), "-f", "S16_LE", "-c", "1"],
+                                        stdin=subprocess.PIPE)
+                playing, player = t, play
+                # A new stream wakes the suspended sink and the first ~0.2 s is
+                # lost: lead with silence so it eats that, not the first word.
+                play.stdin.write(bytes(int(sr * 0.3) * 2))
+            # The pipe buffer lets synthesis of the next sentence overlap playback.
+            play.stdin.write((audio.clip(-1, 1) * 32767).astype("<i2").tobytes())
+        except BrokenPipeError:           # killed by a new press mid-write
+            play = None
 
 
 def die_with_parent():
@@ -319,7 +335,10 @@ def start_whisper():
     sys.exit("whisper-server did not come up")
 
 
-def reply(history, q):
+def reply(history, q, mine):
+    # Runs on its own thread; returns as soon as a newer press makes it stale.
+    # ponytail: a press during this STT can overwrite WAV mid-upload, one
+    # garbled question at worst; per-turn file names if that ever shows.
     pcm = open(RAW, "rb").read()
     if len(pcm) < 16000 * 2 * 0.3:               # under 0.3 s: a stray press
         return
@@ -332,14 +351,14 @@ def reply(history, q):
         w.writeframes(pcm + bytes(16000 * 2 // 2))
     t = time.perf_counter()
     lang, text = stt()
-    if not text or text.startswith("["):         # [BLANK_AUDIO] and friends
+    if not text or text.startswith("[") or turn != mine:   # [BLANK_AUDIO] and friends
         return
     log(f"you ({lang}, {time.perf_counter() - t:.2f}s): {text}")
-    history.append({"role": "user", "content": text})
+    question = {"role": "user", "content": text}
     # ponytail: last 20 messages, summarise if long sessions need memory
     # The model has no clock: without the date it makes one up for tools.
     now = time.strftime("It is %A %d %B %Y, %H:%M.")
-    msgs = [{"role": "system", "content": f"{CFG['system']} {now}"}] + history[-20:]
+    msgs = [{"role": "system", "content": f"{CFG['system']} {now}"}] + history[-20:] + [question]
     t, said, voice, announced = time.perf_counter(), [], lang, False
 
     def say(s):
@@ -348,19 +367,22 @@ def reply(history, q):
             log(f"first sentence after {time.perf_counter() - t:.2f}s")
         said.append(s)
         voice = lang_of(s, voice)
-        q.put((re.sub(r"[*_#`]", "", s), voice))
+        q.put((mine, (re.sub(r"[*_#`]", "", s), voice)))
 
     for rnd in range(CFG["tool_rounds"] + 1):
         calls, buf = {}, ""
         # The last round offers no tools, which forces a spoken answer.
         for delta in ask(msgs, calls, TOOL_SCHEMA if rnd < CFG["tool_rounds"] else None):
+            if turn != mine:
+                log("interrupted")
+                return
             buf += delta
             done, buf = split_sentences(buf)
             for s in done:
                 say(s)
         if buf.strip():
             say(buf.strip())
-        if not calls:
+        if not calls or turn != mine:
             break
         msgs.append({"role": "assistant", "content": "", "tool_calls": [
             {"id": c["id"], "type": "function",
@@ -369,21 +391,27 @@ def reply(history, q):
         phrase = None if announced else announcement(TOOLS, calls, lang)
         if phrase:
             # The question's language: the reply has no text yet to follow.
-            q.put((phrase, lang))
+            q.put((mine, (phrase, lang)))
             # A beat after it: straight into the answer sounds rushed.
-            q.put(0.6)
+            q.put((mine, 0.6))
             announced = True
+        # ponytail: a press does not kill a running tool, its result is dropped
         for c in calls.values():
             result = run_tool(TOOLS, c["name"], c["arguments"])
             log(f"tool {c['name']}({c['arguments']}): {result[:120]!r}")
             msgs.append({"role": "tool", "tool_call_id": c["id"], "content": result})
-    q.put(None)
-    # Tool calls stay inside the turn: history keeps only what was said.
-    history.append({"role": "assistant", "content": " ".join(said)})
+    if turn != mine:
+        log("interrupted")
+        return
+    q.put((mine, None))
+    # Tool calls stay inside the turn: history keeps only what was said. A cut
+    # off exchange is dropped whole, so user and assistant still alternate.
+    history += [question, {"role": "assistant", "content": " ".join(said)}]
     log("assistant:", " ".join(said))
 
 
 def main():
+    global turn
     # SIGTERM must run the cleanup below too, or the recording is left behind.
     signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
     start_whisper()
@@ -405,14 +433,17 @@ def main():
             if on and presenting():
                 log("presentation mode: ignored")
             elif on:
-                # ponytail: no barge-in, a new press does not cut the reply
+                # Barge-in: a press silences the reply now and makes it stale.
+                turn += 1
+                if player:
+                    player.kill()
                 rec = subprocess.Popen(["arecord", "-q", "-t", "raw", "-f", "S16_LE", "-r", "16000",
                                         "-c", "1", RAW], stderr=subprocess.DEVNULL)
             elif rec:
                 rec.send_signal(signal.SIGINT)
                 rec.wait()
                 rec = None
-                reply(history, q)
+                threading.Thread(target=reply, args=(history, q, turn), daemon=True).start()
     except KeyboardInterrupt:
         pass
     finally: