diff --git a/broker/realtime_broker/agent.py b/broker/realtime_broker/agent.py index 93babd7..654f16b 100644 --- a/broker/realtime_broker/agent.py +++ b/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 = [ { diff --git a/broker/realtime_broker/config.py b/broker/realtime_broker/config.py index 2c55e08..7c70f62 100644 --- a/broker/realtime_broker/config.py +++ b/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") diff --git a/broker/realtime_broker/serializer.py b/broker/realtime_broker/serializer.py index a47c618..cab6875 100644 --- a/broker/realtime_broker/serializer.py +++ b/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)) diff --git a/broker/realtime_broker/server.py b/broker/realtime_broker/server.py index 13a0bb2..ab021a5 100644 --- a/broker/realtime_broker/server.py +++ b/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") diff --git a/broker/tools/harness.py b/broker/tools/harness.py index 738e0d0..360e214 100644 --- a/broker/tools/harness.py +++ b/broker/tools/harness.py @@ -425,13 +425,113 @@ async def h_turn_budget_cap(url): } -async def run(url: str, soak: int = 1, only: str | None = None, hygiene: bool = False) -> int: +# ---------------------------------------------------------------------------- +# Wake-interrupt scenarios — run with --wake. Exercise the firmware's +# mid-session wake path: a {"type":"interrupt"} TEXT frame on the open socket +# (what voice_assistant_websocket.interrupt sends) instead of a reconnect. +# Fine against prod hygiene values (W=6/MAX=8): the wake resets the window and +# re-arms the 10s initial grace, so synth turnaround can't race the close. +# ---------------------------------------------------------------------------- +async def w_wake_cuts_reply(url): + # Interrupt as soon as reply audio starts flowing; the tail already in + # flight must stay short (broker flushes its send-ahead), and the same + # socket must answer a fresh question afterwards. + follow = synth("In one short sentence, what is two plus two?") + async with session(url) as c: + await c._stream(synth("Slowly and in detail, tell me a short story about a lighthouse.")) + await c._stream(SILENCE_1S) + c._send_done = time.monotonic() + # wait for first reply audio + first = None + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + try: + msg = await asyncio.wait_for(c.ws.recv(), timeout=deadline - time.monotonic()) + except (asyncio.TimeoutError, websockets.ConnectionClosed): + break + if isinstance(msg, (bytes, bytearray)): + first = time.monotonic() + break + if first is None: + return False, "no reply audio to interrupt", None + await c.ws.send('{"type":"interrupt"}') + tail = bytearray() + while True: # collect the in-flight tail; 1.2s idle = tail over + try: + msg = await asyncio.wait_for(c.ws.recv(), timeout=1.2) + except (asyncio.TimeoutError, websockets.ConnectionClosed): + break + if isinstance(msg, (bytes, bytearray)): + tail.extend(msg) + tail_s = len(tail) / 2 / RATE + if tail_s > 3.0: + return False, f"reply kept flowing {tail_s:.1f}s after interrupt (want <3s)", None + r = await c.ask_pcm(follow) + t = r.transcript().lower() if r.got_audio else "" + ok = r.got_audio and any(w in t for w in ("four", "4")) + return ok, f"tail={tail_s:.1f}s post-wake reply={t!r}", r.first_audio_ms + + +async def w_wake_while_idle(url): + # Wake with nothing in flight (bot idle, listening). Must not error the + # session (response.cancel is guarded/swallowed) and must answer next turn. + follow = synth("In one short sentence, what is the capital of France?") + async with session(url) as c: + r1 = await c.ask("In one short sentence, what is two plus two?") + if not r1.got_audio: + return False, "setup turn got no audio", r1.first_audio_ms + await c.ws.send('{"type":"interrupt"}') + r = await c.ask_pcm(follow) + t = r.transcript().lower() if r.got_audio else "" + ok = r.got_audio and "paris" in t + return ok, f"post-idle-wake reply={t!r}", r.first_audio_ms + + +async def w_wake_bare_no_followup(url): + # The complaint scenario: bot mid-reply, user says the wake word and then + # NOTHING. The reply must stop and stay stopped (explicit response.cancel: + # server_vad never heard the user, so it would not cancel on its own). + async with session(url) as c: + await c._stream(synth("Slowly and in detail, tell me a short story about a lighthouse.")) + await c._stream(SILENCE_1S) + first = None + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + try: + msg = await asyncio.wait_for(c.ws.recv(), timeout=deadline - time.monotonic()) + except (asyncio.TimeoutError, websockets.ConnectionClosed): + break + if isinstance(msg, (bytes, bytearray)): + first = time.monotonic() + break + if first is None: + return False, "no reply audio to interrupt", None + await c.ws.send('{"type":"interrupt"}') + await asyncio.sleep(3.0) # let any in-flight tail land + quiet = await c.listen(3.5) # silence, like a user who said nothing + ok = not quiet.got_audio + return ok, f"audio_after_bare_wake={quiet.seconds:.1f}s (want 0)", None + + +WAKE_SCENARIOS = { + "wake_cuts_reply": w_wake_cuts_reply, + "wake_while_idle": w_wake_while_idle, + "wake_bare_no_followup": w_wake_bare_no_followup, +} + + +async def run( + url: str, soak: int = 1, only: str | None = None, + hygiene: bool = False, wake: bool = False, +) -> int: print(f"== voice-pe broker reliability harness -> {url} ==") if hygiene: print(" hygiene mode: broker must run FOLLOWUP_WINDOW_SECONDS=6 MAX_TURNS_PER_WAKE=2") + if wake: + print(" wake mode: mid-session {\"type\":\"interrupt\"} control frames") if soak > 1: print(f" soak mode: {soak} rounds") - scenario_set = HYGIENE_SCENARIOS if hygiene else SCENARIOS + scenario_set = WAKE_SCENARIOS if wake else HYGIENE_SCENARIOS if hygiene else SCENARIOS scenarios = {only: scenario_set[only]} if only else scenario_set results: list[tuple[str, bool, str, float | None]] = [] latencies: list[float] = [] @@ -470,16 +570,17 @@ def main() -> None: ] url = args[0] if args else "ws://127.0.0.1:8766" hygiene = "--hygiene" in sys.argv + wake = "--wake" in sys.argv soak = 1 if "--soak" in sys.argv: soak = int(sys.argv[sys.argv.index("--soak") + 1]) only = None if "--only" in sys.argv: only = sys.argv[sys.argv.index("--only") + 1] - valid = HYGIENE_SCENARIOS if hygiene else SCENARIOS + valid = WAKE_SCENARIOS if wake else HYGIENE_SCENARIOS if hygiene else SCENARIOS if only not in valid: raise SystemExit(f"unknown scenario {only!r}; one of: {', '.join(valid)}") - raise SystemExit(asyncio.run(run(url, soak, only, hygiene))) + raise SystemExit(asyncio.run(run(url, soak, only, hygiene, wake))) if __name__ == "__main__": diff --git a/firmware/voice_pe_dual.yaml b/firmware/voice_pe_dual.yaml index 019a433..81d0971 100644 --- a/firmware/voice_pe_dual.yaml +++ b/firmware/voice_pe_dual.yaml @@ -1807,25 +1807,35 @@ micro_wake_word: condition: switch.is_off: master_mute_switch then: - # Stop any prior websocket session before starting a new one (they - # share one mic + speaker). Do NOT stop micro_wake_word: it owns the - # shared microphone that the websocket component feeds from. - if: condition: voice_assistant_websocket.is_running: vaws then: - - voice_assistant_websocket.stop: vaws - - delay: 200ms - - if: - condition: - switch.is_on: wake_sound - then: - - script.execute: - id: play_sound - priority: true - sound_file: "wake_word_triggered_sound" - - delay: 300ms - - voice_assistant_websocket.start: vaws + # Mid-session wake = barge-in, not a restart. Keep the socket: + # interrupt cuts local playback and tells the broker to drop + # the in-flight reply and open a fresh turn. Stopping instead + # would fire on_stopped's end cue (goodbye sound on every + # re-wake) and lose the start of the command to the reconnect. + - voice_assistant_websocket.interrupt: vaws + - if: + condition: + switch.is_on: wake_sound + then: + - script.execute: + id: play_sound + priority: true + sound_file: "wake_word_triggered_sound" + else: + - if: + condition: + switch.is_on: wake_sound + then: + - script.execute: + id: play_sound + priority: true + sound_file: "wake_word_triggered_sound" + - delay: 300ms + - voice_assistant_websocket.start: vaws select: - platform: template