Mid-session wake = barge-in, not a reconnect

broker/realtime_broker/agent.py

@@ -11,6 +11,7 @@
 
 import logging
 
+from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.services.mcp_service import MCPClient
 from pipecat.services.openai.realtime.events import (
     AudioConfiguration,
@@ -21,7 +22,6 @@
     TurnDetection,
 )
 from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService
-from pipecat.processors.aggregators.llm_context import LLMContext
 
 from .config import Config
 
@@ -47,6 +47,16 @@ async def _handle_context(self, context: LLMContext) -> None:
         self._llm_needs_conversation_setup = False
         await self._process_completed_function_calls(send_new_results=True)
 
+    async def _handle_evt_error(self, evt) -> None:
+        # A device-wake response.cancel can race response.done; OpenAI then
+        # answers with response_cancel_not_active. Upstream treats every error
+        # event as fatal (ErrorFrame -> session rebuild), which would turn a
+        # harmless race into a dropped conversation. Swallow just that code.
+        if getattr(getattr(evt, "error", None), "code", None) == "response_cancel_not_active":
+            logger.debug("Ignoring benign response.cancel race (no active response)")
+            return
+        await super()._handle_evt_error(evt)
+
 # Custom broker tools, registered with handlers by the server.
 CUSTOM_TOOLS = [
     {

broker/realtime_broker/config.py

@@ -72,7 +72,7 @@ def ha_control_enabled(self) -> bool:
         return bool(self.ha_mcp_url and self.ha_token)
 
     @classmethod
-    def from_env(cls) -> "Config":
+    def from_env(cls) -> Config:
         api_key = os.environ.get("OPENAI_API_KEY")
         if not api_key:
             raise RuntimeError("OPENAI_API_KEY is required")

broker/realtime_broker/serializer.py

@@ -1,12 +1,14 @@
 """WebSocket frame serializer for raw PCM audio.
 
 The device speaks the simplest possible protocol: binary frames are raw
-PCM16 / 24 kHz / mono audio, both directions. (Text frames, if any, are
-control messages handled by the transport, not here.)
+PCM16 / 24 kHz / mono audio, both directions. Text frames are device->broker
+control messages (currently {"type": "interrupt"}, sent on a mid-session
+wake word) and are dispatched to `on_control` instead of the pipeline.
 """
 
 from __future__ import annotations
 
+import json
 import logging
 
 from pipecat.frames.frames import Frame, InputAudioRawFrame, OutputAudioRawFrame
@@ -18,14 +20,36 @@
 
 
 class RawPCMSerializer(FrameSerializer):
-    """Treats binary WebSocket messages as raw PCM16/24k/mono audio."""
+    """Treats binary WebSocket messages as raw PCM16/24k/mono audio.
+
+    `on_control` (assigned by the server after the pipeline exists) receives
+    parsed device control frames; it runs inline in the transport's receive
+    loop, so it observes the same ordering the device sent.
+    """
+
+    on_control = None  # async callable(dict) | None
 
     @property
     def type(self) -> FrameSerializerType:
         return FrameSerializerType.BINARY
 
-    async def deserialize(self, message: bytes) -> InputAudioRawFrame | None:
+    async def deserialize(self, message: bytes | str) -> InputAudioRawFrame | None:
         if not isinstance(message, bytes):
+            if isinstance(message, str) and self.on_control is not None:
+                try:
+                    data = json.loads(message)
+                except ValueError:
+                    logger.warning("Dropping malformed control frame: %r", message[:200])
+                    return None
+                if not isinstance(data, dict):
+                    logger.warning("Dropping non-object control frame: %r", message[:200])
+                    return None
+                # An exception here would kill the transport receive loop and
+                # with it the session's audio input; log and carry on instead.
+                try:
+                    await self.on_control(data)
+                except Exception:
+                    logger.exception("Control frame handler failed")
             return None
         if len(message) % 2 != 0:
             logger.warning("Dropping odd-length audio frame (%d bytes)", len(message))

broker/realtime_broker/server.py

@@ -39,15 +39,14 @@
     LLMContextAggregatorPair,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.llm_service import FunctionCallResultProperties
+from pipecat.services.openai.realtime import events as oai_events
 from pipecat.transports.websocket.server import (
     WebsocketServerParams,
     WebsocketServerTransport,
 )
 from websockets.protocol import State
 
-from pipecat.services.llm_service import FunctionCallResultProperties
-from pipecat.services.openai.realtime import events as oai_events
-
 from . import mcp_client
 from .agent import build_agent, build_audio_input
 from .config import Config
@@ -105,7 +104,7 @@ class _BotPlaybackGate(FrameProcessor):
     goes quiet.
     """
 
-    def __init__(self, service, config: Config, get_ws) -> None:  # noqa: ANN001
+    def __init__(self, service, config: Config, get_ws) -> None:
         super().__init__()
         self._service = service
         self._config = config
@@ -155,6 +154,15 @@ async def _restore_when_quiet(self) -> None:
             self._max_backlog = 0.0
         await self._set_threshold(self._config.vad_threshold)
 
+    def on_device_wake(self) -> None:
+        """Mid-session wake: the device already cut its speaker and flushed
+        its queue, so there is no echo tail to ride out. Rewind the playback
+        clock past the release margin so the mic gate opens for the command
+        immediately instead of vad_release_delay_ms later (which would feed
+        OpenAI silence for the first second of the command)."""
+        release = self._config.vad_release_delay_ms / 1000
+        self._playback_end = asyncio.get_running_loop().time() - release
+
     async def _maybe_flush_device(self) -> None:
         now = asyncio.get_running_loop().time()
         if now >= self._playback_end:
@@ -164,7 +172,7 @@ async def _maybe_flush_device(self) -> None:
             return
         try:
             await ws.send('{"type":"interrupt"}')
-        except Exception:  # noqa: BLE001
+        except Exception:
             logger.exception("barge-in: failed to signal device")
             return
         self._playback_end = now  # device drops its queue on the flush
@@ -229,7 +237,7 @@ async def reset_vad(self, drain: float = 0.4) -> None:
                 await asyncio.sleep(drain)
             await self._service.send_client_event(oai_events.InputAudioBufferClearEvent())
             await self._send_vad(self._threshold)
-        except Exception:  # noqa: BLE001
+        except Exception:
             logger.exception("VAD reset failed")
         finally:
             if self._reset_task is task:
@@ -250,7 +258,7 @@ class _MicInputGate(FrameProcessor):
     real barge-in needs echo cancellation, not just an open mic.
     """
 
-    def __init__(self, gate: "_BotPlaybackGate", config: Config) -> None:
+    def __init__(self, gate: _BotPlaybackGate, config: Config) -> None:
         super().__init__()
         self._gate = gate
         self._config = config
@@ -307,7 +315,7 @@ class _TurnHygiene(FrameProcessor):
     it a response end is invisible.
     """
 
-    def __init__(self, gate: "_BotPlaybackGate", config: Config, get_ws) -> None:  # noqa: ANN001
+    def __init__(self, gate: _BotPlaybackGate, config: Config, get_ws) -> None:
         super().__init__()
         self._gate = gate  # single source of truth for the playback clock
         self._config = config
@@ -355,10 +363,9 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
                 self._awaiting_since = asyncio.get_running_loop().time()
             else:
                 self._awaiting_response = False
-        elif isinstance(frame, (CancelFrame, EndFrame)):
-            if self._watch_task is not None:
-                task, self._watch_task = self._watch_task, None
-                await self.cancel_task(task)
+        elif isinstance(frame, (CancelFrame, EndFrame)) and self._watch_task is not None:
+            task, self._watch_task = self._watch_task, None
+            await self.cancel_task(task)
         await self.push_frame(frame, direction)
 
     def on_device_connect(self) -> None:
@@ -453,7 +460,7 @@ async def _watch(self) -> None:
                 continue  # socket swapped mid-tick (re-wake); fresh state incoming
             try:
                 await ws.send('{"type":"disconnect"}')
-            except Exception:  # noqa: BLE001
+            except Exception:
                 logger.exception("turn hygiene: failed to signal device")
             else:
                 logger.info("turn hygiene: %s; disconnecting device", reason)
@@ -476,7 +483,7 @@ async def run(config: Config) -> None:
     while True:
         try:
             await _serve_session(config, mcp)
-        except Exception:  # noqa: BLE001
+        except Exception:
             logger.exception("Session crashed; rebuilding")
         await asyncio.sleep(0.5)  # let the socket fully release before rebind
 
@@ -571,42 +578,43 @@ def _start_music(config: Config, query: str, speaker: str | None) -> str:
         return "Sorry, I couldn't start the music."
 
 
-async def _serve_session(config: Config, mcp) -> None:  # noqa: ANN001
+async def _serve_session(config: Config, mcp) -> None:
     """Run one OpenAI session until it dies or reaches max age, then tear down."""
     service = await build_agent(config, mcp)
 
+    serializer = RawPCMSerializer()
     transport = WebsocketServerTransport(
         host=config.ws_host,
         port=config.ws_port,
         params=WebsocketServerParams(
-            serializer=RawPCMSerializer(),
+            serializer=serializer,
             audio_in_enabled=True,
             audio_out_enabled=True,
         ),
     )
 
-    async def _get_weather(params):  # noqa: ANN001
+    async def _get_weather(params):
         # Run the blocking HA fetch off the event loop so it can't stall audio.
         weather = await asyncio.to_thread(_fetch_weather, config)
         await params.result_callback(weather)
 
-    async def _end_conversation(params):  # noqa: ANN001
+    async def _end_conversation(params):
         await params.result_callback("Okay, goodbye!")
         ws = getattr(transport.input(), "_websocket", None)
         if ws is not None:
             try:
                 await ws.send('{"type":"disconnect"}')
-            except Exception:  # noqa: BLE001
+            except Exception:
                 logger.exception("end_conversation: failed to signal device")
 
-    async def _play_music(params):  # noqa: ANN001
+    async def _play_music(params):
         args = params.arguments or {}
         msg = await asyncio.to_thread(
             _start_music, config, args.get("query", ""), args.get("speaker")
         )
         await params.result_callback(msg)
 
-    async def _wait_for_user(params):  # noqa: ANN001
+    async def _wait_for_user(params):
         # Non-addressed speech (TV, side conversation, background). Acknowledge
         # the call but suppress the follow-up response (run_llm=False) so the bot
         # stays silent and keeps listening instead of replying to the room.
@@ -648,12 +656,36 @@ async def _wait_for_user(params):  # noqa: ANN001
         ]
     )
 
+    async def _on_device_control(msg: dict) -> None:
+        # Device->broker control frames. "interrupt" = a mid-session wake word:
+        # the firmware cut its own playback and kept the socket instead of
+        # reconnecting. Mirror what a fresh connect does, in place: cancel the
+        # in-flight reply, flush the pipeline (which drops the reply audio the
+        # output transport still has queued — up to ~2s of send-ahead), reopen
+        # the mic, clear stale VAD state, and grant a fresh hygiene window.
+        if msg.get("type") != "interrupt":
+            logger.warning("Unknown device control frame: %r", msg)
+            return
+        logger.info("device wake: in-session interrupt")
+        if service._current_assistant_response is not None:
+            # server_vad only auto-cancels on heard speech, and the mic was
+            # gated during playback — cancel explicitly so a bare wake with no
+            # follow-up command still shuts the reply up. The benign race with
+            # response.done is swallowed in VoicePERealtimeService.
+            await service.send_client_event(oai_events.ResponseCancelEvent())
+        await service.push_interruption_task_frame_and_wait()
+        gate.on_device_wake()
+        asyncio.create_task(gate.reset_vad(drain=0.0))
+        hygiene.on_device_connect()
+
+    serializer.on_control = _on_device_control
+
     loop = asyncio.get_running_loop()
     device_connected = False
     idle_since = loop.time()
 
     @transport.event_handler("on_client_connected")
-    async def _on_connect(_transport, client):  # noqa: ANN001
+    async def _on_connect(_transport, client):
         nonlocal device_connected
         device_connected = True
         logger.info("Device connected: %s", getattr(client, "remote_address", client))
@@ -669,7 +701,7 @@ async def _on_connect(_transport, client):  # noqa: ANN001
         hygiene.on_device_connect()
 
     @transport.event_handler("on_client_disconnected")
-    async def _on_disconnect(_transport, client, *args):  # noqa: ANN001
+    async def _on_disconnect(_transport, client, *args):
         nonlocal device_connected, idle_since
         # When a new connection kicks a lingering old one (re-wake after a
         # WiFi blip: the dead socket never closed), Pipecat swaps the
@@ -770,5 +802,5 @@ async def _on_disconnect(_transport, client, *args):  # noqa: ANN001
                 pass
         except asyncio.CancelledError:
             pass
-        except Exception:  # noqa: BLE001
+        except Exception:
             logger.exception("Runner teardown error")