livekit · dorukdumlu · Sep 15, 2026 · Sep 15, 2026
@@ -315,7 +315,7 @@ async def start(
         # and nothing is lost. wait_remote_track buffers until the video track
         # appears; replace_audio_tail keeps the TranscriptSynchronizer /
         # RecorderAudioOutput chain intact.
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -17,6 +17,7 @@
 
 if TYPE_CHECKING:
     from ..agent_session import AgentSession
+    from ..io import AudioOutput
 
 
 class AudioSegmentEnd:
@@ -72,6 +73,10 @@ def __init__(self) -> None:
         self._wait_avatar_join_task: asyncio.Task[None] | None = None
         self._room: rtc.Room | None = None
         self._agent_session: AgentSession | None = None
+        # audio route installed via _attach_audio_output, and what it replaced,
+        # so aclose can put the previous route back
+        self._installed_audio_output: AudioOutput | None = None
+        self._previous_audio_output: AudioOutput | None = None
 
     @property
     @abstractmethod
@@ -103,6 +108,40 @@ async def start(self, agent_session: AgentSession, room: rtc.Room) -> None:
         else:
             self._room.on("connection_state_changed", self._on_connection_state_changed)
 
+    def _attach_audio_output(self, sink: AudioOutput) -> None:
+        """Route the agent session's audio to the avatar, remembering the previous route.
+
+        Plugins call this from ``start()`` instead of
+        ``agent_session.output.replace_audio_tail(...)`` so that :meth:`aclose`
+        can restore the previous route: a failed avatar start then degrades back
+        to regular audio instead of leaving the agent silent.
+        """
+        assert self._agent_session is not None, "_attach_audio_output requires start() first"
+        self._previous_audio_output = self._agent_session.output.replace_audio_tail(sink)
+        self._installed_audio_output = sink
+
+    def _restore_audio_output(self) -> None:
+        """Undo :meth:`_attach_audio_output`, unless something replaced our sink since."""
+        installed, previous = self._installed_audio_output, self._previous_audio_output
+        self._installed_audio_output = None
+        self._previous_audio_output = None
+        if installed is None or self._agent_session is None:
+            return
+
+        output = self._agent_session.output
+        if output.audio_tail is not installed:
+            # someone re-routed the audio after us; their route wins
+            return
+
+        if previous is not None:
+            output.replace_audio_tail(previous)
+        elif output.audio is installed:
+            # the avatar's sink was the whole chain (installed before the session
+            # started); clear it so a later session.start() sets up room audio
+            output.audio = None
+        # else: a wrapper chain was built on top of the avatar's sink and there is
+        # no earlier route to go back to; leave it in place
+
     async def wait_for_join(self, *, timeout: float | None = 30.0) -> None:
         """Wait until the avatar participant has joined the room and
         published its video track.
@@ -119,6 +158,10 @@ async def wait_for_join(self, *, timeout: float | None = 30.0) -> None:
         await asyncio.wait_for(asyncio.shield(self._wait_avatar_join_task), timeout=timeout)
 
     async def aclose(self) -> None:
+        # first, and before any await: a cancellation mid-close (job-shutdown
+        # deadline) must not leave agent audio routed to the closing avatar
+        self._restore_audio_output()
+
         if self._room is not None and self._room.isconnected():
             job_ctx = get_job_context(required=False)
             if job_ctx is not None:

@@ -783,7 +783,22 @@ def audio(self, sink: AudioOutput | None) -> None:
             else:
                 self._audio_sink.on_detached()
 
-    def replace_audio_tail(self, sink: AudioOutput) -> None:
+    @property
+    def audio_tail(self) -> AudioOutput | None:
+        """The sink at the bottom of the audio chain, below any wrappers.
+
+        This is the sink :meth:`replace_audio_tail` would swap out: the
+        downstream of the :class:`_AudioSinkProxy` when wrappers are present,
+        the whole chain otherwise.
+        """
+        cur = self._audio_sink
+        while cur is not None:
+            if isinstance(cur, _AudioSinkProxy):
+                return cur.next_in_chain
+            cur = cur.next_in_chain
+        return self._audio_sink
+
+    def replace_audio_tail(self, sink: AudioOutput) -> AudioOutput | None:
         """Switch the tail sink at the bottom of the chain, keeping wrappers attached.
 
         Walks the chain looking for a :class:`_AudioSinkProxy` and swaps its
@@ -792,14 +807,21 @@ def replace_audio_tail(self, sink: AudioOutput) -> None:
         when no proxy is present (no wrappers, or the chain hasn't been set up yet).
 
         Use ``self.audio = sink`` instead to replace the entire chain.
+
+        Returns:
+            The sink that was replaced, so the caller can restore it later
+            (see ``AvatarSession.aclose``), or None when there was none.
         """
         cur = self._audio_sink
         while cur is not None:
             if isinstance(cur, _AudioSinkProxy):
+                previous: AudioOutput | None = cur.next_in_chain
                 cur.set_next_in_chain(sink)
-                return
+                return previous
             cur = cur.next_in_chain
+        previous = self._audio_sink
         self.audio = sink
+        return previous
 
     @property
     def transcription(self) -> TextOutput | None:

@@ -122,7 +122,7 @@ async def start(
             )
             self.session_id = session_details.get("sessionId")
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -154,7 +154,7 @@ async def start(
                 },
             )
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -145,7 +145,7 @@ async def _shutdown_session() -> None:
             )
             session_task_mapping[room.name] = self.conversation_id
 
-            agent_session.output.replace_audio_tail(
+            self._attach_audio_output(
                 DataStreamAudioOutput(
                     room=room,
                     destination_identity="listener",

@@ -111,7 +111,7 @@ async def start(
         logger.debug("starting avatar session")
         await self._start_agent(livekit_url, livekit_token)
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -299,7 +299,7 @@ async def _start_local(self, agent_session: AgentSession, room: rtc.Room) -> Non
         )
         await self._avatar_runner.start()
 
-        agent_session.output.replace_audio_tail(audio_buffer)
+        self._attach_audio_output(audio_buffer)
 
     async def _start_cloud(
         self,
@@ -353,7 +353,7 @@ async def _start_cloud(
         logger.debug("starting avatar session")
         await self._start_cloud_agent(livekit_url, livekit_token, room.name)
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -129,7 +129,7 @@ async def start(
             audio_config={"sample_rate": self._audio_config.sample_rate},
         )
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -153,7 +153,7 @@ async def start(
 
         self._room = room
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -137,7 +137,7 @@ async def start(  # type: ignore[override]
         # frames until the video track shows up, so nothing is lost
         # in the gap. replace_audio_tail keeps the TranscriptSynchronizer
         # / RecorderAudioOutput chain intact across hot swaps.
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -176,7 +176,7 @@ def on_agent_session_close(ev: Any) -> None:
         await self._audio_buffer.start()
         self._audio_buffer.on("clear_buffer", self._on_clear_buffer)  # type: ignore[arg-type]
 
-        agent_session.output.replace_audio_tail(self._audio_buffer)
+        self._attach_audio_output(self._audio_buffer)
         self._main_atask = asyncio.create_task(self._main_task(), name="AvatarSession._main_task")
 
     def _on_clear_buffer(self) -> None:

@@ -157,7 +157,7 @@ async def start(
             extra={"session_id": self._session_id, "avatar_id": self._avatar_id},
         )
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -138,7 +138,7 @@ async def start(
         def _on_agent_session_close(_: Any) -> None:
             self._ensure_end_session_task()
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -159,7 +159,7 @@ async def start(
                 f"failed to connect to simli avatar session server returned {avatarConnectionRequest.status} and detail {body}"
             )
             return
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -279,7 +279,7 @@ async def start(
             )
             agent_session.on("close", self._on_session_close)
 
-            agent_session.output.replace_audio_tail(self._audio)
+            self._attach_audio_output(self._audio)
             self._forward_atask = asyncio.create_task(
                 self._forward_audio(), name="spatius_avatar_audio_forwarder"
             )

@@ -206,7 +206,7 @@ async def start(
                 )
                 # replace_audio_tail keeps any wrapper AgentSession.start() adds
                 # later (TranscriptSynchronizer, RecorderAudioOutput) attached.
-                agent_session.output.replace_audio_tail(audio_output)
+                self._attach_audio_output(audio_output)
                 self._audio_output = audio_output
 
                 await self.wait_for_join(timeout=self._config.join_timeout)

@@ -116,7 +116,7 @@ async def start(
             properties={"livekit_ws_url": livekit_url, "livekit_room_token": livekit_token},
         )
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,

@@ -127,7 +127,7 @@ async def start(
         logger.debug("Starting Realtime Avatar Session")
         await self._start_session(livekit_url, livekit_token)
 
-        agent_session.output.replace_audio_tail(
+        self._attach_audio_output(
             DataStreamAudioOutput(
                 room=room,
                 destination_identity=self._avatar_participant_identity,