feat: pipecat rt as new realtime voice provider and aec3 webrtc

CLAUDE.md

@@ -2,6 +2,10 @@
 
 This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
 
+## Coding convention
+
+Avoid commenting anything; just drop short notes for critical functions; other functions just need a description, params note (and/or short example); do not provide full payload or usecase to trigger the function; The same expectation for module layer;
+
 ## Multi-IDE Rules (Cursor + Claude Code)
 
 This repo is developed in both **Cursor** and **Claude Code**. The following rules (from `.cursor/rules/`) apply to all code changes:

hal/config.py

@@ -661,6 +661,7 @@ def _os_cfg_realtime() -> dict:
 _RT_GEMINI: dict = _RT.get("gemini") if isinstance(_RT.get("gemini"), dict) else {}
 _RT_OPENAI: dict = _RT.get("openai") if isinstance(_RT.get("openai"), dict) else {}
 _RT_QWEN: dict = _RT.get("qwen") if isinstance(_RT.get("qwen"), dict) else {}
+_RT_PIPECAT: dict = _RT.get("pipecat") if isinstance(_RT.get("pipecat"), dict) else {}
 
 
 def _rt_str(env_key: str, cfg_val, default: str) -> str:
@@ -683,7 +684,7 @@ def _rt_enabled() -> bool:
 
 
 REALTIME_ENABLED: bool = _rt_enabled()
-REALTIME_PROVIDER: str = _rt_str("HAL_REALTIME_PROVIDER", _RT.get("provider"), "gemini")  # none | gemini | openai | qwen
+REALTIME_PROVIDER: str = _rt_str("HAL_REALTIME_PROVIDER", _RT.get("provider"), "gemini")  # none | gemini | openai | qwen | pipecat
 # When enabled, do not send a voice turn to the realtime agent until an STT
 # interim transcript starts with one of the configured wake phrases. This is a
 # top-level config.json setting because it also gates the non-realtime Go path.
@@ -1216,6 +1217,31 @@ def _rt_enabled() -> bool:
 )
 REALTIME_QWEN_SAMPLE_RATE: int = 16000
 
+# --- Realtime: Pipecat (cascaded) ---
+# Not an audio-native brain: HAL keeps its own STT and TTS and pipecat drives
+# only the middle of the turn (text in -> LLM + tools -> text out), so there is
+# no voice or reasoning knob here. The endpoint is any OpenAI-compatible /v1
+# host; blank values fall back to the AI brain's, which is already such a host.
+REALTIME_PIPECAT_API_KEY: str = (
+    os.environ.get("HAL_PIPECAT_API_KEY", "")
+    or _RT_PIPECAT.get("api_key", "")
+    or _os_cfg_get("llm_api_key", "")
+)
+REALTIME_PIPECAT_BASE_URL: str = (
+    os.environ.get("HAL_PIPECAT_BASE_URL", "")
+    or _RT_PIPECAT.get("base_url", "")
+    or _os_cfg_get("llm_base_url", "")
+)
+REALTIME_PIPECAT_MODEL: str = _rt_str(
+    "HAL_PIPECAT_MODEL", _RT_PIPECAT.get("model"), _os_cfg_get("llm_model", "")
+)
+# Gemini grounding search, offered to the model as a `web_search` tool. Separate
+# key because the gateway above is typically self-hosted and has no search.
+REALTIME_PIPECAT_SEARCH_KEY: str = (
+    os.environ.get("HAL_PIPECAT_GEMINI_KEY", "")
+    or _RT_PIPECAT.get("search_api_key", "")
+)
+
 # --- Realtime: Context manager ---
 OPENCLAW_WORKSPACE_DIR: str = os.environ.get("HAL_OPENCLAW_WORKSPACE_DIR", "/root/.openclaw/workspace")
 HERMES_WORKSPACE_DIR: str = os.environ.get("HAL_HERMES_WORKSPACE_DIR", "/root/.hermes")

hal/drivers/audio_route.py

@@ -142,6 +142,14 @@ def bt_active() -> bool:
 
 
 def _swap_tts(output_idx: Optional[int]) -> None:
+    # A new sink is a new acoustic path — the canceller's buffered reference
+    # belongs to the old one.
+    try:
+        from hal.drivers.voice import aec
+
+        aec.reset()
+    except Exception:
+        pass
     tts = state.tts_service
     if tts is None:
         return

hal/drivers/voice/_internal/config.py

@@ -100,6 +100,25 @@
 WARM_MIC_ECHO_SKIP_MAX_S = float(os.environ.get("HAL_WARM_MIC_ECHO_SKIP_MAX_S", "0.3"))
 
 
+# ---------------------------------------------------------------------------
+# Acoustic echo cancellation (WebRTC AEC3) — see drivers/voice/aec.py.
+# Off by default: it needs the `aec-audio-processing` native binding, which is
+# not a hal dependency. Absent, every AEC entry point is a no-op.
+# ---------------------------------------------------------------------------
+AEC_ENABLED = os.environ.get("HAL_AEC_ENABLED", "false").lower() == "true"
+# Speaker→mic delay hint. AEC3 estimates the real delay itself, but the hint
+# decides how fast it converges. Measured on a lamp (USB mic + USB speaker) the
+# true lag is ~154ms; correcting the hint from 80 to 150 took ERLE from 10.9 to
+# 18.6 dB overall and 6.5 to 14.2 dB during the convergence phase.
+AEC_DELAY_MS = int(os.environ.get("HAL_AEC_DELAY_MS", "150"))
+AEC_NOISE_SUPPRESSION = os.environ.get("HAL_AEC_NS", "true").lower() == "true"
+# Keep cancelling for this long after the last speaker write, then bypass the
+# APM until playback resumes.
+AEC_TAIL_S = float(os.environ.get("HAL_AEC_TAIL_S", "0.5"))
+# Set to a directory to write aec_mic/ref/out.wav for offline ERLE analysis.
+AEC_DUMP_DIR = os.environ.get("HAL_AEC_DUMP_DIR", "")
+
+
 # ---------------------------------------------------------------------------
 # STT keepalive — pre-connect WS before speech is detected to cut latency
 # ---------------------------------------------------------------------------

hal/drivers/voice/_internal/pipecat_turn.py

@@ -0,0 +1,201 @@
+"""Pipecat turn handling — the cascaded twin of realtime_turn.py.
+
+Same contract (drive one turn, return a RealtimeTurnResult) over a different
+input: the finished STT transcript instead of committed audio. There is no
+native-audio path, no look-replay and no WS-recovery retry — a cascaded brain
+has no session to lose, and an empty transcript means there is simply nothing
+to send.
+
+The wait filler, thinking cue and CoT leak filter are shared with realtime_turn
+so both brains feel identical to the user.
+"""
+
+import json
+import logging
+import threading
+from typing import Callable
+
+from hal import config as hal_config
+from hal.pipecat_rt import TextChunk, ToolCall
+from hal.realtime.orchestrator import (
+    DEFAULT_EMOTION_INTENSITY,
+    DELEGATE_TOOL_NAME,
+    EMOTION_TOOL_NAME,
+    RealtimeOrchestrator,
+)
+from hal.drivers.voice._internal.cot_leak_filter import CoTLeakFilter, clean_transcript
+from hal.drivers.voice._internal.realtime_turn import (
+    SENTENCE_ENDS,
+    RealtimeTurnResult,
+    _reply_language_name,
+    _thinking_cue_clear,
+    _thinking_cue_start,
+    _WaitFiller,
+)
+
+logger = logging.getLogger("hal.voice")
+
+
+def _fire_emotion(arguments: str) -> None:
+    """Run express_emotion off-thread so the reply never waits on the face."""
+    try:
+        args: dict = json.loads(arguments) if arguments else {}
+    except (ValueError, TypeError):
+        return
+    emotion: str = str(args.get("emotion", "")).strip().lower()
+    if not emotion:
+        return
+    intensity: float = DEFAULT_EMOTION_INTENSITY
+    try:
+        intensity = max(0.0, min(1.0, float(args.get("intensity", intensity))))
+    except (ValueError, TypeError):
+        pass
+    threading.Thread(
+        target=RealtimeOrchestrator._fire_emotion,
+        args=(emotion, intensity),
+        daemon=True,
+    ).start()
+
+
+def run_pipecat_turn(
+    session,
+    tts,
+    strip_markers: Callable[[str], str],
+    combined: str,
+) -> RealtimeTurnResult:
+    """Send the transcript to the pipecat brain and speak its reply.
+
+    Returns how the turn resolved so the caller can forward (delegate),
+    suppress (handled), or fall back to the main agent.
+    """
+    if not hal_config.REALTIME_ENABLED:
+        return RealtimeTurnResult()
+    if not session.available:
+        logger.warning("[pipecat] enabled but not available — falling back to OS server")
+        return RealtimeTurnResult()
+
+    text: str = (combined or "").strip()
+    if not text:
+        logger.info("[pipecat] empty transcript — nothing to send (no audio path)")
+        return RealtimeTurnResult()
+
+    delegated = False
+    handled = False
+    delegate_msg = ""
+    text_parts: list[str] = []
+    sentence_buf = ""
+    first_sentence_sent = False
+    reply_lang: str = _reply_language_name()
+    leak_filter = CoTLeakFilter(reply_lang)
+    wait_filler = _WaitFiller()
+
+    _thinking_cue_start()
+    wait_filler.arm()
+    try:
+        for event in session.run_turn(text):
+            if isinstance(event, ToolCall):
+                if event.name == DELEGATE_TOOL_NAME:
+                    delegated = True
+                    try:
+                        delegate_msg = str(
+                            json.loads(event.arguments or "{}").get("message", "")
+                        )
+                    except (ValueError, TypeError):
+                        delegate_msg = ""
+                    # The main-agent hop that follows fires its own filler.
+                    wait_filler.cancel()
+                    session.tool_result(
+                        event.call_id, '{"status": "delegated"}', run_llm=False
+                    )
+                    continue
+                if event.name == EMOTION_TOOL_NAME:
+                    _fire_emotion(event.arguments)
+                    # run_llm=True: the face is a side effect, the reply still owes
+                    # the user words.
+                    session.tool_result(event.call_id, '{"status": "ok"}', run_llm=True)
+                    continue
+                logger.warning("[pipecat] unknown tool %r", event.name)
+                session.tool_result(
+                    event.call_id, '{"error": "unknown tool"}', run_llm=True
+                )
+                continue
+
+            if delegated or not isinstance(event, TextChunk):
+                continue
+
+            text_parts.append(event.text)
+            sentence_buf += event.text
+            if tts is not None and sentence_buf.rstrip().endswith(SENTENCE_ENDS):
+                sentence: str = leak_filter.filter_text(strip_markers(sentence_buf))
+                if sentence:
+                    if not first_sentence_sent:
+                        logger.info("[pipecat] First sentence → speak: %r", sentence[:80])
+                        wait_filler.cancel()
+                        if not tts.speak(sentence):
+                            tts.speak_queue(sentence)
+                        first_sentence_sent = True
+                        _thinking_cue_clear()
+                    else:
+                        logger.info(
+                            "[pipecat] Next sentence → speak_queue: %r", sentence[:80]
+                        )
+                        tts.speak_queue(sentence)
+                sentence_buf = ""
+
+        transcript: str = clean_transcript(
+            strip_markers("".join(text_parts)), reply_lang
+        )
+
+        if delegated:
+            logger.info("[pipecat] Delegated → will forward to OS server")
+        else:
+            remaining: str = leak_filter.filter_text(strip_markers(sentence_buf))
+            if remaining and tts is not None:
+                if not first_sentence_sent:
+                    logger.info("[pipecat] Final fragment → speak: %r", remaining[:80])
+                    wait_filler.cancel()
+                    if not tts.speak(remaining):
+                        tts.speak_queue(remaining)
+                    first_sentence_sent = True
+                    _thinking_cue_clear()
+                else:
+                    logger.info(
+                        "[pipecat] Final fragment → speak_queue: %r", remaining[:80]
+                    )
+                    tts.speak_queue(remaining)
+            # Same rule as the audio-native path: only claim the turn when the
+            # device actually spoke, so an empty reply still reaches the main agent.
+            if first_sentence_sent or transcript:
+                handled = True
+                logger.info(
+                    "[pipecat] Chit-chat complete — agent_reply=%r",
+                    transcript[:200] if transcript else "(empty)",
+                )
+                session.save_turn(
+                    user_text=text, agent_text=transcript or "(empty)"
+                )
+            else:
+                logger.info("[pipecat] No output (empty / timeout) — falling back")
+                _thinking_cue_clear()
+                try:
+                    from hal.routes.led import restore_led
+
+                    restore_led()
+                except Exception:
+                    pass
+    except Exception as e:
+        logger.warning("[pipecat] Processing failed: %s — will forward to OS server", e)
+        _thinking_cue_clear()
+        try:
+            session.abort_turn()
+        except Exception:
+            pass
+        return RealtimeTurnResult(delegated=True)
+    finally:
+        wait_filler.cancel()
+        try:
+            session.turn_finished()
+        except Exception:
+            logger.exception("[pipecat] turn bookkeeping failed")
+
+    return RealtimeTurnResult(delegated, handled, transcript, delegate_msg)