feat(voice): add user_away_signal for transcript-based away tracking

livekit-agents/livekit/agents/voice/agent_session.py

@@ -235,6 +235,8 @@ class ExpressiveOptions(TypedDict, total=False):
     speech_steering=DEFAULT_SPEECH_STEERING_OPTIONS,
 )
 
+UserAwaySignal = Literal["audio", "transcript"]
+
 
 def _append_instructions(template: Instructions | str, extra: str) -> Instructions:
     # concatenate the *raw* template text so any {placeholders} survive until render()
@@ -285,6 +287,7 @@ class AgentSessionOptions:
     """sparse endpointing keys the user provided explicitly"""
     max_tool_steps: int
     user_away_timeout: float | None
+    user_away_signal: UserAwaySignal
     transcription_timeout: float | None
     min_consecutive_speech_delay: float
     use_tts_aligned_transcript: bool | None
@@ -389,6 +392,7 @@ def __init__(
         aec_warmup_duration: NotGivenOr[float | None] = NOT_GIVEN,
         ivr_detection: bool = False,
         user_away_timeout: float | None = 15.0,
+        user_away_signal: UserAwaySignal = "audio",
         transcription_timeout: float | None = None,
         session_close_transcript_timeout: float = 2.0,
         # Runtime settings
@@ -478,6 +482,10 @@ def __init__(
             user_away_timeout (float, optional): If set, set the user state as
                 "away" after this amount of time after user and agent are silent.
                 Defaults to ``15.0`` s, set to ``None`` to disable.
+            user_away_signal (Literal["audio", "transcript"], optional): Which user
+                activity holds off the "away" state. ``"audio"`` (default) trusts any
+                detected speech. ``"transcript"`` trusts only transcribed speech, so
+                noise that never becomes text is ignored; it needs a streaming STT.
             transcription_timeout (float, optional): If set, emit a
                 ``user_transcription_timeout`` event when VAD detects user speech
                 during the user's turn but no non-empty final transcript arrives
@@ -568,6 +576,7 @@ def __init__(
             endpointing_overrides=endpointing_overrides,
             max_tool_steps=max_tool_steps,
             user_away_timeout=user_away_timeout,
+            user_away_signal=user_away_signal,
             transcription_timeout=transcription_timeout,
             min_consecutive_speech_delay=min_consecutive_speech_delay,
             tts_text_transforms=(
@@ -1949,7 +1958,11 @@ def _update_agent_state(
                 self._aec_warmup_remaining,
             )
 
-        if state == "listening" and self._user_state == "listening":
+        # the transcript signal ignores speech activity, so noise must not block the re-arm
+        user_idle = self._user_state == "listening" or (
+            self._opts.user_away_signal == "transcript" and self._user_state == "speaking"
+        )
+        if state == "listening" and user_idle:
             self._set_user_away_timer()
         else:
             self._cancel_user_away_timer()
@@ -1962,7 +1975,11 @@ def _update_agent_state(
         )
 
     def _update_user_state(
-        self, state: UserState, *, last_speaking_time: float | None = None
+        self,
+        state: UserState,
+        *,
+        last_speaking_time: float | None = None,
+        by_transcript: bool = False,
     ) -> None:
         # pinned to "speaking" while a `claim_user_turn` is active; voice
         # transitions are recoverable from `_user_silence_event` on release
@@ -1972,6 +1989,14 @@ def _update_user_state(
         if self._user_state == state:
             return
 
+        if (
+            self._opts.user_away_signal == "transcript"
+            and self._user_state == "away"
+            and not by_transcript
+        ):
+            # only a transcript ends "away"; noise would otherwise clear it at once
+            return
+
         last_speaking_time_ns = (
             int(last_speaking_time * 1_000_000_000) if last_speaking_time else None
         )
@@ -1993,10 +2018,11 @@ def _update_user_state(
             self._user_speaking_span.end(end_time=last_speaking_time_ns)
             self._user_speaking_span = None
 
-        if state == "listening" and self._agent_state == "listening":
-            self._set_user_away_timer()
-        else:
-            self._cancel_user_away_timer()
+        if self._opts.user_away_signal == "audio":
+            if state == "listening" and self._agent_state == "listening":
+                self._set_user_away_timer()
+            else:
+                self._cancel_user_away_timer()
 
         old_state = self._user_state
         self._user_state = state
@@ -2022,7 +2048,18 @@ def _user_input_transcribed(self, ev: UserInputTranscribedEvent) -> None:
             # a transcript means stt recovered; reset its error tolerance
             self._stt_error_counts = 0
 
-        if ev.is_final and self.user_state != "speaking":
+        if self._opts.user_away_signal == "transcript":
+            if ev.transcript:
+                # interims count: a long answer must not trip "away" before its final
+                if self.user_state == "away":
+                    # "away" swallowed this turn's speech start, so restore what the
+                    # detector sees: an interim still has a speech end coming, a final may not
+                    self._update_user_state(
+                        "listening" if ev.is_final else "speaking", by_transcript=True
+                    )
+                if self._agent_state == "listening":
+                    self._set_user_away_timer()
+        elif ev.is_final and self.user_state != "speaking":
             if self.user_state == "away":
                 # reset user state from away to listening in case VAD has a miss detection
                 self._update_user_state("listening")

livekit-agents/livekit/agents/voice/remote_session.py

@@ -409,6 +409,7 @@ def _serialize_options(opts: AgentSessionOptions) -> dict[str, str]:
         "interruption": str(dict(opts.interruption)),
         "max_tool_steps": str(opts.max_tool_steps),
         "user_away_timeout": str(opts.user_away_timeout),
+        "user_away_signal": opts.user_away_signal,
         "transcription_timeout": str(opts.transcription_timeout),
         "preemptive_generation": str(dict(opts.preemptive_generation)),
         "min_consecutive_speech_delay": str(opts.min_consecutive_speech_delay),

livekit-agents/livekit/agents/voice/report.py

@@ -61,6 +61,7 @@ def to_dict(self) -> dict:
                 "max_endpointing_delay": self.options.endpointing["max_delay"],
                 "max_tool_steps": self.options.max_tool_steps,
                 "user_away_timeout": self.options.user_away_timeout,
+                "user_away_signal": self.options.user_away_signal,
                 "min_consecutive_speech_delay": self.options.min_consecutive_speech_delay,
                 "preemptive_generation": dict(self.options.preemptive_generation),
                 "recording_options": dict(self.options.recording_options),

tests/test_agent_session.py

@@ -1337,6 +1337,158 @@ async def test_final_transcript_resets_away_timer_when_not_speaking() -> None:
         await _close_test_session(session)
 
 
+def _transcript_away_signal_session(timeout: float) -> AgentSession:
+    session = create_session(
+        FakeActions(),
+        extra_kwargs={"user_away_timeout": timeout, "user_away_signal": "transcript"},
+    )
+    session._agent_state = "listening"
+    session._user_state = "listening"
+    session._set_user_away_timer()
+    return session
+
+
+async def test_audio_away_signal_defers_away_on_untranscribed_speech() -> None:
+    """Default signal: any detected speech re-arms the full window."""
+    session = create_session(FakeActions(), extra_kwargs={"user_away_timeout": 0.3})
+    session._agent_state = "listening"
+    session._user_state = "listening"
+    session._set_user_away_timer()
+    try:
+        await asyncio.sleep(0.2)
+        session._update_user_state("speaking")
+        session._update_user_state("listening")
+
+        await asyncio.sleep(0.2)
+        assert session.user_state == "listening"
+
+        await asyncio.sleep(0.2)
+        assert session.user_state == "away"
+    finally:
+        await _close_test_session(session)
+
+
+async def test_transcript_away_signal_ignores_untranscribed_speech() -> None:
+    """Noise that produces no transcript must not hold off "away" (#6030)."""
+    session = _transcript_away_signal_session(0.3)
+    try:
+        # noise flips, then noise that leaves the user stuck in "speaking"
+        session._update_user_state("speaking")
+        session._update_user_state("listening")
+        session._update_user_state("speaking")
+
+        await asyncio.sleep(0.5)
+        assert session.user_state == "away"
+    finally:
+        await _close_test_session(session)
+
+
+async def test_transcript_away_signal_interim_holds_off_away() -> None:
+    """A long answer must not trip "away" before the final transcript arrives."""
+    states: list[str] = []
+    session = _transcript_away_signal_session(0.3)
+    session.on("user_state_changed", lambda ev: states.append(ev.new_state))
+    try:
+        session._update_user_state("speaking")
+        for _ in range(4):
+            await asyncio.sleep(0.15)
+            session._user_input_transcribed(
+                UserInputTranscribedEvent(transcript="still answering", is_final=False)
+            )
+
+        assert states == ["speaking"]
+    finally:
+        await _close_test_session(session)
+
+
+async def test_transcript_away_signal_away_survives_noise() -> None:
+    """Once away, speech activity alone must not bring the user back."""
+    session = _transcript_away_signal_session(0.2)
+    try:
+        await asyncio.sleep(0.3)
+        assert session.user_state == "away"
+
+        session._update_user_state("speaking")
+        assert session.user_state == "away"
+        session._update_user_state("listening")
+        assert session.user_state == "away"
+    finally:
+        await _close_test_session(session)
+
+
+async def test_transcript_away_signal_interim_ends_away_as_speaking() -> None:
+    """An interim means the user is mid-turn, so end "away" into "speaking"."""
+    session = _transcript_away_signal_session(0.2)
+    try:
+        await asyncio.sleep(0.3)
+        assert session.user_state == "away"
+
+        session._user_input_transcribed(
+            UserInputTranscribedEvent(transcript="i'm here", is_final=False)
+        )
+        assert session.user_state == "speaking"
+        assert session._user_away_timer is not None
+
+        # the speech end that closes this turn is still to come
+        session._update_user_state("listening")
+        assert session.user_state == "listening"
+    finally:
+        await _close_test_session(session)
+
+
+async def test_transcript_away_signal_final_ends_away_as_listening() -> None:
+    """A final may land after the speech end was swallowed, so never strand "speaking"."""
+    session = _transcript_away_signal_session(0.2)
+    try:
+        await asyncio.sleep(0.3)
+        assert session.user_state == "away"
+
+        session._user_input_transcribed(
+            UserInputTranscribedEvent(transcript="i'm here", is_final=True)
+        )
+        assert session.user_state == "listening"
+        assert session._user_away_timer is not None
+    finally:
+        await _close_test_session(session)
+
+
+async def test_transcript_away_signal_away_recovery_waits_for_agent() -> None:
+    """Leaving away while the agent talks must not arm the timer."""
+    session = _transcript_away_signal_session(15.0)
+    try:
+        session._update_user_state("away")
+        session._update_agent_state("speaking")
+
+        session._user_input_transcribed(
+            UserInputTranscribedEvent(transcript="i'm back", is_final=True)
+        )
+        assert session.user_state == "listening"
+        assert session._user_away_timer is None
+
+        session._update_agent_state("listening")
+        assert session._user_away_timer is not None
+    finally:
+        await _close_test_session(session)
+
+
+async def test_transcript_away_signal_rearms_after_agent_turn_during_noise() -> None:
+    """The countdown resumes when the agent idles, even if noise still says speaking."""
+    session = _transcript_away_signal_session(0.3)
+    try:
+        session._update_user_state("speaking")
+
+        session._update_agent_state("speaking")
+        assert session._user_away_timer is None
+
+        session._update_agent_state("listening")
+        assert session._user_away_timer is not None
+
+        await asyncio.sleep(0.5)
+        assert session.user_state == "away"
+    finally:
+        await _close_test_session(session)
+
+
 async def test_stt_error_count_resets_on_user_transcript() -> None:
     from livekit.agents.voice.agent_session import SessionConnectOptions
 

tests/test_remote_session.py

@@ -86,6 +86,7 @@ def _make_mock_session() -> MagicMock:
     options.interruption = MagicMock(__iter__=lambda s: iter([]))
     options.max_tool_steps = 5
     options.user_away_timeout = 30
+    options.user_away_signal = "audio"
     options.preemptive_generation = MagicMock(__iter__=lambda s: iter([]))
     options.min_consecutive_speech_delay = 0.5
     options.use_tts_aligned_transcript = True

tests/test_session_host.py

@@ -542,6 +542,7 @@ def _make_mock_session(self) -> MagicMock:
         options.interruption = MagicMock(__iter__=lambda s: iter([]))
         options.max_tool_steps = 5
         options.user_away_timeout = 30
+        options.user_away_signal = "audio"
         options.preemptive_generation = {"enabled": False}
         options.min_consecutive_speech_delay = 0.5
         options.use_tts_aligned_transcript = True