fix(amd): synthesize the speech start when the end arrives without one

livekit-agents/livekit/agents/voice/amd/classifier.py

@@ -139,6 +139,7 @@ def __init__(
 
         self._llm = llm
         self._prompt = prompt
+        self._listening_started_at: float | None = None
         self._speech_started_at: float | None = None
         self._speech_ended_at: float | None = None
         self._speech_active = False
@@ -174,6 +175,7 @@ def start_listening(self) -> None:
         if self._closed or self._emitted or self._listening:
             return
         self._listening = True
+        self._listening_started_at = time.time()
         if self._no_speech_timer is None:
             self._no_speech_timer = asyncio.get_running_loop().call_later(
                 self._no_speech_threshold,
@@ -204,11 +206,32 @@ def on_user_speech_started(self) -> None:
 
     @_listening_guard
     def on_user_speech_ended(self, silence_duration: float) -> None:
+        speech_ended_at = time.time() - silence_duration
         if self._speech_started_at is None:
-            logger.warning("on_user_speech_ended called before on_user_speech_started")
-            return
+            if self._listening_started_at is None or speech_ended_at <= self._listening_started_at:
+                # the segment started AND ended before the gate opened: this is
+                # pre-answer audio (ringback, early media) the gate is
+                # documented to drop - nothing was heard while listening, so
+                # keep the no-speech timer armed instead of committing a
+                # verdict from zero observed speech
+                logger.debug("dropping user speech that ended before listening began")
+                return
+            # the speech began before listening started but overlaps the
+            # listening window (e.g. AMD attached after the callee was already
+            # mid-greeting, so the start event was dropped by the listening
+            # gate). Synthesize the start from the moment the gate opened
+            # instead of dropping the end signal - previously the classifier
+            # stalled with no timers armed until detection_timeout (#5616),
+            # holding back playout for the full timeout budget.
+            logger.debug(
+                "user speech ended without a start signal; assuming speech since listening began"
+            )
+            if self._no_speech_timer is not None:
+                self._no_speech_timer.cancel()
+                self._no_speech_timer = None
+            self._speech_started_at = self._listening_started_at
 
-        self._speech_ended_at = time.time() - silence_duration
+        self._speech_ended_at = speech_ended_at
         speech_duration = self._speech_ended_at - self._speech_started_at
         self._speech_active = False
         self._arm_eot_timer(delay=max(0, self._max_endpointing_delay - silence_duration))

tests/test_amd_classifier.py

@@ -773,3 +773,98 @@ async def fake_wait_for_track_publication(**_: object) -> SimpleNamespace:
         await detector._setup(session)  # type: ignore[arg-type]
 
         assert calls == 2
+
+
+class TestEndBeforeStart:
+    """Regression tests for #5616: speech that began before listening opened.
+
+    When AMD attaches after the callee has already started speaking (outbound
+    calls answered mid-greeting), the start event is dropped by the listening
+    gate; the later end event used to be discarded too, leaving no timers armed
+    and stalling the verdict until ``detection_timeout`` (7-16s+ of held
+    playout in the report).
+    """
+
+    async def test_end_without_start_synthesizes_start_and_emits(self) -> None:
+        clf = _make_classifier(human_silence_threshold=0.1)
+        clf.start_listening()
+        results: list[AMDPredictionEvent] = []
+        clf.on("amd_prediction", results.append)
+
+        await asyncio.sleep(0.05)
+        # no on_user_speech_started: it fired before the gate opened
+        clf.on_user_speech_ended(silence_duration=0.0)
+
+        # the start is synthesized from the listening gate opening
+        assert clf._speech_started_at is not None
+        assert clf._speech_started_at >= clf._listening_started_at
+        # the verdict path is armed instead of stalling until detection_timeout
+        assert clf._silence_timer is not None
+        assert clf._silence_timer_trigger == "short_speech"
+        # speech clearly happened; the no-speech timer must not fire later
+        assert clf._no_speech_timer is None
+
+        await asyncio.sleep(0.2)
+
+        assert len(results) == 1
+        assert results[0].category == AMDCategory.HUMAN
+        assert results[0].reason == "short_greeting"
+
+    async def test_speech_that_ended_before_listening_is_dropped(self) -> None:
+        # ringback/early-media audio can trip the session VAD before the gate
+        # opens; when the end lands just after start_listening() but the speech
+        # is entirely in the past, it must be dropped like its start was - not
+        # synthesized into a zero-length greeting and an instant HUMAN verdict
+        clf = _make_classifier(human_silence_threshold=0.1)
+        clf.start_listening()
+        results: list[AMDPredictionEvent] = []
+        clf.on("amd_prediction", results.append)
+
+        # the VAD end reports the speech stopped 0.5s ago - before the gate opened
+        clf.on_user_speech_ended(silence_duration=0.5)
+
+        assert clf._speech_started_at is None
+        assert clf._speech_ended_at is None
+        assert clf._silence_timer is None
+        # nothing was heard while listening: keep waiting for real speech
+        assert clf._no_speech_timer is not None
+
+        await asyncio.sleep(0.2)
+        assert results == []
+
+    async def test_synthesized_duration_is_positive_on_overlap(self) -> None:
+        # when the speech genuinely overlaps the listening window the start is
+        # synthesized from the gate opening, so the duration stays positive
+        clf = _make_classifier(human_silence_threshold=0.1)
+        clf.start_listening()
+
+        await asyncio.sleep(0.05)
+        clf.on_user_speech_ended(silence_duration=0.0)
+
+        assert clf._speech_started_at is not None and clf._speech_ended_at is not None
+        assert clf._speech_started_at == clf._listening_started_at
+        assert clf._speech_ended_at > clf._speech_started_at
+        assert clf.speech_duration > 0.0
+
+    async def test_end_before_listening_stays_gated(self) -> None:
+        # before start_listening() everything is a no-op, as documented
+        clf = _make_classifier()
+        clf.on_user_speech_ended(silence_duration=0.0)
+        assert clf._speech_started_at is None
+        assert clf._silence_timer is None
+
+    async def test_normal_ordering_unchanged(self) -> None:
+        clf = _make_classifier(human_silence_threshold=0.1)
+        clf.start_listening()
+        results: list[AMDPredictionEvent] = []
+        clf.on("amd_prediction", results.append)
+
+        clf.on_user_speech_started()
+        started_at = clf._speech_started_at
+        await asyncio.sleep(0.05)
+        clf.on_user_speech_ended(silence_duration=0.0)
+
+        # the real start timestamp is kept, not overwritten by synthesis
+        assert clf._speech_started_at == started_at
+        await asyncio.sleep(0.2)
+        assert len(results) == 1