Clear queued playback audio on WebRTC backend (barge-in) (#1186)

* feat(media): clear queued playback audio on WebRTC backend (barge-in)

GstWebRTCClient.clear_output_buffer() was a no-op, so realtime
conversation barge-in could not stop already-buffered robot speech.

Add a real clear_player() to GstWebRTCClient that flushes the local
audio send chain and POSTs /api/media/clear_incoming_audio so the
daemon also flushes its incoming-audio playback pipeline (where most
of the buffered audio actually sits). Deprecate clear_output_buffer()
on both the WebRTC and local GStreamer backends (warn-only).

Wire the daemon flush through both transports: a new REST endpoint and
a ClearIncomingAudioCmd data-channel command, both routed to
Backend.clear_incoming_audio() -> GstMediaServer.clear_incoming_audio().
Flush the incoming-audio appsrc with reset_time=False to keep the
shared-clock timeline intact. Expose clearIncomingAudio() in the
TypeScript SDK and document it.

Fixes #1108

Assisted-by: Claude:claude-opus-4-8

* refactor(media): make clear_player() part of the AudioBase contract

Promote clear_player() to an abstract method on AudioBase now that both
GStreamerAudio and GstWebRTCClient implement it, so callers can invoke
it without hasattr() guards. Tidy the deprecated clear_output_buffer()
docstrings to imperative mood (ruff D401).

Assisted-by: Claude:claude-opus-4-8

* regenerate api

* feat(webrtc): continuous audio send chain via audiomixer

Insert an audiomixer (fed by a silent live audiotestsrc on a second pad)
into the WebRTC audio send chain so it emits a continuous output stream
between utterances, keeping the Opus encoder / webrtcbin warm. This drops
the 0.5 s warm-up-silence hack and the first-word swallowing it worked
around. A capsfilter after the mixer pins the output to SAMPLE_RATE /
CHANNELS so opusenc/rtpopuspay advertise sprop-maxcapturerate and stereo
encoding-params matching the negotiated webrtcbin OPUS sink pad (the
mixer otherwise defaults to 48 kHz / mono and webrtcbin rejects it).

Push path now marks the first buffer of a cue DISCONT with a running-time
PTS and leaves follow-up buffers untimestamped, letting the mixer place
them contiguously by byte offset. Handle LATENCY bus messages to
redistribute latency after the live elements are added dynamically.

Assisted-by: Claude:claude-opus-4-8

* removing extra latency on the mixer

Author: FabienDanieau

docs/source/API/openapi.json

@@ -1314,6 +1314,29 @@
         }
       }
     },
+    "/api/media/clear_incoming_audio": {
+      "post": {
+        "summary": "Clear Incoming Audio",
+        "description": "Drop audio received from WebRTC clients that is queued for the speaker.\n\nUsed for barge-in so the robot stops speaking already-buffered audio.",
+        "operationId": "clear_incoming_audio_api_media_clear_incoming_audio_post",
+        "responses": {
+          "200": {
+            "description": "Successful Response",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "additionalProperties": {
+                    "type": "string"
+                  },
+                  "type": "object",
+                  "title": "Response Clear Incoming Audio Api Media Clear Incoming Audio Post"
+                }
+              }
+            }
+          }
+        }
+      }
+    },
     "/api/media/wobbling/enable": {
       "post": {
         "summary": "Enable Wobbling",

docs/source/SDK/javascript-sdk.md

@@ -109,6 +109,7 @@ new ReachyMini({
 | `setAntennasDeg(right, left)` | `boolean` | Set antenna positions in degrees (wraps `setTarget`) |
 | `setBodyYawDeg(yaw)` | `boolean` | Set body yaw in degrees (wraps `setTarget`) |
 | `playSound(filename)` | `boolean` | Play a sound file on the robot |
+| `clearIncomingAudio()` | `boolean` | Drop audio queued for the robot speaker (barge-in) |
 | `sendRaw(data)` | `boolean` | Send arbitrary JSON via data channel |
 | `requestState()` | `boolean` | Request a state snapshot |
 | `setAudioMuted(muted)` | — | Mute/unmute robot speaker (local) |

src/reachy_mini/daemon/app/routers/media.py

@@ -95,6 +95,22 @@ async def stop_sound(
     return {"status": "ok"}
 
 
+@router.post("/clear_incoming_audio")
+async def clear_incoming_audio(
+    daemon: Daemon = Depends(get_daemon),
+) -> dict[str, str]:
+    """Drop audio received from WebRTC clients that is queued for the speaker.
+
+    Used for barge-in so the robot stops speaking already-buffered audio.
+    """
+    backend = daemon.backend
+    if backend is None or not backend.ready.is_set():
+        raise HTTPException(status_code=503, detail="Backend not running")
+
+    backend.clear_incoming_audio()
+    return {"status": "ok"}
+
+
 @router.post("/wobbling/enable")
 async def enable_wobbling(
     daemon: Daemon = Depends(get_daemon),

src/reachy_mini/daemon/backend/abstract.py

@@ -30,6 +30,7 @@
     ApplyAudioConfigCmd,
     CancelAudioCmd,
     CancelMoveCmd,
+    ClearIncomingAudioCmd,
     GetHardwareIdCmd,
     GetMicrophoneVolumeCmd,
     GetMotorModeCmd,
@@ -901,6 +902,15 @@ def stop_sound(self) -> None:
         if self._media_server is not None:
             self._media_server.stop_sound()
 
+    def clear_incoming_audio(self) -> None:
+        """Flush incoming WebRTC audio queued for the speaker (barge-in).
+
+        Delegates to the media server.  If the server is not available
+        (no_media mode), this is a no-op.
+        """
+        if self._media_server is not None:
+            self._media_server.clear_incoming_audio()
+
     # Basic move definitions
     INIT_HEAD_POSE = np.eye(4)
 
@@ -1149,6 +1159,10 @@ def _maybe_ignore(field: str) -> bool:
             self.play_sound(cmd.file)
             send_response({"status": "ok", "command": "play_sound"})
 
+        elif isinstance(cmd, ClearIncomingAudioCmd):
+            self.clear_incoming_audio()
+            send_response({"status": "ok", "command": "clear_incoming_audio"})
+
         elif isinstance(cmd, SetSpeechOffsetsCmd):
             offsets = cmd.offsets
             if len(offsets) == 6:

src/reachy_mini/io/protocol.py

@@ -12,7 +12,7 @@
     upload_move_start, upload_move_chunk, upload_move_finish,
     upload_audio_start, upload_audio_chunk, upload_audio_finish,
     play_uploaded_move, cancel_move,
-    play_uploaded_audio, cancel_audio,
+    play_uploaded_audio, cancel_audio, clear_incoming_audio,
     apply_audio_config, read_audio_parameter
 
 Server->Client message types:
@@ -644,6 +644,17 @@ class CancelAudioCmd(BaseModel):
     upload_id: str
 
 
+class ClearIncomingAudioCmd(BaseModel):
+    """Drop incoming WebRTC audio queued for the speaker (barge-in). Fire-and-forget.
+
+    Flushes the daemon's incoming-audio playback pipeline so audio already
+    received from a WebRTC client stops playing promptly. No-op if no audio
+    is currently being received.
+    """
+
+    type: Literal["clear_incoming_audio"] = "clear_incoming_audio"
+
+
 AnyCommand = Annotated[
     SetTargetCmd
     | SetHeadJointsCmd
@@ -684,6 +695,7 @@ class CancelAudioCmd(BaseModel):
     | CancelMoveCmd
     | PlayUploadedAudioCmd
     | CancelAudioCmd
+    | ClearIncomingAudioCmd
     | ApplyAudioConfigCmd
     | ReadAudioParameterCmd,
     Field(discriminator="type"),

src/reachy_mini/media/audio_base.py

@@ -11,6 +11,7 @@
 - ``start_recording()``, ``stop_recording()``
 - ``start_playing()``, ``stop_playing()``
 - ``push_audio_sample()``
+- ``clear_player()``
 - ``play_sound()``
 
 """
@@ -218,6 +219,11 @@ def push_audio_sample(self, data: npt.NDArray[np.float32]) -> None:
         """Push audio data to the output."""
         ...
 
+    @abstractmethod
+    def clear_player(self) -> None:
+        """Drop any queued playback audio immediately (barge-in)."""
+        ...
+
     @abstractmethod
     def play_sound(self, sound_file: str) -> None:
         """Play a sound file."""

src/reachy_mini/media/audio_gstreamer.py

@@ -54,6 +54,7 @@
 import os
 import platform
 import time
+import warnings
 from collections.abc import Callable
 from threading import Thread
 from typing import Optional
@@ -84,13 +85,14 @@
 class GStreamerAudio(AudioBase):
     """Audio implementation using GStreamer.
 
-    Extends ``AudioBase`` with two GStreamer-specific helpers:
+    Extends ``AudioBase`` with a GStreamer-specific helper:
 
-    - ``clear_output_buffer()``: flush queued playback data without stopping
-      the pipeline (no-op by default; useful before refilling the buffer).
     - ``clear_player()``: flush the playback appsrc immediately via GStreamer
       flush events, dropping any queued audio.
 
+    (``clear_output_buffer()`` is deprecated and does nothing; use
+    ``clear_player()`` instead.)
+
     """
 
     PLAYBACK_SINK_BUFFER_TIME_US = 50_000
@@ -430,13 +432,13 @@ def stop_playing(self) -> None:
             self._playbin = None
 
     def clear_output_buffer(self) -> None:
-        """Flush queued playback data so it is not played.
-
-        A low ``set_max_output_buffers`` value may make this unnecessary
-        for most use-cases.
-
-        """
-        pass  # subclasses or future implementations can override
+        """Use :meth:`clear_player` instead. Deprecated; does nothing."""
+        warnings.warn(
+            "clear_output_buffer() is deprecated; use clear_player().",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        self.logger.warning("clear_output_buffer() is deprecated; use clear_player().")
 
     def clear_player(self) -> None:
         """Flush the player's appsrc to drop any queued audio immediately."""

src/reachy_mini/media/media_server.py

@@ -138,6 +138,10 @@ class GstMediaServer:
     # converts whatever the source produces down to this rate before delivery.
     WOBBLER_SAMPLE_RATE = 16_000
 
+    # Name of the appsrc feeding the incoming-audio playback pipeline; used
+    # both when building the pipeline and when flushing it (clear_incoming_audio).
+    INCOMING_AUDIO_SRC_NAME = "audio_in"
+
     def __init__(
         self,
         log_level: str = "INFO",
@@ -387,7 +391,7 @@ def _on_consumer_pad_added(
         self._pipeline_playback.use_clock(sender_clock)
         self._pipeline_playback.set_start_time(Gst.CLOCK_TIME_NONE)
 
-        appsrc = Gst.ElementFactory.make("appsrc", "audio_in")
+        appsrc = Gst.ElementFactory.make("appsrc", self.INCOMING_AUDIO_SRC_NAME)
         appsrc.set_property("format", Gst.Format.TIME)
         appsrc.set_property("is-live", True)
         appsrc.set_property("caps", caps)
@@ -500,6 +504,32 @@ def _cleanup_incoming_audio(self, peer_id: str) -> None:
             playback_pipe.set_state(Gst.State.NULL)
         self._logger.info(f"Cleaned up incoming audio for peer {peer_id}")
 
+    def clear_incoming_audio(self) -> None:
+        """Flush queued/rendering audio in the incoming-audio playback pipeline.
+
+        Used for barge-in: drops audio already received from a WebRTC client
+        and queued for the robot's speaker so the robot stops speaking promptly.
+
+        The playback pipeline shares the sender clock + base-time, so incoming
+        buffer PTS live in that shared running-time; we flush with
+        ``reset_time=False`` to keep the timeline intact (``reset_time=True``
+        would strand future-stamped buffers and stall playback). The pad probe
+        keeps pushing new RTP buffers into the appsrc, which resume in sync.
+        """
+        pipeline = self._pipeline_playback
+        if pipeline is None:
+            self._logger.info("No incoming-audio pipeline to clear.")
+            return
+        appsrc = pipeline.get_by_name(self.INCOMING_AUDIO_SRC_NAME)
+        if appsrc is None:
+            self._logger.warning("Incoming-audio appsrc not found; nothing to flush.")
+            return
+        appsrc.send_event(Gst.Event.new_flush_start())
+        appsrc.send_event(Gst.Event.new_flush_stop(reset_time=False))
+        if self._head_wobbler is not None:
+            self._head_wobbler.reset()
+        self._logger.info("Flushed incoming audio playback")
+
     @property
     def resolution(self) -> tuple[int, int]:
         """Get the current camera resolution as a tuple (width, height)."""