refactor(tts): single persistent track + queue playback + gen-based interrupt

- One LocalAudioTrack per session (not per segment)
- asyncio.Queue for sequential audio playback
- Generation counter for turn-based interruption
- Drain queue on new speech start
- Remove old multi-track _play method

asr_agent/conversation_worker.py

@@ -512,10 +512,37 @@ class Worker:
         sr = 16000
         vad = _SileroVAD()
         busy = False
-        play_tasks: list[asyncio.Task] = []
+
+        # ── Single persistent TTS output track per session ──
+        # One track is reused for all turns so audio never cross-streams
+        # between turns/segments (old code published a new track each call).
+        tts_src = rtc.AudioSource(24000, 1)
+        tts_track = rtc.LocalAudioTrack.create_audio_track("tts", tts_src)
+        await self.room.local_participant.publish_track(tts_track)
+        play_q: "asyncio.Queue[tuple[int, np.ndarray]]" = asyncio.Queue()
+        gen = 0                 # turn generation; bumped to flush in-flight audio
+        cur_gen = {"v": -1}    # generation currently being played by _player
+
+        async def _player():
+            """Drain TTS audio sequentially on the single track."""
+            while True:
+                g, audio = await play_q.get()
+                cur_gen["v"] = g
+                for i in range(0, len(audio), 24000 // 50):
+                    if cur_gen["v"] != g:
+                        break  # a newer turn started; drop this stale audio
+                    c = audio[i: i + 24000 // 50]
+                    await tts_src.capture_frame(rtc.AudioFrame(
+                        data=c.tobytes(), sample_rate=24000,
+                        num_channels=1, samples_per_channel=len(c),
+                    ))
+                    await asyncio.sleep(0.018)
+                cur_gen["v"] = -1
+
+        player_task = asyncio.create_task(_player())
 
         async def _acc():
-            nonlocal busy, play_tasks
+            nonlocal busy, gen
             stream = rtc.AudioStream(track, sample_rate=sr, num_channels=1)
             fc = 0
             async for ev in stream:
@@ -535,13 +562,12 @@ class Worker:
 
                 if was_speech and vad.should_end() and vad.min_speech_met():
                     logger.info("[%s] VAD END blen=%d", sid, len(buf.buffer))
-                    for pt in play_tasks:
-                        if not pt.done():
-                            pt.cancel()
-                    play_tasks.clear()
+                    gen += 1
+                    cur_gen["v"] = -1  # stop current playback immediately
+                    _drain_queue(play_q)
                     busy = True
                     try:
-                        await self._transcribe_and_respond(buf, play_tasks)
+                        await self._transcribe_and_respond(buf, play_q, gen)
                     except Exception:
                         logger.exception("[%s] transcribe failed", sid)
                     busy = False
@@ -550,13 +576,12 @@ class Worker:
 
                 if vad.in_speech and vad.total_samples >= int(MAX_SPEECH_S * sr):
                     logger.info("[%s] VAD MAX blen=%d", sid, len(buf.buffer))
-                    for pt in play_tasks:
-                        if not pt.done():
-                            pt.cancel()
-                    play_tasks.clear()
+                    gen += 1
+                    cur_gen["v"] = -1
+                    _drain_queue(play_q)
                     busy = True
                     try:
-                        await self._transcribe_and_respond(buf, play_tasks)
+                        await self._transcribe_and_respond(buf, play_q, gen)
                     except Exception:
                         logger.exception("[%s] max failed", sid)
                     busy = False
@@ -571,16 +596,19 @@ class Worker:
             pass
         except Exception:
             logger.exception("[%s] _acc failed", sid)
+        finally:
+            player_task.cancel()
+            try:
+                await self.room.local_participant.unpublish_track(tts_track)
+            except Exception:
+                pass
 
-    async def _transcribe_and_respond(self, buf: AudioBuffer, play_tasks: list[asyncio.Task] | None = None):
-        if play_tasks is None:
-            play_tasks = []
+    async def _transcribe_and_respond(self, buf: AudioBuffer, play_q, gen: int):
         all_audio = buf.get_all()
         buf.clear()
         if len(all_audio) <= 1600:
             return
 
-        loop = asyncio.get_running_loop()
         txt = await _asr_transcribe(all_audio, self.asr)
         if not txt:
             return
@@ -588,7 +616,9 @@ class Worker:
         logger.info("ASR: %s", txt[:80])
         await self._send({"type": "utterance", "text": txt, "seq": 0})
 
-        # ── Streaming LLM + concurrent TTS playback ──
+        # ── Streaming LLM + concurrent TTS generation (sequential playback) ──
+        # TTS is generated concurrently per segment, but the player consumes the
+        # queue in order on a single track, so segments never overlap.
         reply_full = ""
         seg = _TTSSegmenter()
 
@@ -599,12 +629,12 @@ class Worker:
                     await self._send({"type": "reply_partial", "text": reply_full, "seq": 0})
                     for s in seg.feed(delta):
                         logger.info("TTS seg(%d): %s", len(s), s[:40])
-                        play_tasks.append(asyncio.create_task(self._tts_and_play(s)))
+                        asyncio.create_task(self._tts_enqueue(s, play_q, gen))
 
                 if is_final:
                     remaining = seg.flush()
                     if remaining:
-                        play_tasks.append(asyncio.create_task(self._tts_and_play(remaining)))
+                        asyncio.create_task(self._tts_enqueue(remaining, play_q, gen))
                     break
         except Exception:
             logger.exception("LLM stream failed")
@@ -618,8 +648,8 @@ class Worker:
             ])
             self.hist[:] = self.hist[-20:]
 
-    async def _tts_and_play(self, text: str):
-        """Stream TTS audio and play chunks as they arrive."""
+    async def _tts_enqueue(self, text: str, play_q, gen: int):
+        """Stream TTS audio for one segment, then enqueue it for playback."""
         try:
             chunks: list[np.ndarray] = []
             async for pcm, sr, is_final in _tts_stream(text):
@@ -630,24 +660,20 @@ class Worker:
             if chunks:
                 audio = np.concatenate(chunks)
                 if len(audio) > 0:
-                    await self._play(audio, 24000)
-        except (asyncio.CancelledError, asyncio.TimeoutError):
+                    await play_q.put((gen, audio))
+        except asyncio.CancelledError:
             pass
         except Exception:
-            logger.exception("TTS+play failed")
-
-    async def _play(self, audio: np.ndarray, sr: int):
-        src = rtc.AudioSource(sr, 1)
-        tk = rtc.LocalAudioTrack.create_audio_track("tts", src)
-        await self.room.local_participant.publish_track(tk)
-        for i in range(0, len(audio), sr // 50):
-            c = audio[i: i + sr // 50]
-            await src.capture_frame(rtc.AudioFrame(
-                data=c.tobytes(), sample_rate=sr,
-                num_channels=1, samples_per_channel=len(c),
-            ))
-            await asyncio.sleep(0.018)
-        return tk
+            logger.exception("TTS failed")
+
+
+def _drain_queue(q: "asyncio.Queue") -> None:
+    """Discard any queued-but-not-yet-played TTS audio."""
+    while not q.empty():
+        try:
+            q.get_nowait()
+        except Exception:
+            break
 
     async def _send(self, msg: dict):
         try: