fix(voice): preserve turns across activity handoffs

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

@@ -174,6 +174,16 @@ class _PreemptiveGeneration:
     created_at: float
 
 
+@dataclass
+class _PreparedUserTurn:
+    """A pipeline turn whose ``on_user_turn_completed`` callback has already run."""
+
+    info: _EndOfTurnInfo
+    user_message: llm.ChatMessage
+    chat_ctx: llm.ChatContext
+    on_user_turn_completed_delay: float
+
+
 @dataclass
 class _PausedSpeechInfo:
     handle: SpeechHandle
@@ -216,6 +226,7 @@ def __init__(self, agent: Agent, sess: AgentSession) -> None:
 
         self._scheduling_atask: asyncio.Task[None] | None = None
         self._user_turn_completed_atask: asyncio.Task[None] | None = None
+        self._handoff_user_turns: list[_EndOfTurnInfo | _PreparedUserTurn] = []
         self._speech_tasks: list[asyncio.Task[Any]] = []
 
         self._preemptive_generation: _PreemptiveGeneration | None = None
@@ -1130,6 +1141,59 @@ async def _start_session(self, *, reuse_resources: _ReusableResources | None = N
             if self.stt.capabilities.chat_context and forward_chat_ctx:
                 self._session.on("conversation_item_added", self.stt._push_conversation_item)
 
+        self._schedule_handoff_user_turns()
+
+    def _queue_handoff_user_turn(self, user_turn: _EndOfTurnInfo | _PreparedUserTurn) -> None:
+        """Keep an accepted pipeline turn until this activity's scheduler is running."""
+        self._handoff_user_turns.append(user_turn)
+
+    def _take_handoff_user_turns(self, source_activity: AgentActivity) -> None:
+        """Move queued turns from the activity being replaced to this activity."""
+        if source_activity._handoff_user_turns:
+            self._handoff_user_turns.extend(source_activity._handoff_user_turns)
+            source_activity._handoff_user_turns.clear()
+
+    def _persist_handoff_user_turns(self) -> None:
+        """Append unconsumed handoff turns using the session-close history semantics."""
+        handoff_user_turns = self._handoff_user_turns
+        self._handoff_user_turns = []
+        for user_turn in handoff_user_turns:
+            if isinstance(user_turn, _PreparedUserTurn):
+                # Its metrics were initialized from ``info`` before the source
+                # callback ran; preserve that finalized message verbatim.
+                user_message = user_turn.user_message
+            else:
+                info = user_turn
+                user_message = llm.ChatMessage(
+                    role="user",
+                    content=[info.new_transcript],
+                    transcript_confidence=info.transcript_confidence,
+                )
+                user_message.metrics = self._init_metrics_from_end_of_turn(info)
+
+            self._agent._chat_ctx.items.append(user_message)
+            self._session._conversation_item_added(user_message)
+
+    def _persist_prepared_user_turn(self, user_turn: _PreparedUserTurn) -> None:
+        """Commit an already-callbacked turn without recreating its message."""
+        self._agent._chat_ctx.items.append(user_turn.user_message)
+        self._session._conversation_item_added(user_turn.user_message)
+
+    def _schedule_handoff_user_turns(self) -> None:
+        """Continue turns accepted by the activity that handed off to this one."""
+        handoff_user_turns = self._handoff_user_turns
+        self._handoff_user_turns = []
+        for user_turn in handoff_user_turns:
+            old_task = self._user_turn_completed_atask
+            if isinstance(user_turn, _PreparedUserTurn):
+                task = self._prepared_user_turn_completed_task(old_task, user_turn)
+                task_name = "AgentActivity._prepared_user_turn_completed_task"
+            else:
+                task = self._user_turn_completed_task(old_task, user_turn)
+                task_name = "AgentActivity._user_turn_completed_task"
+
+            self._user_turn_completed_atask = self._create_speech_task(task, name=task_name)
+
     @tracer.start_as_current_span("drain_agent_activity")
     async def drain(
         self, *, new_activity: AgentActivity | None = None
@@ -1335,6 +1399,7 @@ async def aclose(self) -> None:
 
             self._closed = True
             self._cancel_preemptive_generation()
+            self._persist_handoff_user_turns()
             await self._session._keyterm_detector.aclose()
 
             # on_exit_task should be awaited in `drain`
@@ -2341,6 +2406,11 @@ def on_end_of_turn(self, info: _EndOfTurnInfo) -> bool:
                 extra={"user_input": info.new_transcript},
             )
 
+            if isinstance(self.llm, llm.LLM) and self._session._forward_handoff_user_turn(
+                self, info
+            ):
+                return True
+
             if self._session._closing:
                 # add user input to chat context
                 user_message = llm.ChatMessage(
@@ -2471,6 +2541,10 @@ async def _user_turn_completed_task(
                 "skipping on_user_turn_completed, speech scheduling is paused",
                 extra={"user_input": info.new_transcript},
             )
+            if isinstance(self.llm, llm.LLM) and self._session._forward_handoff_user_turn(
+                self, info
+            ):
+                return
             if self._session._closing:
                 self._agent._chat_ctx.items.append(user_message)
                 self._session._conversation_item_added(user_message)
@@ -2494,42 +2568,112 @@ async def _user_turn_completed_task(
         on_user_turn_completed_delay = time.perf_counter() - start_time
         metrics_report["on_user_turn_completed_delay"] = on_user_turn_completed_delay
 
-        if isinstance(self.llm, llm.RealtimeModel):
+        handoff_user_turn: _PreparedUserTurn | None = None
+        if isinstance(self.llm, llm.LLM):
+            handoff_user_turn = _PreparedUserTurn(
+                info=info,
+                user_message=user_message,
+                chat_ctx=temp_mutable_chat_ctx,
+                on_user_turn_completed_delay=on_user_turn_completed_delay,
+            )
+        elif isinstance(self.llm, llm.RealtimeModel):
             # ignore stt transcription for realtime model
             user_message = None  # type: ignore
         elif self.llm is None:
             return  # skip response if no llm is set
 
+        await self._schedule_reply_after_user_turn(
+            info=info,
+            user_message=user_message,
+            chat_ctx=temp_mutable_chat_ctx,
+            on_user_turn_completed_delay=on_user_turn_completed_delay,
+            handoff_user_turn=handoff_user_turn,
+        )
+
+    @utils.log_exceptions(logger=logger)
+    async def _prepared_user_turn_completed_task(
+        self,
+        old_task: asyncio.Task[None] | None,
+        user_turn: _PreparedUserTurn,
+    ) -> None:
+        """Resume a handed-off pipeline turn after its callback already completed."""
+        if old_task is not None:
+            await old_task
+
+        self._preemptive_generation_count = 0
+        await asyncio.gather(*self._interrupt_background_speeches(force=False))
+
+        # A prepared turn is deliberately pipeline-only. Realtime input ownership
+        # stays with its remote session; committing the local message is safer than
+        # replaying it through a different lifecycle.
+        if not isinstance(self.llm, llm.LLM):
+            self._persist_prepared_user_turn(user_turn)
+            return
+
+        if (current_speech := self._current_speech) is not None:
+            if not current_speech.allow_interruptions:
+                logger.warning(
+                    "skipping reply to handed-off user input, current speech generation cannot be interrupted",
+                    extra={"user_input": user_turn.user_message.raw_text_content},
+                )
+                self._persist_prepared_user_turn(user_turn)
+                return
+            await self._cancel_speech_pause(self._cancel_speech_pause_task)
+            await current_speech.interrupt()
+
+        await self._schedule_reply_after_user_turn(
+            info=user_turn.info,
+            user_message=user_turn.user_message,
+            chat_ctx=user_turn.chat_ctx,
+            on_user_turn_completed_delay=user_turn.on_user_turn_completed_delay,
+            handoff_user_turn=user_turn,
+        )
+
+    async def _schedule_reply_after_user_turn(
+        self,
+        *,
+        info: _EndOfTurnInfo,
+        user_message: llm.ChatMessage | None,
+        chat_ctx: llm.ChatContext,
+        on_user_turn_completed_delay: float,
+        handoff_user_turn: _PreparedUserTurn | None,
+    ) -> None:
+        """Schedule reply/history work after a completed user-turn callback."""
+
         if self._scheduling_paused or self._new_turns_blocked:
             logger.warning(
                 "skipping reply to user input, speech scheduling is paused",
                 extra={"user_input": info.new_transcript},
             )
-            if user_message and self._session._closing:
-                self._agent._chat_ctx.items.append(user_message)
-                self._session._conversation_item_added(user_message)
+            if handoff_user_turn and self._session._forward_handoff_user_turn(
+                self, handoff_user_turn
+            ):
+                return
+            if self._session._closing:
+                if handoff_user_turn:
+                    self._persist_prepared_user_turn(handoff_user_turn)
+                elif user_message:
+                    self._agent._chat_ctx.items.append(user_message)
+                    self._session._conversation_item_added(user_message)
             return
 
         speech_handle: SpeechHandle | None = None
         if preemptive := self._preemptive_generation:
-            # make sure the on_user_turn_completed didn't change some request parameters
-            # otherwise invalidate the preemptive generation
+            # make sure the callback didn't change some request parameters; otherwise
+            # invalidate the preemptive generation just as an ordinary pipeline turn does.
             if (
-                _transcripts_equivalent(
+                user_message is not None
+                and _transcripts_equivalent(
                     preemptive.info.new_transcript, user_message.raw_text_content
                 )
-                and preemptive.chat_ctx.is_equivalent(temp_mutable_chat_ctx)
+                and preemptive.chat_ctx.is_equivalent(chat_ctx)
                 and preemptive.tools == self.tools
                 and preemptive.tool_choice == self._tool_choice
             ):
                 speech_handle = preemptive.speech_handle
-
-                # The pipeline task retains the ChatMessage created for preemptive generation.
-                # Reconcile it with the finalized message before scheduling so conversation
-                # history keeps the final transcript and on_user_turn_completed edits.
                 preemptive.user_message.content = user_message.content.copy()
                 preemptive.user_message.transcript_confidence = user_message.transcript_confidence
-                preemptive.user_message.metrics = metrics_report
+                preemptive.user_message.metrics = user_message.metrics
                 self._schedule_speech(speech_handle, priority=SpeechHandle.SPEECH_PRIORITY_NORMAL)
                 logger.debug(
                     "using preemptive generation",
@@ -2545,19 +2689,15 @@ async def _user_turn_completed_task(
             self._preemptive_generation = None
 
         if speech_handle is None:
-            # Ensure the new message is passed to generate_reply
-            # This preserves the original message_id, making it easier for users to track responses
             speech_handle = self._generate_reply(
                 user_message=user_message,
-                chat_ctx=temp_mutable_chat_ctx,
+                chat_ctx=chat_ctx,
                 input_details=InputDetails(modality="audio"),
             )
 
         if self._user_turn_completed_atask != asyncio.current_task():
-            # If a new user turn has already started, interrupt this one since it's now outdated
-            # (We still create the SpeechHandle and the generate_reply coroutine, otherwise we may
-            # lose data like the beginning of a user speech).
-            # await the interrupt to make sure user message is added to the chat context before the new task starts
+            # A later accepted turn supersedes this one, but still let the pipeline
+            # commit its user message before it is interrupted.
             await speech_handle.interrupt()
 
         metadata: Metadata | None = None

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

@@ -47,7 +47,7 @@
 from . import io, room_io
 from ._utils import _set_participant_attributes
 from .agent import Agent, AgentTask
-from .agent_activity import AgentActivity, _ReusableResources
+from .agent_activity import AgentActivity, _PreparedUserTurn, _ReusableResources
 from .amd import AMD
 from .events import (
     AgentEvent,
@@ -91,6 +91,7 @@
     from ..cli.tcp_console import TcpAudioInput, TcpAudioOutput
     from ..inference import LLMModels, STTModels, TTSModels
     from ..llm import mcp
+    from .audio_recognition import _EndOfTurnInfo
     from .transcription.text_transforms import TextTransforms
 
 
@@ -1708,6 +1709,16 @@ async def _update_activity(
 
                 self._next_activity = agent._activity
 
+            next_activity = self._next_activity
+            assert next_activity is not None
+
+            # A turn can finish after update_agent() blocks its source activity
+            # but before this coroutine creates the replacement. The queue lives
+            # on that source (rather than on the session), so it has one clear
+            # owner and follows chained handoffs without retaining stale activities.
+            if (activity := self._activity) is not None:
+                next_activity._take_handoff_user_turns(activity)
+
             if self._root_span_context is not None:
                 # restore the root span context so on_exit, on_enter, and future turns
                 # are direct children of the root span, not nested under a tool call.
@@ -1718,27 +1729,28 @@ async def _update_activity(
                 previous_activity_v = self._activity
                 if (activity := self._activity) is not None:
                     if previous_activity == "close":
-                        reuse_resources = await activity.drain(new_activity=self._next_activity)
+                        reuse_resources = await activity.drain(new_activity=next_activity)
                         await activity.aclose()
                     elif previous_activity == "pause":
                         reuse_resources = await activity.pause(
                             blocked_tasks=blocked_tasks or [],
-                            new_activity=self._next_activity,
+                            new_activity=next_activity,
                         )
 
                 if self._closing and new_activity == "start":
                     # disallow starting a new activity when the session is closing
                     logger.warning(
-                        f"session is closing, skipping {new_activity} activity of {self._next_activity.agent.id}",
+                        f"session is closing, skipping {new_activity} activity of {next_activity.agent.id}",
                     )
                     if reuse_resources is not None:
                         await reuse_resources.cleanup()
                         reuse_resources = None
+                    next_activity._persist_handoff_user_turns()
                     self._next_activity = None
                     self._activity = None
                     return
 
-                self._activity = self._next_activity
+                self._activity = next_activity
                 self._next_activity = None
 
                 run_state = self._global_run_state
@@ -1763,6 +1775,9 @@ async def _update_activity(
                 elif new_activity == "resume":
                     await self._activity.resume(reuse_resources=reuse_resources)
             except BaseException:
+                # The successor cannot consume these accepted turns, so preserve
+                # them with the same history semantics as session close.
+                next_activity._persist_handoff_user_turns()
                 if reuse_resources is not None:
                     await reuse_resources.cleanup()
                 raise
@@ -1772,6 +1787,34 @@ async def _update_activity(
             assert self._activity._on_enter_task is not None
             await asyncio.shield(self._activity._on_enter_task)
 
+    def _forward_handoff_user_turn(
+        self,
+        source_activity: AgentActivity,
+        user_turn: _EndOfTurnInfo | _PreparedUserTurn,
+    ) -> bool:
+        """Route an accepted pipeline turn across an activity handoff.
+
+        Callers intentionally restrict this to ``llm.LLM``: pipeline turns use
+        this local ChatMessage/speech-scheduler lifecycle, while Realtime models
+        commit their input through the remote session instead.
+
+        This is intentionally limited to an activity transition. A manually
+        drained activity still has no successor, so it keeps its existing
+        behaviour instead of retaining a turn indefinitely.
+        """
+        if self._closing or self._activity is not source_activity:
+            return False
+
+        if self._next_activity is not None:
+            self._next_activity._queue_handoff_user_turn(user_turn)
+            return True
+
+        if self._update_activity_atask is not None and not self._update_activity_atask.done():
+            source_activity._queue_handoff_user_turn(user_turn)
+            return True
+
+        return False
+
     @utils.log_exceptions(logger=logger)
     async def _update_activity_task(
         self, old_task: asyncio.Task[None] | None, agent: Agent