fix(voice): don't duplicate queued speeches on scheduling tie

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -4,6 +4,7 @@
 import contextlib
 import contextvars
 import heapq
+import itertools
 import json
 import time
 from collections.abc import AsyncGenerator, AsyncIterable, Coroutine, Iterator
@@ -344,7 +345,11 @@ def __init__(self, agent: Agent, sess: AgentSession) -> None:
         self._new_turns_blocked = False
 
         self._current_speech: SpeechHandle | None = None
-        self._speech_q: list[tuple[int, float, SpeechHandle]] = []
+        # (-priority, seq, speech): `seq` is strictly increasing, so two entries never
+        # tie on the first two elements and the heap never falls through to comparing
+        # SpeechHandles (which are not orderable)
+        self._speech_q: list[tuple[int, int, SpeechHandle]] = []
+        self._speech_seq: Iterator[int] = itertools.count()
         self._user_silence_event: asyncio.Event = asyncio.Event()
         self._user_silence_event.set()
 
@@ -1983,15 +1988,9 @@ def _schedule_speech(self, speech: SpeechHandle, priority: int, force: bool = Fa
             speech.interrupt(force=True)
             return
 
-        while True:
-            try:
-                # negate the priority to make it a max heap
-                heapq.heappush(self._speech_q, (-priority, time.perf_counter_ns(), speech))
-                break
-            except TypeError:
-                # handle TypeError when identical timestamps cause speech comparison failure
-                # with perf_counter_ns(), collisions should be rare
-                pass
+        # negate the priority to make it a max heap; the sequence number breaks ties
+        # in scheduling order
+        heapq.heappush(self._speech_q, (-priority, next(self._speech_seq), speech))
 
         speech._mark_scheduled()
         self._wake_up_scheduling_task()

tests/test_speech_queue.py

@@ -0,0 +1,48 @@
+"""``AgentActivity._schedule_speech`` must enqueue exactly one entry per speech.
+
+The queue is a heap of ``(-priority, timestamp, speech)`` tuples. Two entries whose
+priority and timestamp tie fall through to comparing the ``SpeechHandle`` objects,
+which are not orderable — ``heapq.heappush`` raises ``TypeError`` after it has
+already appended the item. The retry loop then pushes a *second* copy of the same
+speech, so the queue holds one duplicate and can lose its heap invariant.
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from livekit.agents.voice.agent_activity import AgentActivity
+from livekit.agents.voice.speech_handle import SpeechHandle
+
+from .fake_session import FakeActions, create_session
+from .test_agent_session import MyAgent, _close_test_session
+
+pytestmark = pytest.mark.unit
+
+
+def _make_activity() -> AgentActivity:
+    return AgentActivity(MyAgent(), create_session(FakeActions()))
+
+
+async def test_tied_timestamps_enqueue_each_speech_once(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A clock whose reads tie must not duplicate entries in the speech queue."""
+    activity = _make_activity()
+    activity._scheduling_paused = False
+    # a coarse perf clock (e.g. Windows QPC) can return the same value for
+    # back-to-back scheduling calls
+    monkeypatch.setattr(time, "perf_counter_ns", lambda: 1_000)
+    try:
+        first = SpeechHandle.create()
+        second = SpeechHandle.create()
+        activity._schedule_speech(first, priority=SpeechHandle.SPEECH_PRIORITY_NORMAL)
+        activity._schedule_speech(second, priority=SpeechHandle.SPEECH_PRIORITY_NORMAL)
+
+        queued = [speech for _, _, speech in activity._speech_q]
+        assert len(queued) == 2, f"expected one queue entry per speech, got {queued}"
+        assert sorted(queued, key=id) == sorted([first, second], key=id)
+    finally:
+        await _close_test_session(activity._session)