feat(voice): cut over wearable clients to protocol v1

Author: AnkushMalaker

backends/advanced/src/advanced_omi_backend/controllers/websocket_controller.py

@@ -66,11 +66,6 @@
     SessionStatus,
     SessionStore,
 )
-from advanced_omi_backend.services.device_audio import (
-    is_opus_streaming_client,
-    stop_play_audio,
-    stream_play_audio_as_opus,
-)
 from advanced_omi_backend.services.observability import record_event_sync
 from advanced_omi_backend.services.plugin_service import get_plugin_router
 from advanced_omi_backend.services.response_coordinator import (
@@ -519,10 +514,8 @@ async def subscribe_to_device_downlink(
 ) -> None:
     """Forward backend→device control messages from Redis Pub/Sub to the device WebSocket.
 
-    Any backend component (wake-word service, plugins) can push a message to a
-    specific device by publishing to ``device:downlink:{client_id}``. Each message
-    is a Wyoming-style control frame (e.g. ``{"type": "play-audio", "data": {...}}``)
-    which the HAVPE relay's ``handle_backend_messages`` dispatches to the device.
+    Any backend component can push a bound voice event or non-audio device control
+    message to ``device:downlink:{client_id}``.
 
     Runs as a background task for the lifetime of the WebSocket connection.
     """
@@ -532,7 +525,6 @@ async def subscribe_to_device_downlink(
     channel = str(device_downlink_channel(client_id))
     redis_client = None
     pubsub = None
-    opus_stream = is_opus_streaming_client(client_id_value)
     ack_deadline_tasks: set[asyncio.Task] = set()
 
     try:
@@ -541,10 +533,7 @@ async def subscribe_to_device_downlink(
         responses = ResponseCoordinator(redis_client, voice_sessions)
         pubsub = redis_client.pubsub()
         await pubsub.subscribe(channel)
-        logger.info(
-            f"🔊 Subscribed to device downlink channel: {channel}"
-            f"{' (opus streaming)' if opus_stream else ''}"
-        )
+        logger.info(f"🔊 Subscribed to device downlink channel: {channel}")
 
         while True:
             try:
@@ -607,23 +596,6 @@ async def expire_after(
                                 )
                                 ack_deadline_tasks.add(task)
                                 task.add_done_callback(ack_deadline_tasks.discard)
-                    elif msg_type == "stop-audio":
-                        # Barge-in: stop whatever TTS is playing on the device. For
-                        # Opus clients this cancels the in-flight stream + flushes the
-                        # device; for others there's nothing streaming to cancel, so
-                        # just forward the control frame best-effort.
-                        if opus_stream:
-                            await stop_play_audio(websocket, client_id_value)
-                        else:
-                            await websocket.send_json(payload)
-                    elif opus_stream and msg_type == "play-audio":
-                        # RAM-limited devices can't take a big base64 WAV frame;
-                        # transcode + stream it as small Opus packets instead.
-                        streamed = await stream_play_audio_as_opus(
-                            websocket, payload.get("data") or {}, client_id_value
-                        )
-                        if not streamed:
-                            await websocket.send_json(payload)  # fallback
                     else:
                         await websocket.send_json(payload)
                     data = payload.get("data")

backends/advanced/src/advanced_omi_backend/plugins/services.py

@@ -5,14 +5,14 @@
 (e.g., close a conversation) or with other plugins (e.g., call Home Assistant to toggle lights).
 """
 
-import json
 import logging
 from typing import TYPE_CHECKING, Optional
 
 from advanced_omi_backend.models.conversation import Conversation
 from advanced_omi_backend.redis_factory import create_async_redis
-from advanced_omi_backend.redis_keys import ClientId, device_downlink_channel
 from advanced_omi_backend.services.audio_stream.session_store import SessionStore
+from advanced_omi_backend.services.response_coordinator import ResponseCoordinator
+from advanced_omi_backend.services.voice_sessions import VoiceSessionCoordinator
 from advanced_omi_backend.users import User
 
 from .base import PluginContext, PluginResult
@@ -115,31 +115,18 @@ async def star_conversation(self, session_id: str) -> bool:
         # toggle_star returns a dict on success, JSONResponse on error
         return isinstance(result, dict) and "starred" in result
 
-    async def stop_playback(self, client_id: str) -> bool:
-        """Stop any TTS currently playing on a device (barge-in).
-
-        Publishes a ``stop-audio`` control frame to the device's downlink channel.
-        The WebSocket handler that owns the device connection picks it up and, for
-        Opus-streaming clients, cancels the in-flight stream and tells the device to
-        flush (see ``device_audio.stop_play_audio``). Decoupled via Redis so this
-        works from any process (the button handler runs in the backend, but wake
-        handlers run in the workers).
-
-        Args:
-            client_id: The device/client whose playback should stop.
-
-        Returns:
-            True if the stop request was published.
-        """
-        if not client_id:
-            logger.warning("stop_playback called with no client_id")
-            return False
-        message = json.dumps({"type": "stop-audio", "data": {}})
-        client_ref = ClientId.from_value(client_id)
-        await self._async_redis.publish(
-            str(device_downlink_channel(client_ref)), message
+    async def stop_playback(self, user_id: str, client_id: str) -> bool:
+        """Cancel the client's current protocol-v1 response at its generation fence."""
+        if not user_id or not client_id:
+            raise ValueError("stop_playback requires user_id and client_id")
+        coordinator = ResponseCoordinator(
+            self._async_redis,
+            VoiceSessionCoordinator(self._async_redis),
+        )
+        await coordinator.begin_turn(user_id, client_id, reason="barge_in")
+        logger.info(
+            "Button requested coordinated playback cancellation for %s", client_id
         )
-        logger.info(f"⏹ Requested stop-audio for {client_id}")
         return True
 
     async def call_plugin(