Voice: wait 1.3s before replying; new speech cancels the reply in progress

core/conversation.py

@@ -40,6 +40,7 @@ class Chat:
     matches: list = field(default_factory=list)
     profile: dict = field(default_factory=dict)
     lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
+    cancel: threading.Event | None = field(default=None, repr=False)   # set to stop the turn in progress
 
 
 _chats: dict[str, Chat] = {}
@@ -91,37 +92,58 @@ def _search(chat: Chat, gen, spoken: bool):
 
 def turn(chat_id: str, text: str = "", file: str | Path | None = None, spoken: bool = False):
     chat = get(chat_id)
+    mine = threading.Event()
+    if chat.cancel is not None:
+        chat.cancel.set()      # a newer message replaces a reply still being written; it stops at its next step
     with chat.lock:   # one turn at a time per chat (e.g. a voice turn and a typed turn can't interleave)
-        text = (text or "").strip()
-        label = ("🎤 " if spoken else "") + (text or ("📎 " + Path(file).name if file else ""))
-        chat.history.append({"role": "user", "content": label})
-        yield ("history", chat.history)
-        if chat.mode == "intake":
-            chat.history.append({"role": "assistant", "content": ""})
-            said, st = [], None
-            for kind, data in intake.send_stream(chat.id, text, file=file):
-                if kind == "say":
-                    said.append(data)
-                    chat.history[-1]["content"] = " ".join(said)
-                    yield ("history", chat.history)
-                    if spoken:
-                        yield ("say", data)
-                else:
-                    st = data
-            chat.history[-1]["content"] = st["reply"]
-            chat.profile = {**st["profile"], "_missing": st["missing"], "_done": st["done"]}
-            yield ("history", chat.history)
-            yield ("profile", chat.profile)
-            if not st["done"]:
+        chat.cancel = mine
+        yield from _guarded(chat, mine, _turn(chat, text, file, spoken))
+
+
+def _guarded(chat: Chat, mine: threading.Event, gen):
+    """Pass events through until a newer turn cancels this one; closing the generator stops the work early."""
+    try:
+        for event in gen:
+            if mine.is_set():
                 return
-            chat.mode = "match"
-            yield ("mode", chat.mode)
-            yield from _search(chat, matchmaker.find(chat.id), spoken)
-            if chat.matches:
-                chat.history.append({"role": "assistant", "content": FOLLOW_UP})
+            yield event
+    finally:
+        gen.close()
+        if chat.cancel is mine:
+            chat.cancel = None
+
+
+def _turn(chat: Chat, text: str, file, spoken: bool):
+    text = (text or "").strip()
+    label = ("🎤 " if spoken else "") + (text or ("📎 " + Path(file).name if file else ""))
+    chat.history.append({"role": "user", "content": label})
+    yield ("history", chat.history)
+    if chat.mode == "intake":
+        chat.history.append({"role": "assistant", "content": ""})
+        said, st = [], None
+        for kind, data in intake.send_stream(chat.id, text, file=file):
+            if kind == "say":
+                said.append(data)
+                chat.history[-1]["content"] = " ".join(said)
                 yield ("history", chat.history)
+                if spoken:
+                    yield ("say", data)
+            else:
+                st = data
+        chat.history[-1]["content"] = st["reply"]
+        chat.profile = {**st["profile"], "_missing": st["missing"], "_done": st["done"]}
+        yield ("history", chat.history)
+        yield ("profile", chat.profile)
+        if not st["done"]:
             return
-        yield from _search(chat, matchmaker.chat(chat.id, text), spoken)
+        chat.mode = "match"
+        yield ("mode", chat.mode)
+        yield from _search(chat, matchmaker.find(chat.id), spoken)
+        if chat.matches:
+            chat.history.append({"role": "assistant", "content": FOLLOW_UP})
+            yield ("history", chat.history)
+        return
+    yield from _search(chat, matchmaker.chat(chat.id, text), spoken)
 
 
 def spoken_turn(chat_id: str, audio_path: str):

scripts/chat_demo.py

@@ -130,7 +130,7 @@ async def voice_turn(chat_id: str, request: Request):
 VOICE_JS = r"""
 <script>
 (() => {
-  const SPEECH = 0.018, SPEECH_WHILE_PLAYING = 0.07, END_SILENCE_MS = 800, MIN_SPEECH_MS = 350, PREROLL_S = 0.5;
+  const SPEECH = 0.018, SPEECH_WHILE_PLAYING = 0.07, END_SILENCE_MS = 1300, MIN_SPEECH_MS = 350, PREROLL_S = 0.5;
   const S = {on: false, stream: null, ctx: null, proc: null, rate: 48000, pre: [], cur: null, inSpeech: false,
              speechMs: 0, silenceMs: 0, abort: null, queue: [], playing: null, busy: false};
   const el = id => document.getElementById(id);
@@ -198,7 +198,11 @@ VOICE_JS = r"""
     const rms = Math.sqrt(sum / frame.length), thr = S.playing ? SPEECH_WHILE_PLAYING : SPEECH;
     S.pre.push(frame); while (S.pre.length * frame.length > PREROLL_S * S.rate) S.pre.shift();
     if (rms > thr) {
-      if (!S.inSpeech) { S.inSpeech = true; S.speechMs = 0; S.cur = S.pre.slice(); stopPlayback(); status('Listening…'); }
+      if (!S.inSpeech) {
+        S.inSpeech = true; S.speechMs = 0; S.cur = S.pre.slice(); stopPlayback();
+        if (S.abort) { S.abort.abort(); S.abort = null; S.busy = false; }   // they're talking again: drop the old reply
+        status('Listening…');
+      }
       S.silenceMs = 0;
     } else if (S.inSpeech) { S.silenceMs += ms; }
     if (S.inSpeech) {

tests/test_conversation.py

@@ -39,3 +39,19 @@ def test_spoken_turn_reports_transcription_problems(monkeypatch):
     chat = conversation.start()
     assert list(conversation.spoken_turn(chat.id, "x.wav")) == [("error", voice.NOTHING_HEARD)]
     assert len(chat.history) == 1          # nothing added to the chat for an empty recording
+
+
+def test_new_message_cancels_the_reply_in_progress(monkeypatch):
+    import threading
+    fake_llm(monkeypatch, TURN)
+    chat = conversation.start()
+    first = conversation.turn(chat.id, "I'm Maha", spoken=True)
+    next(first)                                   # first turn is running (holding the chat)
+    done = []
+    t = threading.Thread(target=lambda: done.append(list(conversation.turn(chat.id, "raising 1.5M", spoken=True))))
+    t.start()
+    import time; time.sleep(0.2)                  # the second turn has asked the first to stop
+    rest = list(first)                            # first turn stops at its next step
+    t.join(timeout=5)
+    assert not t.is_alive() and done and rest == []
+    assert chat.history[-2]["content"] == "🎤 raising 1.5M"