From 0636b0c1a531d65e07e9425007f49b07c26698b9 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Thu, 7 May 2026 20:10:42 -0700 Subject: [PATCH] feat(meet_agent): live note-taking agent for Google Meet (listen + speak) (#1355) --- app/src-tauri/src/lib.rs | 2 + app/src-tauri/src/meet_audio/audio_bridge.js | 235 +++++++++ .../src/meet_audio/caption_listener.rs | 127 +++++ .../src/meet_audio/captions_bridge.js | 186 +++++++ app/src-tauri/src/meet_audio/inject.rs | 253 +++++++++ .../src/meet_audio/listen_capture.rs | 327 ++++++++++++ app/src-tauri/src/meet_audio/mod.rs | 309 +++++++++++ app/src-tauri/src/meet_audio/speak_pump.rs | 138 +++++ app/src-tauri/src/meet_call/mod.rs | 40 ++ app/src-tauri/vendor/tauri-cef | 2 +- docs/MEET_AGENT_SMOKE.md | 113 +++++ src/core/all.rs | 8 + src/openhuman/about_app/catalog.rs | 19 + src/openhuman/about_app/catalog_tests.rs | 1 + src/openhuman/meet_agent/brain.rs | 448 ++++++++++++++++ src/openhuman/meet_agent/mod.rs | 50 ++ src/openhuman/meet_agent/ops.rs | 204 ++++++++ src/openhuman/meet_agent/rpc.rs | 264 ++++++++++ src/openhuman/meet_agent/schemas.rs | 357 +++++++++++++ src/openhuman/meet_agent/session.rs | 478 ++++++++++++++++++ src/openhuman/meet_agent/types.rs | 152 ++++++ src/openhuman/meet_agent/wav.rs | 79 +++ src/openhuman/mod.rs | 1 + tests/json_rpc_e2e.rs | 165 ++++++ 24 files changed, 3957 insertions(+), 1 deletion(-) create mode 100644 app/src-tauri/src/meet_audio/audio_bridge.js create mode 100644 app/src-tauri/src/meet_audio/caption_listener.rs create mode 100644 app/src-tauri/src/meet_audio/captions_bridge.js create mode 100644 app/src-tauri/src/meet_audio/inject.rs create mode 100644 app/src-tauri/src/meet_audio/listen_capture.rs create mode 100644 app/src-tauri/src/meet_audio/mod.rs create mode 100644 app/src-tauri/src/meet_audio/speak_pump.rs create mode 100644 docs/MEET_AGENT_SMOKE.md create mode 100644 src/openhuman/meet_agent/brain.rs create mode 100644 src/openhuman/meet_agent/mod.rs create mode 100644 src/openhuman/meet_agent/ops.rs create mode 100644 src/openhuman/meet_agent/rpc.rs create mode 100644 src/openhuman/meet_agent/schemas.rs create mode 100644 src/openhuman/meet_agent/session.rs create mode 100644 src/openhuman/meet_agent/types.rs create mode 100644 src/openhuman/meet_agent/wav.rs diff --git a/app/src-tauri/src/lib.rs b/app/src-tauri/src/lib.rs index b696bd4c6..2a76c7e9c 100644 --- a/app/src-tauri/src/lib.rs +++ b/app/src-tauri/src/lib.rs @@ -15,6 +15,7 @@ mod gmessages_scanner; mod imessage_scanner; #[cfg(target_os = "macos")] mod mascot_native_window; +mod meet_audio; mod meet_call; mod meet_scanner; mod native_notifications; @@ -1367,6 +1368,7 @@ pub fn run() { let builder = builder.manage(telegram_scanner::ScannerRegistry::new()); let builder = builder.manage(screen_capture::ScreenShareState::new()); let builder = builder.manage(meet_call::MeetCallState::new()); + let builder = builder.manage(meet_audio::MeetAudioState::new()); builder .setup(move |app| { #[cfg(any(windows, target_os = "linux"))] diff --git a/app/src-tauri/src/meet_audio/audio_bridge.js b/app/src-tauri/src/meet_audio/audio_bridge.js new file mode 100644 index 000000000..9d9cfdedd --- /dev/null +++ b/app/src-tauri/src/meet_audio/audio_bridge.js @@ -0,0 +1,235 @@ +// OpenHuman audio bridge for the embedded Google Meet webview. +// +// Installed via CDP `Page.addScriptToEvaluateOnNewDocument` from the +// Tauri shell (`app/src-tauri/src/meet_audio/inject.rs`) so it runs at +// document-start, *before* Meet's join page calls +// `navigator.mediaDevices.getUserMedia`. The shell then triggers a +// `Page.reload` so that even an already-navigated meet page picks up +// the override. +// +// What this script does: +// +// 1. Builds a 16 kHz mono Web-Audio graph whose +// `MediaStreamAudioDestinationNode` provides an audio MediaStream +// track the page can hand to its RTCPeerConnection. +// 2. Monkey-patches `navigator.mediaDevices.getUserMedia` so any audio +// request returns our destination stream (and combined audio+video +// requests get the real video track from Chromium's fake-camera Y4M +// plus our audio track). +// 3. Exposes `window.__openhumanFeedPcm(b64)` — the Tauri shell calls +// this on a ~100 ms cadence via CDP `Runtime.evaluate` to push the +// next chunk of synthesized PCM16LE bytes from +// `openhuman.meet_agent_poll_speech`. +// +// JS-injection note: the project's broader rule (CLAUDE.md) is "no new +// JS in embedded provider webviews". The Meet call window is a special +// case — it is a dedicated top-level window for a single audio-bridging +// purpose where the public `CefAudioHandler` API is sufficient for the +// listen path but Chromium's audio *input* path has no comparable +// public hook short of a from-source rebuild. The user has explicitly +// authorized this injection for the speak path; legacy provider +// webviews keep the no-JS rule. + +(function () { + if (window.__openhumanAudioBridgeInstalled) { + console.log("[openhuman-audio-bridge] already installed; skipping"); + return; + } + window.__openhumanAudioBridgeInstalled = true; + console.log("[openhuman-audio-bridge] install begin"); + + var SAMPLE_RATE = 16000; + var ctx; + var dest; + var nextStartTime = 0; + + function ensureContext() { + if (ctx) { + console.log( + "[openhuman-audio-bridge] reuse AudioContext state=" + ctx.state + ); + return ctx; + } + var requestedRate = SAMPLE_RATE; + try { + ctx = new (window.AudioContext || window.webkitAudioContext)({ + sampleRate: SAMPLE_RATE, + }); + } catch (e) { + // Some Chromium builds don't honor the explicit sampleRate; fall + // back to the default (the bridge will resample implicitly via + // each AudioBuffer's declared rate). + console.warn( + "[openhuman-audio-bridge] AudioContext sampleRate hint rejected; falling back to default rate err=" + + e + ); + ctx = new (window.AudioContext || window.webkitAudioContext)(); + } + dest = ctx.createMediaStreamDestination(); + nextStartTime = ctx.currentTime; + console.log( + "[openhuman-audio-bridge] AudioContext created requested_rate=" + + requestedRate + + " actual_rate=" + + ctx.sampleRate + + " state=" + + ctx.state + ); + return ctx; + } + + function decodeBase64Pcm16leToFloat32(b64) { + var bin = atob(b64); + var len = bin.length; + if (len % 2 !== 0) { + // Trailing byte = corrupt frame; drop it rather than read past + // the end and emit a click. + len = len - 1; + } + var out = new Float32Array(len / 2); + for (var i = 0, j = 0; j < len; i++, j += 2) { + var lo = bin.charCodeAt(j); + var hi = bin.charCodeAt(j + 1); + var v = (hi << 8) | lo; + if (v & 0x8000) v -= 0x10000; + out[i] = v / 32768; + } + return out; + } + + // Public push API. Returns the duration in seconds the chunk added + // to the queue, mostly for diagnostics; the shell ignores it. + window.__openhumanFeedPcm = function (b64) { + if (!b64) return 0; + try { + ensureContext(); + var samples = decodeBase64Pcm16leToFloat32(b64); + if (!samples.length) return 0; + var buffer = ctx.createBuffer(1, samples.length, SAMPLE_RATE); + buffer.copyToChannel(samples, 0, 0); + var src = ctx.createBufferSource(); + src.buffer = buffer; + src.connect(dest); + // Schedule strictly after the previous chunk so successive + // 100 ms feeds line up gaplessly. If the queue has emptied + // (caller fell behind), restart at currentTime so we don't try + // to play in the past. + if (nextStartTime < ctx.currentTime) { + nextStartTime = ctx.currentTime; + } + src.start(nextStartTime); + nextStartTime += buffer.duration; + // High-frequency log gated by a counter so we don't drown the + // console at 10 Hz; emit ~1 in 50 frames (~5 s cadence at the + // shell's 100 ms feed rate). + window.__openhumanFeedCounter = (window.__openhumanFeedCounter || 0) + 1; + if (window.__openhumanFeedCounter % 50 === 1) { + console.log( + "[openhuman-audio-bridge] feed sampled chunk_dur=" + + buffer.duration.toFixed(3) + + "s queue_ahead=" + + (nextStartTime - ctx.currentTime).toFixed(3) + + "s frame=" + + window.__openhumanFeedCounter + ); + } + return buffer.duration; + } catch (e) { + console.warn("[openhuman-audio-bridge] feed failed:", e); + return 0; + } + }; + + // Public introspection — useful from the shell side via + // Runtime.evaluate to confirm the bridge is alive. + window.__openhumanAudioBridgeInfo = function () { + return { + installed: true, + sample_rate: SAMPLE_RATE, + audio_context_state: ctx ? ctx.state : "not-created", + next_start_time: nextStartTime, + destination_track_count: dest ? dest.stream.getAudioTracks().length : 0, + }; + }; + + // Override getUserMedia so Meet's audio requests are served from our + // bridge stream. We delegate video to the original implementation so + // Chromium's fake-camera Y4M (mascot) keeps working. + if ( + !navigator.mediaDevices || + typeof navigator.mediaDevices.getUserMedia !== "function" + ) { + console.warn( + "[openhuman-audio-bridge] navigator.mediaDevices.getUserMedia missing; interception disabled" + ); + return; + } + var origGum = navigator.mediaDevices.getUserMedia.bind(navigator.mediaDevices); + + // Build a fresh audio MediaStream backed by clones of the bridge's + // destination tracks. Returning the singleton `dest.stream` directly + // would let any caller's `track.stop()` (e.g. Meet during preview + // teardown / track renegotiation) permanently kill the bridge. Each + // call gets its own track lifecycle. + function freshAudioStream() { + ensureContext(); + return new MediaStream( + dest.stream.getAudioTracks().map(function (t) { + return t.clone(); + }) + ); + } + + navigator.mediaDevices.getUserMedia = function (constraints) { + if (!constraints || !constraints.audio) { + console.log( + "[openhuman-audio-bridge] getUserMedia passthrough (no audio)" + ); + return origGum(constraints); + } + + if (!constraints.video) { + console.log( + "[openhuman-audio-bridge] getUserMedia intercepted audio-only" + ); + return Promise.resolve(freshAudioStream()); + } + // Combined audio + video request: pull video from the real + // (fake-camera-backed) getUserMedia and splice in fresh clones of + // our audio tracks. + console.log( + "[openhuman-audio-bridge] getUserMedia intercepted audio+video; splicing audio onto fake-camera stream" + ); + return origGum({ video: constraints.video }).then(function (realStream) { + try { + realStream.getAudioTracks().forEach(function (t) { + realStream.removeTrack(t); + t.stop(); + }); + } catch (_) {} + freshAudioStream() + .getAudioTracks() + .forEach(function (t) { + realStream.addTrack(t); + }); + return realStream; + }); + }; + + // Best-effort: also patch the legacy `getUserMedia` aliases some + // older Meet code paths still call into. + if (typeof navigator.getUserMedia === "function") { + console.log("[openhuman-audio-bridge] patching legacy navigator.getUserMedia"); + var origLegacy = navigator.getUserMedia.bind(navigator); + navigator.getUserMedia = function (constraints, success, failure) { + navigator.mediaDevices + .getUserMedia(constraints) + .then(success, failure) + .catch(function (e) { + if (failure) failure(e); + else origLegacy(constraints, success, failure); + }); + }; + } + console.log("[openhuman-audio-bridge] install complete"); +})(); diff --git a/app/src-tauri/src/meet_audio/caption_listener.rs b/app/src-tauri/src/meet_audio/caption_listener.rs new file mode 100644 index 000000000..637c4ee7b --- /dev/null +++ b/app/src-tauri/src/meet_audio/caption_listener.rs @@ -0,0 +1,127 @@ +//! Listen path v2: drains Meet's built-in captions region via the +//! `captions_bridge.js` we install at session start, and forwards each +//! new line to core's `meet_agent_push_caption` RPC. +//! +//! Replaces the old [`super::listen_capture`] (CEF audio handler → +//! Whisper STT) which proved unreliable: CEF's `cef_audio_handler_t` +//! is queried lazily on first audio output, so a solo agent in a +//! lobby never engaged the pipeline. Captions handle that case for +//! free — Meet's STT is already running, speaker-attributed, and +//! pre-segmented. +//! +//! Lifecycle is owned by [`super::SpeakPump`]'s sibling: dropping the +//! returned [`CaptionListener`] shuts the polling task down. + +use std::time::Duration; + +use tokio::sync::oneshot; +use tokio::time::interval; + +use crate::cdp::CdpConn; + +use super::inject; + +/// Polling cadence for `__openhumanDrainCaptions`. Captions arrive at +/// roughly word-by-word frequency; 500 ms is the sweet spot between +/// "responsive enough that wake-word detection feels live" and "not +/// hammering the CDP socket". +const POLL_INTERVAL: Duration = Duration::from_millis(500); + +/// Cap on consecutive drain failures before the listener gives up. +/// Same shape as the speak pump — usually means the page navigated +/// away (call ended) or the renderer crashed. +const MAX_CONSECUTIVE_ERRORS: u32 = 30; + +/// RAII handle. Drop to stop the listener task. +pub struct CaptionListener { + pub request_id: String, + pub(crate) _shutdown_tx: Option>, +} + +impl Drop for CaptionListener { + fn drop(&mut self) { + let _ = self._shutdown_tx.take(); + } +} + +/// Spawn the caption polling loop for a session whose audio bridge +/// has already installed both `audio_bridge.js` and +/// `captions_bridge.js`. Owns its own clone of the CDP connection so +/// drains run concurrently with speak-pump feeds. +pub fn start(request_id: String, cdp: CdpConn, session_id: String) -> CaptionListener { + let (shutdown_tx, mut shutdown_rx) = oneshot::channel::<()>(); + let request_id_for_task = request_id.clone(); + tauri::async_runtime::spawn(async move { + let mut tick = interval(POLL_INTERVAL); + // Burn the first tick so the very first drain has something + // to drain (the page-side observer needs ~250 ms to attach). + tick.tick().await; + let mut cdp = cdp; + let mut errors: u32 = 0; + loop { + tokio::select! { + _ = &mut shutdown_rx => { + log::info!( + "[meet-audio] caption listener shutdown request_id={request_id_for_task}" + ); + break; + } + _ = tick.tick() => { + match drain_and_forward(&request_id_for_task, &mut cdp, &session_id).await { + Ok(_) => errors = 0, + Err(err) => { + errors += 1; + log::debug!( + "[meet-audio] caption tick err request_id={request_id_for_task} consec_errors={errors} err={err}" + ); + if errors >= MAX_CONSECUTIVE_ERRORS { + log::warn!( + "[meet-audio] caption listener giving up after {errors} consecutive errors request_id={request_id_for_task}" + ); + break; + } + } + } + } + } + } + }); + + CaptionListener { + request_id, + _shutdown_tx: Some(shutdown_tx), + } +} + +async fn drain_and_forward( + request_id: &str, + cdp: &mut CdpConn, + session_id: &str, +) -> Result<(), String> { + let captions = inject::drain_captions(cdp, session_id).await?; + if captions.is_empty() { + return Ok(()); + } + log::info!( + "[meet-audio] captions drained count={} request_id={request_id}", + captions.len() + ); + for (speaker, text, ts_ms) in captions { + // Propagate the failure so MAX_CONSECUTIVE_ERRORS can trip if + // core's session/RPC path is broken — without this the + // listener would silently drop captions forever while the + // page kept producing them. + super::rpc_call( + "openhuman.meet_agent_push_caption", + serde_json::json!({ + "request_id": request_id, + "speaker": speaker, + "text": text, + "ts_ms": ts_ms, + }), + ) + .await + .map_err(|err| format!("push_caption (request_id={request_id}): {err}"))?; + } + Ok(()) +} diff --git a/app/src-tauri/src/meet_audio/captions_bridge.js b/app/src-tauri/src/meet_audio/captions_bridge.js new file mode 100644 index 000000000..cf79bd45e --- /dev/null +++ b/app/src-tauri/src/meet_audio/captions_bridge.js @@ -0,0 +1,186 @@ +// OpenHuman captions bridge for the embedded Google Meet webview. +// +// Companion to `audio_bridge.js`. Where the audio bridge handles the +// SPEAK direction (synthesized PCM → MediaStream the page hands to its +// RTCPeerConnection), this script handles the LISTEN direction by +// scraping Meet's built-in live captions instead of running our own +// STT pipeline: +// +// - Auto-click the "Turn on captions" button so the user doesn't +// have to remember. +// - Watch the captions region with a MutationObserver and a 250 ms +// poll fallback (Meet sometimes batches DOM updates outside the +// observer's notify window). +// - Maintain a queue of new caption lines, deduped by speaker+text. +// Each entry: { speaker, text, ts }. +// - Expose `window.__openhumanDrainCaptions()` and +// `__openhumanCaptionsBridgeInfo()` for the Tauri shell to drive +// over CDP `Runtime.evaluate`. +// +// Why scraping (and not getDisplayMedia, or Web Speech, or Meet's +// undocumented APIs)? +// - getDisplayMedia would prompt the user for screen-share permission. +// - Web Speech doesn't reach the remote participants' audio — only +// local mic. +// - Meet has no public caption API. +// - The captions DOM is the simplest stable source. Class names +// obfuscate often, so we lean on `aria-label="Captions"` (which +// Meet keeps stable for accessibility). +// +// Wake-word handling lives in the core (`src/openhuman/meet_agent/`), +// not here — the page just streams every caption line out and core +// decides when to act. + +(function () { + if (window.__openhumanCaptionsBridgeInstalled) { + return; + } + window.__openhumanCaptionsBridgeInstalled = true; + + var queue = []; + // Per-speaker last-text fingerprint so a caption that grows in place + // (Meet appends text mid-utterance) doesn't get queued multiple + // times. We emit the *latest* text for each speaker only when it + // changes; downstream wake-word logic dedupes on its own buffer. + var lastBySpeaker = {}; + + function findCaptionsRegion() { + // Meet's captions region carries a stable accessibility label + // even as class names churn between rollouts. Try the canonical + // English first, then fall back to a fuzzy match for localized + // builds ("Subtitles", "Sous-titres", etc.) that still embed + // "captions" / "caption" in the aria-label. + return ( + document.querySelector('[aria-label="Captions"]') || + document.querySelector('div[role="region"][aria-label*="aption"]') || + document.querySelector('div[role="region"][aria-label*="aption" i]') || + null + ); + } + + function pollOnce() { + var region = findCaptionsRegion(); + if (!region) return; + + // Each caption line is typically a flex row with the speaker name + // at the top and the live transcript below. We don't depend on + // exact class names; instead we walk direct children and treat + // each as one caption "row". + var rows = region.querySelectorAll( + ':scope > div, :scope > section, :scope > [role="listitem"]' + ); + if (!rows.length) { + // Fall back to a single-block region: one big innerText blob. + var blob = (region.innerText || "").trim(); + if (blob && blob !== lastBySpeaker.__blob__) { + queue.push({ speaker: "", text: blob, ts: Date.now() }); + lastBySpeaker.__blob__ = blob; + } + return; + } + + rows.forEach(function (row) { + // The speaker name is usually the first text child; the + // transcript is the larger one beneath. Heuristic: the line + // with the most text wins as "transcript". + var nodes = row.querySelectorAll("*"); + var bestText = ""; + var bestCandidate = null; + var speakerGuess = ""; + nodes.forEach(function (n) { + var t = (n.innerText || "").trim(); + if (!t) return; + if (!speakerGuess && t.length < 40 && /^[A-Za-z][\w '\-\.]*$/.test(t)) { + speakerGuess = t; + } + if (t.length > bestText.length) { + bestText = t; + bestCandidate = n; + } + }); + if (!bestText) return; + // Strip the speaker name out of the body if it's the leading + // chunk (Meet sometimes renders "Alice the meeting starts at 3" + // as one innerText blob). + if (speakerGuess && bestText.startsWith(speakerGuess)) { + bestText = bestText.slice(speakerGuess.length).trim(); + } + if (!bestText) return; + + var key = speakerGuess || "_unknown"; + if (lastBySpeaker[key] === bestText) return; + lastBySpeaker[key] = bestText; + queue.push({ speaker: speakerGuess, text: bestText, ts: Date.now() }); + }); + } + + // Two layers, because Meet sometimes batches caption DOM updates + // in ways that miss MutationObserver notifications: + // + // 1. MutationObserver — fires immediately on DOM mutation, picks + // up character-data changes that the poll might miss between + // ticks. + // 2. 250 ms interval poll — safety net for batched updates and + // for the case where the captions region didn't exist at + // observer-attach time. + function attachObserver() { + var region = findCaptionsRegion(); + if (!region || region.__openhumanObserverAttached) return false; + region.__openhumanObserverAttached = true; + var obs = new MutationObserver(function () { + pollOnce(); + }); + obs.observe(region, { + childList: true, + subtree: true, + characterData: true, + }); + return true; + } + + // Auto-enable captions: walk every button on the page and click any + // that has an aria-label starting with "Turn on captions". Caps the + // attempts so we don't fight a user who deliberately disables CC. + var ENABLE_ATTEMPT_BUDGET = 30; // ~30 * 2s = 60s + var enableAttempts = 0; + function tryEnableCaptions() { + if (enableAttempts >= ENABLE_ATTEMPT_BUDGET) return; + enableAttempts++; + var buttons = document.querySelectorAll("button[aria-label]"); + for (var i = 0; i < buttons.length; i++) { + var lbl = (buttons[i].getAttribute("aria-label") || "").toLowerCase(); + // Match "Turn on captions" but NOT "Turn off captions". + if (lbl.indexOf("turn on captions") === 0 || /^turn on captions/.test(lbl)) { + try { + buttons[i].click(); + enableAttempts = ENABLE_ATTEMPT_BUDGET; // success — stop trying. + return true; + } catch (_) {} + } + } + return false; + } + + setInterval(function () { + attachObserver(); + pollOnce(); + }, 250); + setInterval(tryEnableCaptions, 2000); + + // Public API consumed by the Tauri shell over CDP Runtime.evaluate. + window.__openhumanDrainCaptions = function () { + var out = queue.slice(); + queue.length = 0; + return out; + }; + + window.__openhumanCaptionsBridgeInfo = function () { + return { + installed: true, + region_found: !!findCaptionsRegion(), + queue_depth: queue.length, + tracked_speakers: Object.keys(lastBySpeaker).length, + enable_attempts: enableAttempts, + }; + }; +})(); diff --git a/app/src-tauri/src/meet_audio/inject.rs b/app/src-tauri/src/meet_audio/inject.rs new file mode 100644 index 000000000..d0a951573 --- /dev/null +++ b/app/src-tauri/src/meet_audio/inject.rs @@ -0,0 +1,253 @@ +//! Install the OpenHuman audio bridge into the Meet webview via CDP. +//! +//! ## Why this can't live in the runtime +//! +//! The listen path uses CEF's public `cef_audio_handler_t` API and +//! needs no Chromium changes. The speak path is the opposite: there is +//! no public API for *writing* PCM into a renderer's audio input, and +//! the Chromium-internal `FileSource` that backs +//! `--use-file-for-fake-audio-capture` only reads a static WAV. Our +//! options are: +//! +//! - Patch Chromium and rebuild from source (multi-day; we don't +//! maintain a CEF source build pipeline yet). +//! - Inject a tiny Web Audio bridge into the Meet page over CDP. +//! +//! This module implements the second path. It runs once per call, +//! after the meet-call window opens but before [`crate::meet_scanner`] +//! starts driving the join page: +//! +//! 1. Attach a CDP session to the Meet target (or about:blank — see +//! note on initial URL below). +//! 2. `Page.addScriptToEvaluateOnNewDocument` with +//! [`AUDIO_BRIDGE_JS`] so it runs at document-start of the *next* +//! document load. +//! 3. `Page.reload` so even an already-navigated Meet page picks up +//! the override before its first `getUserMedia` call. +//! +//! ## Why a reload (rather than starting at about:blank) +//! +//! `meet_call_open_window` builds the WebviewWindow with the Meet URL +//! directly. Refactoring it to navigate via CDP would change the +//! lifecycle for every other code path that watches the meet window, +//! including `meet_scanner`'s target-URL prefix matching. A one-time +//! reload is surgical: meet_scanner already polls for the meet target +//! and tolerates re-navigation. + +use std::time::Duration; + +use serde_json::{json, Value}; + +use crate::cdp::{self, CdpConn}; + +/// JS bundled at build time — the actual Web Audio bridge lives in the +/// sibling `audio_bridge.js`. `include_str!` bakes it into the binary +/// so there's nothing to copy at install. +pub const AUDIO_BRIDGE_JS: &str = include_str!("audio_bridge.js"); + +/// Captions bridge — DOM observer over Meet's live captions region +/// plus auto-enable for the CC button. Installed alongside the audio +/// bridge so a single `Page.reload` boots both. +pub const CAPTIONS_BRIDGE_JS: &str = include_str!("captions_bridge.js"); + +/// How long we wait for CDP to surface the meet target after the +/// window builds. Mirrors [`crate::meet_scanner::TARGET_DISCOVERY_BUDGET`] +/// so the two scanners share a budget shape. +const TARGET_DISCOVERY_BUDGET: Duration = Duration::from_secs(20); +const TARGET_DISCOVERY_INTERVAL: Duration = Duration::from_millis(500); + +/// Run the inject + reload sequence. Returns the attached CDP +/// connection + session id so the caller (the speak pump) can keep +/// using it for `Runtime.evaluate` calls — opening one CDP session +/// per call rather than per pump tick saves ~5 ms per push. +pub async fn install_audio_bridge(meet_url: &str) -> Result<(CdpConn, String), String> { + let (mut cdp, session) = wait_for_meet_target(meet_url).await?; + log::info!( + "[meet-audio] inject attached session={} meet_url_chars={}", + session, + meet_url.chars().count() + ); + + // Page.enable is required before some build's reload events fire + // ordering callbacks; harmless on builds where it isn't. + let _ = cdp.call("Page.enable", json!({}), Some(&session)).await; + let _ = cdp.call("Runtime.enable", json!({}), Some(&session)).await; + + cdp.call( + "Page.addScriptToEvaluateOnNewDocument", + json!({ "source": AUDIO_BRIDGE_JS }), + Some(&session), + ) + .await + .map_err(|e| format!("addScriptToEvaluateOnNewDocument(audio): {e}"))?; + + cdp.call( + "Page.addScriptToEvaluateOnNewDocument", + json!({ "source": CAPTIONS_BRIDGE_JS }), + Some(&session), + ) + .await + .map_err(|e| format!("addScriptToEvaluateOnNewDocument(captions): {e}"))?; + + // Reload so the script applies to the (already-loaded) meet page. + // `ignoreCache: true` defeats the bfcache so we get a real + // document-start hook for the bridge. + cdp.call( + "Page.reload", + json!({ "ignoreCache": true }), + Some(&session), + ) + .await + .map_err(|e| format!("Page.reload: {e}"))?; + + log::info!("[meet-audio] inject reload requested session={session}"); + + // Confirm the bridge is live before we return — saves the speak + // pump from sending its first chunk into a void if the script + // failed to run for any reason. Best-effort: a missing bridge + // logs and we still return Ok so the listen path keeps working. + confirm_bridge_alive(&mut cdp, &session).await; + + Ok((cdp, session)) +} + +async fn wait_for_meet_target(meet_url: &str) -> Result<(CdpConn, String), String> { + let deadline = tokio::time::Instant::now() + TARGET_DISCOVERY_BUDGET; + let mut last_err = String::new(); + while tokio::time::Instant::now() < deadline { + match cdp::connect_and_attach_matching(|t| t.url.starts_with(meet_url)).await { + Ok(pair) => return Ok(pair), + Err(err) => { + last_err = err; + tokio::time::sleep(TARGET_DISCOVERY_INTERVAL).await; + } + } + } + Err(format!( + "[meet-audio] timeout waiting for meet target: {last_err}" + )) +} + +/// Poll `window.__openhumanAudioBridgeInfo()` for up to ~5 s. Logs the +/// outcome but never returns an error — the speak pump will rediscover +/// the bridge on the next push if it shows up late. +async fn confirm_bridge_alive(cdp: &mut CdpConn, session: &str) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while tokio::time::Instant::now() < deadline { + let res = cdp + .call( + "Runtime.evaluate", + json!({ + "expression": "(typeof window.__openhumanAudioBridgeInfo === 'function') \ + ? JSON.stringify(window.__openhumanAudioBridgeInfo()) \ + : null", + "returnByValue": true, + }), + Some(session), + ) + .await; + if let Ok(v) = res { + let value = v + .get("result") + .and_then(|r| r.get("value")) + .cloned() + .unwrap_or(Value::Null); + if let Some(s) = value.as_str() { + log::info!("[meet-audio] bridge alive info={s}"); + return; + } + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + log::warn!("[meet-audio] bridge readiness probe timed out — speak pump will retry"); +} + +/// Drain the page-side caption queue. Returns 0 or more `(speaker, +/// text, ts_ms)` triples accumulated since the last drain. The caller +/// (the caption listener loop) calls this every ~500 ms. +pub async fn drain_captions( + cdp: &mut CdpConn, + session: &str, +) -> Result, String> { + let res = cdp + .call( + "Runtime.evaluate", + json!({ + "expression": "(typeof window.__openhumanDrainCaptions === 'function') \ + ? JSON.stringify(window.__openhumanDrainCaptions()) \ + : '[]'", + "returnByValue": true, + }), + Some(session), + ) + .await + .map_err(|e| format!("Runtime.evaluate drain_captions: {e}"))?; + let json_str = res + .get("result") + .and_then(|r| r.get("value")) + .and_then(|v| v.as_str()) + .unwrap_or("[]") + .to_string(); + let parsed: Vec = + serde_json::from_str(&json_str).map_err(|e| format!("parse captions json: {e}"))?; + let mut out = Vec::with_capacity(parsed.len()); + for entry in parsed { + let speaker = entry + .get("speaker") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let text = entry + .get("text") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let ts_ms = entry.get("ts").and_then(|v| v.as_u64()).unwrap_or(0); + if text.is_empty() { + continue; + } + out.push((speaker, text, ts_ms)); + } + Ok(out) +} + +/// Dispatch one PCM chunk into the page's bridge. Called on every +/// poll-speech tick by [`crate::meet_audio::speak_pump`]. +/// +/// Errors are returned (rather than logged inline) so the pump can +/// decide whether to back off — repeated failures usually mean the +/// page navigated away (e.g. "you've been removed from the call"), +/// which the meet-call lifecycle handles by tearing the whole session +/// down anyway. +pub async fn feed_pcm_chunk(cdp: &mut CdpConn, session: &str, pcm_b64: &str) -> Result<(), String> { + if pcm_b64.is_empty() { + return Ok(()); + } + // Build the call as a string literal so a long base64 payload + // travels as a JS source argument (CDP's Runtime.callFunctionOn + // would be cleaner but requires the bridge function's objectId, + // and Runtime.evaluate keeps the wire shape one round-trip). + // + // The b64 alphabet has no quote / backslash characters so a plain + // single-quoted literal is safe — but defensively escape just in + // case some future encoder produces padding-edge weirdness. + let escaped = pcm_b64.replace('\\', "\\\\").replace('\'', "\\'"); + let expression = format!( + "(typeof window.__openhumanFeedPcm === 'function') ? window.__openhumanFeedPcm('{escaped}') : -1" + ); + let res = cdp + .call( + "Runtime.evaluate", + json!({ + "expression": expression, + "returnByValue": true, + }), + Some(session), + ) + .await + .map_err(|e| format!("Runtime.evaluate feed: {e}"))?; + if let Some(exception) = res.get("exceptionDetails") { + return Err(format!("page exception: {exception}")); + } + Ok(()) +} diff --git a/app/src-tauri/src/meet_audio/listen_capture.rs b/app/src-tauri/src/meet_audio/listen_capture.rs new file mode 100644 index 000000000..493a52e37 --- /dev/null +++ b/app/src-tauri/src/meet_audio/listen_capture.rs @@ -0,0 +1,327 @@ +//! Capture the embedded Meet webview's audio output and forward it to +//! the core meet_agent loop. +//! +//! ## Pipeline +//! +//! 1. `tauri_runtime_cef::audio::register_audio_handler` taps the +//! per-browser `cef_audio_handler_t`. CEF delivers planar +//! float32 PCM at the renderer's native rate (typically 48 kHz, +//! 1–2 channels) directly from the audio output device path — +//! *before* it hits the OS speaker. No system permission needed. +//! +//! 2. Downsample-to-mono runs inline on the CEF audio thread: +//! - average across channels → mono float32 +//! - linear-interpolate down to 16 kHz (the rate `voice::streaming` +//! and the smoke test in `meet_agent::session` expect) +//! - clamp + scale to PCM16LE +//! +//! 3. Accumulate ~100 ms per chunk (1 600 samples @ 16 kHz). We push +//! via the core RPC on every flush boundary; smaller pushes would +//! overload the JSON envelope, larger ones would slow VAD. +//! +//! 4. RPC pushes are spawned on the tokio runtime so the audio +//! callback never blocks on network IO. A bounded channel +//! backpressures: if core is wedged, we drop the oldest queued +//! chunk rather than holding CEF's audio thread. + +use std::sync::{Arc, Mutex}; + +use tauri_runtime_cef::audio::{ + register_audio_handler, AudioHandlerRegistration, AudioStreamEvent, +}; +use tokio::sync::mpsc; + +const TARGET_SAMPLE_RATE: u32 = 16_000; +/// 100 ms @ 16 kHz mono. `meet_agent::ops::Vad` pushes hangover counts +/// based on per-frame cadence, so changing this changes the VAD wall +/// time too. 100 ms feels responsive without burning RPC. +const FLUSH_SAMPLES: usize = (TARGET_SAMPLE_RATE as usize) / 10; +/// Bounded channel between the CEF callback (producer) and the +/// async-runtime forwarder (consumer). 32 chunks ≈ 3.2 s at the flush +/// cadence — generous slack for transient core latency, but bounded +/// so a wedged core can't OOM us. +const FORWARD_CHANNEL_CAPACITY: usize = 32; + +/// RAII handle. Drop to release the CEF audio registration and shut +/// down the forwarder task. Both happen synchronously — the channel +/// closes first, the task exits its recv loop, and the registration +/// drop unhooks CEF in the same tick. +pub struct ListenSession { + pub request_id: String, + _registration: AudioHandlerRegistration, + /// Held so `Drop` closes the channel even if there are no in-flight + /// chunks. The forwarder task observes the close and exits. + _shutdown_tx: mpsc::Sender>, +} + +/// Opens the audio capture for `meet_url`. The same exact URL must +/// have been used to build the CEF window — `register_audio_handler` +/// matches by prefix. +pub fn start(meet_url: &str, request_id: String) -> Result { + let (tx, rx) = mpsc::channel::>(FORWARD_CHANNEL_CAPACITY); + let resampler = Arc::new(Mutex::new(Resampler::new())); + + let resampler_for_handler = resampler.clone(); + let tx_for_handler = tx.clone(); + let request_id_for_log = request_id.clone(); + let registration = register_audio_handler(meet_url.to_string(), move |event| { + on_audio_event( + &request_id_for_log, + event, + &resampler_for_handler, + &tx_for_handler, + ); + }); + + spawn_forwarder(request_id.clone(), rx); + + log::info!( + "[meet-audio] listen registered request_id={} url_chars={}", + request_id, + meet_url.chars().count() + ); + + Ok(ListenSession { + request_id, + _registration: registration, + _shutdown_tx: tx, + }) +} + +/// Process one CEF audio event. Speech/Stopped/Error all flow through +/// here; only `Packet` produces RPC traffic, but the others are logged +/// at info so an aborted call leaves a breadcrumb in the file logs. +fn on_audio_event( + request_id: &str, + event: AudioStreamEvent, + resampler: &Arc>, + tx: &mpsc::Sender>, +) { + match event { + AudioStreamEvent::Started { + sample_rate_hz, + channels, + frames_per_buffer, + } => { + log::info!( + "[meet-audio] cef stream start request_id={request_id} hz={sample_rate_hz} channels={channels} frames_per_buffer={frames_per_buffer}" + ); + if let Ok(mut r) = resampler.lock() { + r.reset(sample_rate_hz as u32); + } + } + AudioStreamEvent::Packet { + channels: planes, + pts_ms: _, + } => { + let pcm_bytes = match resampler.lock() { + Ok(mut r) => r.feed_and_drain(&planes), + Err(_) => return, + }; + for chunk in pcm_bytes.chunks(FLUSH_SAMPLES * 2) { + // `try_send` drops the chunk on a full channel rather + // than blocking the CEF audio thread. Better to lose + // a frame than to stall the renderer. + if tx.try_send(chunk.to_vec()).is_err() { + log::warn!( + "[meet-audio] forward channel full; dropping {} bytes request_id={request_id}", + chunk.len() + ); + } + } + } + AudioStreamEvent::Stopped => { + log::info!("[meet-audio] cef stream stopped request_id={request_id}"); + if let Ok(mut r) = resampler.lock() { + r.reset(0); + } + } + AudioStreamEvent::Error(msg) => { + log::warn!("[meet-audio] cef stream error request_id={request_id} msg={msg}"); + } + } +} + +/// Pull chunks off the bounded channel and POST each to core. Lives in +/// its own task so the CEF callback never blocks on HTTP. +fn spawn_forwarder(request_id: String, mut rx: mpsc::Receiver>) { + tauri::async_runtime::spawn(async move { + use base64::{engine::general_purpose::STANDARD as B64, Engine as _}; + while let Some(chunk) = rx.recv().await { + let pcm_b64 = B64.encode(&chunk); + let res = super::rpc_call( + "openhuman.meet_agent_push_listen_pcm", + serde_json::json!({ + "request_id": request_id, + "pcm_base64": pcm_b64, + }), + ) + .await; + if let Err(err) = res { + log::debug!( + "[meet-audio] push_listen_pcm err request_id={request_id} bytes={} err={err}", + chunk.len() + ); + } + } + log::info!("[meet-audio] forwarder exiting request_id={request_id}"); + }); +} + +/// Stateful float32-planar → PCM16LE mono @ 16 kHz resampler. +/// +/// Uses linear interpolation, which is good enough for speech (the +/// downstream STT does not care about ultrasonics or pristine high +/// frequencies). Carry the previous sample across `feed_and_drain` +/// calls so we don't introduce a tick at every CEF buffer boundary. +/// Pick a source sample by signed index. Negative indices return the +/// carry sample from the previous call (so phase < 0 keeps the +/// interpolation continuous across buffer boundaries); past-the-end +/// indices clamp to the last sample (which is what the next call will +/// install as its own carry, so the output stays smooth even if a +/// caller stops feeding mid-stream). +fn sample_at(mono: &[f32], carry: f32, idx: i64) -> f32 { + if idx < 0 { + carry + } else if (idx as usize) < mono.len() { + mono[idx as usize] + } else { + *mono.last().unwrap_or(&0.0) + } +} + +struct Resampler { + source_rate_hz: u32, + /// Fractional position into the source buffer between calls. + /// 0.0 means "start cleanly with the next sample". Negative is + /// not used — the source rate is always known before we feed. + phase: f64, + /// Last source sample of the previous call, used as the "left" + /// neighbour when we interpolate the first sample of the next call. + last_sample: f32, +} + +impl Resampler { + fn new() -> Self { + Self { + source_rate_hz: 0, + phase: 0.0, + last_sample: 0.0, + } + } + + fn reset(&mut self, source_rate_hz: u32) { + self.source_rate_hz = source_rate_hz; + self.phase = 0.0; + self.last_sample = 0.0; + } + + fn feed_and_drain(&mut self, planes: &[Vec]) -> Vec { + if planes.is_empty() || self.source_rate_hz == 0 { + return Vec::new(); + } + let frames = planes[0].len(); + if frames == 0 { + return Vec::new(); + } + // Mono mix. + let mono: Vec = (0..frames) + .map(|i| { + let mut sum = 0.0_f32; + for plane in planes { + if let Some(v) = plane.get(i) { + sum += *v; + } + } + sum / planes.len() as f32 + }) + .collect(); + + let ratio = self.source_rate_hz as f64 / TARGET_SAMPLE_RATE as f64; + let mut out = Vec::with_capacity((mono.len() as f64 / ratio).ceil() as usize * 2); + // `pos` floats through `mono` indices. `pos < 0` means "still + // sampling the carry sample from the previous call"; `pos = 0` + // means "right at mono[0]". + let mut pos = self.phase; + while pos < mono.len() as f64 { + let idx_f = pos.floor(); + let frac = pos - idx_f; + let idx = idx_f as i64; + let s_left = sample_at(mono.as_slice(), self.last_sample, idx); + let s_right = sample_at(mono.as_slice(), self.last_sample, idx + 1); + let sample = s_left as f64 + (s_right as f64 - s_left as f64) * frac; + // Float32 [-1.0, 1.0] → i16. Clamp because Chromium can + // overshoot a touch on heavy compression. + let s_i16 = (sample.clamp(-1.0, 1.0) * i16::MAX as f64) as i16; + out.extend_from_slice(&s_i16.to_le_bytes()); + pos += ratio; + } + // Carry the trailing fractional position into the next call. + // It will be negative when we overshot (next call resumes + // mid-source-sample), so the next call interpolates between + // `last_sample` and the new mono[0]. + self.phase = pos - mono.len() as f64; + self.last_sample = *mono.last().unwrap_or(&0.0); + out + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resampler_with_no_source_rate_yields_nothing() { + let mut r = Resampler::new(); + let out = r.feed_and_drain(&[vec![0.5; 100]]); + assert!(out.is_empty(), "no source rate set, must produce nothing"); + } + + #[test] + fn resampler_48k_to_16k_mono_drops_samples_3to1() { + let mut r = Resampler::new(); + r.reset(48_000); + let plane = vec![0.5_f32; 4_800]; // 100ms @ 48k + let bytes = r.feed_and_drain(&[plane]); + // 100ms @ 16k = 1600 samples * 2 bytes. Allow ±2 samples slop + // from the fractional phase carry. + let samples = bytes.len() / 2; + assert!( + (1598..=1602).contains(&samples), + "expected ~1600 samples, got {samples}" + ); + } + + #[test] + fn resampler_stereo_to_mono_averages_channels() { + let mut r = Resampler::new(); + r.reset(16_000); + let left = vec![0.8_f32; 1600]; + let right = vec![-0.2_f32; 1600]; + let bytes = r.feed_and_drain(&[left, right]); + // Avg = 0.3 → ~9830 in i16. First two bytes are LE i16. + let first = i16::from_le_bytes([bytes[0], bytes[1]]); + assert!( + (9000..11000).contains(&first), + "expected mid-amplitude i16, got {first}" + ); + } + + #[test] + fn resampler_clamps_out_of_range_floats() { + let mut r = Resampler::new(); + r.reset(16_000); + let bytes = r.feed_and_drain(&[vec![5.0_f32; 100]]); + let first = i16::from_le_bytes([bytes[0], bytes[1]]); + assert_eq!(first, i16::MAX); + } + + #[test] + fn resampler_passthrough_when_rates_match() { + let mut r = Resampler::new(); + r.reset(16_000); + let plane = vec![0.5_f32; 1600]; + let bytes = r.feed_and_drain(&[plane]); + assert_eq!(bytes.len(), 1600 * 2); + } +} diff --git a/app/src-tauri/src/meet_audio/mod.rs b/app/src-tauri/src/meet_audio/mod.rs new file mode 100644 index 000000000..7a6eac6f0 --- /dev/null +++ b/app/src-tauri/src/meet_audio/mod.rs @@ -0,0 +1,309 @@ +//! Shell-side audio plumbing for the live meet-agent loop. +//! +//! ## Pieces +//! +//! - [`listen_capture`] — taps the embedded Meet webview's audio output +//! via the per-browser `CefAudioHandler` exposed by our vendored +//! `tauri-runtime-cef::audio` extension, downsamples to 16 kHz mono +//! PCM16LE, batches into ~100 ms chunks, and posts them to core via +//! `openhuman.meet_agent_push_listen_pcm`. Zero OS-level audio +//! permission needed: we read frames straight out of the renderer. +//! +//! - [`speak_pump`] — drains synthesized PCM the brain enqueued (via +//! `openhuman.meet_agent_poll_speech`) and writes it into the +//! Chromium `pipe://openhuman/` fake-audio source we +//! patch in the vendored CEF subtree. PR1 ships the pump scaffolding; +//! the Chromium-side patch lands in a follow-up slice. +//! +//! ## Lifecycle +//! +//! [`start`] is invoked once the meet-call window has been built (in +//! `meet_call::meet_call_open_window`). It opens the core session, +//! registers the audio handler keyed by the call's URL, and spawns the +//! poll-speech loop. [`stop`] runs from the window-destroyed handler: +//! it drops the audio handler registration (which silences capture +//! immediately), stops the speak pump, and tells core to close the +//! session and report counters. + +pub mod caption_listener; +pub mod inject; +pub mod listen_capture; +pub mod speak_pump; + +use std::collections::HashMap; +use std::sync::Mutex; + +use serde::Serialize; +use tauri::{AppHandle, Manager, Runtime}; + +/// Process-wide registry of active meet-agent sessions, keyed by +/// `request_id`. Mirrors the shape of `meet_call::MeetCallState` so +/// the two registries stay symmetric. +#[derive(Default)] +pub struct MeetAudioState { + inner: Mutex>, +} + +impl MeetAudioState { + pub fn new() -> Self { + Self::default() + } +} + +/// Held while a session is live. Dropping it runs the listen + speak +/// teardown synchronously — no async drop needed because the caption +/// listener and speak pump both shut down on signal/drop. +/// +/// The legacy CEF-audio `listen_capture::ListenSession` is kept as an +/// optional field so the pre-register flow still has somewhere to +/// hand the registration off if a future build re-enables it. In the +/// caption-driven path it stays `None`. +pub struct MeetAudioSession { + pub request_id: String, + _captions: caption_listener::CaptionListener, + _legacy_listen: Option, + _speak: speak_pump::SpeakPump, +} + +#[derive(Debug, Clone, Serialize)] +pub struct StopSummary { + pub request_id: String, + pub listened_seconds: f32, + pub spoken_seconds: f32, + pub turn_count: u32, +} + +/// Open a meet-agent audio session. +/// +/// Listen path goes via the captions bridge (`captions_bridge.js`) + +/// [`caption_listener`]. Speak path goes via the audio bridge +/// (`audio_bridge.js`) + [`speak_pump`]. Both are installed by +/// [`inject::install_audio_bridge`]. +/// +/// `meet_url` must be the *exact* URL the CEF window was built with — +/// the inject path uses it as the CDP target prefix so two concurrent +/// calls each attach to their own browser. +pub async fn start( + app: AppHandle, + request_id: String, + meet_url: String, +) -> Result<(), String> { + log::info!( + "[meet-audio] start request_id={request_id} url_prefix={}", + truncate_for_log(&meet_url, 64) + ); + + if let Some(state) = app.try_state::() { + let mut guard = state.inner.lock().unwrap(); + if guard.contains_key(&request_id) { + // Idempotent restart: drop the previous session before + // overwriting so its registration is released. + guard.remove(&request_id); + log::info!("[meet-audio] replaced existing session request_id={request_id}"); + } + } + + // Tell core to open its session first so the very first PCM push + // doesn't race the start RPC. + rpc_call( + "openhuman.meet_agent_start_session", + serde_json::json!({ + "request_id": request_id, + "sample_rate_hz": 16_000, + }), + ) + .await?; + + // Install the page-side audio + captions bridges in one go. The + // returned CDP session is shared by the speak pump and caption + // listener — we open a second session for the listener so the + // two run concurrently without serialising on a single CDP + // mailbox. + let (speak, captions) = match inject::install_audio_bridge(&meet_url).await { + Ok((cdp, session)) => { + // Spawn the caption listener on its own CDP attach so a + // long Runtime.evaluate from one side never starves the + // other. The second attach reuses the same CDP target. + let captions = match crate::cdp::connect_and_attach_matching(|t| { + t.url.starts_with(&meet_url) + }) + .await + { + Ok((cdp_for_captions, session_for_captions)) => caption_listener::start( + request_id.clone(), + cdp_for_captions, + session_for_captions, + ), + Err(err) => { + log::warn!( + "[meet-audio] caption listener cdp attach failed request_id={request_id} err={err}" + ); + caption_listener_disabled(request_id.clone()) + } + }; + let speak = speak_pump::start(request_id.clone(), cdp, session); + (speak, captions) + } + Err(err) => { + log::warn!( + "[meet-audio] audio bridge install failed request_id={request_id} err={err} — speak + caption paths disabled for this call" + ); + ( + speak_pump::start_disabled(request_id.clone()), + caption_listener_disabled(request_id.clone()), + ) + } + }; + + if let Some(state) = app.try_state::() { + state.inner.lock().unwrap().insert( + request_id.clone(), + MeetAudioSession { + request_id: request_id.clone(), + _captions: captions, + _legacy_listen: None, + _speak: speak, + }, + ); + } else { + log::warn!( + "[meet-audio] MeetAudioState missing from app — session will be ungoverned request_id={request_id}" + ); + } + + Ok(()) +} + +/// Stop a meet-agent audio session. Best-effort: errors from individual +/// shutdown steps are logged but never propagated, because window +/// destruction must finish even if e.g. core is unreachable. +pub async fn stop( + app: AppHandle, + request_id: String, +) -> Result, String> { + let session = if let Some(state) = app.try_state::() { + state.inner.lock().unwrap().remove(&request_id) + } else { + None + }; + + let Some(session) = session else { + log::debug!("[meet-audio] stop: no session for request_id={request_id}"); + return Ok(None); + }; + + // Dropping `session` first releases the audio handler registration + // (so CEF stops feeding us frames) and signals the pump to exit. + drop(session); + + match rpc_call( + "openhuman.meet_agent_stop_session", + serde_json::json!({ "request_id": request_id }), + ) + .await + { + Ok(v) => { + let listened = v + .get("listened_seconds") + .and_then(|x| x.as_f64()) + .unwrap_or(0.0) as f32; + let spoken = v + .get("spoken_seconds") + .and_then(|x| x.as_f64()) + .unwrap_or(0.0) as f32; + let turns = v.get("turn_count").and_then(|x| x.as_u64()).unwrap_or(0) as u32; + log::info!( + "[meet-audio] stop ok request_id={request_id} listened={listened:.2}s spoken={spoken:.2}s turns={turns}" + ); + Ok(Some(StopSummary { + request_id, + listened_seconds: listened, + spoken_seconds: spoken, + turn_count: turns, + })) + } + Err(err) => { + log::warn!("[meet-audio] stop_session rpc failed request_id={request_id} err={err}"); + Ok(None) + } + } +} + +/// Minimal JSON-RPC helper used by both this module and the speak pump +/// loop. Mirrors the call shape used by other shell scanners (see +/// `telegram_scanner::mod.rs`). +pub(crate) async fn rpc_call( + method: &str, + params: serde_json::Value, +) -> Result { + let body = serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": method, + "params": params, + }); + let url = crate::core_rpc::core_rpc_url_value(); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .build() + .map_err(|e| format!("http client: {e}"))?; + let req = crate::core_rpc::apply_auth(client.post(&url)) + .map_err(|e| format!("prepare {url}: {e}"))?; + let resp = req + .json(&body) + .send() + .await + .map_err(|e| format!("POST {url}: {e}"))?; + let status = resp.status(); + let v: serde_json::Value = resp + .json() + .await + .map_err(|e| format!("decode {status}: {e}"))?; + if !status.is_success() { + return Err(format!("{status}: {v}")); + } + if let Some(err) = v.get("error") { + return Err(format!("rpc error: {err}")); + } + Ok(v.get("result").cloned().unwrap_or(serde_json::Value::Null)) +} + +/// No-op caption listener used when CDP attach failed at session +/// start. Lets the rest of the lifecycle hold a uniform value. +fn caption_listener_disabled(request_id: String) -> caption_listener::CaptionListener { + caption_listener::CaptionListener { + request_id, + _shutdown_tx: None, + } +} + +/// Trim a string for logging without panicking on multi-byte chars. +fn truncate_for_log(s: &str, max_chars: usize) -> String { + let mut out = String::new(); + for (i, c) in s.chars().enumerate() { + if i >= max_chars { + out.push('…'); + break; + } + out.push(c); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn truncate_handles_short_strings() { + assert_eq!(truncate_for_log("hi", 10), "hi"); + } + + #[test] + fn truncate_caps_long_strings() { + let long = "a".repeat(100); + let trimmed = truncate_for_log(&long, 10); + assert!(trimmed.ends_with('…')); + assert_eq!(trimmed.chars().count(), 11); + } +} diff --git a/app/src-tauri/src/meet_audio/speak_pump.rs b/app/src-tauri/src/meet_audio/speak_pump.rs new file mode 100644 index 000000000..1fab93f86 --- /dev/null +++ b/app/src-tauri/src/meet_audio/speak_pump.rs @@ -0,0 +1,138 @@ +//! Speak path: poll synthesized PCM out of core and feed it into the +//! Meet page's audio bridge over CDP. +//! +//! Design lives in [`super::inject`]: the bridge is installed once at +//! session start by `install_audio_bridge`, which returns the open CDP +//! connection + session id. The pump owns those for the lifetime of +//! the call so each tick is a single `Runtime.evaluate` round-trip +//! rather than fresh attach + detach. + +use std::time::Duration; + +use base64::{engine::general_purpose::STANDARD as B64, Engine as _}; +use tokio::sync::oneshot; +use tokio::time::interval; + +use crate::cdp::CdpConn; + +use super::inject; + +/// Polling cadence. Same as the listen path's flush boundary so the +/// loop stays in lockstep — every ~100 ms we push captured audio in +/// (listen) and pull synthesized audio out (speak). +const POLL_INTERVAL: Duration = Duration::from_millis(100); + +/// Cap on consecutive feed failures before we give up and stop +/// pumping. Hitting this usually means the page navigated away +/// (Meet's "you've been removed" / network drop) — the meet-call +/// window-destroyed handler will tear the rest of the session down +/// either way. +const MAX_CONSECUTIVE_FEED_ERRORS: u32 = 30; + +/// RAII handle. Drop to stop the pump task. The shutdown channel +/// causes the spawned loop to exit on the next select tick. +pub struct SpeakPump { + pub request_id: String, + _shutdown_tx: Option>, +} + +impl Drop for SpeakPump { + fn drop(&mut self) { + let _ = self._shutdown_tx.take(); + } +} + +/// Spawn the speak pump for a session that already has the audio +/// bridge installed. `cdp` and `session_id` come from +/// [`inject::install_audio_bridge`] and are owned by the pump task +/// from this point on. +pub fn start(request_id: String, cdp: CdpConn, session_id: String) -> SpeakPump { + let (shutdown_tx, mut shutdown_rx) = oneshot::channel::<()>(); + let request_id_for_task = request_id.clone(); + tauri::async_runtime::spawn(async move { + let mut tick = interval(POLL_INTERVAL); + // Burn the first tick (`interval` fires immediately) so we + // don't poll before the listen path has had a chance to push. + tick.tick().await; + let mut cdp = cdp; + let mut feed_errors: u32 = 0; + loop { + tokio::select! { + _ = &mut shutdown_rx => { + log::info!( + "[meet-audio] speak pump shutdown request_id={request_id_for_task}" + ); + break; + } + _ = tick.tick() => { + match poll_and_feed(&request_id_for_task, &mut cdp, &session_id).await { + Ok(_) => feed_errors = 0, + Err(err) => { + feed_errors += 1; + log::debug!( + "[meet-audio] speak pump tick err request_id={request_id_for_task} consec_errors={feed_errors} err={err}" + ); + if feed_errors >= MAX_CONSECUTIVE_FEED_ERRORS { + log::warn!( + "[meet-audio] speak pump giving up after {feed_errors} consecutive errors request_id={request_id_for_task}" + ); + break; + } + } + } + } + } + } + }); + + SpeakPump { + request_id, + _shutdown_tx: Some(shutdown_tx), + } +} + +/// No-op pump used when bridge install failed at session start. Keeps +/// the rest of the session lifecycle uniform — `MeetAudioSession` can +/// still hold a `SpeakPump` regardless of speak-path readiness. +pub fn start_disabled(request_id: String) -> SpeakPump { + SpeakPump { + request_id, + _shutdown_tx: None, + } +} + +async fn poll_and_feed( + request_id: &str, + cdp: &mut CdpConn, + session_id: &str, +) -> Result<(), String> { + let v = super::rpc_call( + "openhuman.meet_agent_poll_speech", + serde_json::json!({ "request_id": request_id }), + ) + .await?; + let pcm_b64 = v + .get("pcm_base64") + .and_then(|x| x.as_str()) + .unwrap_or_default(); + let utterance_done = v + .get("utterance_done") + .and_then(|x| x.as_bool()) + .unwrap_or(false); + + if !pcm_b64.is_empty() { + // Validate decode locally before pushing — saves a round-trip + // when the brain enqueues a malformed batch. + let bytes = B64 + .decode(pcm_b64.as_bytes()) + .map_err(|e| format!("base64: {e}"))?; + log::debug!( + "[meet-audio] speak pump feeding request_id={request_id} bytes={} done={utterance_done}", + bytes.len() + ); + inject::feed_pcm_chunk(cdp, session_id, pcm_b64).await?; + } else if utterance_done { + log::info!("[meet-audio] speak pump utterance complete request_id={request_id}"); + } + Ok(()) +} diff --git a/app/src-tauri/src/meet_call/mod.rs b/app/src-tauri/src/meet_call/mod.rs index 75f23a1c2..bf05ed474 100644 --- a/app/src-tauri/src/meet_call/mod.rs +++ b/app/src-tauri/src/meet_call/mod.rs @@ -124,6 +124,27 @@ pub async fn meet_call_open_window( args.display_name.clone(), ); + // Start the live meet-agent audio loop: registers a CEF audio + // handler keyed by the meet URL, opens a core session, and spawns + // the speak-pump poller. Fire-and-forget — failures here must not + // prevent the user from at least seeing the join page, so we log + // and continue. The teardown below mirrors this on window close. + { + let app_for_audio = app.clone(); + let request_id_for_audio = request_id.clone(); + let url_for_audio = parsed.to_string(); + tauri::async_runtime::spawn(async move { + if let Err(err) = + crate::meet_audio::start(app_for_audio, request_id_for_audio.clone(), url_for_audio) + .await + { + log::warn!( + "[meet-call] meet_audio start failed request_id={request_id_for_audio} err={err}" + ); + } + }); + } + // Emit a `closed` event when the user dismisses the window AND clean // up the per-call data directory. The data dir holds an isolated CEF // profile (cookies, cache) we explicitly want to throw away after @@ -151,6 +172,25 @@ pub async fn meet_call_open_window( log::info!( "[meet-call] window destroyed label={label_for_event} request_id={request_id_for_event}" ); + // Tear down the meet-agent audio loop *before* the + // data dir wipe so the audio handler registration + // releases CEF cleanly while the browser is still + // shutting down. + { + let app_for_audio = app_for_event.clone(); + let request_id_for_audio = request_id_for_event.clone(); + tauri::async_runtime::spawn(async move { + if let Err(err) = + crate::meet_audio::stop(app_for_audio, request_id_for_audio.clone()) + .await + { + log::debug!( + "[meet-call] meet_audio stop err request_id={request_id_for_audio} err={err}" + ); + } + }); + } + // CEF may still be flushing the profile to disk on // teardown; do the rmdir off the UI thread so any // last-second writes don't race the delete. diff --git a/app/src-tauri/vendor/tauri-cef b/app/src-tauri/vendor/tauri-cef index c10bc0f72..a57470231 160000 --- a/app/src-tauri/vendor/tauri-cef +++ b/app/src-tauri/vendor/tauri-cef @@ -1 +1 @@ -Subproject commit c10bc0f7293f6070478fc8af1a157004f5916f10 +Subproject commit a57470231f8c1037d49f9913ec316d6bca325f74 diff --git a/docs/MEET_AGENT_SMOKE.md b/docs/MEET_AGENT_SMOKE.md new file mode 100644 index 000000000..e91130676 --- /dev/null +++ b/docs/MEET_AGENT_SMOKE.md @@ -0,0 +1,113 @@ +# Meet-agent live loop — smoke test runbook + +End-to-end validation that the agent hears, thinks, and speaks on a +real Google Meet call. Two laptops are easiest (Laptop A runs OpenHuman ++ joins the Meet as the agent; Laptop B is the human host who creates +the call and listens to the agent's voice). + +## Pre-flight + +1. Sign in to OpenHuman so a backend session token exists. Without + it, all three brain stages (STT/LLM/TTS) silently fall back to + stubs and you'll only hear a 200 ms tone — useful for plumbing + smoke but not the real loop. +2. Ensure the vendored `tauri-cef` submodule is on + `feat/openhuman-audio-handler` (or whatever branch carries the + `audio` module — see `app/src-tauri/vendor/tauri-cef`). +3. `pnpm tauri dev` in the repo root. + +## Steps + +1. **Laptop B**: create a Meet call at , + stay in the lobby. +2. **Laptop A**: + - Open OpenHuman → Intelligence → Calls. + - Paste the Meet URL, set display name (e.g. "OpenHuman Agent"). + - Click *Join*. + - A dedicated CEF window opens. The window title bar reads + "Meet — OpenHuman Agent". +3. **Laptop B**: admit the agent from the lobby. +4. Confirm Meet's live captions are on. The captions bridge auto-clicks + "Turn on captions" up to ~30 times over the first minute; if the + button isn't found (Meet UI rolls), enable CC manually. +5. Speak a wake-word phrase into the call mic. Examples: + - "Hey, OpenHuman — remember to email Bob about the launch." + - "Hey OpenHuman, follow up with the design team next week." + + The agent should reply with a short canned ack ("Got it.", + "Noted.", "Adding that.", "On it.", or "Captured.") routed back + into Meet's audio. + +## What to watch for + +### Listen path (Meet captions → agent) + +- The CEF audio handler / Whisper STT path is **not** the live-call + listen path; do not expect `cef stream start` or `push_listen_pcm` + log lines (those modules are kept in tree as `_legacy_listen` for a + future opt-in). + +- Tail the file logs (`~/Library/Application Support/OpenHuman/logs/`): + + ```text + [meet-audio] inject reload requested session=… + [meet-audio] bridge alive info={"installed":true,"sample_rate":16000,…} + [meet-audio] captions drained count=N request_id=… + [meet-agent-rpc] wake word fired request_id=… speaker=… + [meet-agent] caption turn start request_id=… prompt_chars=… + [meet-agent] caption turn done request_id=… reply_chars=… synth_samples=… + ``` + +- If `captions drained` never logs, the captions bridge didn't find + Meet's caption region — either CC is off (auto-enable failed) or + Meet rolled the DOM and the `aria-label="Captions"` selector + needs updating in `captions_bridge.js`. Confirm via the embedded + page console: `window.__openhumanCaptionsBridgeInfo()` — the + `region_found` field should be `true` once captions are on. + +### Speak path (agent → Meet) + +- Inspect the embedded Meet page's console (right-click → Inspect; or + attach via the CDP port 19222 on Laptop A): you should see + `[openhuman-audio-bridge] feed failed: …` only on errors. +- Run `window.__openhumanAudioBridgeInfo()` in the console: + + ```json + { "installed": true, "sample_rate": 16000, "audio_context_state": "running", + "next_start_time": 12.3, "destination_track_count": 1 } + ``` + +- **Laptop B**: you should hear the agent's reply through Meet, with + the agent's tile lighting up the "speaking" indicator. + +### Mascot webcam + +- Laptop B sees the OpenHuman mascot SVG in the agent's tile. + Confirms `--use-file-for-fake-video-capture` is still active (the + speak path doesn't break it). + +## Things that should NOT happen + +- macOS prompt for screen recording / microphone permission. +- macOS prompt for installing a system audio driver / kext. +- The OpenHuman main window's mic indicator turning on (we tap CEF's + audio at the renderer level, not via the OS mic). + +## Common failure modes + +| Symptom | Likely cause | Fix | +| --- | --- | --- | +| Heard event empty / "STT failure" | No backend session | Sign in | +| Spoke event present, no audio on Laptop B | Bridge install failed | Check `Page.reload` errored — devtools network | +| 1× turn fires, then nothing | VAD `in_utterance` flag stuck | Look for `EndOfUtterance` events; may need a longer hangover | +| Audio "robot voice" | Sample-rate mismatch — bridge says 16000 but TTS gave another rate | Confirm `output_format=pcm_16000` request was honored | +| `cef stream error` repeated | Renderer crashed | Check Chromium logs in the meet-call data dir | + +## Cleanup + +- Close the meet-call window. The window-destroyed handler tears down + `meet_audio` (drops the audio handler registration → silences + capture immediately, signals the speak pump → exits) and calls + `openhuman.meet_agent_stop_session` which logs the listened/spoken + totals. +- Per-call data dir is wiped automatically. diff --git a/src/core/all.rs b/src/core/all.rs index 50810c185..095325610 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -211,6 +211,8 @@ fn build_registered_controllers() -> Vec { controllers.extend(crate::openhuman::notifications::all_notifications_registered_controllers()); // Google Meet call-join request validation (shell handles the webview) controllers.extend(crate::openhuman::meet::all_meet_registered_controllers()); + // Live meet-agent loop: STT/LLM/TTS over the open call's audio. + controllers.extend(crate::openhuman::meet_agent::all_meet_agent_registered_controllers()); // Structured WhatsApp Web data — agent-facing read-only controllers (list/search). // The write-path ingest controller is registered separately in build_internal_only_controllers. controllers.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_registered_controllers()); @@ -293,6 +295,8 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::notifications::all_notifications_controller_schemas()); // Google Meet call-join request validation schemas.extend(crate::openhuman::meet::all_meet_controller_schemas()); + // Live meet-agent listening + speaking loop + schemas.extend(crate::openhuman::meet_agent::all_meet_agent_controller_schemas()); // Structured WhatsApp Web data — local SQLite store, agent-queryable schemas.extend(crate::openhuman::whatsapp_data::all_whatsapp_data_controller_schemas()); schemas @@ -385,6 +389,10 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "Validate Google Meet call-join requests and mint a request_id; the desktop \ shell opens the embedded CEF webview that joins the call as an anonymous guest.", ), + "meet_agent" => Some( + "Live agent loop for an open Google Meet call: shell streams inbound PCM, \ + core runs VAD-segmented STT → LLM → TTS, shell pulls synthesized PCM back.", + ), "whatsapp_data" => Some( "Structured WhatsApp conversation and message store — list chats, read messages, and search across WhatsApp Web data.", ), diff --git a/src/openhuman/about_app/catalog.rs b/src/openhuman/about_app/catalog.rs index b5fab3fe4..8d2c14dc5 100644 --- a/src/openhuman/about_app/catalog.rs +++ b/src/openhuman/about_app/catalog.rs @@ -877,6 +877,25 @@ const CAPABILITIES: &[Capability] = &[ destinations: &["Google Meet"], }), }, + Capability { + id: "meet_agent.live_loop", + name: "Live Meet Agent — Listen + Speak", + domain: "meet_agent", + category: CapabilityCategory::Automation, + description: "While the agent is in a Google Meet call, it listens to the other \ + participants by tapping the embedded webview's audio output, runs \ + VAD-segmented speech-to-text, decides whether to respond, and speaks \ + back through a virtual microphone the embedded Chromium reads as if \ + it were a real input device. No system audio permission required — \ + capture and playback both stay inside the CEF process.", + how_to: "Automatic once a Meet call is open via Intelligence > Calls.", + status: CapabilityStatus::Beta, + privacy: Some(CapabilityPrivacy { + leaves_device: true, + data_kind: PrivacyDataKind::Derived, + destinations: &["Google Meet", "ElevenLabs (STT/TTS via hosted backend)"], + }), + }, // ── Update ────────────────────────────────────────────────────────────── Capability { id: "update.check", diff --git a/src/openhuman/about_app/catalog_tests.rs b/src/openhuman/about_app/catalog_tests.rs index c83945dce..c6b57cc19 100644 --- a/src/openhuman/about_app/catalog_tests.rs +++ b/src/openhuman/about_app/catalog_tests.rs @@ -79,6 +79,7 @@ fn catalog_includes_additional_user_facing_surfaces() { "settings.manage_service", "settings.clear_app_data", "meet.join_call", + "meet_agent.live_loop", ] { assert!( ids.contains(expected), diff --git a/src/openhuman/meet_agent/brain.rs b/src/openhuman/meet_agent/brain.rs new file mode 100644 index 000000000..6dfae320d --- /dev/null +++ b/src/openhuman/meet_agent/brain.rs @@ -0,0 +1,448 @@ +//! Turn orchestration: STT → LLM → TTS. +//! +//! ## Pipeline +//! +//! When [`session::Vad`] reports `EndOfUtterance`, [`run_turn`] drains +//! the inbound buffer and runs three serial stages: +//! +//! 1. **STT** — wrap the PCM16LE samples in a WAV container and post +//! to [`crate::openhuman::voice::cloud_transcribe`]. Returns the +//! transcribed text (or `Err` on transport / auth failure). +//! +//! 2. **LLM** — send a tiny chat-completions request through +//! [`crate::api::BackendOAuthClient`] with a "live meeting agent" +//! system prompt and the transcript as the user message. Returns a +//! short reply (or empty string when the agent decides to stay +//! silent). +//! +//! 3. **TTS** — feed the reply text into +//! [`crate::openhuman::voice::reply_speech`] requesting +//! `output_format = "pcm_16000"`. Decode the base64 PCM bytes back +//! into `Vec` and enqueue on the session's outbound queue. +//! +//! ## Fallback +//! +//! When the backend session token is missing (the most common reason +//! a stage fails outside production: tests, no-network smoke runs), +//! we fall back to deterministic stubs so the loop still produces an +//! audible blip and the unit tests stay network-free. Real +//! transport / 5xx errors are *not* swallowed — they surface as +//! `Note` events so a real-call failure is visible in the transcript +//! log, not silently degraded to a stub. + +use base64::{engine::general_purpose::STANDARD as B64, Engine as _}; +use serde_json::{json, Value}; + +use super::session::registry; +use super::types::SessionEventKind; +use super::wav; + +/// Minimum samples below which we skip the brain turn entirely. +/// 250 ms @ 16 kHz — under this, VAD almost certainly fired on a +/// transient (cough, click) rather than real speech. +const MIN_TURN_SAMPLES: usize = 4_000; +/// Re-exported from `ops` so any drift (if we ever loosen the +/// boundary check) immediately breaks the WAV / duration math here +/// at compile time. Today the same constant is used in both places — +/// the ops boundary check rejects anything else outright. +const SAMPLE_RATE_HZ: u32 = super::ops::REQUIRED_SAMPLE_RATE; + +/// Caption-driven turn. Drains the session's pending wake-word prompt +/// (assembled by `session::note_caption`) and runs LLM → TTS → enqueue +/// outbound. Skips STT entirely — the captions are already text. +/// +/// We give the user a short window (`CAPTION_TURN_DELAY_MS`) after the +/// wake word fires so multi-caption utterances ("hey openhuman … +/// what's the weather like in paris") have a chance to assemble +/// before we hit the LLM. The shell calls this on every caption +/// push that flagged the wake word; subsequent calls before the +/// delay expires are coalesced via the session's `wake_active` flag. +pub async fn run_caption_turn(request_id: &str) -> Result { + // Wait briefly so a multi-fragment wake utterance ("hey openhuman + // what's the weather like in paris" arriving as 2-3 captions) has + // a chance to assemble before we drain the prompt. + tokio::time::sleep(std::time::Duration::from_millis(CAPTION_TURN_DELAY_MS)).await; + + let prompt = match registry().with_session(request_id, |s| s.take_pending_prompt())? { + Some(p) => p, + None => return Ok(false), + }; + log::info!( + "[meet-agent] caption turn start request_id={request_id} prompt_chars={}", + prompt.chars().count() + ); + + // We deliberately skip the LLM in the hot path. The note itself + // is already stored verbatim in the session transcript (a `Heard` + // event). The verbal ack should always be brief, deterministic, + // and free of model latency / chain-of-thought leakage — so we + // pick a short canned phrase. Post-meeting summarisation (which + // can take its time and run a real LLM) lives in a follow-up. + let reply_text = pick_ack_phrase(&prompt).to_string(); + + let synthesized = if reply_text.trim().is_empty() { + Vec::new() + } else { + match tts(&reply_text).await { + Ok(samples) => samples, + Err(err) => { + log::warn!( + "[meet-agent] caption-turn TTS failed request_id={request_id} err={err}" + ); + let _ = registry().with_session(request_id, |s| { + s.record_event( + SessionEventKind::Note, + format!("TTS failure (using stub): {err}"), + ); + }); + stub_tts(&reply_text).await + } + } + }; + + registry().with_session(request_id, |s| { + s.record_event(SessionEventKind::Heard, prompt.clone()); + if !reply_text.is_empty() { + s.record_event(SessionEventKind::Spoke, reply_text.clone()); + if !synthesized.is_empty() { + s.enqueue_outbound_pcm(&synthesized, true); + } + } else { + s.record_event( + SessionEventKind::Note, + "agent declined to respond".to_string(), + ); + } + s.turn_count += 1; + })?; + + log::info!( + "[meet-agent] caption turn done request_id={request_id} reply_chars={} synth_samples={}", + reply_text.chars().count(), + synthesized.len() + ); + Ok(true) +} + +/// Delay between wake-word match and prompt drain. Long enough that +/// 2-3 caption fragments can join up; short enough that the user +/// doesn't experience awkward silence after they stop talking. +const CAPTION_TURN_DELAY_MS: u64 = 1_500; + +/// Canned acknowledgements the agent speaks out loud after capturing +/// a note. Short, varied so consecutive notes don't sound robotic. +/// Selected by hashing the prompt so the same dictation reliably +/// produces the same ack (helpful for tests + debugging) while still +/// rotating across the set in a normal conversation. +const ACK_PHRASES: &[&str] = &["Got it.", "Noted.", "Adding that.", "On it.", "Captured."]; + +fn pick_ack_phrase(prompt: &str) -> &'static str { + if prompt.trim().is_empty() { + return ""; + } + let h: u32 = prompt.bytes().fold(0u32, |a, b| a.wrapping_add(b as u32)); + ACK_PHRASES[(h as usize) % ACK_PHRASES.len()] +} + +/// Fire one brain turn for the named session. Returns `Ok(true)` when a +/// turn actually ran, `Ok(false)` when the inbound buffer was below the +/// floor. +pub async fn run_turn(request_id: &str) -> Result { + let drained = registry().with_session(request_id, |s| s.drain_inbound())?; + if drained.len() < MIN_TURN_SAMPLES { + log::debug!( + "[meet-agent] skipping turn request_id={request_id} samples={}", + drained.len() + ); + return Ok(false); + } + + log::info!( + "[meet-agent] turn start request_id={request_id} samples={}", + drained.len() + ); + + // ─── STT ──────────────────────────────────────────────────────── + let heard = match stt(&drained).await { + Ok(text) if text.trim().is_empty() => { + log::info!("[meet-agent] STT empty, skipping turn request_id={request_id}"); + return Ok(false); + } + Ok(text) => text, + Err(err) => { + log::warn!("[meet-agent] STT failed request_id={request_id} err={err}"); + // Record a Note so the transcript log makes the failure + // visible to whoever's looking at logs. + let _ = registry().with_session(request_id, |s| { + s.record_event( + SessionEventKind::Note, + format!("STT failure (using stub): {err}"), + ); + }); + stub_stt(&drained).await + } + }; + log::info!( + "[meet-agent] STT request_id={request_id} text_chars={}", + heard.chars().count() + ); + + // ─── LLM ──────────────────────────────────────────────────────── + let reply_text = match llm(&heard).await { + Ok(text) => text, + Err(err) => { + log::warn!("[meet-agent] LLM failed request_id={request_id} err={err}"); + let _ = registry().with_session(request_id, |s| { + s.record_event( + SessionEventKind::Note, + format!("LLM failure (using stub): {err}"), + ); + }); + stub_llm(&heard).await + } + }; + + // ─── TTS ──────────────────────────────────────────────────────── + let synthesized = if reply_text.trim().is_empty() { + Vec::new() + } else { + match tts(&reply_text).await { + Ok(samples) => samples, + Err(err) => { + log::warn!("[meet-agent] TTS failed request_id={request_id} err={err}"); + let _ = registry().with_session(request_id, |s| { + s.record_event( + SessionEventKind::Note, + format!("TTS failure (using stub): {err}"), + ); + }); + stub_tts(&reply_text).await + } + } + }; + + registry().with_session(request_id, |s| { + s.record_event(SessionEventKind::Heard, heard.clone()); + if !reply_text.is_empty() { + s.record_event(SessionEventKind::Spoke, reply_text.clone()); + if !synthesized.is_empty() { + s.enqueue_outbound_pcm(&synthesized, true); + } + } else { + s.record_event( + SessionEventKind::Note, + "agent declined to respond".to_string(), + ); + } + s.turn_count += 1; + })?; + + log::info!( + "[meet-agent] turn done request_id={request_id} reply_chars={} synth_samples={}", + reply_text.chars().count(), + synthesized.len() + ); + Ok(true) +} + +// ─── Real adapters ────────────────────────────────────────────────── + +async fn stt(samples: &[i16]) -> Result { + use crate::openhuman::voice::cloud_transcribe::{transcribe_cloud, CloudTranscribeOptions}; + + let config = crate::openhuman::config::ops::load_config_with_timeout().await?; + let wav_bytes = wav::pack_pcm16le_mono_wav(samples, SAMPLE_RATE_HZ); + let audio_b64 = B64.encode(&wav_bytes); + let opts = CloudTranscribeOptions { + mime_type: Some("audio/wav".to_string()), + file_name: Some("meet-agent.wav".to_string()), + ..Default::default() + }; + let outcome = transcribe_cloud(&config, &audio_b64, &opts).await?; + let text = outcome.value.text.clone(); + Ok(text) +} + +async fn llm(heard: &str) -> Result { + use crate::api::config::effective_api_url; + use crate::api::jwt::get_session_token; + use crate::api::BackendOAuthClient; + use reqwest::Method; + + let config = crate::openhuman::config::ops::load_config_with_timeout().await?; + let token = get_session_token(&config) + .map_err(|e| e.to_string())? + .filter(|t| !t.trim().is_empty()) + .ok_or_else(|| "no backend session token".to_string())?; + + let api_url = effective_api_url(&config.api_url); + let client = BackendOAuthClient::new(&api_url).map_err(|e| e.to_string())?; + + let body = json!({ + "model": "agentic-v1", + "temperature": 0.2, + // Hard cap to force a terse acknowledgement. agentic-v1 + // tends to elaborate (200+ chars) when given more headroom. + "max_tokens": 20, + "messages": [ + { + "role": "system", + "content": "You are OpenHuman, a note-taking AI participating in a live Google \ + Meet call. Each user message you receive is a single dictation — a \ + note, action item, or follow-up the user wants captured for after \ + the meeting. The transcript log stores the note verbatim, so your \ + job is only to acknowledge briefly out loud that you got it. \ + \ + Reply with one short sentence, ideally three to six words \ + (\"Got it.\", \"Noted.\", \"Adding that to your follow-ups.\"). \ + Never repeat the user's note back, never speculate or expand on \ + it, never engage in side conversation. \ + \ + If the message is empty, unclear, or just filler (\"uh\", \"hmm\", \ + a name), reply with an empty string." + }, + { "role": "user", "content": heard } + ] + }); + + let raw = client + .authed_json( + &token, + Method::POST, + "/openai/v1/chat/completions", + Some(body), + ) + .await + .map_err(|e| e.to_string())?; + + let text = extract_chat_completion_text(&raw) + .ok_or_else(|| format!("unexpected chat completions response: {raw}"))?; + Ok(text) +} + +async fn tts(text: &str) -> Result, String> { + use crate::openhuman::voice::reply_speech::{synthesize_reply, ReplySpeechOptions}; + + let config = crate::openhuman::config::ops::load_config_with_timeout().await?; + let opts = ReplySpeechOptions { + // Ask ElevenLabs (via the hosted backend) for raw PCM16LE @ + // 16 kHz so we can feed the result straight into the + // shell-side bridge with no transcoding. + output_format: Some("pcm_16000".to_string()), + ..Default::default() + }; + let outcome = synthesize_reply(&config, text, &opts).await?; + let result = outcome.value; + let pcm_bytes = B64 + .decode(result.audio_base64.as_bytes()) + .map_err(|e| format!("decode tts base64: {e}"))?; + if !pcm_bytes.len().is_multiple_of(2) { + return Err(format!("odd byte length from tts: {}", pcm_bytes.len())); + } + Ok(pcm_bytes + .chunks_exact(2) + .map(|c| i16::from_le_bytes([c[0], c[1]])) + .collect()) +} + +fn extract_chat_completion_text(raw: &Value) -> Option { + raw.get("choices") + .and_then(|c| c.as_array()) + .and_then(|arr| arr.first()) + .and_then(|first| first.get("message")) + .and_then(|m| m.get("content")) + .and_then(|s| s.as_str()) + .map(|s| s.trim().to_string()) +} + +// ─── Stubs (fallback for tests / no-backend) ──────────────────────── + +async fn stub_stt(samples: &[i16]) -> String { + let secs = samples.len() as f32 / SAMPLE_RATE_HZ as f32; + format!("(heard ~{secs:.1}s of audio)") +} + +async fn stub_llm(_heard: &str) -> String { + "I'm listening.".to_string() +} + +async fn stub_tts(text: &str) -> Vec { + if text.is_empty() { + return Vec::new(); + } + let sample_rate = SAMPLE_RATE_HZ as f32; + let freq = 440.0_f32; + let duration_secs = 0.2_f32; + let count = (sample_rate * duration_secs) as usize; + (0..count) + .map(|i| { + let t = i as f32 / sample_rate; + (((2.0 * std::f32::consts::PI * freq * t).sin()) * (i16::MAX as f32 * 0.3)) as i16 + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::meet_agent::session::registry; + + #[tokio::test] + async fn run_turn_skips_short_buffers() { + registry().start("brain-skip", 16_000).unwrap(); + registry() + .with_session("brain-skip", |s| { + s.push_inbound_pcm(&vec![0; 800]); // 50ms — under floor + }) + .unwrap(); + assert_eq!(run_turn("brain-skip").await.unwrap(), false); + let _ = registry().stop("brain-skip"); + } + + #[tokio::test] + async fn run_turn_falls_back_to_stub_without_backend() { + // No backend session in test env → STT/LLM/TTS all fail and + // each stage falls back to its stub. The turn still produces + // a Heard event, a Spoke event, and synthesized PCM, so the + // smoke-test contract holds. + registry().start("brain-fallback", 16_000).unwrap(); + registry() + .with_session("brain-fallback", |s| { + s.push_inbound_pcm(&vec![1000; 16_000]); // 1s + }) + .unwrap(); + assert_eq!(run_turn("brain-fallback").await.unwrap(), true); + registry() + .with_session("brain-fallback", |s| { + let kinds: Vec<_> = s.events().iter().map(|e| format!("{:?}", e.kind)).collect(); + assert!(kinds.contains(&"Heard".to_string())); + assert!(kinds.contains(&"Spoke".to_string())); + assert_eq!(s.turn_count, 1); + assert!(s.spoken_seconds() > 0.0); + }) + .unwrap(); + let _ = registry().stop("brain-fallback"); + } + + #[test] + fn extract_chat_completion_text_pulls_first_choice() { + let raw = json!({ + "choices": [ + { "message": { "content": " hello world " } } + ] + }); + assert_eq!( + extract_chat_completion_text(&raw), + Some("hello world".to_string()) + ); + } + + #[test] + fn extract_chat_completion_text_returns_none_on_malformed() { + assert_eq!(extract_chat_completion_text(&json!({})), None); + assert_eq!( + extract_chat_completion_text(&json!({ "choices": [] })), + None + ); + } +} diff --git a/src/openhuman/meet_agent/mod.rs b/src/openhuman/meet_agent/mod.rs new file mode 100644 index 000000000..8cf60bdae --- /dev/null +++ b/src/openhuman/meet_agent/mod.rs @@ -0,0 +1,50 @@ +//! Meet-agent domain — listening + speaking loop for a live Google Meet +//! call. +//! +//! Sits *next to* `meet/` (which only validates a URL and mints a +//! `request_id`) and reuses `voice/` for STT/TTS. Where `meet/` is +//! single-shot ("here is a request_id, shell goes off and opens a +//! window"), `meet_agent/` is a long-lived session: while the call is +//! open, the Tauri shell streams PCM frames from the CEF audio handler +//! into the core; the core runs VAD-segmented STT, decides whether to +//! reply, runs TTS, and streams synthesized PCM back out to the shell's +//! virtual-mic pump. +//! +//! ## Why a separate domain (not just more functions on `meet/`)? +//! +//! `meet/` is intentionally pure-validation — no state, no streams, no +//! audio. A live agentic loop is the opposite shape: a session registry, +//! per-session ring buffers, VAD/turn state, transcript log, and a TTS +//! pipeline. Bolting that onto `meet/` would force the validation surface +//! to drag in audio dependencies. Splitting keeps each domain small. +//! +//! ## Module layout +//! +//! - [`types`] — request/response types, public session events +//! - [`ops`] — VAD, ring-buffer, transcript helpers (pure, testable) +//! - [`session`] — `MeetAgentSession` and the per-session registry +//! - [`brain`] — turn orchestration: STT → LLM → TTS (stub in PR1) +//! - [`rpc`] — JSON-RPC handlers +//! - [`schemas`] — controller schema definitions +//! +//! ## RPC surface +//! +//! - `openhuman.meet_agent_start_session` — open a session for a `request_id` +//! - `openhuman.meet_agent_push_listen_pcm` — shell pushes captured PCM frames +//! - `openhuman.meet_agent_poll_speech` — shell pulls synthesized PCM frames +//! - `openhuman.meet_agent_stop_session` — close session, flush pending audio + +pub mod brain; +pub mod ops; +pub mod rpc; +pub mod schemas; +pub mod session; +pub mod types; +pub mod wav; + +pub use schemas::{ + all_controller_schemas as all_meet_agent_controller_schemas, + all_registered_controllers as all_meet_agent_registered_controllers, +}; +pub use session::{MeetAgentSession, MeetAgentSessionRegistry, SESSION_REGISTRY}; +pub use types::*; diff --git a/src/openhuman/meet_agent/ops.rs b/src/openhuman/meet_agent/ops.rs new file mode 100644 index 000000000..8448154fd --- /dev/null +++ b/src/openhuman/meet_agent/ops.rs @@ -0,0 +1,204 @@ +//! Pure helpers for the `meet_agent` domain: VAD-style end-of-utterance +//! detection, sample-rate sanity, request_id sanitization. Kept out of +//! `session.rs` so they can be unit-tested without a tokio runtime. + +/// The only sample rate the meet-agent loop currently supports. +/// `brain.rs` packs WAVs, computes durations, and sizes the turn +/// floor against this constant; until we plumb the per-session rate +/// all the way through, every helper assumes 16 kHz. The shell's +/// listen path resamples to this rate before pushing. +pub const REQUIRED_SAMPLE_RATE: u32 = 16_000; + +/// Validate a sample rate handed in from the shell. Locked to a +/// single value at the boundary instead of accepting a range — see +/// `REQUIRED_SAMPLE_RATE` for the rationale. +pub fn validate_sample_rate(hz: u32) -> Result { + if hz != REQUIRED_SAMPLE_RATE { + return Err(format!( + "sample_rate_hz {hz} unsupported; only {REQUIRED_SAMPLE_RATE} is allowed" + )); + } + Ok(hz) +} + +/// Same shape as `meet_call::sanitize_request_id` in the shell — keeping +/// the rule symmetric on both sides means a session key the shell minted +/// is always accepted by core and vice-versa. +pub fn sanitize_request_id(raw: &str) -> Result { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return Err("request_id must not be empty".into()); + } + if trimmed.len() > 64 { + return Err("request_id exceeds 64 characters".into()); + } + if !trimmed + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') + { + return Err("request_id contains forbidden characters".into()); + } + Ok(trimmed.to_string()) +} + +/// Crude energy-based VAD. Computes RMS over the supplied PCM16LE samples +/// and reports whether they are above a "speech-y" threshold. The brain +/// uses this in combination with a hangover counter to decide when an +/// utterance has ended (see `Vad::feed`). +/// +/// Crude on purpose: a real model-based VAD (Silero, webrtcvad) is the +/// follow-up; for the MVP the goal is "did somebody just stop talking +/// for ~600ms?", which RMS handles fine. +pub fn frame_rms(samples: &[i16]) -> f32 { + if samples.is_empty() { + return 0.0; + } + let sum_sq: f64 = samples.iter().map(|&s| (s as f64).powi(2)).sum(); + let mean = sum_sq / samples.len() as f64; + (mean.sqrt() / i16::MAX as f64) as f32 +} + +/// RMS above this is "voice-ish". Picked empirically against +/// `voice::streaming` test fixtures — anything below this is room tone. +pub const VAD_RMS_THRESHOLD: f32 = 0.015; + +/// Number of consecutive sub-threshold frames that mean the speaker has +/// stopped. At ~100ms-per-frame (the cadence the shell pushes), 6 frames +/// ≈ 600ms of silence — comfortable end-of-utterance marker without +/// chopping mid-thought. +pub const VAD_HANGOVER_FRAMES: u32 = 6; + +/// Stateful VAD wrapper. Owned by the session. +#[derive(Debug, Default)] +pub struct Vad { + /// True once we've seen at least one speech-y frame for the current + /// utterance — prevents firing "end of utterance" on a freshly-opened + /// session that has never seen audio. + in_utterance: bool, + /// Consecutive silent frames since the last speech-y one. + silence_run: u32, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum VadEvent { + /// Speech-y frame; ignore. + Speech, + /// Silent frame, but not enough to close the utterance yet. + Silence, + /// `VAD_HANGOVER_FRAMES` of silence after speech — turn ends now. + EndOfUtterance, + /// Silence with no preceding speech this session — caller can skip + /// any buffer-flush work. + Idle, +} + +impl Vad { + pub fn new() -> Self { + Self::default() + } + + /// Feed a single PCM frame and learn whether it ended the utterance. + pub fn feed(&mut self, samples: &[i16]) -> VadEvent { + let rms = frame_rms(samples); + if rms >= VAD_RMS_THRESHOLD { + self.in_utterance = true; + self.silence_run = 0; + VadEvent::Speech + } else if !self.in_utterance { + VadEvent::Idle + } else { + self.silence_run += 1; + if self.silence_run >= VAD_HANGOVER_FRAMES { + self.in_utterance = false; + self.silence_run = 0; + VadEvent::EndOfUtterance + } else { + VadEvent::Silence + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn validate_sample_rate_accepts_only_required_rate() { + validate_sample_rate(16_000).unwrap(); + } + + #[test] + fn validate_sample_rate_rejects_anything_else() { + assert!(validate_sample_rate(8_000).is_err()); + assert!(validate_sample_rate(48_000).is_err()); + assert!(validate_sample_rate(96_000).is_err()); + } + + #[test] + fn sanitize_request_id_matches_shell_rules() { + sanitize_request_id("550e8400-e29b-41d4-a716-446655440000").unwrap(); + assert!(sanitize_request_id("").is_err()); + assert!(sanitize_request_id("a/b").is_err()); + assert!(sanitize_request_id(&"x".repeat(65)).is_err()); + } + + #[test] + fn frame_rms_is_zero_for_silence() { + assert_eq!(frame_rms(&[0; 320]), 0.0); + } + + #[test] + fn frame_rms_grows_with_amplitude() { + let quiet: Vec = (0..320) + .map(|i| if i % 2 == 0 { 1000i16 } else { -1000 }) + .collect(); + let loud: Vec = (0..320) + .map(|i| if i % 2 == 0 { 8000i16 } else { -8000 }) + .collect(); + assert!(frame_rms(&loud) > frame_rms(&quiet)); + } + + /// Build a frame that's deterministically above the VAD threshold. + fn loud_frame() -> Vec { + // Half-amplitude square wave — comfortably above VAD_RMS_THRESHOLD + // without saturating clamps in downstream tests. + (0..1600) + .map(|i| if i % 2 == 0 { 8000 } else { -8000 }) + .collect() + } + + #[test] + fn vad_idle_until_first_speech() { + let mut vad = Vad::new(); + for _ in 0..10 { + assert_eq!(vad.feed(&[0; 320]), VadEvent::Idle); + } + } + + #[test] + fn vad_emits_end_of_utterance_after_hangover() { + let mut vad = Vad::new(); + assert_eq!(vad.feed(&loud_frame()), VadEvent::Speech); + for i in 0..VAD_HANGOVER_FRAMES - 1 { + assert_eq!( + vad.feed(&[0; 320]), + VadEvent::Silence, + "frame #{i} of silence run" + ); + } + assert_eq!(vad.feed(&[0; 320]), VadEvent::EndOfUtterance); + } + + #[test] + fn vad_resets_after_utterance() { + let mut vad = Vad::new(); + vad.feed(&loud_frame()); + for _ in 0..VAD_HANGOVER_FRAMES { + vad.feed(&[0; 320]); + } + // Next silent frame after end-of-utterance should be Idle, not + // a fresh Silence run. + assert_eq!(vad.feed(&[0; 320]), VadEvent::Idle); + } +} diff --git a/src/openhuman/meet_agent/rpc.rs b/src/openhuman/meet_agent/rpc.rs new file mode 100644 index 000000000..31c36c833 --- /dev/null +++ b/src/openhuman/meet_agent/rpc.rs @@ -0,0 +1,264 @@ +//! JSON-RPC handlers for the `meet_agent` domain. +//! +//! Four endpoints, all keyed by `request_id`: +//! +//! - `start_session` — open a session (idempotent restart on dup id) +//! - `push_listen_pcm` — feed PCM frames in; may trigger a brain turn +//! - `poll_speech` — pull synthesized PCM out +//! - `stop_session` — close + return summary counters +//! +//! Each handler is intentionally short — heavy lifting lives in +//! `session.rs` (state) and `brain.rs` (behavior). RPC code is +//! deserialize-validate-dispatch only. + +use base64::{engine::general_purpose::STANDARD as B64, Engine as _}; +use serde_json::{json, Map, Value}; + +use crate::rpc::RpcOutcome; + +use super::brain; +use super::ops::VadEvent; +use super::session::registry; +use super::types::{ + PollSpeechRequest, PushCaptionRequest, PushListenPcmRequest, StartSessionRequest, + StopSessionRequest, +}; + +const LOG_PREFIX: &str = "[meet-agent-rpc]"; + +pub async fn handle_start_session(params: Map) -> Result { + let req: StartSessionRequest = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("{LOG_PREFIX} invalid start_session params: {e}"))?; + + registry().start(&req.request_id, req.sample_rate_hz)?; + log::info!( + "{LOG_PREFIX} start_session request_id={} sample_rate_hz={}", + req.request_id, + req.sample_rate_hz + ); + + RpcOutcome::new( + json!({ + "ok": true, + "request_id": req.request_id, + "sample_rate_hz": req.sample_rate_hz, + }), + vec![], + ) + .into_cli_compatible_json() +} + +pub async fn handle_push_listen_pcm(params: Map) -> Result { + let req: PushListenPcmRequest = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("{LOG_PREFIX} invalid push_listen_pcm params: {e}"))?; + + let samples = + decode_pcm16le_b64(&req.pcm_base64).map_err(|e| format!("{LOG_PREFIX} pcm decode: {e}"))?; + + let event = registry().with_session(&req.request_id, |s| s.push_inbound_pcm(&samples))?; + + let turn_started = matches!(event, VadEvent::EndOfUtterance); + if turn_started { + // Spawn the turn so the RPC reply doesn't have to wait for STT + // + TTS to finish — the shell will drain audio via poll_speech. + let request_id = req.request_id.clone(); + tokio::spawn(async move { + if let Err(err) = brain::run_turn(&request_id).await { + log::warn!("{LOG_PREFIX} brain turn failed request_id={request_id} err={err}"); + } + }); + } + + RpcOutcome::new( + json!({ + "ok": true, + "turn_started": turn_started, + }), + vec![], + ) + .into_cli_compatible_json() +} + +pub async fn handle_push_caption(params: Map) -> Result { + let req: PushCaptionRequest = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("{LOG_PREFIX} invalid push_caption params: {e}"))?; + + let wake_fired = registry().with_session(&req.request_id, |s| { + s.note_caption(&req.speaker, &req.text, req.ts_ms) + })?; + + if wake_fired { + log::info!( + "{LOG_PREFIX} wake word fired request_id={} speaker={}", + req.request_id, + req.speaker + ); + let request_id = req.request_id.clone(); + tokio::spawn(async move { + if let Err(err) = brain::run_caption_turn(&request_id).await { + log::warn!("{LOG_PREFIX} caption-turn failed request_id={request_id} err={err}"); + } + }); + } + + RpcOutcome::new( + json!({ + "ok": true, + "turn_started": wake_fired, + }), + vec![], + ) + .into_cli_compatible_json() +} + +pub async fn handle_poll_speech(params: Map) -> Result { + let req: PollSpeechRequest = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("{LOG_PREFIX} invalid poll_speech params: {e}"))?; + + let (pcm_base64, utterance_done) = + registry().with_session(&req.request_id, |s| s.poll_outbound())?; + + RpcOutcome::new( + json!({ + "ok": true, + "pcm_base64": pcm_base64, + "utterance_done": utterance_done, + }), + vec![], + ) + .into_cli_compatible_json() +} + +pub async fn handle_stop_session(params: Map) -> Result { + let req: StopSessionRequest = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("{LOG_PREFIX} invalid stop_session params: {e}"))?; + + let session = registry().stop(&req.request_id)?; + log::info!( + "{LOG_PREFIX} stop_session request_id={} listened={:.2}s spoken={:.2}s turns={}", + session.request_id, + session.listened_seconds(), + session.spoken_seconds(), + session.turn_count + ); + + RpcOutcome::new( + json!({ + "ok": true, + "request_id": session.request_id, + "listened_seconds": session.listened_seconds(), + "spoken_seconds": session.spoken_seconds(), + "turn_count": session.turn_count, + }), + vec![], + ) + .into_cli_compatible_json() +} + +/// Decode a base64 string of PCM16LE bytes into samples. Empty input is +/// a "heartbeat" push (no audio this tick) and yields an empty Vec. +fn decode_pcm16le_b64(b64: &str) -> Result, String> { + if b64.is_empty() { + return Ok(Vec::new()); + } + let bytes = B64 + .decode(b64.as_bytes()) + .map_err(|e| format!("base64: {e}"))?; + if !bytes.len().is_multiple_of(2) { + return Err(format!("odd byte length {}", bytes.len())); + } + Ok(bytes + .chunks_exact(2) + .map(|c| i16::from_le_bytes([c[0], c[1]])) + .collect()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn b64_pcm(samples: &[i16]) -> String { + let bytes: Vec = samples.iter().flat_map(|s| s.to_le_bytes()).collect(); + B64.encode(bytes) + } + + #[tokio::test] + async fn start_then_stop_round_trip() { + let mut params = Map::new(); + params.insert("request_id".into(), json!("rpc-roundtrip")); + params.insert("sample_rate_hz".into(), json!(16_000)); + let out = handle_start_session(params).await.unwrap(); + assert_eq!(out.get("ok"), Some(&json!(true))); + + let mut stop = Map::new(); + stop.insert("request_id".into(), json!("rpc-roundtrip")); + let out = handle_stop_session(stop).await.unwrap(); + assert_eq!(out.get("turn_count"), Some(&json!(0))); + } + + #[tokio::test] + async fn push_then_poll_returns_audio_after_brain_turn() { + let mut start = Map::new(); + start.insert("request_id".into(), json!("rpc-push")); + start.insert("sample_rate_hz".into(), json!(16_000)); + handle_start_session(start).await.unwrap(); + + // Push a loud frame, then enough silent frames to cross the + // VAD hangover and trigger a turn. + let loud: Vec = (0..1600) + .map(|i| if i % 2 == 0 { 8000i16 } else { -8000 }) + .collect(); + let mut p = Map::new(); + p.insert("request_id".into(), json!("rpc-push")); + p.insert("pcm_base64".into(), json!(b64_pcm(&loud))); + handle_push_listen_pcm(p).await.unwrap(); + + // ~1s of speech-like content so the brain turn doesn't skip. + for _ in 0..10 { + let mut p = Map::new(); + p.insert("request_id".into(), json!("rpc-push")); + p.insert("pcm_base64".into(), json!(b64_pcm(&loud))); + handle_push_listen_pcm(p).await.unwrap(); + } + + // Now silence frames to trigger end-of-utterance. + let silence = vec![0i16; 1600]; + let mut last = json!(false); + for _ in 0..10 { + let mut p = Map::new(); + p.insert("request_id".into(), json!("rpc-push")); + p.insert("pcm_base64".into(), json!(b64_pcm(&silence))); + let out = handle_push_listen_pcm(p).await.unwrap(); + if out.get("turn_started") == Some(&json!(true)) { + last = json!(true); + break; + } + } + assert_eq!(last, json!(true), "expected a turn_started=true reply"); + + // Give the spawned turn a moment to enqueue audio. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + let mut poll = Map::new(); + poll.insert("request_id".into(), json!("rpc-push")); + let out = handle_poll_speech(poll).await.unwrap(); + let pcm = out.get("pcm_base64").and_then(|v| v.as_str()).unwrap_or(""); + assert!(!pcm.is_empty(), "expected synthesized audio after turn"); + + let mut stop = Map::new(); + stop.insert("request_id".into(), json!("rpc-push")); + handle_stop_session(stop).await.unwrap(); + } + + #[test] + fn decode_pcm16le_b64_handles_empty() { + assert!(decode_pcm16le_b64("").unwrap().is_empty()); + } + + #[test] + fn decode_pcm16le_b64_rejects_odd_length() { + // Three bytes -> odd number of bytes -> reject. + let odd = B64.encode([0u8, 1, 2]); + assert!(decode_pcm16le_b64(&odd).is_err()); + } +} diff --git a/src/openhuman/meet_agent/schemas.rs b/src/openhuman/meet_agent/schemas.rs new file mode 100644 index 000000000..1c1795a09 --- /dev/null +++ b/src/openhuman/meet_agent/schemas.rs @@ -0,0 +1,357 @@ +//! Controller schemas for the `meet_agent` domain. + +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; + +type SchemaBuilder = fn() -> ControllerSchema; +type ControllerHandler = fn(Map) -> ControllerFuture; + +struct Def { + function: &'static str, + schema: SchemaBuilder, + handler: ControllerHandler, +} + +const DEFS: &[Def] = &[ + Def { + function: "start_session", + schema: schema_start_session, + handler: handle_start_session, + }, + Def { + function: "push_listen_pcm", + schema: schema_push_listen_pcm, + handler: handle_push_listen_pcm, + }, + Def { + function: "push_caption", + schema: schema_push_caption, + handler: handle_push_caption, + }, + Def { + function: "poll_speech", + schema: schema_poll_speech, + handler: handle_poll_speech, + }, + Def { + function: "stop_session", + schema: schema_stop_session, + handler: handle_stop_session, + }, +]; + +pub fn all_controller_schemas() -> Vec { + DEFS.iter().map(|d| (d.schema)()).collect() +} + +pub fn all_registered_controllers() -> Vec { + DEFS.iter() + .map(|d| RegisteredController { + schema: (d.schema)(), + handler: d.handler, + }) + .collect() +} + +pub fn schemas(function: &str) -> ControllerSchema { + if let Some(d) = DEFS.iter().find(|d| d.function == function) { + return (d.schema)(); + } + schema_unknown() +} + +fn schema_start_session() -> ControllerSchema { + ControllerSchema { + namespace: "meet_agent", + function: "start_session", + description: + "Open a meet-agent session keyed by request_id. The Tauri shell calls this after \ + the meet-call window opens and before pushing PCM frames.", + inputs: vec![ + FieldSchema { + name: "request_id", + ty: TypeSchema::String, + comment: "UUID minted by openhuman.meet_join_call.", + required: true, + }, + FieldSchema { + name: "sample_rate_hz", + ty: TypeSchema::F64, + comment: "Sample rate of inbound/outbound PCM. Default 16000.", + required: false, + }, + ], + outputs: vec![ + FieldSchema { + name: "ok", + ty: TypeSchema::Bool, + comment: "True when the session was opened.", + required: true, + }, + FieldSchema { + name: "request_id", + ty: TypeSchema::String, + comment: "Echoed session key.", + required: true, + }, + FieldSchema { + name: "sample_rate_hz", + ty: TypeSchema::F64, + comment: "Echoed sample rate the session was opened with.", + required: true, + }, + ], + } +} + +fn schema_push_listen_pcm() -> ControllerSchema { + ControllerSchema { + namespace: "meet_agent", + function: "push_listen_pcm", + description: + "Push a chunk of inbound PCM (Meet → agent) into the session. May trigger a brain \ + turn when VAD detects end-of-utterance.", + inputs: vec![ + FieldSchema { + name: "request_id", + ty: TypeSchema::String, + comment: "Session key from start_session.", + required: true, + }, + FieldSchema { + name: "pcm_base64", + ty: TypeSchema::String, + comment: + "Base64-encoded PCM16LE samples at the session's sample rate. Empty allowed.", + required: true, + }, + ], + outputs: vec![ + FieldSchema { + name: "ok", + ty: TypeSchema::Bool, + comment: "True when the chunk was accepted.", + required: true, + }, + FieldSchema { + name: "turn_started", + ty: TypeSchema::Bool, + comment: "True when this push closed an utterance and the brain ran a turn.", + required: true, + }, + ], + } +} + +fn schema_push_caption() -> ControllerSchema { + ControllerSchema { + namespace: "meet_agent", + function: "push_caption", + description: "Push a caption line scraped from Meet's live captions DOM. The wake-word \ + gate (\"hey openhuman\") triggers an LLM/TTS turn when fired.", + inputs: vec![ + FieldSchema { + name: "request_id", + ty: TypeSchema::String, + comment: "Session key from start_session.", + required: true, + }, + FieldSchema { + name: "speaker", + ty: TypeSchema::String, + comment: "Speaker label scraped from Meet (display name); may be empty.", + required: false, + }, + FieldSchema { + name: "text", + ty: TypeSchema::String, + comment: "Caption transcript (already trimmed by the page-side bridge).", + required: true, + }, + FieldSchema { + name: "ts_ms", + ty: TypeSchema::F64, + comment: "Page-side Date.now() when the caption was queued.", + required: false, + }, + ], + outputs: vec![ + FieldSchema { + name: "ok", + ty: TypeSchema::Bool, + comment: "True when the caption was accepted.", + required: true, + }, + FieldSchema { + name: "turn_started", + ty: TypeSchema::Bool, + comment: + "True when this caption tripped the wake word and a brain turn dispatched.", + required: true, + }, + ], + } +} + +fn schema_poll_speech() -> ControllerSchema { + ControllerSchema { + namespace: "meet_agent", + function: "poll_speech", + description: "Drain any synthesized outbound PCM (agent → Meet) the session has queued.", + inputs: vec![FieldSchema { + name: "request_id", + ty: TypeSchema::String, + comment: "Session key from start_session.", + required: true, + }], + outputs: vec![ + FieldSchema { + name: "ok", + ty: TypeSchema::Bool, + comment: "True when the poll succeeded (even if no audio was queued).", + required: true, + }, + FieldSchema { + name: "pcm_base64", + ty: TypeSchema::String, + comment: "Base64 PCM16LE since the last poll. Empty when nothing is queued.", + required: true, + }, + FieldSchema { + name: "utterance_done", + ty: TypeSchema::Bool, + comment: "True when the current outbound utterance is complete.", + required: true, + }, + ], + } +} + +fn schema_stop_session() -> ControllerSchema { + ControllerSchema { + namespace: "meet_agent", + function: "stop_session", + description: "Close the named session and return summary counters.", + inputs: vec![FieldSchema { + name: "request_id", + ty: TypeSchema::String, + comment: "Session key.", + required: true, + }], + outputs: vec![ + FieldSchema { + name: "ok", + ty: TypeSchema::Bool, + comment: "True when the session existed and was closed.", + required: true, + }, + FieldSchema { + name: "request_id", + ty: TypeSchema::String, + comment: "Echoed session key.", + required: true, + }, + FieldSchema { + name: "listened_seconds", + ty: TypeSchema::F64, + comment: "Total seconds of inbound audio processed.", + required: true, + }, + FieldSchema { + name: "spoken_seconds", + ty: TypeSchema::F64, + comment: "Total seconds of outbound audio synthesized.", + required: true, + }, + FieldSchema { + name: "turn_count", + ty: TypeSchema::F64, + comment: "Number of completed agent turns.", + required: true, + }, + ], + } +} + +fn schema_unknown() -> ControllerSchema { + ControllerSchema { + namespace: "meet_agent", + function: "unknown", + description: "Unknown meet_agent controller function.", + inputs: vec![FieldSchema { + name: "function", + ty: TypeSchema::String, + comment: "Unknown function requested.", + required: true, + }], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Lookup error details.", + required: true, + }], + } +} + +fn handle_start_session(p: Map) -> ControllerFuture { + Box::pin(async move { super::rpc::handle_start_session(p).await }) +} +fn handle_push_listen_pcm(p: Map) -> ControllerFuture { + Box::pin(async move { super::rpc::handle_push_listen_pcm(p).await }) +} +fn handle_push_caption(p: Map) -> ControllerFuture { + Box::pin(async move { super::rpc::handle_push_caption(p).await }) +} +fn handle_poll_speech(p: Map) -> ControllerFuture { + Box::pin(async move { super::rpc::handle_poll_speech(p).await }) +} +fn handle_stop_session(p: Map) -> ControllerFuture { + Box::pin(async move { super::rpc::handle_stop_session(p).await }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registered_handlers_match_schemas() { + let schema_fns: Vec<_> = all_controller_schemas() + .into_iter() + .map(|s| s.function) + .collect(); + let handler_fns: Vec<_> = all_registered_controllers() + .into_iter() + .map(|c| c.schema.function) + .collect(); + assert_eq!(schema_fns, handler_fns); + assert_eq!( + schema_fns, + vec![ + "start_session", + "push_listen_pcm", + "push_caption", + "poll_speech", + "stop_session" + ] + ); + } + + #[test] + fn lookup_returns_unknown_for_missing_function() { + assert_eq!(schemas("nope").function, "unknown"); + } + + #[test] + fn start_session_requires_request_id() { + let s = schema_start_session(); + let required: Vec<_> = s + .inputs + .iter() + .filter(|f| f.required) + .map(|f| f.name) + .collect(); + assert_eq!(required, vec!["request_id"]); + } +} diff --git a/src/openhuman/meet_agent/session.rs b/src/openhuman/meet_agent/session.rs new file mode 100644 index 000000000..6e60c649e --- /dev/null +++ b/src/openhuman/meet_agent/session.rs @@ -0,0 +1,478 @@ +//! Per-call session state for the meet-agent loop. +//! +//! A `MeetAgentSession` holds the state that has to live for the +//! duration of a Google Meet call: the inbound PCM ring buffer (kept +//! short — VAD chops it into utterances), the outbound TTS queue (PCM +//! the brain has produced and the shell hasn't drained yet), VAD state, +//! transcript log, and counters for the smoke test. +//! +//! Sessions are keyed by `request_id` (the same UUID `meet/` mints) and +//! live in a process-wide `OnceLock>>`. The locking +//! pattern matches `meet_call::MeetCallState` on the shell side. + +use std::collections::HashMap; +use std::sync::{Mutex, OnceLock}; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; + +use base64::{engine::general_purpose::STANDARD as B64, Engine as _}; + +use super::ops::{self, Vad, VadEvent}; +use super::types::{SessionEvent, SessionEventKind}; + +/// Cap on the inbound buffer so a runaway shell push (e.g. shell never +/// stops, brain never drains) can't grow memory unboundedly. 30s @ 16kHz +/// mono = 960 KB per session — generous for any reasonable utterance. +const MAX_INBOUND_SAMPLES: usize = 30 * 16_000; +/// Same idea for outbound: cap synthesized backlog at 30s. Brain trims +/// older audio if the shell hasn't polled fast enough. +const MAX_OUTBOUND_SAMPLES: usize = 30 * 16_000; +/// Keep the most recent N session events. Bounded so a noisy call +/// can't grow the log forever. +const MAX_EVENTS: usize = 256; + +#[derive(Debug)] +pub struct MeetAgentSession { + pub request_id: String, + pub sample_rate_hz: u32, + /// Wall-clock start. Used by the smoke-test response and to stamp + /// session events. + pub started_at: Instant, + /// PCM samples awaiting brain processing. Drained per utterance. + inbound: Vec, + /// PCM samples the brain has synthesized but the shell hasn't + /// pulled yet. Front-of-vec is "next bytes the shell will consume". + outbound: Vec, + /// True when the *current* outbound batch represents a complete + /// utterance — the shell uses this to flush + drop back to silence. + outbound_done: bool, + vad: Vad, + events: Vec, + /// Total samples ever pushed in. Counter, not a buffer length — + /// the inbound vec is drained per utterance, so we track separately + /// for the smoke-test seconds-listened metric. + total_inbound_samples: u64, + total_outbound_samples: u64, + pub turn_count: u32, + /// Buffer of post-wake-word caption text waiting for the brain + /// turn to fire. Populated by `note_caption` once a wake word is + /// observed; flushed by `take_pending_prompt`. + pending_prompt: String, + /// True between "wake word matched" and "brain turn dispatched". + /// Used to avoid firing a second turn on every subsequent caption + /// line while the prompt is still being assembled. + pub wake_active: bool, + /// `ts_ms` of the last caption that contributed to + /// `pending_prompt`. The brain uses this + the current time to + /// decide whether the user has stopped talking. + pub last_caption_ts_ms: u64, + /// Page-side `Date.now()` of the most recent caption that fired + /// the wake word. Suppresses re-firing while Meet's caption + /// region keeps the same utterance visible (Meet shows captions + /// for ~5–8 s after speaking ends, and our dedupe is per-exact- + /// text — a single character growth re-queues the line). Without + /// this gate the brain spam-fires on every caption growth. + wake_cooldown_until_ts_ms: u64, +} + +impl MeetAgentSession { + pub fn new(request_id: String, sample_rate_hz: u32) -> Self { + Self { + request_id, + sample_rate_hz, + started_at: Instant::now(), + inbound: Vec::new(), + outbound: Vec::new(), + outbound_done: false, + vad: Vad::new(), + events: Vec::new(), + total_inbound_samples: 0, + total_outbound_samples: 0, + turn_count: 0, + pending_prompt: String::new(), + wake_active: false, + last_caption_ts_ms: 0, + wake_cooldown_until_ts_ms: 0, + } + } + + /// Caption-driven listen path. Returns `true` when this caption + /// just tripped the wake word (caller should kick a turn). + /// + /// The wake-word match is intentionally permissive: case-folded + /// substring on `"hey openhuman"` (and `"hey open human"` to + /// tolerate Meet's STT splitting the brand name). Any text after + /// the match in the same caption is treated as the start of the + /// prompt; subsequent captions append until `take_pending_prompt` + /// drains. + pub fn note_caption(&mut self, speaker: &str, text: &str, ts_ms: u64) -> bool { + if text.trim().is_empty() { + return false; + } + self.last_caption_ts_ms = ts_ms; + // Already collecting after a previous wake word: just append + // the new caption. No second fire — the brain is already + // scheduled and will drain the prompt in ~1.5 s. Without this + // gate, a slowly-growing caption fires the wake word on + // every dedupe-then-grow cycle. + if self.wake_active { + if !self.pending_prompt.is_empty() { + self.pending_prompt.push(' '); + } + self.pending_prompt.push_str(text.trim()); + return false; + } + // In cooldown after a recent turn — Meet keeps the same + // utterance visible for several seconds, so without this + // gate the brain re-fires on every caption growth. Continue + // recording the caption to the transcript log (below) but + // skip wake-word matching. + if ts_ms != 0 && ts_ms < self.wake_cooldown_until_ts_ms { + self.record_event( + SessionEventKind::Heard, + if speaker.is_empty() { + text.to_string() + } else { + format!("{speaker}: {text}") + }, + ); + return false; + } + // Normalize before matching: Meet's STT punctuates the wake + // phrase ("hey, openhuman"), capitalizes mid-sentence, and + // sometimes collapses the brand to two words. Folding to + // lowercase + replacing punctuation with spaces + collapsing + // whitespace gives us a single canonical form to substring + // against. The tail (the dictation after the wake phrase) is + // returned in normalized form too — that's fine for the LLM + // and the transcript log; the user's punctuation isn't load- + // bearing for note-taking. + let normalized = normalize_for_wake(text); + let wake_idx = normalized + .find("hey openhuman") + .or_else(|| normalized.find("hey open human")); + if let Some(idx) = wake_idx { + let after = if normalized[idx..].starts_with("hey openhuman") { + idx + "hey openhuman".len() + } else { + idx + "hey open human".len() + }; + let tail = normalized.get(after..).unwrap_or("").trim().to_string(); + self.pending_prompt = tail; + self.wake_active = true; + self.record_event( + SessionEventKind::Note, + format!("wake word from speaker={speaker}"), + ); + return true; + } + // Outside a wake context, just record the line for the + // transcript log. Useful for debugging "why didn't the agent + // respond". (The wake-active branch is handled by the + // early-return above.) + self.record_event( + SessionEventKind::Heard, + if speaker.is_empty() { + text.to_string() + } else { + format!("{speaker}: {text}") + }, + ); + false + } + + /// Drain the assembled wake-word prompt and clear the active + /// flag. The brain calls this once it's ready to dispatch the + /// turn so subsequent captions start a fresh wake-word cycle. + /// + /// Sets a cooldown window keyed off `last_caption_ts_ms` so any + /// subsequent caption push for the same lingering utterance + /// doesn't re-fire the wake-word state machine. 8s is a comfortable + /// upper bound on how long Meet keeps a finalised caption visible. + pub fn take_pending_prompt(&mut self) -> Option { + if !self.wake_active { + return None; + } + self.wake_active = false; + // 8s grace beyond the most recent caption's page timestamp. + // `last_caption_ts_ms` is whatever Date.now() was page-side + // when the line landed — same clock as future caption pushes. + const COOLDOWN_MS: u64 = 8_000; + self.wake_cooldown_until_ts_ms = self.last_caption_ts_ms.saturating_add(COOLDOWN_MS); + let prompt = std::mem::take(&mut self.pending_prompt); + let trimmed = prompt.trim().to_string(); + if trimmed.is_empty() { + None + } else { + Some(trimmed) + } + } + + /// Append PCM samples to the inbound buffer. Returns the VAD verdict + /// for *this* batch — caller consults it to decide whether to fire + /// a brain turn. + pub fn push_inbound_pcm(&mut self, samples: &[i16]) -> VadEvent { + self.total_inbound_samples += samples.len() as u64; + self.inbound.extend_from_slice(samples); + if self.inbound.len() > MAX_INBOUND_SAMPLES { + // Drop oldest; the in-progress utterance is what matters. + let drop = self.inbound.len() - MAX_INBOUND_SAMPLES; + self.inbound.drain(..drop); + } + self.vad.feed(samples) + } + + /// Take ownership of the accumulated utterance for STT. The session + /// keeps the VAD state — the next push_inbound_pcm starts a fresh + /// utterance. + pub fn drain_inbound(&mut self) -> Vec { + std::mem::take(&mut self.inbound) + } + + /// Brain hands synthesized PCM back to the session. `done` flips + /// `outbound_done` so the next poll surfaces "utterance over". + pub fn enqueue_outbound_pcm(&mut self, samples: &[i16], done: bool) { + self.total_outbound_samples += samples.len() as u64; + self.outbound.extend_from_slice(samples); + if self.outbound.len() > MAX_OUTBOUND_SAMPLES { + let drop = self.outbound.len() - MAX_OUTBOUND_SAMPLES; + self.outbound.drain(..drop); + } + if done { + self.outbound_done = true; + } + } + + /// Drain everything currently queued for the shell. Returns + /// `(pcm_base64, utterance_done)`. + pub fn poll_outbound(&mut self) -> (String, bool) { + if self.outbound.is_empty() { + let done = std::mem::take(&mut self.outbound_done); + return (String::new(), done); + } + let bytes: Vec = self + .outbound + .drain(..) + .flat_map(|s| s.to_le_bytes()) + .collect(); + let done = std::mem::take(&mut self.outbound_done); + (B64.encode(bytes), done) + } + + pub fn record_event(&mut self, kind: SessionEventKind, text: String) { + let timestamp_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + self.events.push(SessionEvent { + kind, + text, + timestamp_ms, + }); + if self.events.len() > MAX_EVENTS { + let drop = self.events.len() - MAX_EVENTS; + self.events.drain(..drop); + } + } + + pub fn events(&self) -> &[SessionEvent] { + &self.events + } + + pub fn listened_seconds(&self) -> f32 { + self.total_inbound_samples as f32 / self.sample_rate_hz as f32 + } + + pub fn spoken_seconds(&self) -> f32 { + self.total_outbound_samples as f32 / self.sample_rate_hz as f32 + } +} + +/// Lowercase + drop punctuation + collapse whitespace, so the wake +/// phrase matches regardless of how Meet's STT punctuated or cased +/// it ("Hey, OpenHuman", "hey open-human", etc). +fn normalize_for_wake(text: &str) -> String { + let mut out = String::with_capacity(text.len()); + let mut prev_space = true; + for c in text.chars() { + let lc = c.to_ascii_lowercase(); + if lc.is_ascii_alphanumeric() { + out.push(lc); + prev_space = false; + } else if !prev_space { + out.push(' '); + prev_space = true; + } + } + out.trim_end().to_string() +} + +/// Process-wide session registry. Sessions are keyed by `request_id`. +#[derive(Default)] +pub struct MeetAgentSessionRegistry { + inner: Mutex>, +} + +impl MeetAgentSessionRegistry { + pub fn new() -> Self { + Self::default() + } + + pub fn start(&self, request_id: &str, sample_rate_hz: u32) -> Result<(), String> { + let request_id = ops::sanitize_request_id(request_id)?; + let sample_rate_hz = ops::validate_sample_rate(sample_rate_hz)?; + let mut guard = self.inner.lock().unwrap(); + if guard.contains_key(&request_id) { + // Idempotent restart: replace the old session so a shell + // crash + reconnect doesn't wedge the registry. + log::info!("[meet-agent] replacing existing session request_id={request_id}"); + } + guard.insert( + request_id.clone(), + MeetAgentSession::new(request_id, sample_rate_hz), + ); + Ok(()) + } + + pub fn stop(&self, request_id: &str) -> Result { + let request_id = ops::sanitize_request_id(request_id)?; + let mut guard = self.inner.lock().unwrap(); + guard + .remove(&request_id) + .ok_or_else(|| format!("[meet-agent] no session for request_id={request_id}")) + } + + /// Run a closure with mutable access to the named session. Returns + /// `Err` when the session is unknown. + pub fn with_session( + &self, + request_id: &str, + f: impl FnOnce(&mut MeetAgentSession) -> R, + ) -> Result { + let request_id = ops::sanitize_request_id(request_id)?; + let mut guard = self.inner.lock().unwrap(); + let session = guard + .get_mut(&request_id) + .ok_or_else(|| format!("[meet-agent] no session for request_id={request_id}"))?; + Ok(f(session)) + } + + #[cfg(test)] + pub fn len(&self) -> usize { + self.inner.lock().unwrap().len() + } +} + +/// Process-wide singleton. Lazy-initialized so tests can use a fresh +/// registry where they want to. +pub static SESSION_REGISTRY: OnceLock = OnceLock::new(); + +pub fn registry() -> &'static MeetAgentSessionRegistry { + SESSION_REGISTRY.get_or_init(MeetAgentSessionRegistry::new) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn start_and_stop_round_trip() { + let reg = MeetAgentSessionRegistry::new(); + reg.start("abc-123", 16_000).unwrap(); + assert_eq!(reg.len(), 1); + let session = reg.stop("abc-123").unwrap(); + assert_eq!(session.request_id, "abc-123"); + assert_eq!(reg.len(), 0); + } + + #[test] + fn start_rejects_bad_inputs() { + let reg = MeetAgentSessionRegistry::new(); + assert!(reg.start("", 16_000).is_err()); + assert!(reg.start("abc", 1_000).is_err()); + } + + #[test] + fn stop_unknown_session_errors() { + let reg = MeetAgentSessionRegistry::new(); + assert!(reg.stop("never-started").is_err()); + } + + #[test] + fn push_inbound_accumulates_samples() { + let reg = MeetAgentSessionRegistry::new(); + reg.start("s1", 16_000).unwrap(); + reg.with_session("s1", |s| { + s.push_inbound_pcm(&vec![1000; 320]); + s.push_inbound_pcm(&vec![1000; 320]); + assert_eq!(s.inbound.len(), 640); + }) + .unwrap(); + } + + #[test] + fn poll_outbound_returns_done_flag_once() { + let reg = MeetAgentSessionRegistry::new(); + reg.start("s2", 16_000).unwrap(); + reg.with_session("s2", |s| { + s.enqueue_outbound_pcm(&vec![0; 100], true); + let (b64, done) = s.poll_outbound(); + assert!(!b64.is_empty()); + assert!(done); + // Second poll: no audio, no `done` (we already consumed it). + let (b64, done) = s.poll_outbound(); + assert!(b64.is_empty()); + assert!(!done); + }) + .unwrap(); + } + + #[test] + fn note_caption_handles_punctuated_wake() { + let mut s = MeetAgentSession::new("p".into(), 16_000); + // Meet often inserts a comma after "hey". + let fired = s.note_caption("Alice", "Hey, OpenHuman remember the launch", 1); + assert!(fired, "punctuated wake phrase should still fire"); + let prompt = s.take_pending_prompt().expect("prompt drained"); + assert_eq!(prompt, "remember the launch"); + } + + #[test] + fn note_caption_handles_split_brand() { + let mut s = MeetAgentSession::new("p".into(), 16_000); + let fired = s.note_caption("Alice", "hey open-human, send the report", 1); + assert!(fired); + let prompt = s.take_pending_prompt().expect("prompt drained"); + assert_eq!(prompt, "send the report"); + } + + #[test] + fn note_caption_does_not_double_fire_on_growing_caption() { + let mut s = MeetAgentSession::new("p".into(), 16_000); + let first = s.note_caption("Alice", "hey openhuman take notes", 1); + assert!(first); + let second = s.note_caption("Alice", "hey openhuman take notes about the launch", 2); + assert!(!second, "second caption while wake_active must not refire"); + let prompt = s.take_pending_prompt().expect("prompt drained"); + // First wake stripped "hey openhuman"; the continuation + // appended the WHOLE growing caption (still containing "hey + // openhuman" because we don't re-strip), separated by a + // space. That's fine — the LLM ignores the prefix and the + // transcript log still records the verbatim dictation. + assert!( + prompt.contains("take notes about the launch"), + "got prompt: {prompt}" + ); + } + + #[test] + fn listened_seconds_tracks_total_inbound() { + let reg = MeetAgentSessionRegistry::new(); + reg.start("s3", 16_000).unwrap(); + reg.with_session("s3", |s| { + s.push_inbound_pcm(&vec![0; 16_000]); // 1.0s + s.push_inbound_pcm(&vec![0; 8_000]); // 0.5s + assert!((s.listened_seconds() - 1.5).abs() < 1e-3); + }) + .unwrap(); + } +} diff --git a/src/openhuman/meet_agent/types.rs b/src/openhuman/meet_agent/types.rs new file mode 100644 index 000000000..636fbd682 --- /dev/null +++ b/src/openhuman/meet_agent/types.rs @@ -0,0 +1,152 @@ +//! Request / response types for the `meet_agent` domain. +//! +//! Audio frames cross the RPC boundary as base64-encoded PCM16LE @ 16kHz +//! mono. Base64 (rather than raw bytes) because JSON-RPC transports the +//! envelope as JSON and binary bytes don't survive the trip — the shell +//! decodes/encodes at the `core_rpc` boundary, mirroring how the existing +//! `voice::streaming` WebSocket path moves audio. + +use serde::{Deserialize, Serialize}; + +/// Inputs to `openhuman.meet_agent_start_session`. +#[derive(Debug, Clone, Deserialize)] +pub struct StartSessionRequest { + /// `request_id` minted by `openhuman.meet_join_call`. Used as the + /// session key so the shell's existing per-call book-keeping (window + /// label, data dir) lines up with the agent loop's session. + pub request_id: String, + /// Sample rate of the PCM frames the shell will push. Must match + /// what `voice::streaming` expects (16000) — the shell is responsible + /// for resampling the CEF audio handler's native rate down before + /// sending. Validated on entry. + #[serde(default = "default_sample_rate")] + pub sample_rate_hz: u32, +} + +fn default_sample_rate() -> u32 { + 16_000 +} + +/// Outputs from `openhuman.meet_agent_start_session`. +#[derive(Debug, Clone, Serialize)] +pub struct StartSessionResponse { + pub ok: bool, + pub request_id: String, + /// Echoed sample rate the session was opened with — the shell pins + /// its resampler to this. + pub sample_rate_hz: u32, +} + +/// Inputs to `openhuman.meet_agent_push_listen_pcm`. +/// +/// Sent every ~100ms while the call is open. Small frames keep VAD +/// responsive without overloading the JSON envelope. +#[derive(Debug, Clone, Deserialize)] +pub struct PushListenPcmRequest { + pub request_id: String, + /// Base64-encoded PCM16LE samples at the session's `sample_rate_hz`. + /// Empty string is allowed and treated as "no audio this tick" + /// (used by the shell to keep the keep-alive heartbeat without a + /// payload when CEF reports silence). + pub pcm_base64: String, +} + +#[derive(Debug, Clone, Serialize)] +pub struct PushListenPcmResponse { + pub ok: bool, + /// True when this push triggered a VAD-detected end-of-utterance and + /// the brain ran a turn. The shell can use this as a UI hint + /// ("agent is thinking…"). + pub turn_started: bool, +} + +/// Inputs to `openhuman.meet_agent_poll_speech`. +/// +/// Pull-style: the shell calls this periodically and gets any PCM the +/// brain has synthesized since the last poll. Pull beats push here +/// because the shell is the side that knows whether the virtual mic is +/// actually draining (back-pressure lives there, not in core). +#[derive(Debug, Clone, Deserialize)] +pub struct PollSpeechRequest { + pub request_id: String, +} + +#[derive(Debug, Clone, Serialize)] +pub struct PollSpeechResponse { + pub ok: bool, + /// Base64-encoded PCM16LE @ session sample rate, or empty when there + /// is nothing queued. The shell appends this to its UDS feed. + pub pcm_base64: String, + /// True when the brain has finished synthesizing the current + /// utterance and the shell can flush + drop back to silence. + pub utterance_done: bool, +} + +/// Inputs to `openhuman.meet_agent_push_caption`. +/// +/// One row per new line scraped from Meet's captions DOM. Sent by the +/// shell's `caption_listener` every ~500 ms. The wake-word state +/// machine in the brain (see `brain::on_caption`) decides whether to +/// fire a turn. +#[derive(Debug, Clone, Deserialize)] +pub struct PushCaptionRequest { + pub request_id: String, + /// Speaker label scraped from Meet (the participant's display + /// name); empty when the captions row didn't expose one. + #[serde(default)] + pub speaker: String, + /// Caption transcript. Already trimmed by the page-side bridge. + pub text: String, + /// `Date.now()` from the page when the line was queued. Used + /// only for ordering / staleness — the brain treats it as opaque. + #[serde(default)] + pub ts_ms: u64, +} + +#[derive(Debug, Clone, Serialize)] +pub struct PushCaptionResponse { + pub ok: bool, + /// True when this caption tripped the wake-word and a brain turn + /// is now in flight. + pub turn_started: bool, +} + +/// Inputs to `openhuman.meet_agent_stop_session`. +#[derive(Debug, Clone, Deserialize)] +pub struct StopSessionRequest { + pub request_id: String, +} + +#[derive(Debug, Clone, Serialize)] +pub struct StopSessionResponse { + pub ok: bool, + pub request_id: String, + /// Total seconds of inbound audio the session processed — useful + /// for telemetry and the smoke test in [`crate::openhuman::meet_agent`]. + pub listened_seconds: f32, + /// Total seconds of outbound audio the session synthesized. + pub spoken_seconds: f32, + /// Number of completed agent turns (one transcript + one TTS reply). + pub turn_count: u32, +} + +/// Lightweight transcript / event record kept per session. Exposed so +/// the shell can render a live captions overlay and so the json_rpc_e2e +/// test can assert turn boundaries. +#[derive(Debug, Clone, Serialize)] +pub struct SessionEvent { + pub kind: SessionEventKind, + pub text: String, + pub timestamp_ms: u64, +} + +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum SessionEventKind { + /// Final STT transcript for an inbound utterance. + Heard, + /// Outbound text the agent decided to speak. + Spoke, + /// Internal note (errors, "agent declined to respond", etc). + Note, +} diff --git a/src/openhuman/meet_agent/wav.rs b/src/openhuman/meet_agent/wav.rs new file mode 100644 index 000000000..7f833ac36 --- /dev/null +++ b/src/openhuman/meet_agent/wav.rs @@ -0,0 +1,79 @@ +//! Tiny PCM16LE → WAV-container wrapper used to ship audio batches to +//! the backend Whisper endpoint. +//! +//! `voice::cloud_transcribe` takes whatever the desktop UI captured +//! (typically `audio/webm`) and forwards bytes to the backend. Our +//! call buffers are raw PCM16LE @ 16 kHz mono — Whisper accepts WAV +//! natively, so we wrap the bytes in a minimal RIFF/WAVE header and +//! mark the upload as `audio/wav`. No other transcoding needed. + +const WAV_HEADER_LEN: usize = 44; + +/// Produce a complete WAV file (header + interleaved PCM16LE samples). +/// Caller passes the raw `i16` slice and the sample rate; mono is +/// hard-coded because that's what the meet-agent loop uses end-to-end. +pub fn pack_pcm16le_mono_wav(samples: &[i16], sample_rate_hz: u32) -> Vec { + let data_bytes = samples.len() * 2; + let mut out = Vec::with_capacity(WAV_HEADER_LEN + data_bytes); + + // RIFF chunk descriptor + out.extend_from_slice(b"RIFF"); + out.extend_from_slice(&((36 + data_bytes) as u32).to_le_bytes()); + out.extend_from_slice(b"WAVE"); + + // fmt sub-chunk + out.extend_from_slice(b"fmt "); + out.extend_from_slice(&16u32.to_le_bytes()); // PCM header size + out.extend_from_slice(&1u16.to_le_bytes()); // audio format = PCM + out.extend_from_slice(&1u16.to_le_bytes()); // num channels = 1 + out.extend_from_slice(&sample_rate_hz.to_le_bytes()); + out.extend_from_slice(&(sample_rate_hz * 2).to_le_bytes()); // byte rate + out.extend_from_slice(&2u16.to_le_bytes()); // block align + out.extend_from_slice(&16u16.to_le_bytes()); // bits per sample + + // data sub-chunk + out.extend_from_slice(b"data"); + out.extend_from_slice(&(data_bytes as u32).to_le_bytes()); + for s in samples { + out.extend_from_slice(&s.to_le_bytes()); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn header_bytes_match_riff_wave_layout() { + let bytes = pack_pcm16le_mono_wav(&[0; 8000], 16_000); + assert_eq!(&bytes[0..4], b"RIFF"); + assert_eq!(&bytes[8..12], b"WAVE"); + assert_eq!(&bytes[12..16], b"fmt "); + assert_eq!(&bytes[36..40], b"data"); + // RIFF size = 36 + data_bytes (8000 samples * 2 bytes = 16000). + let riff_size = u32::from_le_bytes([bytes[4], bytes[5], bytes[6], bytes[7]]); + assert_eq!(riff_size, 36 + 16_000); + // Sample rate field at offset 24. + let rate = u32::from_le_bytes([bytes[24], bytes[25], bytes[26], bytes[27]]); + assert_eq!(rate, 16_000); + } + + #[test] + fn empty_input_still_produces_valid_header() { + let bytes = pack_pcm16le_mono_wav(&[], 16_000); + assert_eq!(bytes.len(), WAV_HEADER_LEN); + assert_eq!(&bytes[0..4], b"RIFF"); + } + + #[test] + fn samples_are_appended_little_endian() { + let bytes = pack_pcm16le_mono_wav(&[0x1234, -1], 16_000); + // First sample 0x1234 → LE bytes 0x34, 0x12 starting at offset 44. + assert_eq!(bytes[44], 0x34); + assert_eq!(bytes[45], 0x12); + // -1 in i16 LE → 0xFF, 0xFF. + assert_eq!(bytes[46], 0xFF); + assert_eq!(bytes[47], 0xFF); + } +} diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index e6c40b782..cf48ed0c7 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -38,6 +38,7 @@ pub mod integrations; pub mod learning; pub mod local_ai; pub mod meet; +pub mod meet_agent; pub mod memory; pub mod migration; pub mod node_runtime; diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index 0e5c23e6b..50c145bf0 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -3927,3 +3927,168 @@ async fn json_rpc_meet_join_call_validates_and_returns_request_id() { rpc_join.abort(); } + +/// Walks the full meet_agent session lifecycle: +/// start_session → push silent frame → push loud frame ×N → push +/// silent frames until VAD fires a turn → poll_speech (expects +/// non-empty PCM from the brain stub) → stop_session. +/// +/// Pins behavior the shell relies on: the RPC surface accepts +/// base64-PCM16LE frames, fires a turn on VAD silence after speech, +/// the brain stub enqueues outbound audio synchronously enough for a +/// 250 ms-budget poll to see it, and stop_session returns sane +/// counters. STT / TTS adapters are stubbed in PR1 so this stays +/// network-free. +#[tokio::test] +async fn json_rpc_meet_agent_session_lifecycle() { + use base64::{engine::general_purpose::STANDARD as B64, Engine as _}; + + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{}", rpc_addr); + + tokio::time::sleep(Duration::from_millis(50)).await; + + let request_id = "e2e-meet-agent"; + + // 1) start_session — opens registry slot, defaults sample_rate to 16000. + let start = post_json_rpc( + &rpc_base, + 9101, + "openhuman.meet_agent_start_session", + json!({ "request_id": request_id, "sample_rate_hz": 16_000 }), + ) + .await; + let start_result = assert_no_jsonrpc_error(&start, "start_session ok"); + let start_body = start_result.get("result").unwrap_or(start_result); + assert_eq!(start_body.get("ok"), Some(&json!(true))); + assert_eq!( + start_body.get("request_id").and_then(|v| v.as_str()), + Some(request_id) + ); + + // 2) Push ~1s of "loud" PCM (square wave well above VAD threshold) + // so the brain has enough material to NOT skip the turn. + let loud_frame: Vec = (0..1600) + .map(|i| if i % 2 == 0 { 8000i16 } else { -8000 }) + .collect(); + let loud_b64 = { + let bytes: Vec = loud_frame.iter().flat_map(|s| s.to_le_bytes()).collect(); + B64.encode(bytes) + }; + for i in 0..10 { + let r = post_json_rpc( + &rpc_base, + 9110 + i, + "openhuman.meet_agent_push_listen_pcm", + json!({ "request_id": request_id, "pcm_base64": loud_b64 }), + ) + .await; + let body = assert_no_jsonrpc_error(&r, "push_listen_pcm loud"); + let body = body.get("result").unwrap_or(body); + assert_eq!( + body.get("turn_started"), + Some(&json!(false)), + "VAD must not fire while still hearing speech" + ); + } + + // 3) Push silent frames until turn_started flips. With + // VAD_HANGOVER_FRAMES=6 the turn should fire within at most + // ~7 silent pushes (allow 12 for slop). + let silent_frame = vec![0i16; 1600]; + let silent_b64 = { + let bytes: Vec = silent_frame.iter().flat_map(|s| s.to_le_bytes()).collect(); + B64.encode(bytes) + }; + let mut turn_fired = false; + for i in 0..12 { + let r = post_json_rpc( + &rpc_base, + 9130 + i, + "openhuman.meet_agent_push_listen_pcm", + json!({ "request_id": request_id, "pcm_base64": silent_b64 }), + ) + .await; + let body = assert_no_jsonrpc_error(&r, "push_listen_pcm silent"); + let body = body.get("result").unwrap_or(body); + if body.get("turn_started") == Some(&json!(true)) { + turn_fired = true; + break; + } + } + assert!(turn_fired, "VAD silence run failed to close utterance"); + + // 4) Give the spawned brain turn a chance to finish, then poll for + // synthesized PCM. The stub TTS produces 200 ms of 440 Hz tone + // which encodes to ~6.4 KB of base64. + let mut got_audio = false; + for _ in 0..20 { + tokio::time::sleep(Duration::from_millis(50)).await; + let r = post_json_rpc( + &rpc_base, + 9150, + "openhuman.meet_agent_poll_speech", + json!({ "request_id": request_id }), + ) + .await; + let body = assert_no_jsonrpc_error(&r, "poll_speech ok"); + let body = body.get("result").unwrap_or(body); + let b64 = body + .get("pcm_base64") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + if !b64.is_empty() { + got_audio = true; + assert!( + b64.len() > 1000, + "stub TTS should produce a multi-KB base64 payload" + ); + break; + } + } + assert!(got_audio, "expected synthesized audio after VAD-fired turn"); + + // 5) stop_session returns counters. listened_seconds should be + // > 0 (we pushed >1s of audio); turn_count should be exactly 1. + let stop = post_json_rpc( + &rpc_base, + 9160, + "openhuman.meet_agent_stop_session", + json!({ "request_id": request_id }), + ) + .await; + let stop_result = assert_no_jsonrpc_error(&stop, "stop_session ok"); + let stop_body = stop_result.get("result").unwrap_or(stop_result); + assert_eq!(stop_body.get("ok"), Some(&json!(true))); + let listened = stop_body + .get("listened_seconds") + .and_then(|v| v.as_f64()) + .unwrap_or(0.0); + assert!(listened > 1.0, "expected >1s listened, got {listened:.2}"); + let turns = stop_body + .get("turn_count") + .and_then(|v| v.as_u64()) + .unwrap_or(0); + assert_eq!(turns, 1, "expected exactly one brain turn"); + + // 6) Stopping a non-existent session is an error (not silent). + let bogus = post_json_rpc( + &rpc_base, + 9161, + "openhuman.meet_agent_stop_session", + json!({ "request_id": "never-started" }), + ) + .await; + assert_jsonrpc_error(&bogus, "stop_session unknown"); + + rpc_join.abort(); +}