Update 2026-07-09 17:48:37

config/voice_config.json

@@ -1,6 +1,5 @@
 {
   "_description": "Tunables for voice/* modules. Loaded via core.config_loader.load('voice').",
-
   "sanad_voice": {
     "_comment": "voice/sanad_voice.py — main live voice subprocess. Gemini API credentials (api_key, model, voice_name) come from core_config.json's gemini_defaults — single source of truth.",
     "mic_gain": 1.0,
@@ -10,9 +9,10 @@
     "session_timeout_sec": 660,
     "max_reconnect_delay_sec": 30,
     "max_consecutive_errors": 10,
-    "no_messages_timeout_sec": 30
+    "no_messages_timeout_sec": 180,
+    "pure_voice": false,
+    "internet_search": false
   },
-
   "mic_udp": {
     "_comment": "G1 built-in mic — UDP multicast subscriber",
     "group": "239.168.123.161",
@@ -21,41 +21,40 @@
     "read_timeout_sec": 0.04,
     "socket_timeout_sec": 1.0
   },
-
   "speaker": {
     "_comment": "G1 built-in speaker — AudioClient.PlayStream wrapper",
     "app_name": "sanad",
     "begin_stream_pause_sec": 0.15,
-    "wait_finish_margin_sec": 0.3
+    "wait_finish_margin_sec": 0.3,
+    "jitter_prebuffer_sec": 0.3,
+    "jitter_first_flush_sec": 0.6,
+    "jitter_max_lead_sec": 0.6
   },
-
   "vad": {
     "_comment": "Gemini Live server-side voice-activity-detection config",
     "start_sensitivity": "START_SENSITIVITY_HIGH",
     "end_sensitivity": "END_SENSITIVITY_LOW",
     "prefix_padding_ms": 20,
-    "silence_duration_ms": 200
+    "silence_duration_ms": 200,
+    "direct_answer": false
   },
-
   "barge_in": {
     "threshold": 500,
     "loud_chunks_needed": 3,
     "cooldown_sec": 0.3,
     "echo_suppress_below": 500,
-    "ai_speak_grace_sec": 0.15
+    "ai_speak_grace_sec": 0.15,
+    "external_barge_in": false
   },
-
   "recording": {
     "enabled": true,
     "dir_relative": "data/recordings"
   },
-
   "typed_replay": {
     "_comment": "voice/typed_replay.py — max_text_len comes from dashboard.api_input",
     "monitor_chunk_size": 512,
     "monitor_tail_sec": 0.2
   },
-
   "live_voice_loop": {
     "_comment": "voice/live_voice_loop.py — arm phrase dispatcher. arm_txt filename comes from core.script_files.arm_phrases",
     "trigger_log_size": 100,
@@ -63,7 +62,6 @@
     "deferred_default": false,
     "trigger_enabled_default": false
   },
-
   "local_tts": {
     "_comment": "voice/local_tts.py — offline Coqui TTS",
     "model_subdir": "speecht5_tts_clartts_ar",
@@ -72,4 +70,4 @@
     "sample_rate": 16000,
     "channels": 1
   }
-}
+}

dashboard/routes/_safe_io.py

@@ -58,13 +58,43 @@ def check_upload_size(content: bytes, max_bytes: int = MAX_UPLOAD_BYTES) -> None
 
 
 def atomic_write_bytes(path: Path, data: bytes) -> None:
-    """Write bytes atomically via tempfile + os.replace."""
+    """Write bytes atomically via tempfile + os.replace.
+
+    Fallback: a bind-MOUNTED single file (docker `-v host.txt:/app/x.txt`)
+    cannot be renamed over — os.replace raises EBUSY (errno 16) because the
+    mount pins the inode. In that case fall back to an in-place truncate+write
+    (not atomic, but the only way to persist through a file mount)."""
     path.parent.mkdir(parents=True, exist_ok=True)
+    # Carry the target's mode/owner onto the replacement — mkstemp creates
+    # 0600 root, and dashboards run as root in containers over host-owned
+    # bind mounts: without this every save locks the host user out (root:600).
+    try:
+        st = path.stat()
+    except OSError:
+        st = None
     fd, tmp = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
     try:
         with os.fdopen(fd, "wb") as f:
             f.write(data)
+        try:
+            if st is not None:
+                os.chmod(tmp, st.st_mode & 0o7777)
+                os.chown(tmp, st.st_uid, st.st_gid)
+            else:
+                os.chmod(tmp, 0o664)
+        except OSError:
+            pass  # chown needs root; keep the write even if it fails
         os.replace(tmp, path)
+    except OSError as exc:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+        if exc.errno == 16:  # EBUSY — path is a bind-mounted file
+            with open(path, "wb") as f:
+                f.write(data)
+            return
+        raise
     except Exception:
         try:
             os.unlink(tmp)

dashboard/routes/audio_control.py

@@ -627,6 +627,37 @@ def _kill_audio_daemon(flavour: str) -> dict:
                 "stderr": "timeout (>5s)"}
 
 
+def _remote_pulse_mode() -> bool:
+    """True when this process talks to ANOTHER machine-scope PulseAudio —
+    the containerized deployments (P1–P4) run as root with PULSE_SERVER
+    pointed at the host's socket. Restarting that daemon is not ours to do."""
+    return bool(os.environ.get("PULSE_SERVER")) or os.path.exists("/.dockerenv")
+
+
+def _cycle_remote_pulse() -> dict:
+    """Suspend→resume every sink/source on the remote (host) PulseAudio.
+    A suspend cycle forces the ALSA device rebind and clears wedged
+    streams — the useful part of a daemon restart, minus the restart."""
+    cycled: list[str] = []
+    for kind, flag in (("sinks", "suspend-sink"), ("sources", "suspend-source")):
+        try:
+            r = subprocess.run(["pactl", "list", "short", kind], check=False,
+                               capture_output=True, text=True, timeout=5.0)
+            names = [ln.split("\t")[1] for ln in (r.stdout or "").splitlines()
+                     if "\t" in ln]
+        except (FileNotFoundError, subprocess.SubprocessError):
+            names = []
+        for name in names:
+            if name.endswith(".monitor"):
+                continue
+            for state in ("1", "0"):
+                subprocess.run(["pactl", flag, name, state], check=False,
+                               capture_output=True, text=True, timeout=5.0)
+            cycled.append(name)
+    return {"cmd": "remote pulse suspend/resume cycle", "returncode": 0,
+            "stderr": "", "cycled": cycled}
+
+
 def _wait_for_pactl(deadline_s: float = 5.0, interval_s: float = 0.2) -> bool:
     """Poll `pactl info` until it returns 0 or the deadline expires."""
     import time as _time
@@ -646,7 +677,8 @@ async def reset_audio_subsystem():
     is being selected. **Does NOT recover a kernel-side missing USB capture
     descriptor** — for that symptom use /api/audio/usb-reset.
     """
-    if os.geteuid() == 0:
+    remote_pulse = _remote_pulse_mode()
+    if os.geteuid() == 0 and not remote_pulse:
         raise HTTPException(
             403, "Refusing to reset audio as root — Sanad must run as the "
             "unitree user so the per-user PulseAudio session is reachable.",
@@ -687,9 +719,16 @@ async def reset_audio_subsystem():
                     if acquired and play_lock is not None:
                         play_lock.release()
 
-            flavour = _detect_pa_flavour()
+            if remote_pulse:
-            kill_info = _kill_audio_daemon(flavour)
+                # Containerized: the daemon is the HOST's — never kill it.
-            came_back = _wait_for_pactl(deadline_s=5.0)
+                # Cycle its devices over the socket instead.
+                flavour = "pulse-remote"
+                kill_info = _cycle_remote_pulse()
+                came_back = _wait_for_pactl(deadline_s=5.0)
+            else:
+                flavour = _detect_pa_flavour()
+                kill_info = _kill_audio_daemon(flavour)
+                came_back = _wait_for_pactl(deadline_s=5.0)
             if not came_back and flavour == "pulse":
                 # autospawn may be disabled — try an explicit start.
                 try:

dashboard/routes/live_subprocess.py

@@ -41,6 +41,237 @@ async def set_record(on: bool = Query(...)):
     return {"ok": True, "record_enabled": st.record_enabled}
 
 
+_VOICE_OPTS_PATH = BASE_DIR / "data" / ".voice_options.json"
+
+
+def _read_voice_opts() -> dict:
+    import json
+    try:
+        with open(_VOICE_OPTS_PATH, encoding="utf-8") as f:
+            d = json.load(f)
+        return d if isinstance(d, dict) else {}
+    except Exception:
+        return {}
+
+
+@router.get("/interrupt-option")
+async def get_interrupt_option():
+    """Voice barge-in on external speakers (JBL / Beats Pill). Resolution:
+    dashboard toggle (data/.voice_options.json) > SANAD_EXTERNAL_BARGE_IN env
+    > voice_config.json barge_in.external_barge_in > off."""
+    import os
+    opts = _read_voice_opts()
+    if "external_barge_in" in opts:
+        return {"enabled": bool(opts["external_barge_in"]), "source": "dashboard"}
+    env = os.environ.get("SANAD_EXTERNAL_BARGE_IN", "").strip().lower()
+    if env in ("1", "true", "yes"):
+        return {"enabled": True, "source": "env"}
+    try:
+        from Project.Sanad.core.config_loader import section
+        if bool((section("voice", "barge_in") or {}).get("external_barge_in", False)):
+            return {"enabled": True, "source": "config"}
+    except Exception:
+        pass
+    return {"enabled": False, "source": "default"}
+
+
+@router.post("/interrupt-option")
+async def set_interrupt_option(on: bool = Query(...)):
+    """Persist the interrupt option (survives rebuilds — data/ is mounted) and
+    restart the live session so the voice child picks it up."""
+    import json
+    import tempfile
+    import os as _os
+    opts = _read_voice_opts()
+    opts["external_barge_in"] = bool(on)
+    _VOICE_OPTS_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp = tempfile.mkstemp(dir=str(_VOICE_OPTS_PATH.parent),
+                               prefix=".voice_options.", suffix=".tmp")
+    with _os.fdopen(fd, "w", encoding="utf-8") as f:
+        json.dump(opts, f, indent=2)
+    _os.replace(tmp, _VOICE_OPTS_PATH)
+
+    restarted = False
+    from Project.Sanad.main import live_sub
+    if live_sub is not None:
+        try:
+            if (live_sub.status() or {}).get("running"):
+                await asyncio.to_thread(live_sub.stop)
+                await asyncio.to_thread(live_sub.start)
+                restarted = True
+        except Exception:
+            pass
+    return {"ok": True, "enabled": bool(on), "restarted": restarted}
+
+
+@router.get("/direct-answer-option")
+async def get_direct_answer_option():
+    """'Direct answer' = Gemini replies as soon as the speaker pauses (Live-API
+    end-of-speech sensitivity HIGH). Resolution: dashboard > env > config."""
+    import os
+    opts = _read_voice_opts()
+    if "direct_answer" in opts:
+        return {"enabled": bool(opts["direct_answer"]), "source": "dashboard"}
+    if os.environ.get("SANAD_DIRECT_ANSWER", "").strip().lower() in ("1", "true", "yes"):
+        return {"enabled": True, "source": "env"}
+    try:
+        from Project.Sanad.core.config_loader import section
+        if bool((section("voice", "vad") or {}).get("direct_answer", False)):
+            return {"enabled": True, "source": "config"}
+    except Exception:
+        pass
+    return {"enabled": False, "source": "default"}
+
+
+@router.post("/direct-answer-option")
+async def set_direct_answer_option(on: bool = Query(...)):
+    """Persist the direct-answer option and restart the live session."""
+    import json
+    import tempfile
+    import os as _os
+    opts = _read_voice_opts()
+    opts["direct_answer"] = bool(on)
+    _VOICE_OPTS_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp = tempfile.mkstemp(dir=str(_VOICE_OPTS_PATH.parent),
+                               prefix=".voice_options.", suffix=".tmp")
+    with _os.fdopen(fd, "w", encoding="utf-8") as f:
+        json.dump(opts, f, indent=2)
+    _os.replace(tmp, _VOICE_OPTS_PATH)
+    restarted = False
+    from Project.Sanad.main import live_sub
+    if live_sub is not None:
+        try:
+            if (live_sub.status() or {}).get("running"):
+                await asyncio.to_thread(live_sub.stop)
+                await asyncio.to_thread(live_sub.start)
+                restarted = True
+        except Exception:
+            pass
+    return {"ok": True, "enabled": bool(on), "restarted": restarted}
+
+
+@router.get("/pure-voice-option")
+async def get_pure_voice_option():
+    """PURE VOICE = the live session built exactly like standalone Sanadv1:
+    no tools, no state injections, persona only, no lip-sync markers — the
+    fastest conversation pipe. Resolution: dashboard > env > config."""
+    import os
+    opts = _read_voice_opts()
+    if "pure_voice" in opts:
+        return {"enabled": bool(opts["pure_voice"]), "source": "dashboard"}
+    if os.environ.get("SANAD_PURE_VOICE", "").strip().lower() in ("1", "true", "yes"):
+        return {"enabled": True, "source": "env"}
+    try:
+        from Project.Sanad.core.config_loader import section
+        if bool((section("voice", "sanad_voice") or {}).get("pure_voice", False)):
+            return {"enabled": True, "source": "config"}
+    except Exception:
+        pass
+    return {"enabled": False, "source": "default"}
+
+
+@router.post("/pure-voice-option")
+async def set_pure_voice_option(on: bool = Query(...)):
+    """Persist the pure-voice option and restart the live session."""
+    import json
+    import tempfile
+    import os as _os
+    opts = _read_voice_opts()
+    opts["pure_voice"] = bool(on)
+    _VOICE_OPTS_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp = tempfile.mkstemp(dir=str(_VOICE_OPTS_PATH.parent),
+                               prefix=".voice_options.", suffix=".tmp")
+    with _os.fdopen(fd, "w", encoding="utf-8") as f:
+        json.dump(opts, f, indent=2)
+    _os.replace(tmp, _VOICE_OPTS_PATH)
+    restarted = False
+    from Project.Sanad.main import live_sub
+    if live_sub is not None:
+        try:
+            if (live_sub.status() or {}).get("running"):
+                await asyncio.to_thread(live_sub.stop)
+                await asyncio.to_thread(live_sub.start)
+                restarted = True
+        except Exception:
+            pass
+    return {"ok": True, "enabled": bool(on), "restarted": restarted}
+
+
+@router.get("/internet-search-option")
+async def get_internet_search_option():
+    """INTERNET SEARCH = the live session gets the search_internet tool so Gemini
+    can look up current facts (prices, latest phones, live offers, news) via
+    Google Search grounding, announcing the search first. Resolution:
+    dashboard > env > config."""
+    import os
+    opts = _read_voice_opts()
+    if "internet_search" in opts:
+        return {"enabled": bool(opts["internet_search"]), "source": "dashboard"}
+    if os.environ.get("SANAD_INTERNET_SEARCH", "").strip().lower() in ("1", "true", "yes"):
+        return {"enabled": True, "source": "env"}
+    try:
+        from Project.Sanad.core.config_loader import section
+        if bool((section("voice", "sanad_voice") or {}).get("internet_search", False)):
+            return {"enabled": True, "source": "config"}
+    except Exception:
+        pass
+    return {"enabled": False, "source": "default"}
+
+
+@router.post("/internet-search-option")
+async def set_internet_search_option(on: bool = Query(...)):
+    """Persist the internet-search option and restart the live session so the
+    tool set (and persona addendum) is rebaked."""
+    import json
+    import tempfile
+    import os as _os
+    opts = _read_voice_opts()
+    opts["internet_search"] = bool(on)
+    _VOICE_OPTS_PATH.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp = tempfile.mkstemp(dir=str(_VOICE_OPTS_PATH.parent),
+                               prefix=".voice_options.", suffix=".tmp")
+    with _os.fdopen(fd, "w", encoding="utf-8") as f:
+        json.dump(opts, f, indent=2)
+    _os.replace(tmp, _VOICE_OPTS_PATH)
+    restarted = False
+    from Project.Sanad.main import live_sub
+    if live_sub is not None:
+        try:
+            if (live_sub.status() or {}).get("running"):
+                await asyncio.to_thread(live_sub.stop)
+                await asyncio.to_thread(live_sub.start)
+                restarted = True
+        except Exception:
+            pass
+    return {"ok": True, "enabled": bool(on), "restarted": restarted}
+
+
+@router.post("/answer-now")
+async def answer_now():
+    """Force Gemini to reply immediately instead of waiting out the pause
+    detection (no-op while it is already speaking)."""
+    sub = _sub_or_503()
+    if not (sub.status() or {}).get("running"):
+        raise HTTPException(409, "Live session not running")
+    if not hasattr(sub, "send_answer_now"):
+        raise HTTPException(501, "engine build lacks send_answer_now")
+    await asyncio.to_thread(sub.send_answer_now)
+    return {"ok": True}
+
+
+@router.post("/interrupt")
+async def interrupt_speech():
+    """Instantly stop the AI mid-sentence and go back to listening (works
+    regardless of the barge-in option; no-op if it isn't speaking)."""
+    sub = _sub_or_503()
+    if not (sub.status() or {}).get("running"):
+        raise HTTPException(409, "Live session not running")
+    if not hasattr(sub, "send_interrupt"):
+        raise HTTPException(501, "engine build lacks send_interrupt")
+    await asyncio.to_thread(sub.send_interrupt)
+    return {"ok": True}
+
+
 @router.post("/start")
 async def start_subprocess():
     live_sub = _sub_or_503()

dashboard/routes/voice.py

@@ -173,10 +173,12 @@ async def update_api_key(payload: ApiKeyPayload):
         raise HTTPException(400, "API key cannot be empty.")
     if len(key) < 20:
         raise HTTPException(400, "API key looks too short.")
-    if not key.startswith("AIza"):
+    # Google AI Studio issues classic "AIza..." keys and (since late 2025)
+    # new-format "AQ...." keys — both are plain x-goog-api-key API keys.
+    if not key.startswith(("AIza", "AQ.")):
         raise HTTPException(
             400,
-            "Gemini API keys normally start with 'AIza'. "
+            "Gemini API keys start with 'AIza' (classic) or 'AQ.' (new format). "
             "Double-check you're pasting a Google AI Studio key.",
         )
 

dashboard/static/index.html

@@ -302,7 +302,7 @@
     </div>
     <div style="margin-top:.7rem">
       <div style="display:flex;justify-content:space-between;align-items:center">
-        <label>Speaker Volume (G1 / JBL / Anker)</label>
+        <label>Speaker Volume (G1 / JBL / Anker / Beats Pill)</label>
         <span id="g1-vol-label" style="font-size:.72rem;color:var(--dim)">–</span>
       </div>
       <div class="row" style="margin-top:.25rem;gap:.3rem;align-items:center">
@@ -314,7 +314,7 @@
         <button class="btn btn-ghost btn-sm" onclick="setG1Vol(100,this)" title="Full">100</button>
       </div>
       <div id="g1-vol-status" style="font-size:.65rem;color:var(--dim);margin-top:.2rem">
-        Controls the ACTIVE speaker — G1 chest (DDS) + the selected PulseAudio sink (JBL/Anker). Applies live.
+        Controls the ACTIVE speaker — G1 chest (DDS) + the selected PulseAudio sink (JBL / Anker / Beats Pill / any plugged). Applies live.
       </div>
     </div>
     <div style="margin-top:.6rem">
@@ -405,9 +405,15 @@
     <div class="row">
       <button class="btn btn-success" onclick="startLiveSub(this)">Start</button>
       <button class="btn btn-danger" onclick="stopLiveSub(this)">Stop</button>
+      <button class="btn" style="background:#7c3aed;color:#fff" onclick="instantInterrupt()" title="Instantly stop Sanad mid-sentence and go back to listening (works even with the Interrupt voice option off).">⏹ Interrupt now</button>
+      <button class="btn" style="background:#0e9f6e;color:#fff" onclick="answerNow()" title="Make Sanad answer RIGHT NOW instead of waiting for the end-of-speech pause.">➤ Answer now</button>
       <button id="ls-cam-btn" class="btn btn-sm btn-ghost" onclick="toggleGeminiCamera(this)" title="Stream camera frames to Gemini Live — same toggle as the Recognition tab">Camera: --</button>
       <button id="ls-rec-btn" class="btn btn-sm btn-ghost" onclick="toggleAutoRecord(this)" title="Auto-save every conversation turn to data/recordings/. Toggling takes effect live — no session restart.">Rec: --</button>
       <span id="ls-state" class="badge"></span>
+      <span id="ls-pure" class="badge" onclick="togglePureOpt()" style="cursor:pointer" title="Pure voice: fastest replies (Sanadv1-style — no tools, no standing state, persona only). TOGGLE-AWARE: features you enable (Camera vision, Movement) still work live and announce themselves; disable them and it returns to pure speech. Mask expressions stay off while Pure is on.">Pure: …</span>
+      <span id="ls-direct" class="badge" onclick="toggleDirectOpt()" style="cursor:pointer" title="Direct answer: Sanad replies the moment you pause (fast turn-taking). OFF = waits longer to be sure you finished (better for slow/hesitant speakers). Toggling restarts the live session.">Direct: …</span>
+      <span id="ls-interrupt" class="badge" onclick="toggleInterruptOpt()" style="cursor:pointer" title="Voice interrupt (barge-in) on external speakers (JBL / Beats Pill): speak loudly for ~1s to cut Sanad mid-sentence. Echo-safe — the mic to Gemini stays gated while he speaks. Toggling restarts the live session.">Interrupt: …</span>
+      <span id="ls-search" class="badge" onclick="toggleSearchOpt()" style="cursor:pointer" title="Internet search (Google grounding): Sanad can look up CURRENT facts — today's prices, the latest phones, live Etisalat offers, news. When he needs to, he says 'let me search the internet', looks it up, then answers. OFF = pure voice, no search, fastest replies (recommended for a busy event). Toggling restarts the live session.">Search: …</span>
       <span id="ls-pausemode" class="badge" onclick="toggleLiveHoldBadge()" style="cursor:pointer" title="Pause mode (same as the Saved Records 'Keep Gemini paused' toggle). Auto = pause only during a record, resume after. Manual = Gemini stays paused until you switch back. Click to toggle.">Pause: Auto</span>
       <button class="btn btn-sm mic-mute-shortcut btn-success" onclick="toggleMic()" style="margin-left:auto">Mic: LIVE</button>
       <button class="btn btn-sm spk-mute-shortcut btn-success" onclick="toggleSpeaker()">Speaker: LIVE</button>
@@ -2068,6 +2074,90 @@ function updateLiveHoldUI(hold){
     pm.className='badge '+(hold?'badge-warn':'badge-ok');
   }
 }
+let _pureOpt=null;
+function _paintPureBadge(){
+  const b=document.getElementById('ls-pure');
+  if(!b)return;
+  b.textContent='Pure: '+(_pureOpt===null?'…':(_pureOpt?'ON':'OFF'));
+  b.className='badge '+(_pureOpt?'badge-ok':'');
+}
+async function refreshPureOpt(){
+  try{const r=await api('GET','/api/live-subprocess/pure-voice-option');_pureOpt=!!r.enabled;_paintPureBadge();}catch(e){}
+}
+async function togglePureOpt(){
+  const want=!_pureOpt;
+  try{
+    const r=await api('POST','/api/live-subprocess/pure-voice-option?on='+want);
+    _pureOpt=!!r.enabled;_paintPureBadge();
+    toast('Pure voice '+(_pureOpt?'ON — Sanadv1-style session':'OFF — full features')+(r.restarted?' (session restarted)':''));
+  }catch(e){toast('Pure toggle failed: '+(e.message||e));}
+}
+setTimeout(refreshPureOpt,1000);
+let _searchOpt=null;
+function _paintSearchBadge(){
+  const b=document.getElementById('ls-search');
+  if(!b)return;
+  b.textContent='Search: '+(_searchOpt===null?'…':(_searchOpt?'ON':'OFF'));
+  b.className='badge '+(_searchOpt?'badge-ok':'');
+}
+async function refreshSearchOpt(){
+  try{const r=await api('GET','/api/live-subprocess/internet-search-option');_searchOpt=!!r.enabled;_paintSearchBadge();}catch(e){}
+}
+async function toggleSearchOpt(){
+  const want=!_searchOpt;
+  try{
+    const r=await api('POST','/api/live-subprocess/internet-search-option?on='+want);
+    _searchOpt=!!r.enabled;_paintSearchBadge();
+    toast('Internet search '+(_searchOpt?'ON — Sanad can look up current info':'OFF — pure voice, no search')+(r.restarted?' (session restarted)':''));
+  }catch(e){toast('Search toggle failed: '+(e.message||e));}
+}
+setTimeout(refreshSearchOpt,1050);
+let _dirOpt=null;
+function _paintDirectBadge(){
+  const b=document.getElementById('ls-direct');
+  if(!b)return;
+  b.textContent='Direct: '+(_dirOpt===null?'…':(_dirOpt?'ON':'OFF'));
+  b.className='badge '+(_dirOpt?'badge-ok':'');
+}
+async function refreshDirectOpt(){
+  try{const r=await api('GET','/api/live-subprocess/direct-answer-option');_dirOpt=!!r.enabled;_paintDirectBadge();}catch(e){}
+}
+async function toggleDirectOpt(){
+  const want=!_dirOpt;
+  try{
+    const r=await api('POST','/api/live-subprocess/direct-answer-option?on='+want);
+    _dirOpt=!!r.enabled;_paintDirectBadge();
+    toast('Direct answer '+(_dirOpt?'ON':'OFF')+(r.restarted?' — live session restarted':''));
+  }catch(e){toast('Direct toggle failed: '+(e.message||e));}
+}
+setTimeout(refreshDirectOpt,950);
+let _intOpt=null;
+function _paintInterruptBadge(){
+  const b=document.getElementById('ls-interrupt');
+  if(!b)return;
+  b.textContent='Interrupt: '+(_intOpt===null?'…':(_intOpt?'ON':'OFF'));
+  b.className='badge '+(_intOpt?'badge-ok':'');
+}
+async function refreshInterruptOpt(){
+  try{const r=await api('GET','/api/live-subprocess/interrupt-option');_intOpt=!!r.enabled;_paintInterruptBadge();}catch(e){}
+}
+async function answerNow(){
+  try{await api('POST','/api/live-subprocess/answer-now');toast('Answering now…');}
+  catch(e){toast('Answer-now failed: '+(e.message||e));}
+}
+async function instantInterrupt(){
+  try{await api('POST','/api/live-subprocess/interrupt');toast('Interrupted — listening');}
+  catch(e){toast('Interrupt failed: '+(e.message||e));}
+}
+async function toggleInterruptOpt(){
+  const want=!_intOpt;
+  try{
+    const r=await api('POST','/api/live-subprocess/interrupt-option?on='+want);
+    _intOpt=!!r.enabled;_paintInterruptBadge();
+    toast('Interrupt '+(_intOpt?'ON':'OFF')+(r.restarted?' — live session restarted':''));
+  }catch(e){toast('Interrupt toggle failed: '+(e.message||e));}
+}
+setTimeout(refreshInterruptOpt,900);
 async function refreshPlaybackStatus(){
   try{
     const s=await api('GET','/api/records/playback-status');

gemini/script.py

@@ -55,11 +55,23 @@ _MODEL = os.environ.get(
     "SANAD_GEMINI_MODEL",
     "gemini-2.5-flash-native-audio-preview-12-2025",
 )
-_MIC_GAIN = _SV.get("mic_gain", 1.0)
+# Mic gain: amplifies mic samples before they reach Gemini — the single
+# biggest lever for slow/failed speech detection at a distance (weak audio
+# transcribes but often yields NO reply turn). Env override for per-robot
+# tuning; watch for clipping if the speaker stands close (>3.0 rarely wise).
+try:
+    _MIC_GAIN = float(os.environ.get("SANAD_MIC_GAIN", "") or _SV.get("mic_gain", 1.0))
+except ValueError:
+    _MIC_GAIN = _SV.get("mic_gain", 1.0)
 _SESSION_TIMEOUT = _SV.get("session_timeout_sec", 660)
 _MAX_RECONNECT_DELAY = _SV.get("max_reconnect_delay_sec", 30)
 _MAX_CONSECUTIVE_ERRORS = _SV.get("max_consecutive_errors", 10)
-_NO_MESSAGES_TIMEOUT = _SV.get("no_messages_timeout_sec", 30)
+# How long with ZERO messages from Gemini before we declare the session dead.
+# In a QUIET room Gemini legitimately sends nothing — 30s was killing healthy
+# sessions every ~40s (reconnect churn + unsolicited greeting + a 1-3s deaf
+# window each time). REAL deaths are caught fast by the send path ("mic send
+# failed: 1011 ..."), so this can be patient.
+_NO_MESSAGES_TIMEOUT = _SV.get("no_messages_timeout_sec", 180)
 # Extra mic-gate time after the AI stops, on loud external-speaker profiles
 # (JBL) — covers the speaker buffer + room reverb so it doesn't hear its tail.
 _ECHO_TAIL_SEC = _SV.get("echo_tail_sec", 0.6)
@@ -77,6 +89,67 @@ _JBL_BARGE_CHUNKS = _SV.get("jbl_barge_chunks", 9)
 # FADED (a gap between words/numbers). In that window barge-in drops to a low,
 # sensitive bar so the user can interrupt; while audio is flowing it stays high.
 _JBL_BLEED_FADE_SEC = _SV.get("jbl_bleed_fade_sec", 0.5)
+# OPTION: allow voice barge-in on loud external speakers (JBL / Beats Pill).
+# The mic feed to Gemini STAYS fully gated while the AI speaks (echo-safe);
+# this only enables the LOCAL energy detector to cut the speaker, using the
+# stricter external parameters (threshold×margin, longer grace, more sustained
+# chunks) so speaker bleed can't false-trigger. Env > config; default OFF.
+_EXTERNAL_BARGE_IN = (
+    os.environ.get("SANAD_EXTERNAL_BARGE_IN", "").strip().lower() in ("1", "true", "yes")
+    or bool(_BI.get("external_barge_in", False))
+)
+# OPTION: "direct answer" — reply as soon as the speaker pauses. Flips the
+# Live-API end-of-speech sensitivity from LOW (conservative: waits to be sure
+# you finished, feels slow) to HIGH (immediate turn-taking; may cut off very
+# hesitant speakers mid-thought). Env > config; dashboard file has final say.
+_DIRECT_ANSWER = (
+    os.environ.get("SANAD_DIRECT_ANSWER", "").strip().lower() in ("1", "true", "yes")
+    or bool(_VAD.get("direct_answer", False))
+)
+
+# OPTION: PURE VOICE — make the live session EXACTLY like standalone Sanadv1:
+# no tools (no function-calling deliberation), no state injections, persona
+# only (no face addendum), no lip-sync markers, no frame/state loops. The
+# fastest possible conversation pipe; mask/nav/vision hooks are dormant.
+_PURE_VOICE = (
+    os.environ.get("SANAD_PURE_VOICE", "").strip().lower() in ("1", "true", "yes")
+    or bool(_SV.get("pure_voice", False))
+)
+
+# OPTION: INTERNET SEARCH (Google Search grounding). When ON, the live session
+# gets ONE extra tool, `search_internet(query)`, so the model can look up
+# CURRENT facts (today's prices, the newest phones, live Etisalat offers, news).
+# The tool description tells the model to SAY it will check first, then call the
+# tool — so the robot announces "let me search the internet", the parent runs
+# the grounded generateContent call in a worker thread, and the model speaks the
+# result. Default OFF → pure voice stays tool-free and Sanadv1-fast; only when
+# a search actually fires does the (announced) grounding delay occur.
+_INTERNET_SEARCH = (
+    os.environ.get("SANAD_INTERNET_SEARCH", "").strip().lower() in ("1", "true", "yes")
+    or bool(_SV.get("internet_search", False))
+)
+# Model used for the grounded web-search call (generateContent, NOT the live
+# native-audio model). gemini-3.5-flash grounds well and cites the source site.
+_SEARCH_MODEL = os.environ.get("SANAD_SEARCH_MODEL", "gemini-3.5-flash")
+
+# The dashboard toggles (data/.voice_options.json, written by the
+# /api/live-subprocess/*-option endpoints, which then restart this child)
+# have the FINAL say over env/config.
+try:
+    import json as _vo_json
+    with open(BASE_DIR / "data" / ".voice_options.json", encoding="utf-8") as _vo_f:
+        _vo = _vo_json.load(_vo_f)
+    if isinstance(_vo, dict):
+        if "external_barge_in" in _vo:
+            _EXTERNAL_BARGE_IN = bool(_vo["external_barge_in"])
+        if "direct_answer" in _vo:
+            _DIRECT_ANSWER = bool(_vo["direct_answer"])
+        if "pure_voice" in _vo:
+            _PURE_VOICE = bool(_vo["pure_voice"])
+        if "internet_search" in _vo:
+            _INTERNET_SEARCH = bool(_vo["internet_search"])
+except Exception:
+    pass
 
 _CHUNK_BYTES = CHUNK_SIZE * 2
 _SILENCE_PCM = b"\x00" * _CHUNK_BYTES
@@ -88,6 +161,15 @@ _SILENCE_PCM = b"\x00" * _CHUNK_BYTES
 # safe to read from the asyncio loops.
 _INPUT_PAUSED = threading.Event()
 
+# Set by an "interrupt:" stdin command (the dashboard's Interrupt-now button).
+# The mic loop consumes it: stops the speaker mid-sentence, back to listening.
+_MANUAL_INTERRUPT = threading.Event()
+
+# Set by an "answernow:" stdin command (the dashboard's Answer-now button).
+# The mic loop feeds ~0.6s of silence so the server VAD closes the user's turn
+# immediately → Gemini answers right away instead of waiting out the pause.
+_FORCE_ANSWER = threading.Event()
+
 # ── Recognition (camera + face gallery) tunables ──
 _RECOG_STATE_PATH = Path(os.environ.get(
     "SANAD_RECOGNITION_STATE_PATH",
@@ -195,6 +277,72 @@ def _nav_function_declarations() -> list:
     ]
 
 
+def _search_function_declaration() -> list:
+    """Live tool: search the internet (Google Search grounding) for current facts."""
+    S, T = types.Schema, types.Type
+    return [
+        types.FunctionDeclaration(
+            name="search_internet",
+            description=(
+                "Search the LIVE internet for CURRENT, real-time facts you cannot "
+                "otherwise know: today's prices, the newest phone / device models, "
+                "current Etisalat / e& offers and promotions, product availability, "
+                "or recent news. IMPORTANT: BEFORE you call this, first SAY one "
+                "short sentence out loud telling the user you will check — e.g. "
+                "\"Let me search the internet for that.\" / "
+                "\"دعني أبحث لك على الإنترنت.\" — "
+                "then call this with a clear English search query. After the "
+                "result returns, answer the user naturally from it. Do NOT say the "
+                "tool name out loud. Only use this for genuinely current / "
+                "time-sensitive questions, not for things you already know."
+            ),
+            parameters=S(type=T.OBJECT, properties={
+                "query": S(type=T.STRING,
+                           description="The web-search query, phrased in English."),
+            }, required=["query"]),
+        ),
+    ]
+
+
+# Appended to the persona ONLY when internet search is enabled, so the model
+# knows it can (and how to) look things up. Kept short — the tool description
+# carries the detail.
+_SEARCH_PROMPT_ADDENDUM = (
+    "\n\nYou can look things up on the live internet with the search_internet "
+    "tool. Use it whenever the user asks about current prices, the latest phones "
+    "or devices, today's Etisalat/e& offers, availability, or recent news — "
+    "anything you can't be certain is up to date. ALWAYS say a short 'let me "
+    "check the internet' line first (in the user's language), then call the tool, "
+    "then answer from the result. Keep answers concise and, for exact prices, add "
+    "that a sales advisor can confirm the final figure. "
+    "CRITICAL: after the search result returns, CONTINUE where you left off — "
+    "NEVER repeat or re-say any sentence you already spoke, including the 'let me "
+    "check the internet' lead-in. If your reply is complete, stop and wait."
+)
+
+
+def _run_grounding_search(query: str, api_key: str) -> dict:
+    """Blocking Google-Search-grounded generateContent call (run in a thread).
+    Returns {ok, answer} for the model to speak, or {ok: False, reason}."""
+    try:
+        client = genai.Client(api_key=api_key)
+        resp = client.models.generate_content(
+            model=_SEARCH_MODEL,
+            contents=query,
+            config=types.GenerateContentConfig(
+                tools=[types.Tool(google_search=types.GoogleSearch())],
+            ),
+        )
+        answer = (getattr(resp, "text", "") or "").strip()
+        if not answer:
+            return {"ok": False, "reason": "empty_result"}
+        # Cap length so the spoken reply stays reasonable.
+        return {"ok": True, "answer": answer[:1500]}
+    except Exception as exc:
+        log.warning("internet search failed: %s", exc)
+        return {"ok": False, "reason": "search_failed", "detail": str(exc)[:200]}
+
+
 # Emotions Gemini can show on the LED face (a subset of the mask's frames that
 # read as feelings — the talk/blink/gaze frames are driven automatically).
 _FACE_EMOTIONS = ("smile", "laugh", "heart", "love", "sad", "surprised",
@@ -221,7 +369,10 @@ _FACE_PROMPT_ADDENDUM = (
     "Instagram, or to see/show your social media, ALWAYS call show_social with "
     "'bu_sunaidah' (@bu.sunaidah) or 'yslootahtech' (@yslootahtech) to display "
     "the QR on your face. These tools are silent — never say the tool name, the "
-    "emotion, or any bracket marker out loud."
+    "emotion, or any bracket marker out loud. CRITICAL: after any tool returns, "
+    "CONTINUE exactly where you left off — NEVER repeat or re-say sentences you "
+    "already spoke before the tool call. If your reply was already complete, do "
+    "not speak again; wait silently for the user."
 )
 
 
@@ -239,7 +390,8 @@ def _face_function_declarations() -> list:
                 "surprised when astonished, confused when you didn't understand, wink "
                 "when joking, sad when empathizing, cool when playful, sleepy when "
                 "tired, angry only rarely. Your mouth already lip-syncs on its own — "
-                "this is ONLY the emotion, not the mouth."
+                "this is ONLY the emotion, not the mouth. After this tool returns, never "
+                "repeat what you already said — continue your sentence or stay silent."
             ),
             parameters=S(type=T.OBJECT, properties={
                 "emotion": S(type=T.STRING, enum=list(_FACE_EMOTIONS),
@@ -301,7 +453,7 @@ _PROFILE_LOCK = threading.Lock()
 _PROFILE_PENDING: dict = {"id": None, "reason": ""}
 
 _VALID_PROFILES = {"builtin", "anker", "anker_powerconf",
-                   "hollyland_builtin", "jbl_builtin_mic"}
+                   "hollyland_builtin", "jbl_builtin_mic", "beats_pill"}
 
 
 def _stdin_watcher() -> None:
@@ -334,6 +486,10 @@ def _stdin_watcher() -> None:
                     with _LATEST_FRAME_LOCK:
                         _LATEST_FRAME["bytes"] = data
                         _LATEST_FRAME["ts"] = time.time()
+            elif line.startswith("interrupt:"):
+                _MANUAL_INTERRUPT.set()
+            elif line.startswith("answernow:"):
+                _FORCE_ANSWER.set()
             elif line.startswith("state:"):
                 try:
                     payload = json.loads(line[len("state:"):])
@@ -417,7 +573,13 @@ class GeminiBrain:
         self._swap_lock: Optional[asyncio.Lock] = None  # built in run()
         self._recorder = recorder
         self._voice = voice_name or GEMINI_VOICE
-        self._system_prompt = (system_prompt or "") + _FACE_PROMPT_ADDENDUM
+        # PURE VOICE: persona only — no face/tool addendum (Sanadv1 parity).
+        self._system_prompt = ((system_prompt or "") if _PURE_VOICE
+                               else (system_prompt or "") + _FACE_PROMPT_ADDENDUM)
+        # Internet-search addendum is independent of pure voice — the search
+        # tool is available in either mode when the option is enabled.
+        if _INTERNET_SEARCH:
+            self._system_prompt += _SEARCH_PROMPT_ADDENDUM
         self._api_key = GEMINI_API_KEY
         self._stop_flag = asyncio.Event()
         # per-session state (reset in the outer reconnect loop)
@@ -511,6 +673,10 @@ class GeminiBrain:
                     if self._swap_lock is None:
                         self._swap_lock = asyncio.Lock()
 
+                    # PURE VOICE is TOGGLE-AWARE: all loops run (cheap when
+                    # their feature is off) — pure-ness is enforced per-feature
+                    # (announce guards + tools) so enabling Camera/Movement in
+                    # the dashboard works live, and disabling returns to pure.
                     try:
                         await asyncio.wait_for(
                             asyncio.gather(
@@ -573,10 +739,15 @@ class GeminiBrain:
                         types.StartSensitivity,
                         _VAD.get("start_sensitivity", "START_SENSITIVITY_HIGH"),
                     ),
-                    end_of_speech_sensitivity=getattr(
+                    # "Direct answer" option: HIGH = reply the moment the
-                        types.EndSensitivity,
+                    # speaker pauses; otherwise the configured (LOW) default
-                        _VAD.get("end_sensitivity", "END_SENSITIVITY_LOW"),
+                    # waits longer to be sure the speaker finished.
-                    ),
+                    end_of_speech_sensitivity=(
+                        types.EndSensitivity.END_SENSITIVITY_HIGH
+                        if _DIRECT_ANSWER else getattr(
+                            types.EndSensitivity,
+                            _VAD.get("end_sensitivity", "END_SENSITIVITY_LOW"),
+                        )),
                     prefix_padding_ms=_VAD.get("prefix_padding_ms", 20),
                     silence_duration_ms=_VAD.get("silence_duration_ms", 200),
                 ),
@@ -591,11 +762,28 @@ class GeminiBrain:
             # stop_navigation). Disable with SANAD_NAV_TOOLS=0.
             # Native function-calling: nav tools (if enabled) + the always-on
             # expressive-face / social-QR tools (set_expression / show_social).
-            tools=[types.Tool(function_declarations=(
+            # PURE VOICE: no tools (the reply-latency tax) — EXCEPT nav tools
-                (_nav_function_declarations() if _NAV_TOOLS_ENABLED else [])
+            # when movement is enabled (walking commands need them). Face
-                + _face_function_declarations()))],
+            # tools stay off in pure.
+            tools=self._build_tools(),
         )
 
+    def _build_tools(self) -> Any:
+        """Assemble the live-session function tools. PURE VOICE keeps this None
+        (the reply-latency tax) UNLESS movement is on (nav tools) or internet
+        search is on (the search tool). Face tools stay off in pure voice."""
+        decls: list = []
+        # Nav tools: whenever NOT (pure and not moving) and nav is enabled.
+        if _NAV_TOOLS_ENABLED and (self._movement_enabled or not _PURE_VOICE):
+            decls += _nav_function_declarations()
+        # Face/social tools: only outside pure voice.
+        if not _PURE_VOICE:
+            decls += _face_function_declarations()
+        # Internet-search tool: whenever the option is enabled (either mode).
+        if _INTERNET_SEARCH:
+            decls += _search_function_declaration()
+        return [types.Tool(function_declarations=decls)] if decls else None
+
     # ─── state helpers ────────────────────────────────────
 
     def _reset_turn_state(self) -> None:
@@ -635,6 +823,7 @@ class GeminiBrain:
         loop = asyncio.get_event_loop()
         loud_count = 0
         last_activity = time.time()
+        force_silence_until = 0.0   # Answer-now button: silence burst window
 
         while not self._done.is_set() and not self._stop_flag.is_set():
             try:
@@ -649,6 +838,27 @@ class GeminiBrain:
             data = samples.tobytes()
             energy = _audio_energy(data)
             now = time.time()
+
+            # Dashboard Answer-now button: feed the server VAD a silence burst
+            # so it closes the user's turn NOW → immediate reply.
+            if _FORCE_ANSWER.is_set():
+                _FORCE_ANSWER.clear()
+                if not self._speaking:
+                    force_silence_until = now + 0.6
+                    log.info("ANSWER NOW (dashboard) — forcing turn end")
+
+            # Dashboard Interrupt-now button: cut speech instantly AND stay
+            # SILENT until the user speaks — arming the greeting suppression
+            # also swallows any in-flight or reconnect-greeting audio, so a
+            # press while "listening" can never be followed by Sanad talking
+            # on his own.
+            if _MANUAL_INTERRUPT.is_set():
+                _MANUAL_INTERRUPT.clear()
+                log.info("MANUAL INTERRUPT (dashboard)%s",
+                         "" if self._speaking else " — idle; muted until user speaks")
+                self._interrupt("manual")
+                self._barge_block_until = now + cooldown
+                self._suppress_greeting = True
             # On the JBL (loud external speaker) the head mic hears the robot's
             # OWN voice as loud as the user. We FULLY gate the mic to Gemini while
             # it speaks (+ a short echo tail) so it NEVER hears itself, and we
@@ -657,24 +867,38 @@ class GeminiBrain:
             # model. (Reliable JBL interrupt needs AEC; the only PulseAudio mic is
             # dead, so that's separate work.) The chest speaker (builtin) keeps
             # light quiet-frame suppression + working barge-in (firmware AEC).
-            full_gate = "jbl" in (self._current_profile_id or "")
+            # The Beats Pill (USB) is the same regime as the JBL: loud external
+            # speaker with a raw AEC-less mic inches away — the mic feed to
+            # Gemini is ALWAYS fully gated while speaking (echo-safe). Voice
+            # barge-in on external speakers is an OPTION (_EXTERNAL_BARGE_IN):
+            # the LOCAL detector cuts the speaker using stricter parameters.
+            external = any(t in (self._current_profile_id or "")
+                           for t in ("jbl", "beats"))
+            full_gate = external
+            allow_barge = (not external) or _EXTERNAL_BARGE_IN
 
-            # Barge-in: sustained user energy cuts the AI — chest profile only.
+            # Barge-in: sustained user energy cuts the AI.
-            if self._speaking and not full_gate and now >= self._barge_block_until:
+            if self._speaking and allow_barge and now >= self._barge_block_until:
-                if (now - self._ai_speak_start) >= grace:
+                eff_grace = _JBL_BARGE_GRACE if external else grace
-                    if energy > threshold:
+                eff_threshold = threshold * _JBL_BLEED_MARGIN if external else threshold
+                eff_chunks = _JBL_BARGE_CHUNKS if external else chunks_needed
+                if (now - self._ai_speak_start) >= eff_grace:
+                    if energy > eff_threshold:
                         loud_count += 1
                     else:
                         loud_count = max(0, loud_count - 1)
-                    if loud_count > chunks_needed:
+                    if loud_count > eff_chunks:
-                        log.info("BARGE-IN (e=%d)", energy)
+                        log.info("BARGE-IN (e=%d, external=%s)", energy, external)
                         self._interrupt("barge-in")
                         loud_count = 0
                         self._barge_block_until = now + cooldown
 
             # Echo suppression: mask the mic so the model doesn't hear its own bleed.
             send_data = data
-            if _INPUT_PAUSED.is_set():
+            if now < force_silence_until:
+                # Answer-now: mask the mic so the server VAD sees pure silence.
+                send_data = _SILENCE_PCM
+            elif _INPUT_PAUSED.is_set():
                 # Paused for a record playback — feed silence so Gemini neither
                 # hears the record nor keeps talking over it.
                 send_data = _SILENCE_PCM
@@ -835,7 +1059,8 @@ class GeminiBrain:
                                     # mouth-open level (0..3) from this chunk's RMS,
                                     # throttled. Parsed by GeminiSubprocess._reader_loop.
                                     _mnow = time.time()
-                                    if _mnow - getattr(self, "_mouth_t", 0.0) >= 0.08:
+                                    if not _PURE_VOICE and \
+                                            _mnow - getattr(self, "_mouth_t", 0.0) >= 0.08:
                                         _rms = (float(np.sqrt(np.mean(
                                             audio.astype(np.float32) ** 2))) if audio.size else 0.0)
                                         # Lower thresholds bias the mouth more open
@@ -892,6 +1117,10 @@ class GeminiBrain:
 
     async def _announce_vision_state(self, session: Any, enabled: bool,
                                      is_toggle: bool) -> None:
+        if _PURE_VOICE and not enabled and not is_toggle:
+            # PURE: skip the "disabled" standing-context injection — only
+            # ENABLED features get announced/used (toggles always announce).
+            return
         if is_toggle and enabled:
             text = (
                 "[VISION ON] Your camera was just enabled — you can now see "
@@ -935,6 +1164,10 @@ class GeminiBrain:
 
     async def _announce_facerec_state(self, session: Any, enabled: bool,
                                       is_toggle: bool) -> None:
+        if _PURE_VOICE and not enabled and not is_toggle:
+            # PURE: skip the "disabled" standing-context injection — only
+            # ENABLED features get announced/used (toggles always announce).
+            return
         if is_toggle and enabled:
             text = (
                 "[FACE RECOGNITION ON] Face recognition was just enabled — "
@@ -979,6 +1212,10 @@ class GeminiBrain:
 
     async def _announce_zonerec_state(self, session: Any, enabled: bool,
                                       is_toggle: bool) -> None:
+        if _PURE_VOICE and not enabled and not is_toggle:
+            # PURE: skip the "disabled" standing-context injection — only
+            # ENABLED features get announced/used (toggles always announce).
+            return
         if is_toggle and enabled:
             text = (
                 "[ZONE RECOGNITION ON] You were just given the zones and places "
@@ -1140,6 +1377,13 @@ class GeminiBrain:
                         "places": r.get("places", [])}
             if name == "stop_navigation":
                 return await asyncio.to_thread(_nav_api, "POST", "/api/nav/cancel", None)
+            if name == "search_internet":
+                query = str(args.get("query") or "").strip()
+                if not query:
+                    return {"ok": False, "reason": "no_query"}
+                log.info("[[SEARCH:%s]]", query)  # breadcrumb for the dashboard
+                return await asyncio.to_thread(
+                    _run_grounding_search, query, self._api_key)
             if name == "set_expression":
                 emotion = str(args.get("emotion") or "").strip().lower()
                 if emotion not in _FACE_EMOTIONS:
@@ -1165,6 +1409,10 @@ class GeminiBrain:
 
     async def _announce_movement_state(self, session: Any, enabled: bool,
                                        is_toggle: bool) -> None:
+        if _PURE_VOICE and not enabled and not is_toggle:
+            # PURE: skip the "disabled" standing-context injection — only
+            # ENABLED features get announced/used (toggles always announce).
+            return
         if is_toggle and enabled:
             text = (
                 "[MOVEMENT ON] Walking is now enabled — you can move when the "
@@ -1490,6 +1738,12 @@ class GeminiBrain:
                 await self._announce_movement_state(
                     session, self._movement_enabled, is_toggle=True,
                 )
+                if _PURE_VOICE:
+                    # PURE: tools are baked per-session (nav tools only while
+                    # movement is on) — cycle the session so the toggle takes
+                    # effect on the function-calling side too.
+                    log.info("pure-voice: cycling session to rebake nav tools")
+                    self._done.set()
 
             # Auto-record toggle — flip the recorder live (no session restart).
             if new_state.record_enabled != last_state.record_enabled:
@@ -1520,6 +1774,12 @@ class GeminiBrain:
             await asyncio.sleep(max(period, backoff))
             if not self._vision_enabled:
                 continue
+            # Never compete with the INCOMING audio stream while the AI is
+            # speaking — a frame upload on the shared websocket delays audio
+            # chunks, which breaks the chest speaker (tiny firmware buffer;
+            # the pulse/Pill path hides it behind its own buffering).
+            if self._speaking or (time.time() - self._last_ai_audio) < 0.5:
+                continue
             with _LATEST_FRAME_LOCK:
                 data = _LATEST_FRAME.get("bytes")
                 ts = _LATEST_FRAME.get("ts", 0.0)

gemini/subprocess.py

@@ -38,7 +38,9 @@ _FRAME_FORWARD_INTERVAL_S = float(_LS_CFG.get("frame_forward_interval_sec", 0.5)
 
 # Audio profile watcher — poll pactl for the Anker USB device at this
 # interval, send "profile:<json>" to the child on every state change.
-_AUDIO_WATCH_INTERVAL_S = float(_LS_CFG.get("audio_watch_interval_sec", 1.5))
+# 5s: each poll shells out to pactl several times (detect_plugged_profiles) —
+# 1.5s was a constant subprocess storm for a rare event (device hot-plug).
+_AUDIO_WATCH_INTERVAL_S = float(_LS_CFG.get("audio_watch_interval_sec", 5.0))
 
 # The Anker profile id, as defined in voice/audio_devices.py. When this
 # profile is fully plugged (both sink + source present), we switch the
@@ -562,7 +564,7 @@ class GeminiSubprocess:
         if the process isn't running or stdin is closed."""
         pid = (profile_id or "").strip().lower()
         if pid not in {"builtin", "anker", "anker_powerconf",
-                       "hollyland_builtin", "jbl_builtin_mic"}:
+                       "hollyland_builtin", "jbl_builtin_mic", "beats_pill"}:
             log.warning("send_profile: ignoring unknown profile %r", profile_id)
             return
         payload: dict[str, Any] = {"id": pid}
@@ -580,6 +582,16 @@ class GeminiSubprocess:
         owns the chest speaker, then resumes. No-op if not running."""
         self._send_stdin("pause:%d\n" % (1 if paused else 0))
 
+    def send_interrupt(self) -> None:
+        """Instantly stop the AI's current speech (dashboard Interrupt-now
+        button) — the child stops the speaker and returns to listening."""
+        self._send_stdin("interrupt:1\n")
+
+    def send_answer_now(self) -> None:
+        """Force Gemini to answer immediately (dashboard Answer-now button) —
+        the child feeds the server VAD a short silence burst to close the turn."""
+        self._send_stdin("answernow:1\n")
+
     def _audio_watcher(self) -> None:
         """Background thread — poll pactl for the Anker USB device, signal
         the child on every plug/unplug edge transition.

vision/camera.py

@@ -23,6 +23,7 @@ from __future__ import annotations
 
 import base64
 import os
+import re
 import threading
 import time
 from typing import Optional
@@ -386,6 +387,46 @@ class CameraDaemon:
             log.info("USB camera unavailable: %s", exc)
             return None
 
+        # Pinned DEVICE PATH — strongest pin. A /dev/v4l/by-path/... symlink is
+        # keyed to the physical USB port + interface, so it survives the
+        # re-enumeration that silently moves raw /dev/videoN indices around
+        # (observed on the D435i: colour node drifted 12→6→5 across replugs).
+        # Resolve the symlink to the current /dev/videoN and open that.
+        dev_path = os.environ.get("SANAD_CAMERA_DEVICE", "").strip()
+        if dev_path:
+            # Glob patterns allowed: a D435i that renegotiates USB3→USB2
+            # changes its bus/port prefix, but the INTERFACE suffix is
+            # invariant — e.g. "/dev/v4l/by-path/*-usb-*:1.3-video-index0"
+            # always names the RGB capture node wherever it enumerates.
+            import glob as _glob
+            if any(c in dev_path for c in "*?["):
+                candidates = sorted(_glob.glob(dev_path))
+                if not candidates:
+                    log.warning("USB camera: SANAD_CAMERA_DEVICE pattern %s "
+                                "matched nothing — falling back to index pin "
+                                "/ scan", dev_path)
+            else:
+                candidates = [dev_path]
+            for cand in candidates:
+                real = os.path.realpath(cand)
+                m = re.match(r"^/dev/video(\d+)$", real)
+                if not m:
+                    log.warning("USB camera: %s does not resolve to a "
+                                "/dev/videoN node (got %s)", cand, real)
+                    continue
+                backend = self._open_usb_index(int(m.group(1)), w, h, f, cv2)
+                if backend is not None:
+                    fw, fh = backend["frame_wh"]
+                    log.info("USB camera: pinned by path %s → /dev/video%d "
+                             "(%dx%d, %s)", cand, backend["index"], fw, fh,
+                             "colour" if backend["is_color"] else "grayscale/IR")
+                    return backend
+                log.warning("USB camera: pinned path %s (→ %s) unusable",
+                            cand, real)
+            if candidates:
+                log.warning("USB camera: no SANAD_CAMERA_DEVICE candidate "
+                            "usable — falling back to index pin / scan")
+
         # Pinned index — accept whatever it is (colour or not).
         explicit = os.environ.get("SANAD_CAMERA_USB_INDEX", "").strip()
         if explicit.isdigit():

voice/audio_devices.py

@@ -50,6 +50,20 @@ class AudioProfile:
     description: str = ""
     sink_sample_rate: int = 0  # 0 = use device default
     source_sample_rate: int = 0
+    # True → the profile's REAL mic is the G1 firmware UDP-multicast stream
+    # (voice/audio_io), so a missing PulseAudio source must not disqualify it.
+    # Some G1 units never expose the chest mic to the Jetson's PulseAudio
+    # (I2S capture reads silence) while the UDP mic works fine — detection
+    # then falls back to a synthetic "udp:" source (see apply_selection).
+    udp_mic_fallback: bool = False
+    # True → the profile's REAL speaker is the G1 firmware chest speaker driven
+    # over DDS (voice/audio_io BuiltinSpeaker.PlayStream), NOT a PulseAudio sink.
+    # The Jetson's platform-sound card is input-only on this robot (it has no
+    # output sink at all), so a missing PulseAudio sink must NOT disqualify the
+    # chest profile — the chest is always usable. Detection then falls back to
+    # a synthetic "dds:" sink (see apply_selection, which skips
+    # `pactl set-default-sink` for it, exactly like the "udp:" source).
+    dds_speaker_fallback: bool = False
 
 
 # Built-in device profiles.
@@ -77,6 +91,8 @@ PROFILES: list[AudioProfile] = [
         sink_pattern="platform-sound",
         source_pattern="alsa_input.platform-sound",
         description="Jetson / G1 built-in audio chip. (Default)",
+        udp_mic_fallback=True,   # chest mic = G1 firmware UDP stream, not pulse
+        dds_speaker_fallback=True,  # chest speaker = G1 DDS PlayStream, not pulse
     ),
     AudioProfile(
         id="hollyland_builtin",
@@ -92,6 +108,18 @@ PROFILES: list[AudioProfile] = [
         source_pattern="powerconf,anker",
         description="Anker PowerConf USB conference unit — mic + speaker on the same device.",
     ),
+    AudioProfile(
+        id="beats_pill",
+        label="Beats Pill (mic + speaker)",
+        # USB-C: kernel names the card "Pill"; pulse sinks/sources carry
+        # "Beats_Pill" (auto-probed or the manual module-alsa-sink/-source
+        # loads from /etc/pulse/default.pa — pulse fails to probe the Pill's
+        # output profile on some kernels). Bluetooth: bluez sink.
+        sink_pattern="beats_pill,beats,pill",
+        source_pattern="beats_pill,beats,pill",
+        description="Beats Pill (USB-C or Bluetooth) — speaker + speakerphone mic.",
+        udp_mic_fallback=True,   # mic falls back to the G1 UDP mic if absent
+    ),
     AudioProfile(
         id="jbl_builtin_mic",
         label="JBL speaker + built-in mic",
@@ -101,6 +129,7 @@ PROFILES: list[AudioProfile] = [
         # The JBL has NO microphone → input stays on the G1 built-in mic.
         source_pattern="alsa_input.platform-sound",
         description="JBL Bluetooth speaker for output + the G1 built-in microphone for input (the JBL has no mic).",
+        udp_mic_fallback=True,   # builtin mic = G1 UDP stream on units without the alsa source
     ),
 ]
 
@@ -488,6 +517,10 @@ def ensure_card_input_capable(card_pattern: str) -> bool:
     return switched_any
 
 
+# Profiles that already logged the (permanent) UDP-mic fallback notice.
+_UDP_FALLBACK_LOGGED: set[str] = set()
+
+
 def detect_plugged_profiles() -> list[dict[str, Any]]:
     """Return all profiles whose sink AND source are currently plugged in.
 
@@ -507,6 +540,17 @@ def detect_plugged_profiles() -> list[dict[str, Any]]:
     refreshed_sources = False
     for prof in PROFILES:
         sink = find_first_match(sinks, prof.sink_pattern)
+        if not sink and prof.dds_speaker_fallback:
+            # The profile's real speaker is the G1 chest driven over DDS
+            # (AudioClient.PlayStream), not a PulseAudio sink. On this robot the
+            # Jetson audio card is input-only (no output sink exists at all), so
+            # a missing pulse sink must NOT disqualify the chest — it is always
+            # usable. Synthesise a "dds:" sink so the profile stays detected and
+            # selectable; apply_selection() skips set-default-sink for "dds:".
+            sink = {
+                "name": "dds:g1-chest-speaker",
+                "description": "G1 firmware chest speaker (DDS, no PulseAudio sink)",
+            }
         if not sink:
             continue
         src = find_first_match(sources, prof.source_pattern, exclude_monitors=True)
@@ -535,6 +579,24 @@ def detect_plugged_profiles() -> list[dict[str, Any]]:
                 log.info("detect_plugged_profiles: %s source resolved via "
                          "PyAudio fallback (pactl missed it): %s",
                          prof.id, src.get("name", "?"))
+        if src is None and prof.udp_mic_fallback:
+            # The profile's real mic is the G1 firmware UDP-multicast stream
+            # (voice/audio_io) — no PulseAudio source needed. Some G1 units
+            # never expose the chest mic to the Jetson's pulse (I2S capture
+            # reads silence); the profile is still fully functional, so a
+            # synthetic source keeps it selectable. apply_selection() skips
+            # `pactl set-default-source` for "udp:" names (like "pyaudio:").
+            src = {
+                "name": "udp:g1-multicast-mic",
+                "description": "G1 firmware UDP-multicast mic (no PulseAudio source)",
+            }
+            via = "udp-builtin"
+            # This path is the steady state on chest-mic robots and the
+            # detector polls every few seconds — log once, not per poll.
+            if prof.id not in _UDP_FALLBACK_LOGGED:
+                _UDP_FALLBACK_LOGGED.add(prof.id)
+                log.info("detect_plugged_profiles: %s using UDP-mic fallback "
+                         "(no matching pulse source)", prof.id)
         if sink and src:
             detected.append({
                 "profile": asdict(prof),
@@ -726,7 +788,11 @@ def apply_selection(sink: str, source: str) -> dict[str, Any]:
     """
     errors: list[str] = []
     if sink:
-        if not set_default_sink(sink):
+        if sink.startswith("dds:"):
+            log.info("apply_selection: sink is synthetic (%s) — skipping "
+                     "pactl set-default-sink. The G1 chest speaker plays via "
+                     "DDS (AudioClient.PlayStream), not PulseAudio.", sink)
+        elif not set_default_sink(sink):
             errors.append(f"set-default-sink failed: {sink}")
         else:
             try:
@@ -734,10 +800,11 @@ def apply_selection(sink: str, source: str) -> dict[str, Any]:
             except (FileNotFoundError, subprocess.SubprocessError):
                 pass
     if source:
-        if source.startswith("pyaudio:"):
+        if source.startswith(("pyaudio:", "udp:")):
-            log.info("apply_selection: source is PyAudio-direct (%s) — "
+            log.info("apply_selection: source is synthetic (%s) — "
-                     "skipping pactl set-default-source. Live mic path "
+                     "skipping pactl set-default-source. PyAudio-direct mics "
-                     "uses PortAudio device match; pactl defaults stay put.",
+                     "use PortAudio device match; udp: is the G1 firmware "
+                     "multicast mic. pactl defaults stay put.",
                      source)
         elif not set_default_source(source):
             errors.append(f"set-default-source failed: {source}")

voice/audio_io.py

@@ -31,6 +31,7 @@ import json
 import socket
 import struct
 import subprocess
+import queue
 import threading
 import time
 from abc import ABC, abstractmethod
@@ -211,7 +212,16 @@ class BuiltinMic(Mic):
 
 
 class BuiltinSpeaker(Speaker):
-    """G1 robot's built-in speaker via AudioClient.PlayStream (16 kHz mono)."""
+    """G1 robot's built-in speaker via AudioClient.PlayStream (16 kHz mono).
+
+    JITTER-BUFFERED: chunks are enqueued and a dedicated writer thread sends
+    them to the firmware, prebuffering ~0.3s before the first RPC and never
+    running more than ~0.6s ahead of real time. Gemini's audio arrives in
+    BURSTS (WiFi jitter, shared-websocket contention) — the old direct
+    per-chunk RPC mapped every arrival gap to an audible dropout on the
+    chest speaker's tiny firmware buffer, while the pulse (Pill/Anker) path
+    hid it behind PulseAudio's own buffering. This gives the chest the same
+    tolerance."""
 
     HARDWARE_RATE = 16_000
 
@@ -224,10 +234,16 @@ class BuiltinSpeaker(Speaker):
         self._app_name = app_name or _SP_CFG.get("app_name", "sanad")
         self._begin_pause = _SP_CFG.get("begin_stream_pause_sec", 0.15)
         self._finish_margin = _SP_CFG.get("wait_finish_margin_sec", 0.3)
+        # Jitter buffer tuning (see class docstring).
+        self._prebuffer_sec = _SP_CFG.get("jitter_prebuffer_sec", 0.30)
+        self._first_flush_sec = _SP_CFG.get("jitter_first_flush_sec", 0.60)
+        self._max_lead_sec = _SP_CFG.get("jitter_max_lead_sec", 0.60)
         self._stop_flag = threading.Event()
         self._stream_id: Optional[str] = None
         self._total_sent = 0.0
         self._play_start = 0.0
+        self._q: "queue.Queue[Optional[bytes]]" = queue.Queue()
+        self._writer: Optional[threading.Thread] = None
 
     def _stop_play_api(self) -> None:
         try:
@@ -241,13 +257,79 @@ class BuiltinSpeaker(Speaker):
         except Exception:
             log.warning("BuiltinSpeaker AUDIO_STOP_PLAY failed")
 
+    def _drain_queue(self) -> None:
+        try:
+            while True:
+                self._q.get_nowait()
+        except queue.Empty:
+            pass
+
     def begin_stream(self) -> None:
+        # Tear down any previous writer cleanly before starting a new turn.
+        self._stop_flag.set()
+        self._drain_queue()
+        if self._writer is not None and self._writer.is_alive():
+            self._q.put(None)
+            self._writer.join(timeout=1.0)
         self._stop_flag.clear()
+        self._drain_queue()
         self._stop_play_api()
         time.sleep(self._begin_pause)
         self._stream_id = f"s_{int(time.time() * 1000)}"
         self._total_sent = 0.0
         self._play_start = time.time()
+        self._writer = threading.Thread(
+            target=self._writer_loop, args=(self._stream_id,),
+            name="builtin-speaker-writer", daemon=True)
+        self._writer.start()
+
+    def _writer_loop(self, stream_id: str) -> None:
+        rate = float(self.HARDWARE_RATE)
+        sent_sec = 0.0
+        started = False
+        t0 = time.time()
+        batch: list = []
+        buffered = 0.0
+        while not self._stop_flag.is_set() and self._stream_id == stream_id:
+            try:
+                item = self._q.get(timeout=0.05)
+            except queue.Empty:
+                item = ()
+            if item is None:
+                return
+            if not started:
+                if item:
+                    batch.append(item)
+                    buffered += len(item) / 2 / rate
+                if buffered >= self._prebuffer_sec or (
+                        batch and (time.time() - t0) >= self._first_flush_sec):
+                    self._play_start = time.time()
+                    started = True
+                    for b in batch:
+                        if self._stop_flag.is_set() or self._stream_id != stream_id:
+                            return
+                        try:
+                            self._ac.PlayStream(self._app_name, stream_id, b)
+                        except Exception as exc:
+                            log.warning("PlayStream failed: %s", exc)
+                        sent_sec += len(b) / 2 / rate
+                    batch = []
+                continue
+            if not item:
+                continue
+            # Pace: never run more than max_lead ahead of real time — keeps
+            # the firmware queue short (fast interrupts) while OUR queue is
+            # the elastic reservoir that absorbs bursty arrival.
+            lead = sent_sec - (time.time() - self._play_start)
+            if lead > self._max_lead_sec:
+                time.sleep(min(lead - self._max_lead_sec, 0.25))
+                if self._stop_flag.is_set() or self._stream_id != stream_id:
+                    return
+            try:
+                self._ac.PlayStream(self._app_name, stream_id, item)
+            except Exception as exc:
+                log.warning("PlayStream failed: %s", exc)
+            sent_sec += len(item) / 2 / rate
 
     def send_chunk(self, pcm: PCMLike, source_rate: int) -> None:
         if self._stop_flag.is_set():
@@ -256,20 +338,22 @@ class BuiltinSpeaker(Speaker):
         if arr.size < 10:
             return
         hw = _resample_int16(arr, source_rate, self.HARDWARE_RATE)
-        self._ac.PlayStream(self._app_name, self._stream_id, hw.tobytes())
+        self._q.put(hw.tobytes())
         self._total_sent += len(hw) / self.HARDWARE_RATE
 
     def wait_finish(self) -> None:
-        elapsed = time.time() - self._play_start
+        # Wait for the queue to drain AND real-time playback to elapse.
-        remaining = self._total_sent - elapsed + self._finish_margin
+        while not self._stop_flag.is_set():
-        waited = 0.0
+            elapsed = time.time() - self._play_start
-        while waited < remaining and not self._stop_flag.is_set():
+            remaining = self._total_sent - elapsed + self._finish_margin
+            if self._q.empty() and remaining <= 0:
+                break
             time.sleep(0.1)
-            waited += 0.1
         self._stop_play_api()
 
     def stop(self) -> None:
         self._stop_flag.set()
+        self._drain_queue()
         self._stop_play_api()
 
     @property
@@ -815,9 +899,13 @@ _PROFILE_ALIASES = {
     "hollyland_builtin": "hollyland_builtin",
     "jbl": "jbl_builtin_mic",
     "jbl_builtin_mic": "jbl_builtin_mic",
+    "beats": "beats_pill",
+    "beats_pill": "beats_pill",
+    "pill": "beats_pill",
 }
 
-SUPPORTED_PROFILES = ("builtin", "anker", "hollyland_builtin", "jbl_builtin_mic")
+SUPPORTED_PROFILES = ("builtin", "anker", "hollyland_builtin", "jbl_builtin_mic",
+                      "beats_pill")
 
 
 @dataclass
@@ -885,6 +973,13 @@ class AudioIO:
             # pacat is used because PyAudio's 'pulse' device is unavailable in
             # this env. Neither backend needs the AudioClient.
             return BuiltinMic(), PulseStreamSpeaker(label="JBL", sink_pattern="jbl,bluez")
+        if resolved == "beats_pill":
+            # Beats Pill speaker via pacat → its PulseAudio sink + the G1
+            # built-in DDS mic — same regime as the JBL (loud external speaker,
+            # full mic-gate while speaking; see gemini/script.py). The Pill's
+            # own USB mic stays available to pulse for dashboard recordings.
+            return BuiltinMic(), PulseStreamSpeaker(
+                label="Beats Pill", sink_pattern="beats_pill,beats,pill")
         raise AssertionError(f"unhandled resolved profile: {resolved!r}")
 
     @classmethod

voice/movement_dispatch.py

@@ -108,6 +108,11 @@ class MovementDispatcher:
     def _load(self):
         try:
             data = json.loads(self._instruction_path.read_text(encoding="utf-8"))
+        except FileNotFoundError:
+            # Normal on robots with no movement phrase pack — not an error.
+            log.info("no %s — movement dispatcher inert (0 phrases)",
+                     self._instruction_path.name)
+            data = {}
         except Exception as exc:
             log.error("could not load %s: %s — dispatcher inert", self._instruction_path, exc)
             data = {}

voice/sanad_voice.py

@@ -146,7 +146,8 @@ _MOVEMENT_PROMPT_RULES = (
     "(you receive a [MOVEMENT ON] or [MOVEMENT STATUS] note). When movement is "
     "OFF, never confirm a motion — tell the user to enable movement from the "
     "dashboard.\n"
-    "When movement is ON and the user addresses you by name (Bousandah / بوسنده) AND "
+    "When movement is ON and the user addresses you by name (Sanad / سند / "
+    "Bousandah / بوسنده) AND "
     "asks you to move, reply with ONE short confirmation phrase per requested "
     "motion, in the SAME language, in the order asked. Use these EXACT shapes — "
     "they are what triggers the motion:\n"