diff --git a/Cargo.lock b/Cargo.lock index b4986b67b4..0a924f4e36 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1040,6 +1040,18 @@ dependencies = [ "webbrowser", ] +[[package]] +name = "buzz-audio-client" +version = "0.1.0" +dependencies = [ + "futures-util", + "neteq", + "opus", + "serde_json", + "tokio", + "tokio-tungstenite 0.29.0", +] + [[package]] name = "buzz-audit" version = "0.1.0" @@ -6244,6 +6256,19 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "neteq" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57fbe9601485d95750187c92344fa056e75055027b00e375c53d9f4b8b883112" +dependencies = [ + "log", + "ringbuf", + "serde", + "thiserror 1.0.69", + "web-time", +] + [[package]] name = "netlink-packet-core" version = "0.8.1" @@ -7000,6 +7025,24 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" +[[package]] +name = "opus" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33718946cc77d4032911d4efe03a66dbcbfbd2bb16c3da06aaeadcc637c32216" +dependencies = [ + "opusic-sys", +] + +[[package]] +name = "opusic-sys" +version = "0.7.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9d1ecdf206421bc74343ab3bb2f30ad2abbfee41fa341f7181fecbaf957769a" +dependencies = [ + "cmake", +] + [[package]] name = "ordered-float" version = "2.10.1" @@ -8577,6 +8620,15 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "ringbuf" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79abed428d1fd2a128201cec72c5f6938e2da607c6f3745f769fabea399d950a" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "rmcp" version = "1.8.0" @@ -10708,6 +10760,7 @@ dependencies = [ "futures-util", "log", "rustls", + "rustls-native-certs", "rustls-pki-types", "tokio", "tokio-rustls", diff --git a/Cargo.toml b/Cargo.toml index d27c63d234..df77ab7cb3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,6 +31,7 @@ members = [ "crates/buzz-mesh-smoke", "crates/buzz-dev-mcp", "crates/buzz-voice", + "crates/buzz-audio-client", "crates/buzz-backend-kubernetes", "crates/buzz-datastore-tracing", "examples/countdown-bot", diff --git a/Dockerfile b/Dockerfile index 2c1e8adbb5..1f10b0f260 100644 --- a/Dockerfile +++ b/Dockerfile @@ -54,6 +54,7 @@ FROM chef AS builder RUN apt-get update \ && apt-get install -y --no-install-recommends \ build-essential \ + cmake \ pkg-config \ libssl-dev \ ca-certificates \ diff --git a/Dockerfile.push-gateway b/Dockerfile.push-gateway index 5b0aca8bd3..2b899c89e5 100644 --- a/Dockerfile.push-gateway +++ b/Dockerfile.push-gateway @@ -13,7 +13,7 @@ RUN cargo chef prepare --recipe-path recipe.json FROM chef AS builder ARG BUZZ_PUSH_CARGO_FEATURES="" RUN apt-get update \ - && apt-get install -y --no-install-recommends build-essential pkg-config libssl-dev ca-certificates \ + && apt-get install -y --no-install-recommends build-essential cmake pkg-config libssl-dev ca-certificates \ && rm -rf /var/lib/apt/lists/* COPY --from=planner /build/recipe.json recipe.json RUN cargo chef cook --release --recipe-path recipe.json --features "$BUZZ_PUSH_CARGO_FEATURES" diff --git a/Justfile b/Justfile index 3e7ead1528..8d107af69e 100644 --- a/Justfile +++ b/Justfile @@ -397,6 +397,8 @@ test-unit: #!/usr/bin/env bash set -euo pipefail ./scripts/test-ensure-local-relay-key.sh + # Shared Huddle transport, codec, and loopback authentication regressions. + cargo test -p buzz-audio-client if command -v cargo-nextest &>/dev/null; then cargo nextest run -p buzz-core -p buzz-auth --lib cargo nextest run -p buzz-audit --lib diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 985b29bee7..026b2344a0 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -3384,29 +3384,85 @@ mod tests { #[tokio::test] async fn idle_resets_on_stdout_activity() { - // Send valid JSON (session/update notifications) to reset the idle timer. - // Non-JSON lines no longer reset idle — only valid JSON notifications do. - let mut client = spawn_script( - r#"for i in {1..10}; do echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"agent_thought_chunk","content":{"text":"thinking"}}}}'; sleep 0.05; done; sleep 10"#, - ) + assert_activity_resets_idle(serde_json::json!({ + "sessionUpdate": "agent_thought_chunk", "content": { "text": "thinking" } + })) .await; - let max_dur = std::time::Duration::from_secs(10); - let hard_deadline = tokio::time::Instant::now() + max_dur; - let start = std::time::Instant::now(); - let result = client - .read_until_response_with_idle_timeout( + } + + /// Exercise the real stdout reader with a paused clock. Advance only after + /// each notification is observed, so OS scheduling cannot consume the idle + /// window. Each update must extend the deadline; silence must still expire. + async fn assert_activity_resets_idle(update: serde_json::Value) { + use tokio::io::AsyncWriteExt; + + let mut client = + spawn_script(r#"while IFS= read -r line; do printf '%s\n' "$line"; done"#).await; + let mut stdin = client.stdin.take().expect("test stdin"); + let observer = ObserverHandle::in_process(); + let mut events = observer.subscribe(); + client.set_observer(Some(observer), 0); + // Prevent Tokio from auto-advancing while waiting for real pipe I/O. + let clock_guard = tokio::spawn(async { + loop { + tokio::task::yield_now().await; + } + }); + tokio::time::pause(); + let idle = std::time::Duration::from_secs(1); + let hard = std::time::Duration::from_secs(10); + let hard_deadline = tokio::time::Instant::now() + hard; + let mut frame = serde_json::json!({ + "jsonrpc": "2.0", "method": "session/update", "params": { "update": update } + }) + .to_string(); + frame.push('\n'); + { + let read = client.read_until_response_with_idle_timeout( "test", 999, - std::time::Duration::from_millis(200), + idle, hard_deadline, - max_dur, - ) - .await; - let elapsed = start.elapsed(); - // 10 messages × 50ms = ~500ms of activity, then idle timeout fires after 200ms more - assert!(elapsed >= std::time::Duration::from_millis(400)); - assert!(elapsed < std::time::Duration::from_secs(3)); - assert!(matches!(result, Err(AcpError::IdleTimeout(_)))); + hard, + ); + tokio::pin!(read); + assert!(futures_util::poll!(&mut read).is_pending()); + for _ in 0..3 { + tokio::time::advance(std::time::Duration::from_millis(750)).await; + stdin + .write_all(frame.as_bytes()) + .await + .expect("write activity"); + let watchdog = std::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + assert!( + futures_util::poll!(&mut read).is_pending(), + "activity must extend the idle deadline" + ); + if let Ok(event) = events.try_recv() { + assert_eq!(event.kind, "acp_read"); + break; + } + assert!( + std::time::Instant::now() < watchdog, + "activity was not read from the pipe" + ); + tokio::task::yield_now().await; + } + } + tokio::time::advance(idle - std::time::Duration::from_millis(1)).await; + assert!( + futures_util::poll!(&mut read).is_pending(), + "the last activity must reset the full idle window" + ); + tokio::time::advance(std::time::Duration::from_millis(1)).await; + // Pipe I/O is done. Let Tokio drive the timer wheel, whose wakeup + // can round the deadline up to the next millisecond. + clock_guard.abort(); + assert!(matches!(read.await, Err(AcpError::IdleTimeout(duration)) if duration == idle)); + } + tokio::time::resume(); + client.shutdown().await; } #[tokio::test] @@ -3581,33 +3637,7 @@ mod tests { #[tokio::test] async fn keepalive_resets_idle_past_deadline() { - // Keepalive session/update lines every 50ms against a 250ms idle deadline. - // The turn should survive well past the idle deadline (proves the fix). - let mut client = spawn_script( - r#"for i in {1..20}; do echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"keepalive"}}}'; sleep 0.05; done; sleep 10"#, - ) - .await; - let max_dur = std::time::Duration::from_secs(10); - let hard_deadline = tokio::time::Instant::now() + max_dur; - let start = std::time::Instant::now(); - let result = client - .read_until_response_with_idle_timeout( - "test", - 999, - std::time::Duration::from_millis(250), - hard_deadline, - max_dur, - ) - .await; - let elapsed = start.elapsed(); - // 20 keepalives × 50ms = ~1000ms of activity, then idle fires after 250ms more. - // Must survive well past the 250ms deadline. - assert!( - elapsed >= std::time::Duration::from_millis(750), - "keepalive should reset idle past the deadline; elapsed only {elapsed:?}" - ); - assert!(elapsed < std::time::Duration::from_secs(5)); - assert!(matches!(result, Err(AcpError::IdleTimeout(_)))); + assert_activity_resets_idle(serde_json::json!({ "sessionUpdate": "keepalive" })).await; } #[tokio::test] diff --git a/crates/buzz-agent/tests/fake_llm.rs b/crates/buzz-agent/tests/fake_llm.rs index 2451deae08..86c469f982 100644 --- a/crates/buzz-agent/tests/fake_llm.rs +++ b/crates/buzz-agent/tests/fake_llm.rs @@ -956,10 +956,18 @@ async fn steer_rejected_when_no_active_run() { async fn steer_rejected_on_run_id_mismatch() { // A live run, but the caller targets a stale/wrong run id → invalid_params, // so the client falls back to cancel+merge instead of injecting blind. - let (url, _captures) = spawn_capturing_fake_llm(vec![ - openai_tool_call("call_x", "fake__noop", json!({})), - openai_text("done"), - ]) + // Hold the provider response until the rejection is observed. Otherwise + // the turn can finish before the client sends or receives the steer. + let (gate_tx, gate_rx) = tokio::sync::oneshot::channel(); + let responses = vec![CannedResponse { + status: 200, + body: openai_text("done"), + }]; + let (url, _) = spawn_gated_capturing_fake_llm( + responses, + Arc::new(Mutex::new(Vec::new())), + Arc::new(Mutex::new(Some(gate_rx))), + ) .await; let mut h = Harness::spawn(&url).await; let sid = init_session(&mut h).await; @@ -983,21 +991,14 @@ async fn steer_rejected_on_run_id_mismatch() { ) .await; - let mut saw_reject = false; - for _ in 0..40 { - let v = h.recv().await; - if v["id"] == json!(s_id) { - assert_eq!( - v["error"]["code"], -32602, - "mismatched runId must be rejected" - ); - saw_reject = true; - } else if v["id"] == json!(p_id) { - // Turn finishes normally regardless of the rejected steer. - break; - } - } - assert!(saw_reject, "run-id mismatch was not rejected"); + let rejected = h.recv_until(|v| v["id"] == json!(s_id)).await; + assert_eq!( + rejected["error"]["code"], -32602, + "mismatched runId must be rejected" + ); + gate_tx.send(()).expect("provider gate should remain live"); + let completed = h.recv_until(|v| v["id"] == json!(p_id)).await; + assert_eq!(completed["result"]["stopReason"], "end_turn"); h.shutdown().await; } diff --git a/crates/buzz-audio-client/Cargo.toml b/crates/buzz-audio-client/Cargo.toml new file mode 100644 index 0000000000..8b0a591925 --- /dev/null +++ b/crates/buzz-audio-client/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "buzz-audio-client" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Shared Huddle audio transport and codecs without a desktop application" + +[features] +default = ["native-roots"] +native-roots = ["tokio-tungstenite/rustls-tls-native-roots"] +webpki-roots = ["tokio-tungstenite/rustls-tls-webpki-roots"] + +[dependencies] +futures-util = { version = "0.3", features = ["sink"] } +# Keep the audited unused ringbuf dependency pinned until NetEQ removes it. +neteq = { version = "=0.8.3", default-features = false } +opus = "0.4" +serde_json = "1" +tokio = { version = "1", features = ["net", "time"] } +tokio-tungstenite = { version = "0.29", default-features = false, features = ["connect", "handshake"] } + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt", "sync", "test-util"] } diff --git a/crates/buzz-audio-client/README.md b/crates/buzz-audio-client/README.md new file mode 100644 index 0000000000..f847c98576 --- /dev/null +++ b/crates/buzz-audio-client/README.md @@ -0,0 +1,42 @@ +# Shared Huddle audio client + +`buzz-audio-client` contains the Huddle v2 audio handshake, wire format, Opus +sender, and per-peer NetEQ jitter buffer used by Buzz desktop clients. It does +not depend on Tauri, the v0 desktop application, a relay server, identity stores, +or speech models. A consumer can select this package from a pinned `block/buzz` +Git revision without building the rest of the workspace. + +The host validates the destination and supplies an asynchronous NIP-42 signer +to `connection::connect`. The crate bounds connection and admission, validates +the initial roster, and returns a socket owned by the host. Dropping the pending +future or socket releases it; the crate creates no detached tasks. Hosts retain +room creation, membership checks, mute, device routing, connection supervision, +and application teardown. Audio admission alone does not grant chat access. + +`AudioEncoder` accepts up to 960 finite mono samples at 48 kHz and produces a +20 ms v2 packet. `PeerJitterBuffer` consumes the sequence/timestamp/Opus fields +from `wire::parse_relay_frame` and produces 480 samples per 10 ms playout tick. +Hosts maintain one receiver per peer, discard its state on leave or roster +reassignment, bound playback queues, and stop idle peers after buffered audio +drains. V2 has no occupancy epoch in media, so a delayed packet from a previous +occupant cannot be distinguished after index reuse. + +The default `native-roots` feature selects system TLS trust. Hosts that use +WebPKI roots select `default-features = false, features = ["webpki-roots"]`; +hosts can also pass their configured TLS connector. Signing and trust policy +remain host-owned. + +Opus 0.4 uses the maintained `opusic-sys` backend. NetEQ is pinned to 0.8.3: +its published source does not use its declared `ringbuf` dependency, which has +advisory RUSTSEC-2026-0293. The root security policy documents that unreachable +path and forbids other consumers of vulnerable ringbuf versions. Re-audit or +remove that exception when updating NetEQ; 0.9.1 still declares the old ringbuf. + +Run `cargo test -p buzz-audio-client` and +`cargo clippy -p buzz-audio-client --all-targets -- -D warnings` from the root +with Hermit active. Tests use real loopback WebSockets and Opus audio, without a +live community or microphone. Client integration still requires a native +two-person call, mute, leave, cancellation, and device-switching tryout. + +STT/TTS providers, model installation, and speech HTTP endpoints are intentionally +outside this transport crate. diff --git a/crates/buzz-audio-client/src/connection.rs b/crates/buzz-audio-client/src/connection.rs new file mode 100644 index 0000000000..193a32c5fe --- /dev/null +++ b/crates/buzz-audio-client/src/connection.rs @@ -0,0 +1,165 @@ +//! Bounded NIP-42 audio handshake. Signing and TLS policy belong to the host. + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use std::{collections::BTreeSet, future::Future, time::Duration}; +use tokio_tungstenite::{ + connect_async_tls_with_config, + tungstenite::{protocol::WebSocketConfig, Message}, + Connector, MaybeTlsStream, WebSocketStream, +}; + +/// Authenticated audio socket; dropping it closes the connection. +pub type AudioSocket = WebSocketStream>; + +/// Relay roster entry. Epoch is optional on released v2 relays. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Peer { + /// Relay-assigned media index. + pub index: u8, + /// Hex Nostr public key. + pub pubkey: String, + /// Control-plane occupancy epoch, zero on legacy relays. + pub epoch: u8, +} + +/// Successful audio admission, without granting message read/write authority. +pub struct Joined { + /// Host-owned connection, ready for media and control messages. + pub socket: AudioSocket, + /// Index assigned to this publisher. + pub index: u8, + /// Initial room roster. + pub peers: Vec, +} + +/// Parse a bounded roster without truncating out-of-range indices or epochs. +pub fn parse_peers(value: &Value) -> Result, String> { + let peers = value["peers"].as_array().ok_or("Invalid Huddle roster")?; + if peers.len() > 256 { + return Err("Huddle roster exceeds protocol capacity".into()); + } + let mut indices = BTreeSet::new(); + peers + .iter() + .map(|p| { + let index = p["peer_index"] + .as_u64() + .and_then(|i| u8::try_from(i).ok()) + .ok_or("Invalid Huddle peer index")?; + let pubkey = p["pubkey"] + .as_str() + .filter(|s| s.len() == 64 && s.bytes().all(|c| c.is_ascii_hexdigit())) + .ok_or("Invalid Huddle peer key")?; + let epoch = match p.get("epoch") { + None => 0, + Some(e) => e + .as_u64() + .and_then(|i| u8::try_from(i).ok()) + .ok_or("Invalid Huddle peer epoch")?, + }; + if !indices.insert(index) { + return Err("Duplicate Huddle peer".into()); + } + Ok(Peer { + index, + pubkey: pubkey.into(), + epoch, + }) + }) + .collect() +} + +async fn control(socket: &mut AudioSocket) -> Result { + loop { + match socket.next().await { + Some(Ok(Message::Text(text))) => { + let value: Value = + serde_json::from_str(&text).map_err(|_| "Invalid Huddle response")?; + if value["type"] == "error" { + return Err(value["message"] + .as_str() + .unwrap_or("Huddle rejected by relay") + .chars() + .take(240) + .collect()); + } + return Ok(value); + } + Some(Ok(Message::Ping(data))) => socket + .send(Message::Pong(data)) + .await + .map_err(|_| "Huddle disconnected")?, + Some(Ok(Message::Pong(_))) => {} + _ => return Err("Huddle disconnected during admission".into()), + } + } +} + +/// Connect and authenticate using a host-provided asynchronous signer. +/// +/// The host validates the destination before calling, and constructs the NIP-42 +/// event (including its exact relay tag) from the challenge. No private key is +/// accepted. Dropping this future cancels connection/admission; the entire +/// operation is bounded to 15 seconds, including signing. +pub async fn connect( + audio_url: &str, + parent: Option<&str>, + connector: Option, + sign: F, +) -> Result +where + F: FnOnce(String) -> Fut, + Fut: Future>, +{ + tokio::time::timeout(Duration::from_secs(15), async { + let config = WebSocketConfig::default() + .max_message_size(Some(65_536)) + .max_frame_size(Some(65_536)); + let (mut socket, _) = + connect_async_tls_with_config(audio_url, Some(config), false, connector) + .await + .map_err(|e| format!("audio WS connect failed: {e}"))?; + let challenge = tokio::time::timeout(Duration::from_secs(5), control(&mut socket)) + .await + .map_err(|_| "Huddle challenge timed out")??; + if challenge["type"] != "challenge" { + return Err("Invalid Huddle handshake".into()); + } + let challenge = challenge["challenge"] + .as_str() + .filter(|s| !s.is_empty() && s.len() <= 4096) + .ok_or("Invalid Huddle challenge")? + .to_owned(); + let event = sign(challenge).await?; + socket + .send(Message::Text( + json!({ + "type": "auth", "event": event, "parent_channel_id": parent, + "protocol_version": crate::wire::PROTOCOL_VERSION, + }) + .to_string() + .into(), + )) + .await + .map_err(|_| "Huddle authentication failed")?; + let joined = tokio::time::timeout(Duration::from_secs(5), control(&mut socket)) + .await + .map_err(|_| "Huddle admission timed out")??; + if joined["type"] != "joined" { + return Err("Huddle admission was not confirmed".into()); + } + let index = joined["peer_index"] + .as_u64() + .and_then(|i| u8::try_from(i).ok()) + .ok_or("Invalid Huddle publisher index")?; + let peers = parse_peers(&joined)?; + Ok(Joined { + socket, + index, + peers, + }) + }) + .await + .map_err(|_| "Huddle connection timed out")? +} diff --git a/crates/buzz-audio-client/src/encoder.rs b/crates/buzz-audio-client/src/encoder.rs new file mode 100644 index 0000000000..342187b07c --- /dev/null +++ b/crates/buzz-audio-client/src/encoder.rs @@ -0,0 +1,65 @@ +//! Stateful 48 kHz mono Opus sender shared by human and agent publishers. + +use crate::wire::{audio_level_dbov, FrameHeader, FLAG_DTX, V2_HEADER_LEN}; + +/// Samples in one 20 ms, 48 kHz mono frame. +pub const FRAME_SAMPLES: usize = 960; + +/// Opus encoder and v2 sequence/media clock for one connection. +pub struct AudioEncoder { + opus: opus::Encoder, + sequence: u16, + timestamp: u32, + output: Vec, +} + +impl AudioEncoder { + /// Create the standard 32 kbit/s VoIP encoder with discontinuous transmission. + pub fn new() -> Result { + let mut opus = opus::Encoder::new(48_000, opus::Channels::Mono, opus::Application::Voip) + .map_err(|e| format!("opus encoder: {e}"))?; + opus.set_bitrate(opus::Bitrate::Bits(32_000)) + .map_err(|e| format!("opus bitrate: {e}"))?; + opus.set_dtx(true).map_err(|e| format!("opus dtx: {e}"))?; + Ok(Self { + opus, + sequence: 0, + timestamp: 0, + output: vec![0; 4_000], + }) + } + + /// Encode up to 20 ms of finite mono PCM; pad the final short frame with silence. + /// + /// Returns the eight-byte v2 header followed by Opus. Empty/oversized or + /// non-finite input is rejected without advancing the media clock. + pub fn encode(&mut self, samples: &[f32]) -> Result, String> { + if samples.is_empty() + || samples.len() > FRAME_SAMPLES + || samples.iter().any(|s| !s.is_finite()) + { + return Err("Invalid Huddle audio frame".into()); + } + let mut padded = [0.0; FRAME_SAMPLES]; + padded[..samples.len()].copy_from_slice(samples); + let n = self + .opus + .encode_float(&padded, &mut self.output) + .map_err(|e| format!("opus encode: {e}"))?; + if n == 0 { + return Err("Opus produced an empty frame".into()); + } + let header = FrameHeader { + seq: self.sequence, + ts_48k: self.timestamp, + level_dbov: audio_level_dbov(samples), + flags: if n <= 2 { FLAG_DTX } else { 0 }, + }; + let mut frame = Vec::with_capacity(V2_HEADER_LEN + n); + frame.extend_from_slice(&header.encode()); + frame.extend_from_slice(&self.output[..n]); + self.sequence = self.sequence.wrapping_add(1); + self.timestamp = self.timestamp.wrapping_add(FRAME_SAMPLES as u32); + Ok(frame) + } +} diff --git a/crates/buzz-audio-client/src/jitter.rs b/crates/buzz-audio-client/src/jitter.rs new file mode 100644 index 0000000000..ff1c36077a --- /dev/null +++ b/crates/buzz-audio-client/src/jitter.rs @@ -0,0 +1,285 @@ +//! Per-peer NetEQ-style jitter buffer for huddle remote audio. +//! +//! Wraps the [`neteq`] crate's `NetEq` state machine and registers a thin +//! `AudioDecoder` impl over our `opus` dependency. Default +//! features on `neteq` are disabled so we don't pull in axum/cpal/clap; this +//! module owns the Opus side of the trait instead. +//! +//! ## Lifecycle +//! +//! One [`PeerJitterBuffer`] per active remote peer. It owns: +//! * a `NetEq` configured for 48 kHz mono with conservative jitter bounds, +//! * a stable synthetic RTP SSRC (the relay-assigned `peer_index`), and +//! * a 16-bit sequence number that wraps with the wire protocol's `seq`. +//! +//! On `joined` / first packet, construct one. On `left` / rejoin, drop and +//! recreate so NetEq's internal delay state doesn't carry over. +//! +//! ## Decode path +//! +//! Each inbound Opus frame becomes a `neteq::AudioPacket` with: +//! * sequence number from the protocol v2 header, +//! * 48 kHz RTP-style media timestamp (frame_index * 960 for 20 ms frames), +//! * a synthetic SSRC == peer_index, +//! * payload type [`OPUS_PAYLOAD_TYPE`] (any value, only NetEq uses it to +//! dispatch to its registered decoder), +//! * the raw Opus bytes as `payload`, +//! * `duration_ms = 20` (our encoder frame size). +//! +//! The playout side calls [`PeerJitterBuffer::get_audio`] on a 10 ms tick; +//! NetEq returns 10 ms of decoded PCM with adaptive-delay / expand / accelerate +//! decisions already applied. For 20 ms input frames at 48 kHz mono, that's +//! 480 f32 samples per call. + +use neteq::{ + codec::AudioDecoder, neteq::SpeechType, AudioPacket, NetEq, NetEqConfig, NetEqError, RtpHeader, +}; + +/// RTP payload type used internally. NetEq only needs it to look up the decoder +/// registered via `register_decoder`; the relay forwards Opus payloads opaquely +/// so the wire never carries a payload-type byte. +pub const OPUS_PAYLOAD_TYPE: u8 = 111; + +/// Audio sample rate for huddle (Opus VoIP, mono). +pub const SAMPLE_RATE_HZ: u32 = 48_000; +/// One channel (mono). +pub const CHANNELS: u8 = 1; +/// Opus encoder frame size in milliseconds. The wire format is one Opus packet +/// per 20 ms frame; the receiver still drains NetEq at 10 ms granularity. +pub const FRAME_DURATION_MS: u32 = 20; +/// 48 kHz media-time increment per encoder frame. +pub const FRAME_TIMESTAMP_DELTA: u32 = SAMPLE_RATE_HZ / 1000 * FRAME_DURATION_MS; // 960 +/// NetEq returns 10 ms frames; this is the sample count per `get_audio` call. +/// Kept as a documented constant for the protocol-v2 follow-up and consumers +/// that want to size their own buffers around the playout contract. +#[allow(dead_code)] +pub const PLAYOUT_SAMPLES: usize = (SAMPLE_RATE_HZ as usize / 1000) * 10; // 480 + +/// Minimum jitter delay NetEq is allowed to converge to (ms). +const MIN_DELAY_MS: u32 = 40; +/// Maximum jitter delay NetEq is allowed to converge to (ms). +const MAX_DELAY_MS: u32 = 200; +/// Buffer cap in packets — generous, but bounded so a runaway sender can't +/// drive memory growth. +const MAX_PACKETS_IN_BUFFER: usize = 50; + +/// Thin `AudioDecoder` over `opus::Decoder` so `neteq` can drive Opus payloads. +/// +/// NetEq's `decode` trait is sample-based, not FEC-aware (no `fec` bool). The +/// stock `neteq` `NativeOpusDecoder` (gated behind the `native` feature, which +/// pulls in cpal/axum) calls `decode_float(..., false)` unconditionally; this +/// matches that behavior so we don't need the feature. +/// +/// Receive-side FEC ("decode this frame's redundant copy on a known-lost prior +/// frame") would require either a trait change upstream or NetEq calling decode +/// twice with different `fec` values — out of scope for the initial 10-person +/// fix. Tracked as a follow-up alongside encoder-side `set_inband_fec`. +struct OpusFrameDecoder { + inner: opus::Decoder, + sample_rate: u32, + channels: u8, + /// Scratch buffer reused across calls to avoid per-frame allocation. + scratch: Vec, +} + +impl OpusFrameDecoder { + fn new(sample_rate: u32, channels: u8) -> Result { + let opus_channels = match channels { + 1 => opus::Channels::Mono, + 2 => opus::Channels::Stereo, + n => return Err(format!("unsupported channel count: {n}")), + }; + let inner = opus::Decoder::new(sample_rate, opus_channels) + .map_err(|e| format!("opus decoder init: {e}"))?; + // Opus can carry 120 ms even though our own sender uses 20 ms mono. + // Keep the decoder bounded while accepting other compatible senders. + let cap = (sample_rate as usize / 1000) * 120 * channels as usize; + Ok(Self { + inner, + sample_rate, + channels, + scratch: vec![0.0; cap], + }) + } +} + +impl AudioDecoder for OpusFrameDecoder { + fn sample_rate(&self) -> u32 { + self.sample_rate + } + + fn channels(&self) -> u8 { + self.channels + } + + fn decode(&mut self, encoded: &[u8]) -> Result, NetEqError> { + match self.inner.decode_float(encoded, &mut self.scratch, false) { + Ok(samples_per_channel) => { + let total = samples_per_channel * self.channels as usize; + Ok(self.scratch[..total].to_vec()) + } + Err(e) => Err(NetEqError::DecoderError(format!("opus decode: {e}"))), + } + } +} + +/// One peer's jitter buffer + decoder pair. +pub struct PeerJitterBuffer { + neteq: NetEq, + ssrc: u32, +} + +impl PeerJitterBuffer { + /// Construct a new jitter buffer for the peer identified by `peer_index`. + /// + /// `peer_index` is the relay-assigned 0..=254 stable index for the peer's + /// lifetime in the room; we promote it to a 32-bit synthetic SSRC so NetEq + /// has a stable stream identity for its delay-manager state. + pub fn new(peer_index: u8) -> Result { + let config = NetEqConfig { + sample_rate: SAMPLE_RATE_HZ, + channels: CHANNELS, + max_packets_in_buffer: MAX_PACKETS_IN_BUFFER, + min_delay_ms: MIN_DELAY_MS, + max_delay_ms: MAX_DELAY_MS, + ..Default::default() + }; + let mut neteq = NetEq::new(config)?; + let decoder = + OpusFrameDecoder::new(SAMPLE_RATE_HZ, CHANNELS).map_err(NetEqError::DecoderError)?; + neteq.register_decoder(OPUS_PAYLOAD_TYPE, Box::new(decoder)); + Ok(Self { + neteq, + ssrc: peer_index as u32, + }) + } + + /// Insert a received Opus packet. + /// + /// `seq` and `ts_48k` come from the protocol v2 header (sender-authored, + /// 16-bit wrapping sequence + 48 kHz RTP-style media time). + pub fn insert_packet( + &mut self, + seq: u16, + ts_48k: u32, + opus_payload: &[u8], + ) -> Result<(), NetEqError> { + let header = RtpHeader::new(seq, ts_48k, self.ssrc, OPUS_PAYLOAD_TYPE, false); + let packet = AudioPacket::new( + header, + opus_payload.to_vec(), + SAMPLE_RATE_HZ, + CHANNELS, + FRAME_DURATION_MS, + ); + self.neteq.insert_packet(packet) + } + + /// Drain one 10 ms playout frame. Always succeeds: NetEq emits PLC / + /// comfort-noise / silence rather than an error when there's nothing to + /// play. + /// + /// Returns `(samples, vad_active)`. `samples.len() == PLAYOUT_SAMPLES`. + pub fn get_audio(&mut self) -> Result<(Vec, bool), NetEqError> { + let frame = self.neteq.get_audio()?; + let vad = frame.vad_activity && !matches!(frame.speech_type, SpeechType::Expand); + Ok((frame.samples, vad)) + } + + /// True when NetEq's buffer has no packets queued and the next + /// `get_audio` call would emit expand/silence. Used by + /// `huddle::playout` as part of the idle-peer gate — if a peer has + /// gone quiet (`last_packet_at` is stale) *and* there's nothing + /// buffered to drain, we stop appending frames to that peer's rodio + /// Player. The check is conservative: a peer who just sent a burst + /// then disconnected may have buffered audio that should still play + /// out, so this method drives that decision. + pub fn is_empty(&self) -> bool { + self.neteq.is_empty() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn accepts_a_repacketized_120_ms_opus_packet() { + let mut encoder = + opus::Encoder::new(48_000, opus::Channels::Mono, opus::Application::Voip).unwrap(); + let mut packet = vec![0; 4000]; + let size = encoder.encode_float(&[0.0; 2880], &mut packet).unwrap(); + packet.truncate(size); + let mut combined = vec![0; 8000]; + let size = opus::Repacketizer::new() + .unwrap() + .combine(&[&packet, &packet], &mut combined) + .unwrap(); + let mut decoder = OpusFrameDecoder::new(48_000, 1).unwrap(); + let samples = decoder.decode(&combined[..size]).unwrap(); + assert_eq!(samples.len(), 5760); + assert!(samples.iter().all(|s| s.is_finite())); + } + + fn opus_silence_frame() -> Vec { + // Encode 20 ms of silence at 48 kHz mono so the test doesn't rely on + // synthetic Opus byte sequences. + let mut encoder = opus::Encoder::new( + SAMPLE_RATE_HZ, + opus::Channels::Mono, + opus::Application::Voip, + ) + .expect("opus encoder"); + let pcm = vec![0.0_f32; FRAME_TIMESTAMP_DELTA as usize]; + let mut out = vec![0u8; 4000]; + let n = encoder + .encode_float(&pcm, &mut out) + .expect("opus encode silence"); + out.truncate(n); + out + } + + #[test] + fn jitter_buffer_construction_succeeds_for_each_peer_index() { + for idx in [0u8, 1, 7, 42, 254] { + let jb = PeerJitterBuffer::new(idx).expect("construct jitter buffer"); + assert!(jb.is_empty(), "peer {idx} should start empty"); + // SSRC is the peer_index as u32 — verifies the stable-identity contract. + assert_eq!(jb.ssrc, idx as u32); + } + } + + #[test] + fn insert_packet_then_get_audio_returns_playout_frame() { + let mut jb = PeerJitterBuffer::new(3).expect("jitter buffer"); + let payload = opus_silence_frame(); + + // Insert a few sequential packets so NetEq has enough buffer to start + // emitting Normal-decoded frames instead of pure Expand/silence. + for i in 0..6u16 { + let ts = (i as u32) * FRAME_TIMESTAMP_DELTA; + jb.insert_packet(i, ts, &payload).expect("insert"); + } + + // Pull a playout frame — should be 10 ms = 480 mono samples. + let (samples, _vad) = jb.get_audio().expect("get_audio"); + assert_eq!( + samples.len(), + PLAYOUT_SAMPLES, + "NetEq must emit exactly one 10 ms frame per get_audio call", + ); + } + + #[test] + fn empty_buffer_get_audio_still_produces_a_frame() { + // Even with no packets inserted, NetEq's contract is to always emit + // 10 ms — as Expand/silence — so the playout clock keeps ticking. + let mut jb = PeerJitterBuffer::new(0).expect("jitter buffer"); + let (samples, vad) = jb.get_audio().expect("get_audio"); + assert_eq!(samples.len(), PLAYOUT_SAMPLES); + assert!( + !vad, + "empty buffer should not report voice activity (Expand frame)", + ); + } +} diff --git a/crates/buzz-audio-client/src/lib.rs b/crates/buzz-audio-client/src/lib.rs new file mode 100644 index 0000000000..14917b9500 --- /dev/null +++ b/crates/buzz-audio-client/src/lib.rs @@ -0,0 +1,11 @@ +//! Huddle audio primitives shared by desktop clients. +//! +//! The host retains identity, room admission, device routing, UI, and speech models. +//! This crate owns the v2 handshake, wire format, Opus encoding and receive jitter +//! buffering. It neither loads private keys nor starts background tasks. +#![deny(unsafe_code)] + +pub mod connection; +pub mod encoder; +pub mod jitter; +pub mod wire; diff --git a/crates/buzz-audio-client/src/wire.rs b/crates/buzz-audio-client/src/wire.rs new file mode 100644 index 0000000000..bcf9c007c2 --- /dev/null +++ b/crates/buzz-audio-client/src/wire.rs @@ -0,0 +1,346 @@ +//! Huddle audio wire protocol — versioning + frame header. +//! +//! ## v1 (legacy) +//! +//! Client → relay: `` +//! Relay → client: `` +//! +//! No per-frame metadata; receiver synthesizes sequence/timestamp on arrival. +//! Kept for backward compatibility — relay still admits v1 clients into +//! v1-pinned rooms — but new clients speak v2 while deployed relays remain +//! capped at the released v2 contract. +//! +//! ## v2 (compatibility contract) +//! +//! Client → relay: `` +//! Relay → client: `` +//! +//! Protocol v2 does not carry v3's occupancy epoch in media frames. The +//! control-plane roster still resets decoder and playout state when an index is +//! reassigned, but v2 cannot fence a delayed packet from the previous occupant +//! after that reassignment. +//! +//! Header layout (8 bytes, network byte order, big-endian): +//! +//! ```text +//! byte 0..=1 : seq u16 wrapping sequence number, +1 per packet +//! byte 2..=5 : ts_48k u32 sender RTP-style 48 kHz media time, +//! +960 per 20 ms encoded frame +//! byte 6 : level_dbov i8 audio level in dBov, range [-127, 0] +//! byte 7 : flags u8 bit 0 = DTX/comfort frame; other bits +//! MUST be ignored on decode +//! ``` +//! +//! Notes for reviewers (Max's spec): +//! * Header is fixed-size. No extension mechanism in v2 — future fields go +//! into a v3 with a new pinned version. +//! * `level_dbov` is client-authored telemetry. The relay parses it for +//! logging/active-speaker hints, clamps invalid values into range, and +//! **never** uses it for trust decisions (admission, moderation, etc.). +//! * Negotiation lives in the WS auth message (`protocol_version: 2`), not +//! in any bit of `flags`. Mixed-version rooms are rejected at the relay +//! with `upgrade_required`. + +/// Wire protocol version this client speaks. Bumped only when the frame +/// layout itself changes; the relay tracks pinned per-room. +pub const PROTOCOL_VERSION: u8 = 2; + +/// Length of the v2 per-frame header in bytes. +pub const V2_HEADER_LEN: usize = 8; + +/// Flag bit indicating a DTX / comfort-noise frame. Receivers MAY skip +/// arrival accounting for these; encoders MUST set it for any Opus packet +/// they tag as DTX. Unset on normal speech frames. +pub const FLAG_DTX: u8 = 0x01; + +/// Reserved bit-mask. Any bits outside this set MUST be ignored by the +/// receiver — they're available for the next minor protocol bump within v2. +#[allow(dead_code)] // Public constant for diagnostics / future flag additions. +pub const RESERVED_FLAG_MASK: u8 = !FLAG_DTX; + +/// Parsed view of one v2 header. Cheap (Copy). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FrameHeader { + /// Sender-authored sequence number; wraps every 2^16 frames (~21.8 min + /// at 50 Hz). + pub seq: u16, + /// Sender-authored 48 kHz RTP-style media timestamp. + pub ts_48k: u32, + /// Audio level in dBov. Always parsed into the canonical `-127..=0` + /// range — out-of-range inputs are clamped to `-127` (silence). + pub level_dbov: i8, + /// Raw flags byte. Use [`Self::is_dtx`] / explicit masks; bits outside + /// `FLAG_DTX | reserved` are not inspected. + pub flags: u8, +} + +impl FrameHeader { + /// Encode `self` into 8 bytes, network byte order. Suitable for direct + /// concatenation onto an Opus payload. + pub fn encode(self) -> [u8; V2_HEADER_LEN] { + let mut out = [0u8; V2_HEADER_LEN]; + out[0..2].copy_from_slice(&self.seq.to_be_bytes()); + out[2..6].copy_from_slice(&self.ts_48k.to_be_bytes()); + out[6] = self.level_dbov as u8; + out[7] = self.flags; + out + } + + /// Parse a v2 header from the leading 8 bytes of `bytes`. Returns + /// `(header, remainder)` so callers can pass the remainder straight to + /// the Opus decoder. + /// + /// `level_dbov` is clamped into `-127..=0`; values outside that range + /// are coerced to `-127`. Per Max's spec, malformed VU metadata must + /// *not* cause the audio frame to be dropped — only the metric is + /// suppressed. + pub fn parse(bytes: &[u8]) -> Option<(Self, &[u8])> { + if bytes.len() < V2_HEADER_LEN { + return None; + } + let seq = u16::from_be_bytes([bytes[0], bytes[1]]); + let ts_48k = u32::from_be_bytes([bytes[2], bytes[3], bytes[4], bytes[5]]); + let raw_level = bytes[6] as i8; + let level_dbov = if (-127..=0).contains(&raw_level) { + raw_level + } else { + // Out-of-range telemetry is coerced to "silent floor"; the + // audio packet is still delivered. This matches Max's "don't + // drop audio on bad VU" guidance. + -127 + }; + let flags = bytes[7]; + Some(( + Self { + seq, + ts_48k, + level_dbov, + flags, + }, + &bytes[V2_HEADER_LEN..], + )) + } + + /// True if [`FLAG_DTX`] is set. + #[allow(dead_code)] + pub fn is_dtx(&self) -> bool { + (self.flags & FLAG_DTX) != 0 + } +} + +/// Parse a complete relay-to-client v2 frame. +/// +/// The released v2 contract has exactly one relay-authored prefix byte: the +/// sender's peer index. A non-empty Opus payload must follow the fixed header. +pub fn parse_relay_frame(bytes: &[u8]) -> Option<(u8, FrameHeader, &[u8])> { + let (&peer_index, framed_audio) = bytes.split_first()?; + let (header, opus_payload) = FrameHeader::parse(framed_audio)?; + if opus_payload.is_empty() { + return None; + } + Some((peer_index, header, opus_payload)) +} + +/// Compute a dBov audio level for a normalized f32 PCM frame. +/// +/// "dBov" is RMS expressed in dB relative to full scale (where full scale = +/// peak amplitude 1.0), clamped into the v2 header's canonical `-127..=0` +/// range. A silent frame returns `-127`; full-scale ±1.0 returns `0`. +/// +/// This is cheap (one pass over the samples, one log) and runs once per +/// 20 ms frame, so it's nowhere near the audio fast-path budget. +pub fn audio_level_dbov(samples: &[f32]) -> i8 { + if samples.is_empty() { + return -127; + } + let mean_square: f64 = samples + .iter() + .map(|&s| { + let v = s as f64; + v * v + }) + .sum::() + / samples.len() as f64; + if mean_square <= 0.0 { + return -127; + } + let rms = mean_square.sqrt(); + let db = 20.0 * rms.log10(); + // Clamp into the canonical range. -127 = silence floor, 0 = full scale. + if !db.is_finite() || db <= -127.0 { + -127 + } else if db >= 0.0 { + 0 + } else { + db.round() as i8 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn round_trip_preserves_fields() { + let h = FrameHeader { + seq: 0xABCD, + ts_48k: 0x12_34_56_78, + level_dbov: -40, + flags: FLAG_DTX, + }; + let bytes = h.encode(); + let (parsed, tail) = FrameHeader::parse(&bytes).expect("parse"); + assert_eq!(parsed, h); + assert!(tail.is_empty()); + assert!(parsed.is_dtx()); + } + + /// Header is fixed 8 bytes; anything shorter is rejected. + #[test] + fn parse_rejects_short_input() { + for len in 0..V2_HEADER_LEN { + let buf = vec![0u8; len]; + assert!( + FrameHeader::parse(&buf).is_none(), + "{len} bytes must be rejected", + ); + } + } + + /// Trailing bytes pass through as the remainder, which is what the + /// receive path hands to the Opus decoder. + #[test] + fn parse_returns_payload_remainder() { + let mut buf = FrameHeader { + seq: 1, + ts_48k: 960, + level_dbov: -20, + flags: 0, + } + .encode() + .to_vec(); + buf.extend_from_slice(b"opus-bytes"); + let (_h, tail) = FrameHeader::parse(&buf).expect("parse"); + assert_eq!(tail, b"opus-bytes"); + } + + #[test] + fn relay_frame_uses_the_v2_one_byte_peer_prefix() { + let header = FrameHeader { + seq: 0x0102, + ts_48k: 960, + level_dbov: -20, + flags: 0, + }; + let mut frame = vec![7]; + frame.extend_from_slice(&header.encode()); + frame.extend_from_slice(b"opus"); + + let (peer_index, parsed_header, opus_payload) = + parse_relay_frame(&frame).expect("valid v2 relay frame"); + assert_eq!(peer_index, 7); + assert_eq!(parsed_header, header); + assert_eq!(opus_payload, b"opus"); + } + + #[test] + fn relay_frame_rejects_a_missing_opus_payload() { + let mut frame = vec![7]; + frame.extend_from_slice( + &FrameHeader { + seq: 1, + ts_48k: 960, + level_dbov: -20, + flags: 0, + } + .encode(), + ); + + assert!(parse_relay_frame(&frame).is_none()); + } + + /// Bytes in big-endian network order, matching Max's spec. This pins + /// the byte layout against accidental endianness changes. + #[test] + fn encoded_byte_order_is_network_byte_order() { + let h = FrameHeader { + seq: 0x0102, + ts_48k: 0x0304_0506, + level_dbov: -1, + flags: 0, + }; + let bytes = h.encode(); + assert_eq!(bytes[0..2], [0x01, 0x02]); + assert_eq!(bytes[2..6], [0x03, 0x04, 0x05, 0x06]); + assert_eq!(bytes[6], 0xFF); // -1 as i8 -> 0xFF + assert_eq!(bytes[7], 0x00); + } + + /// Per Max: malformed `level_dbov` must not drop the frame; only the + /// metric is suppressed. We coerce to -127 (silence floor) and keep + /// the payload available. + #[test] + fn out_of_range_level_dbov_is_clamped_not_rejected() { + // i8(127) is +127 which is outside [-127, 0]. + let mut bytes = FrameHeader { + seq: 7, + ts_48k: 7 * 960, + level_dbov: 0, + flags: 0, + } + .encode(); + bytes[6] = 0x7F; // raw +127 — out of range + let (parsed, _) = FrameHeader::parse(&bytes).expect("parse must succeed"); + assert_eq!(parsed.level_dbov, -127, "clamped to silence floor"); + assert_eq!(parsed.seq, 7, "other fields preserved"); + } + + #[test] + fn audio_level_dbov_clamps_silence_to_minus_127() { + assert_eq!(audio_level_dbov(&[]), -127); + assert_eq!(audio_level_dbov(&[0.0_f32; 960]), -127); + } + + #[test] + fn audio_level_dbov_full_scale_returns_zero() { + let frame: Vec = (0..960).map(|_| 1.0_f32).collect(); + assert_eq!(audio_level_dbov(&frame), 0); + } + + #[test] + fn audio_level_dbov_is_in_canonical_range_for_realistic_speech() { + // A 1 kHz sine at amplitude 0.3 — typical normalized speech peak. + // RMS ≈ 0.3 / sqrt(2) ≈ 0.212 → ≈ -13 dBov. + let frame: Vec = (0..960) + .map(|i| { + let t = i as f32 / 48_000.0; + 0.3 * (2.0 * std::f32::consts::PI * 1_000.0 * t).sin() + }) + .collect(); + let db = audio_level_dbov(&frame); + assert!( + (-20..=-8).contains(&db), + "expected -20..=-8 dBov for 0.3-amp sine, got {db}", + ); + } + + /// Reserved flag bits MUST be ignored on parse (no error, just preserved + /// for forward-compat). The contract is that *receivers* don't inspect + /// them; the parse path keeps them so introspection still works. + #[test] + fn reserved_flag_bits_are_preserved_for_inspection() { + let mut bytes = FrameHeader { + seq: 0, + ts_48k: 0, + level_dbov: 0, + flags: 0, + } + .encode(); + bytes[7] = 0b1010_1010; // FLAG_DTX bit clear, reserved bits set + let (parsed, _) = FrameHeader::parse(&bytes).expect("parse"); + assert_eq!(parsed.flags, 0b1010_1010); + assert!(!parsed.is_dtx()); + // Caller can still check which reserved bits were set if needed. + assert_eq!(parsed.flags & RESERVED_FLAG_MASK, 0b1010_1010); + } +} diff --git a/crates/buzz-audio-client/tests/codec.rs b/crates/buzz-audio-client/tests/codec.rs new file mode 100644 index 0000000000..5f15e39699 --- /dev/null +++ b/crates/buzz-audio-client/tests/codec.rs @@ -0,0 +1,56 @@ +use buzz_audio_client::{ + encoder::{AudioEncoder, FRAME_SAMPLES}, + jitter::PeerJitterBuffer, + wire::{parse_relay_frame, FrameHeader, V2_HEADER_LEN}, +}; + +#[test] +fn encoded_media_plays_through_the_shared_receiver() { + let mut encoder = AudioEncoder::new().unwrap(); + let mut receiver = PeerJitterBuffer::new(7).unwrap(); + let samples: Vec = (0..FRAME_SAMPLES) + .map(|n| (n as f32 * 0.1).sin() * 0.1) + .collect(); + for seq in 0..8 { + let payload = encoder.encode(&samples).unwrap(); + let frame = [vec![7], payload].concat(); + let (peer, header, opus) = parse_relay_frame(&frame).unwrap(); + assert_eq!(peer, 7); + assert_eq!(header.seq, seq); + assert_eq!(header.ts_48k, u32::from(seq) * 960); + receiver + .insert_packet(header.seq, header.ts_48k, opus) + .unwrap(); + } + let mut energy = 0.0_f32; + for _ in 0..20 { + let (pcm, _) = receiver.get_audio().unwrap(); + assert_eq!(pcm.len(), 480); + assert!(pcm.iter().all(|s| s.is_finite())); + energy += pcm.iter().map(|s| s * s).sum::(); + } + assert!( + energy > 0.01, + "voice must survive actual Opus and jitter decoding" + ); +} + +#[test] +fn invalid_input_does_not_advance_clock_and_short_frames_are_padded() { + let mut encoder = AudioEncoder::new().unwrap(); + for samples in [vec![], vec![0.; 961], vec![f32::NAN], vec![f32::INFINITY]] { + assert!(encoder.encode(&samples).is_err()); + } + let encoded = encoder.encode(&[0.1; 120]).unwrap(); + let (header, _) = FrameHeader::parse(&encoded).unwrap(); + assert_eq!(header.seq, 0); + assert_eq!(header.ts_48k, 0); + let mut decoder = opus::Decoder::new(48000, opus::Channels::Mono).unwrap(); + let mut pcm = [0.0; 5760]; + assert_eq!( + decoder + .decode_float(&encoded[V2_HEADER_LEN..], &mut pcm, false) + .unwrap(), + 960 + ); +} diff --git a/crates/buzz-audio-client/tests/connection.rs b/crates/buzz-audio-client/tests/connection.rs new file mode 100644 index 0000000000..c67c24b764 --- /dev/null +++ b/crates/buzz-audio-client/tests/connection.rs @@ -0,0 +1,133 @@ +use buzz_audio_client::connection::{connect, parse_peers}; +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use std::time::Duration; +use tokio::{net::TcpListener, sync::oneshot}; +use tokio_tungstenite::{accept_async, tungstenite::Message}; + +#[tokio::test] +async fn authenticates_with_host_signer_and_preserves_v2_media() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}/huddle/room/audio", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut socket = accept_async(stream).await.unwrap(); + socket.send(Message::Ping(vec![7].into())).await.unwrap(); + socket + .send(Message::Text( + json!({"type":"challenge","challenge":"host-challenge"}) + .to_string() + .into(), + )) + .await + .unwrap(); + let auth = loop { + match socket.next().await.unwrap().unwrap() { + Message::Text(text) => break serde_json::from_str::(&text).unwrap(), + Message::Pong(bytes) => assert_eq!(bytes.as_ref(), &[7]), + other => panic!("unexpected {other:?}"), + } + }; + assert_eq!( + auth, + json!({"type":"auth","event":{"signed":"host-challenge"},"parent_channel_id":"parent","protocol_version":2}) + ); + socket.send(Message::Text(json!({"type":"joined","peer_index":0,"peers":[{"peer_index":1,"pubkey":"a".repeat(64)}]}).to_string().into())).await.unwrap(); + socket + .send(Message::Binary(vec![1, 2, 3].into())) + .await + .unwrap(); + assert_eq!( + socket.next().await.unwrap().unwrap(), + Message::Binary(vec![4, 5, 6].into()) + ); + }); + let mut joined = connect(&url, Some("parent"), None, |challenge| async move { + assert_eq!(challenge, "host-challenge"); + Ok(json!({"signed":challenge})) + }) + .await + .unwrap(); + assert_eq!(joined.index, 0); + assert_eq!(joined.peers[0].epoch, 0); + assert_eq!(joined.peers[0].pubkey, "a".repeat(64)); + assert_eq!( + joined.socket.next().await.unwrap().unwrap(), + Message::Binary(vec![1, 2, 3].into()) + ); + joined + .socket + .send(Message::Binary(vec![4, 5, 6].into())) + .await + .unwrap(); + server.await.unwrap(); +} + +#[tokio::test] +async fn invalid_challenge_never_reaches_signer() { + for challenge in [ + json!({"type":"joined","challenge":"wrong"}), + json!({"type":"challenge","challenge":""}), + json!({"type":"challenge","challenge":"x".repeat(4097)}), + ] { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut socket = accept_async(stream).await.unwrap(); + socket + .send(Message::Text(challenge.to_string().into())) + .await + .unwrap(); + assert!(socket + .next() + .await + .is_none_or(|r| r.is_err() || matches!(r, Ok(Message::Close(_))))); + }); + assert!(connect(&url, None, None, |_| async { + panic!("must not sign an invalid challenge") + }) + .await + .is_err()); + server.await.unwrap(); + } +} + +#[tokio::test] +async fn cancelling_admission_drops_the_socket() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let (ready, received) = oneshot::channel(); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut socket = accept_async(stream).await.unwrap(); + ready.send(()).unwrap(); + let result = tokio::time::timeout(Duration::from_secs(2), socket.next()) + .await + .unwrap(); + assert!(result.is_none_or(|r| r.is_err() || matches!(r, Ok(Message::Close(_))))); + }); + let client = tokio::spawn(async move { + connect(&url, None, None, |_| async { panic!("no challenge") }).await + }); + received.await.unwrap(); + client.abort(); + assert!(matches!(client.await, Err(e) if e.is_cancelled())); + server.await.unwrap(); +} + +#[test] +fn rejects_invalid_rosters_without_truncation() { + for peers in [ + json!([{"peer_index":256,"pubkey":"a".repeat(64)}]), + json!([{"peer_index":1,"pubkey":"bad"}]), + json!([{"peer_index":1,"pubkey":"a".repeat(64),"epoch":256}]), + json!([{"peer_index":1,"pubkey":"a".repeat(64)},{"peer_index":1,"pubkey":"b".repeat(64)}]), + ] { + assert!(parse_peers(&json!({"peers":peers})).is_err()); + } + assert!(parse_peers( + &json!({"peers":vec![json!({"peer_index":1,"pubkey":"a".repeat(64)});257]}) + ) + .is_err()); +} diff --git a/deny.toml b/deny.toml index 74e83fb351..00b4f66ae4 100644 --- a/deny.toml +++ b/deny.toml @@ -20,6 +20,12 @@ ignore = [ # fork uses the same crate, so removing it means migrating YAML parsing in # buzz-agent, buzz-persona, buzz-relay, buzz-workflow and desktop. { id = "RUSTSEC-2024-0449", reason = "transitive via serde_yaml; unmaintained, not vulnerable; removal needs a YAML parser migration" }, + # NetEQ =0.8.3 declares ringbuf but its published Rust sources never use it, + # so Consumer::skip/clear and their panic-unsafe Drop path are unreachable. + # NetEQ 0.9.1 still declares ringbuf 0.3; ringbuf itself is fixed in 0.5.2. + # Keep the exact NetEQ pin and the wrapper ban below together. Re-audit or + # remove this exception when NetEQ removes/updates the unused dependency. + { id = "RUSTSEC-2026-0293", reason = "unused dependency of audited, exactly pinned neteq 0.8.3; no Rust references to ringbuf; other consumers forbidden by bans.deny" }, ] [licenses] @@ -95,3 +101,6 @@ ignore = true [bans] multiple-versions = "warn" wildcards = "allow" +deny = [ + { name = "ringbuf", version = "<0.5.2", wrappers = ["neteq"], reason = "only audited NetEQ may declare this unused vulnerable dependency; all other consumers must use a patched release" }, +] diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index cab1aa500e..7ba160f8c8 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -598,17 +598,6 @@ dependencies = [ "num-traits", ] -[[package]] -name = "audiopus_sys" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62314a1546a2064e033665d658e88c620a62904be945f8147e6b16c3db9f8651" -dependencies = [ - "cmake", - "log", - "pkg-config", -] - [[package]] name = "autocfg" version = "1.5.1" @@ -1095,6 +1084,18 @@ dependencies = [ "webbrowser", ] +[[package]] +name = "buzz-audio-client" +version = "0.1.0" +dependencies = [ + "futures-util", + "neteq", + "opus", + "serde_json", + "tokio", + "tokio-tungstenite 0.29.0", +] + [[package]] name = "buzz-core" version = "0.1.0" @@ -1128,6 +1129,7 @@ dependencies = [ "base64 0.22.1", "block2", "buzz-agent", + "buzz-audio-client", "buzz-core", "buzz-media", "buzz-persona", @@ -1157,14 +1159,12 @@ dependencies = [ "mesh-llm-node", "mesh-llm-sdk", "mesh-llm-system", - "neteq", "nostr 0.44.7", "notify-rust", "objc2", "objc2-app-kit", "objc2-foundation", "objc2-user-notifications", - "opus", "percent-encoding", "plist", "png 0.18.1", @@ -2387,7 +2387,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.118", ] [[package]] @@ -7174,11 +7174,20 @@ checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" [[package]] name = "opus" -version = "0.3.1" +version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d3809943dff6fbad5f0484449ea26bdb9cb7d8efdf26ed50d3c7f227f69eb5c" +checksum = "33718946cc77d4032911d4efe03a66dbcbfbd2bb16c3da06aaeadcc637c32216" dependencies = [ - "audiopus_sys", + "opusic-sys", +] + +[[package]] +name = "opusic-sys" +version = "0.7.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9d1ecdf206421bc74343ab3bb2f30ad2abbfee41fa341f7181fecbaf957769a" +dependencies = [ + "cmake", ] [[package]] @@ -11114,7 +11123,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", diff --git a/desktop/src-tauri/Cargo.toml b/desktop/src-tauri/Cargo.toml index b7b31117f7..fdff759d0f 100644 --- a/desktop/src-tauri/Cargo.toml +++ b/desktop/src-tauri/Cargo.toml @@ -93,8 +93,6 @@ tokio-tungstenite = { version = "0.29", features = ["rustls-tls-webpki-roots"] } tokio-util = { version = "0.7", features = ["rt"] } bytes = "1" futures-util = "0.3" -opus = "0.3" -neteq = { version = "0.8", default-features = false } serde = { version = "1", features = ["derive"] } serde_json = "1" serde_yaml = "0.9" @@ -112,6 +110,7 @@ buzz_core_pkg = { package = "buzz-core", path = "../../crates/buzz-core" } buzz_persona_pkg = { package = "buzz-persona", path = "../../crates/buzz-persona" } buzz_sdk_pkg = { package = "buzz-sdk", path = "../../crates/buzz-sdk" } buzz_agent_pkg = { package = "buzz-agent", path = "../../crates/buzz-agent" } +buzz_audio_client = { package = "buzz-audio-client", path = "../../crates/buzz-audio-client", default-features = false, features = ["webpki-roots"] } buzz_voice_pkg = { package = "buzz-voice", path = "../../crates/buzz-voice" } buzz_terminal = { package = "buzz-terminal", path = "crates/buzz-terminal" } buzz_ws_client_pkg = { package = "buzz-ws-client", path = "../../crates/buzz-ws-client" } diff --git a/desktop/src-tauri/src/huddle/jitter.rs b/desktop/src-tauri/src/huddle/jitter.rs index ba10a5b4fc..44302db1c2 100644 --- a/desktop/src-tauri/src/huddle/jitter.rs +++ b/desktop/src-tauri/src/huddle/jitter.rs @@ -1,267 +1,2 @@ -//! Per-peer NetEQ-style jitter buffer for huddle remote audio. -//! -//! Wraps the [`neteq`] crate's `NetEq` state machine and registers a thin -//! `AudioDecoder` impl over our existing `opus = "0.3"` dependency. Default -//! features on `neteq` are disabled so we don't pull in axum/cpal/clap; this -//! module owns the Opus side of the trait instead. -//! -//! ## Lifecycle -//! -//! One [`PeerJitterBuffer`] per active remote peer. It owns: -//! * a `NetEq` configured for 48 kHz mono with conservative jitter bounds, -//! * a stable synthetic RTP SSRC (the relay-assigned `peer_index`), and -//! * a 16-bit sequence number that wraps with the wire protocol's `seq`. -//! -//! On `joined` / first packet, construct one. On `left` / rejoin, drop and -//! recreate so NetEq's internal delay state doesn't carry over. -//! -//! ## Decode path -//! -//! Each inbound Opus frame becomes a `neteq::AudioPacket` with: -//! * sequence number from the protocol v2 header, -//! * 48 kHz RTP-style media timestamp (frame_index * 960 for 20 ms frames), -//! * a synthetic SSRC == peer_index, -//! * payload type [`OPUS_PAYLOAD_TYPE`] (any value, only NetEq uses it to -//! dispatch to its registered decoder), -//! * the raw Opus bytes as `payload`, -//! * `duration_ms = 20` (our encoder frame size). -//! -//! The playout side calls [`PeerJitterBuffer::get_audio`] on a 10 ms tick; -//! NetEq returns 10 ms of decoded PCM with adaptive-delay / expand / accelerate -//! decisions already applied. For 20 ms input frames at 48 kHz mono, that's -//! 480 f32 samples per call. - -use neteq::{ - codec::AudioDecoder, neteq::SpeechType, AudioPacket, NetEq, NetEqConfig, NetEqError, RtpHeader, -}; - -/// RTP payload type used internally. NetEq only needs it to look up the decoder -/// registered via `register_decoder`; the relay forwards Opus payloads opaquely -/// so the wire never carries a payload-type byte. -pub const OPUS_PAYLOAD_TYPE: u8 = 111; - -/// Audio sample rate for huddle (Opus VoIP, mono). -pub const SAMPLE_RATE_HZ: u32 = 48_000; -/// One channel (mono). -pub const CHANNELS: u8 = 1; -/// Opus encoder frame size in milliseconds. The wire format is one Opus packet -/// per 20 ms frame; the receiver still drains NetEq at 10 ms granularity. -pub const FRAME_DURATION_MS: u32 = 20; -/// 48 kHz media-time increment per encoder frame. -pub const FRAME_TIMESTAMP_DELTA: u32 = SAMPLE_RATE_HZ / 1000 * FRAME_DURATION_MS; // 960 -/// NetEq returns 10 ms frames; this is the sample count per `get_audio` call. -/// Kept as a documented constant for the protocol-v2 follow-up and consumers -/// that want to size their own buffers around the playout contract. -#[allow(dead_code)] -pub const PLAYOUT_SAMPLES: usize = (SAMPLE_RATE_HZ as usize / 1000) * 10; // 480 - -/// Minimum jitter delay NetEq is allowed to converge to (ms). -const MIN_DELAY_MS: u32 = 40; -/// Maximum jitter delay NetEq is allowed to converge to (ms). -const MAX_DELAY_MS: u32 = 200; -/// Buffer cap in packets — generous, but bounded so a runaway sender can't -/// drive memory growth. -const MAX_PACKETS_IN_BUFFER: usize = 50; - -/// Thin `AudioDecoder` over `opus::Decoder` so `neteq` can drive Opus payloads. -/// -/// NetEq's `decode` trait is sample-based, not FEC-aware (no `fec` bool). The -/// stock `neteq` `NativeOpusDecoder` (gated behind the `native` feature, which -/// pulls in cpal/axum) calls `decode_float(..., false)` unconditionally; this -/// matches that behavior so we don't need the feature. -/// -/// Receive-side FEC ("decode this frame's redundant copy on a known-lost prior -/// frame") would require either a trait change upstream or NetEq calling decode -/// twice with different `fec` values — out of scope for the initial 10-person -/// fix. Tracked as a follow-up alongside encoder-side `set_inband_fec`. -struct OpusFrameDecoder { - inner: opus::Decoder, - sample_rate: u32, - channels: u8, - /// Scratch buffer reused across calls to avoid per-frame allocation. - scratch: Vec, -} - -impl OpusFrameDecoder { - fn new(sample_rate: u32, channels: u8) -> Result { - let opus_channels = match channels { - 1 => opus::Channels::Mono, - 2 => opus::Channels::Stereo, - n => return Err(format!("unsupported channel count: {n}")), - }; - let inner = opus::Decoder::new(sample_rate, opus_channels) - .map_err(|e| format!("opus decoder init: {e}"))?; - // Largest plausible frame: 60 ms at 48 kHz stereo. We always use - // 20 ms mono in practice; this is just a one-shot allocation cap. - let cap = (sample_rate as usize / 1000) * 60 * channels as usize; - Ok(Self { - inner, - sample_rate, - channels, - scratch: vec![0.0; cap], - }) - } -} - -impl AudioDecoder for OpusFrameDecoder { - fn sample_rate(&self) -> u32 { - self.sample_rate - } - - fn channels(&self) -> u8 { - self.channels - } - - fn decode(&mut self, encoded: &[u8]) -> Result, NetEqError> { - match self.inner.decode_float(encoded, &mut self.scratch, false) { - Ok(samples_per_channel) => { - let total = samples_per_channel * self.channels as usize; - Ok(self.scratch[..total].to_vec()) - } - Err(e) => Err(NetEqError::DecoderError(format!("opus decode: {e}"))), - } - } -} - -/// One peer's jitter buffer + decoder pair. -pub struct PeerJitterBuffer { - neteq: NetEq, - ssrc: u32, -} - -impl PeerJitterBuffer { - /// Construct a new jitter buffer for the peer identified by `peer_index`. - /// - /// `peer_index` is the relay-assigned 0..=254 stable index for the peer's - /// lifetime in the room; we promote it to a 32-bit synthetic SSRC so NetEq - /// has a stable stream identity for its delay-manager state. - pub fn new(peer_index: u8) -> Result { - let config = NetEqConfig { - sample_rate: SAMPLE_RATE_HZ, - channels: CHANNELS, - max_packets_in_buffer: MAX_PACKETS_IN_BUFFER, - min_delay_ms: MIN_DELAY_MS, - max_delay_ms: MAX_DELAY_MS, - ..Default::default() - }; - let mut neteq = NetEq::new(config)?; - let decoder = - OpusFrameDecoder::new(SAMPLE_RATE_HZ, CHANNELS).map_err(NetEqError::DecoderError)?; - neteq.register_decoder(OPUS_PAYLOAD_TYPE, Box::new(decoder)); - Ok(Self { - neteq, - ssrc: peer_index as u32, - }) - } - - /// Insert a received Opus packet. - /// - /// `seq` and `ts_48k` come from the protocol v2 header (sender-authored, - /// 16-bit wrapping sequence + 48 kHz RTP-style media time). - pub fn insert_packet( - &mut self, - seq: u16, - ts_48k: u32, - opus_payload: &[u8], - ) -> Result<(), NetEqError> { - let header = RtpHeader::new(seq, ts_48k, self.ssrc, OPUS_PAYLOAD_TYPE, false); - let packet = AudioPacket::new( - header, - opus_payload.to_vec(), - SAMPLE_RATE_HZ, - CHANNELS, - FRAME_DURATION_MS, - ); - self.neteq.insert_packet(packet) - } - - /// Drain one 10 ms playout frame. Always succeeds: NetEq emits PLC / - /// comfort-noise / silence rather than an error when there's nothing to - /// play. - /// - /// Returns `(samples, vad_active)`. `samples.len() == PLAYOUT_SAMPLES`. - pub fn get_audio(&mut self) -> Result<(Vec, bool), NetEqError> { - let frame = self.neteq.get_audio()?; - let vad = frame.vad_activity && !matches!(frame.speech_type, SpeechType::Expand); - Ok((frame.samples, vad)) - } - - /// True when NetEq's buffer has no packets queued and the next - /// `get_audio` call would emit expand/silence. Used by - /// `huddle::playout` as part of the idle-peer gate — if a peer has - /// gone quiet (`last_packet_at` is stale) *and* there's nothing - /// buffered to drain, we stop appending frames to that peer's rodio - /// Player. The check is conservative: a peer who just sent a burst - /// then disconnected may have buffered audio that should still play - /// out, so this method drives that decision. - pub fn is_empty(&self) -> bool { - self.neteq.is_empty() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn opus_silence_frame() -> Vec { - // Encode 20 ms of silence at 48 kHz mono so the test doesn't rely on - // synthetic Opus byte sequences. - let mut encoder = opus::Encoder::new( - SAMPLE_RATE_HZ, - opus::Channels::Mono, - opus::Application::Voip, - ) - .expect("opus encoder"); - let pcm = vec![0.0_f32; FRAME_TIMESTAMP_DELTA as usize]; - let mut out = vec![0u8; 4000]; - let n = encoder - .encode_float(&pcm, &mut out) - .expect("opus encode silence"); - out.truncate(n); - out - } - - #[test] - fn jitter_buffer_construction_succeeds_for_each_peer_index() { - for idx in [0u8, 1, 7, 42, 254] { - let jb = PeerJitterBuffer::new(idx).expect("construct jitter buffer"); - assert!(jb.is_empty(), "peer {idx} should start empty"); - // SSRC is the peer_index as u32 — verifies the stable-identity contract. - assert_eq!(jb.ssrc, idx as u32); - } - } - - #[test] - fn insert_packet_then_get_audio_returns_playout_frame() { - let mut jb = PeerJitterBuffer::new(3).expect("jitter buffer"); - let payload = opus_silence_frame(); - - // Insert a few sequential packets so NetEq has enough buffer to start - // emitting Normal-decoded frames instead of pure Expand/silence. - for i in 0..6u16 { - let ts = (i as u32) * FRAME_TIMESTAMP_DELTA; - jb.insert_packet(i, ts, &payload).expect("insert"); - } - - // Pull a playout frame — should be 10 ms = 480 mono samples. - let (samples, _vad) = jb.get_audio().expect("get_audio"); - assert_eq!( - samples.len(), - PLAYOUT_SAMPLES, - "NetEq must emit exactly one 10 ms frame per get_audio call", - ); - } - - #[test] - fn empty_buffer_get_audio_still_produces_a_frame() { - // Even with no packets inserted, NetEq's contract is to always emit - // 10 ms — as Expand/silence — so the playout clock keeps ticking. - let mut jb = PeerJitterBuffer::new(0).expect("jitter buffer"); - let (samples, vad) = jb.get_audio().expect("get_audio"); - assert_eq!(samples.len(), PLAYOUT_SAMPLES); - assert!( - !vad, - "empty buffer should not report voice activity (Expand frame)", - ); - } -} +//! Shared Huddle jitter implementation. +pub use buzz_audio_client::jitter::*; diff --git a/desktop/src-tauri/src/huddle/relay_api.rs b/desktop/src-tauri/src/huddle/relay_api.rs index 190397aa05..53c757b986 100644 --- a/desktop/src-tauri/src/huddle/relay_api.rs +++ b/desktop/src-tauri/src/huddle/relay_api.rs @@ -10,9 +10,10 @@ //! → recv loop: WS binary frame → Opus decode (per-peer) → rodio playback //! ``` +use buzz_audio_client::encoder::{AudioEncoder, FRAME_SAMPLES}; use futures_util::{SinkExt, StreamExt}; use std::sync::{atomic::AtomicBool, Arc}; -use tokio_tungstenite::{connect_async, tungstenite::Message as WsMsg}; +use tokio_tungstenite::tungstenite::Message as WsMsg; use tokio_util::sync::CancellationToken; use uuid::Uuid; @@ -38,9 +39,6 @@ pub(crate) fn parse_channel_uuid(channel_id: &str) -> Result { Uuid::parse_str(channel_id).map_err(|_| format!("invalid channel UUID: {channel_id}")) } -/// Handshake timeout — matches the server's AUTH_TIMEOUT (5 s). -const HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); - fn build_audio_auth_event( keys: &nostr::Keys, relay_url: &str, @@ -79,99 +77,24 @@ async fn connect_authenticated_audio_socket( use nostr::JsonUtil; let ws_url = format!("{relay_url}/huddle/{channel_id}/audio"); - let (ws_stream, _) = connect_async(&ws_url) - .await - .map_err(|e| format!("audio WS connect failed: {e}"))?; - let (mut ws_tx, mut ws_rx) = ws_stream.split(); - - let challenge = tokio::time::timeout(HANDSHAKE_TIMEOUT, async { - loop { - match ws_rx.next().await { - Some(Ok(WsMsg::Text(text))) => { - let value: serde_json::Value = serde_json::from_str(&text) - .map_err(|e| format!("bad challenge JSON: {e}"))?; - if value["type"] == "challenge" { - break value["challenge"] - .as_str() - .ok_or_else(|| "missing challenge string".to_string()) - .map(str::to_string); - } - } - Some(Ok(WsMsg::Close(_))) | None => { - break Err("connection closed before challenge".into()); - } - _ => continue, - } - } - }) - .await - .map_err(|_| "timeout waiting for challenge from relay".to_string())??; - - let event = build_audio_auth_event(keys, relay_url, &challenge, auth_tag_json)?; - let event_json: serde_json::Value = serde_json::from_str(&event.as_json()) - .map_err(|e| format!("failed to serialize auth event: {e}"))?; - let auth_msg = serde_json::json!({ - "type": "auth", - "event": event_json, - "parent_channel_id": parent_channel_id, - // Use the released v2 contract while deployed relays remain capped at - // v2. Relay-to-client media therefore has a one-byte peer-index prefix; - // see huddle::wire for the compatibility tradeoff. - "protocol_version": super::wire::PROTOCOL_VERSION, - }); - ws_tx - .send(WsMsg::Text(auth_msg.to_string().into())) - .await - .map_err(|e| format!("send auth: {e}"))?; - - let (peer_index, initial_peers) = tokio::time::timeout(HANDSHAKE_TIMEOUT, async { - loop { - match ws_rx.next().await { - Some(Ok(WsMsg::Text(text))) => { - let value: serde_json::Value = serde_json::from_str(&text).unwrap_or_default(); - match value["type"].as_str() { - Some("joined") => { - let peers = value["peers"] - .as_array() - .map(|peers| { - peers - .iter() - .filter_map(|peer| { - Some(( - peer["peer_index"].as_u64()? as u8, - peer["pubkey"].as_str()?.to_string(), - // Absent `epoch` (legacy relay) degrades - // to 0 so the fence becomes a no-op rather - // than rejecting every frame. - peer["epoch"].as_u64().unwrap_or(0) as u8, - )) - }) - .collect() - }) - .unwrap_or_default(); - let peer_index = value["peer_index"] - .as_u64() - .and_then(|index| u8::try_from(index).ok()) - .ok_or_else(|| "joined message missing peer index".to_string())?; - break Ok((peer_index, peers)); - } - Some("error") => { - break Err(format!("audio relay auth error: {}", value["message"])); - } - _ => continue, - } - } - Some(Ok(WsMsg::Close(_))) | None => { - break Err("connection closed before joined".into()); - } - _ => continue, - } - } - }) - .await - .map_err(|_| "timeout waiting for joined from relay".to_string())??; - - Ok((ws_tx, ws_rx, peer_index, initial_peers)) + let joined = buzz_audio_client::connection::connect( + &ws_url, + parent_channel_id, + None, + |challenge| async move { + let event = build_audio_auth_event(keys, relay_url, &challenge, auth_tag_json)?; + serde_json::from_str(&event.as_json()) + .map_err(|e| format!("failed to serialize auth event: {e}")) + }, + ) + .await?; + let (tx, rx) = joined.socket.split(); + let peers = joined + .peers + .into_iter() + .map(|p| (p.index, p.pubkey, p.epoch)) + .collect(); + Ok((tx, rx, joined.index, peers)) } /// Connect to the relay's audio WebSocket and run the Opus encode/decode pipeline. @@ -364,21 +287,9 @@ async fn run_tts_audio_publisher( epoch: Arc, speaker_generation: Arc, ) -> Result<(), String> { - use super::wire::{audio_level_dbov, FrameHeader, V2_HEADER_LEN}; use std::sync::atomic::Ordering; - let mut encoder = opus::Encoder::new(48_000, opus::Channels::Mono, opus::Application::Voip) - .map_err(|error| format!("tts opus encoder: {error}"))?; - encoder - .set_bitrate(opus::Bitrate::Bits(32_000)) - .map_err(|error| format!("tts opus bitrate: {error}"))?; - encoder - .set_dtx(true) - .map_err(|error| format!("tts opus dtx: {error}"))?; - - let mut sequence = 0_u16; - let mut timestamp_48k = 0_u32; - let mut encoded = vec![0_u8; 4_000]; + let mut encoder = AudioEncoder::new()?; let mut queue = std::collections::VecDeque::::new(); let mut send_tick = tokio::time::interval(std::time::Duration::from_millis(20)); send_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); @@ -397,30 +308,11 @@ async fn run_tts_audio_publisher( queue.pop_front(); } let Some(frame) = queue.pop_front() else { continue }; - let level = audio_level_dbov(&frame.samples_48k); - let encoded_len = encoder - .encode_float(&frame.samples_48k, &mut encoded) - .map_err(|error| format!("tts opus encode: {error}"))?; - if encoded_len == 0 { - continue; - } - let flags = if encoded_len <= 2 { super::wire::FLAG_DTX } else { 0 }; - let header = FrameHeader { - seq: sequence, - ts_48k: timestamp_48k, - level_dbov: level, - flags, - } - .encode(); - let mut payload = Vec::with_capacity(V2_HEADER_LEN + encoded_len); - payload.extend_from_slice(&header); - payload.extend_from_slice(&encoded[..encoded_len]); + let payload = encoder.encode(&frame.samples_48k)?; ws_tx .send(WsMsg::Binary(payload.into())) .await .map_err(|error| format!("tts audio send: {error}"))?; - sequence = sequence.wrapping_add(1); - timestamp_48k = timestamp_48k.wrapping_add(super::jitter::FRAME_TIMESTAMP_DELTA); } message = ws_rx.next() => { match message { @@ -481,14 +373,7 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String output_device_name, } = args; - let mut encoder = opus::Encoder::new(48000, opus::Channels::Mono, opus::Application::Voip) - .map_err(|e| format!("opus encoder: {e}"))?; - encoder - .set_bitrate(opus::Bitrate::Bits(32000)) - .map_err(|e| format!("opus bitrate: {e}"))?; - encoder - .set_dtx(true) - .map_err(|e| format!("opus dtx: {e}"))?; + let encoder = AudioEncoder::new()?; let sink_handle = super::audio_output::open_output_sink_by_name(output_device_name.as_deref())?; @@ -498,16 +383,7 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String let cancel_send = cancel.clone(); let send_task = tokio::spawn(async move { - use super::wire::{audio_level_dbov, FrameHeader, V2_HEADER_LEN}; let mut encoder = encoder; // Move encoder into task. - const FRAME_SAMPLES: usize = 960; - let mut out_buf = vec![0u8; 4000]; - // Per-frame wire-protocol state. We send v2 frames now: each Opus - // payload is preceded by an 8-byte header carrying our own seq + - // 48 kHz timestamp + audio level + flags. - let mut seq: u16 = 0; - let mut ts_48k: u32 = 0; - loop { let pcm_bytes = { use futures_util::future::Either; @@ -530,47 +406,15 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String let mut tx = ws_tx_send.lock().await; for chunk in samples.chunks(FRAME_SAMPLES) { - // dBov is computed from the pre-encode PCM. Opus DTX may - // produce a 1-2 byte comfort packet; computing level from - // the encoded payload would be meaningless. - let level = audio_level_dbov(chunk); - let encode_result = if chunk.len() == FRAME_SAMPLES { - encoder.encode_float(chunk, &mut out_buf) - } else { - let mut padded = chunk.to_vec(); - padded.resize(FRAME_SAMPLES, 0.0); - encoder.encode_float(&padded, &mut out_buf) - }; - let n = match encode_result { - Ok(n) => n, - Err(e) => { - eprintln!("buzz-desktop: opus encode error: {e}"); + let frame = match encoder.encode(chunk) { + Ok(frame) => frame, + Err(error) => { + eprintln!("buzz-desktop: opus encode error: {error}"); continue; } }; - if n > 0 { - // Opus DTX packets are very small (≤2 bytes). Flag them - // explicitly so the receiver can elide DTX from speaker - // detection without re-parsing the Opus payload. - let flags = if n <= 2 { super::wire::FLAG_DTX } else { 0 }; - let header = FrameHeader { - seq, - ts_48k, - level_dbov: level, - flags, - } - .encode(); - - // Build the v2 wire frame: 8-byte header + Opus payload. - let mut frame = Vec::with_capacity(V2_HEADER_LEN + n); - frame.extend_from_slice(&header); - frame.extend_from_slice(&out_buf[..n]); - if tx.send(WsMsg::Binary(frame.into())).await.is_err() { - return; // WS closed. - } - - seq = seq.wrapping_add(1); - ts_48k = ts_48k.wrapping_add(super::jitter::FRAME_TIMESTAMP_DELTA); + if tx.send(WsMsg::Binary(frame.into())).await.is_err() { + return; // WS closed. } } } diff --git a/desktop/src-tauri/src/huddle/wire.rs b/desktop/src-tauri/src/huddle/wire.rs index bcf9c007c2..e872d70aff 100644 --- a/desktop/src-tauri/src/huddle/wire.rs +++ b/desktop/src-tauri/src/huddle/wire.rs @@ -1,346 +1,2 @@ -//! Huddle audio wire protocol — versioning + frame header. -//! -//! ## v1 (legacy) -//! -//! Client → relay: `` -//! Relay → client: `` -//! -//! No per-frame metadata; receiver synthesizes sequence/timestamp on arrival. -//! Kept for backward compatibility — relay still admits v1 clients into -//! v1-pinned rooms — but new clients speak v2 while deployed relays remain -//! capped at the released v2 contract. -//! -//! ## v2 (compatibility contract) -//! -//! Client → relay: `` -//! Relay → client: `` -//! -//! Protocol v2 does not carry v3's occupancy epoch in media frames. The -//! control-plane roster still resets decoder and playout state when an index is -//! reassigned, but v2 cannot fence a delayed packet from the previous occupant -//! after that reassignment. -//! -//! Header layout (8 bytes, network byte order, big-endian): -//! -//! ```text -//! byte 0..=1 : seq u16 wrapping sequence number, +1 per packet -//! byte 2..=5 : ts_48k u32 sender RTP-style 48 kHz media time, -//! +960 per 20 ms encoded frame -//! byte 6 : level_dbov i8 audio level in dBov, range [-127, 0] -//! byte 7 : flags u8 bit 0 = DTX/comfort frame; other bits -//! MUST be ignored on decode -//! ``` -//! -//! Notes for reviewers (Max's spec): -//! * Header is fixed-size. No extension mechanism in v2 — future fields go -//! into a v3 with a new pinned version. -//! * `level_dbov` is client-authored telemetry. The relay parses it for -//! logging/active-speaker hints, clamps invalid values into range, and -//! **never** uses it for trust decisions (admission, moderation, etc.). -//! * Negotiation lives in the WS auth message (`protocol_version: 2`), not -//! in any bit of `flags`. Mixed-version rooms are rejected at the relay -//! with `upgrade_required`. - -/// Wire protocol version this client speaks. Bumped only when the frame -/// layout itself changes; the relay tracks pinned per-room. -pub const PROTOCOL_VERSION: u8 = 2; - -/// Length of the v2 per-frame header in bytes. -pub const V2_HEADER_LEN: usize = 8; - -/// Flag bit indicating a DTX / comfort-noise frame. Receivers MAY skip -/// arrival accounting for these; encoders MUST set it for any Opus packet -/// they tag as DTX. Unset on normal speech frames. -pub const FLAG_DTX: u8 = 0x01; - -/// Reserved bit-mask. Any bits outside this set MUST be ignored by the -/// receiver — they're available for the next minor protocol bump within v2. -#[allow(dead_code)] // Public constant for diagnostics / future flag additions. -pub const RESERVED_FLAG_MASK: u8 = !FLAG_DTX; - -/// Parsed view of one v2 header. Cheap (Copy). -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct FrameHeader { - /// Sender-authored sequence number; wraps every 2^16 frames (~21.8 min - /// at 50 Hz). - pub seq: u16, - /// Sender-authored 48 kHz RTP-style media timestamp. - pub ts_48k: u32, - /// Audio level in dBov. Always parsed into the canonical `-127..=0` - /// range — out-of-range inputs are clamped to `-127` (silence). - pub level_dbov: i8, - /// Raw flags byte. Use [`Self::is_dtx`] / explicit masks; bits outside - /// `FLAG_DTX | reserved` are not inspected. - pub flags: u8, -} - -impl FrameHeader { - /// Encode `self` into 8 bytes, network byte order. Suitable for direct - /// concatenation onto an Opus payload. - pub fn encode(self) -> [u8; V2_HEADER_LEN] { - let mut out = [0u8; V2_HEADER_LEN]; - out[0..2].copy_from_slice(&self.seq.to_be_bytes()); - out[2..6].copy_from_slice(&self.ts_48k.to_be_bytes()); - out[6] = self.level_dbov as u8; - out[7] = self.flags; - out - } - - /// Parse a v2 header from the leading 8 bytes of `bytes`. Returns - /// `(header, remainder)` so callers can pass the remainder straight to - /// the Opus decoder. - /// - /// `level_dbov` is clamped into `-127..=0`; values outside that range - /// are coerced to `-127`. Per Max's spec, malformed VU metadata must - /// *not* cause the audio frame to be dropped — only the metric is - /// suppressed. - pub fn parse(bytes: &[u8]) -> Option<(Self, &[u8])> { - if bytes.len() < V2_HEADER_LEN { - return None; - } - let seq = u16::from_be_bytes([bytes[0], bytes[1]]); - let ts_48k = u32::from_be_bytes([bytes[2], bytes[3], bytes[4], bytes[5]]); - let raw_level = bytes[6] as i8; - let level_dbov = if (-127..=0).contains(&raw_level) { - raw_level - } else { - // Out-of-range telemetry is coerced to "silent floor"; the - // audio packet is still delivered. This matches Max's "don't - // drop audio on bad VU" guidance. - -127 - }; - let flags = bytes[7]; - Some(( - Self { - seq, - ts_48k, - level_dbov, - flags, - }, - &bytes[V2_HEADER_LEN..], - )) - } - - /// True if [`FLAG_DTX`] is set. - #[allow(dead_code)] - pub fn is_dtx(&self) -> bool { - (self.flags & FLAG_DTX) != 0 - } -} - -/// Parse a complete relay-to-client v2 frame. -/// -/// The released v2 contract has exactly one relay-authored prefix byte: the -/// sender's peer index. A non-empty Opus payload must follow the fixed header. -pub fn parse_relay_frame(bytes: &[u8]) -> Option<(u8, FrameHeader, &[u8])> { - let (&peer_index, framed_audio) = bytes.split_first()?; - let (header, opus_payload) = FrameHeader::parse(framed_audio)?; - if opus_payload.is_empty() { - return None; - } - Some((peer_index, header, opus_payload)) -} - -/// Compute a dBov audio level for a normalized f32 PCM frame. -/// -/// "dBov" is RMS expressed in dB relative to full scale (where full scale = -/// peak amplitude 1.0), clamped into the v2 header's canonical `-127..=0` -/// range. A silent frame returns `-127`; full-scale ±1.0 returns `0`. -/// -/// This is cheap (one pass over the samples, one log) and runs once per -/// 20 ms frame, so it's nowhere near the audio fast-path budget. -pub fn audio_level_dbov(samples: &[f32]) -> i8 { - if samples.is_empty() { - return -127; - } - let mean_square: f64 = samples - .iter() - .map(|&s| { - let v = s as f64; - v * v - }) - .sum::() - / samples.len() as f64; - if mean_square <= 0.0 { - return -127; - } - let rms = mean_square.sqrt(); - let db = 20.0 * rms.log10(); - // Clamp into the canonical range. -127 = silence floor, 0 = full scale. - if !db.is_finite() || db <= -127.0 { - -127 - } else if db >= 0.0 { - 0 - } else { - db.round() as i8 - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn round_trip_preserves_fields() { - let h = FrameHeader { - seq: 0xABCD, - ts_48k: 0x12_34_56_78, - level_dbov: -40, - flags: FLAG_DTX, - }; - let bytes = h.encode(); - let (parsed, tail) = FrameHeader::parse(&bytes).expect("parse"); - assert_eq!(parsed, h); - assert!(tail.is_empty()); - assert!(parsed.is_dtx()); - } - - /// Header is fixed 8 bytes; anything shorter is rejected. - #[test] - fn parse_rejects_short_input() { - for len in 0..V2_HEADER_LEN { - let buf = vec![0u8; len]; - assert!( - FrameHeader::parse(&buf).is_none(), - "{len} bytes must be rejected", - ); - } - } - - /// Trailing bytes pass through as the remainder, which is what the - /// receive path hands to the Opus decoder. - #[test] - fn parse_returns_payload_remainder() { - let mut buf = FrameHeader { - seq: 1, - ts_48k: 960, - level_dbov: -20, - flags: 0, - } - .encode() - .to_vec(); - buf.extend_from_slice(b"opus-bytes"); - let (_h, tail) = FrameHeader::parse(&buf).expect("parse"); - assert_eq!(tail, b"opus-bytes"); - } - - #[test] - fn relay_frame_uses_the_v2_one_byte_peer_prefix() { - let header = FrameHeader { - seq: 0x0102, - ts_48k: 960, - level_dbov: -20, - flags: 0, - }; - let mut frame = vec![7]; - frame.extend_from_slice(&header.encode()); - frame.extend_from_slice(b"opus"); - - let (peer_index, parsed_header, opus_payload) = - parse_relay_frame(&frame).expect("valid v2 relay frame"); - assert_eq!(peer_index, 7); - assert_eq!(parsed_header, header); - assert_eq!(opus_payload, b"opus"); - } - - #[test] - fn relay_frame_rejects_a_missing_opus_payload() { - let mut frame = vec![7]; - frame.extend_from_slice( - &FrameHeader { - seq: 1, - ts_48k: 960, - level_dbov: -20, - flags: 0, - } - .encode(), - ); - - assert!(parse_relay_frame(&frame).is_none()); - } - - /// Bytes in big-endian network order, matching Max's spec. This pins - /// the byte layout against accidental endianness changes. - #[test] - fn encoded_byte_order_is_network_byte_order() { - let h = FrameHeader { - seq: 0x0102, - ts_48k: 0x0304_0506, - level_dbov: -1, - flags: 0, - }; - let bytes = h.encode(); - assert_eq!(bytes[0..2], [0x01, 0x02]); - assert_eq!(bytes[2..6], [0x03, 0x04, 0x05, 0x06]); - assert_eq!(bytes[6], 0xFF); // -1 as i8 -> 0xFF - assert_eq!(bytes[7], 0x00); - } - - /// Per Max: malformed `level_dbov` must not drop the frame; only the - /// metric is suppressed. We coerce to -127 (silence floor) and keep - /// the payload available. - #[test] - fn out_of_range_level_dbov_is_clamped_not_rejected() { - // i8(127) is +127 which is outside [-127, 0]. - let mut bytes = FrameHeader { - seq: 7, - ts_48k: 7 * 960, - level_dbov: 0, - flags: 0, - } - .encode(); - bytes[6] = 0x7F; // raw +127 — out of range - let (parsed, _) = FrameHeader::parse(&bytes).expect("parse must succeed"); - assert_eq!(parsed.level_dbov, -127, "clamped to silence floor"); - assert_eq!(parsed.seq, 7, "other fields preserved"); - } - - #[test] - fn audio_level_dbov_clamps_silence_to_minus_127() { - assert_eq!(audio_level_dbov(&[]), -127); - assert_eq!(audio_level_dbov(&[0.0_f32; 960]), -127); - } - - #[test] - fn audio_level_dbov_full_scale_returns_zero() { - let frame: Vec = (0..960).map(|_| 1.0_f32).collect(); - assert_eq!(audio_level_dbov(&frame), 0); - } - - #[test] - fn audio_level_dbov_is_in_canonical_range_for_realistic_speech() { - // A 1 kHz sine at amplitude 0.3 — typical normalized speech peak. - // RMS ≈ 0.3 / sqrt(2) ≈ 0.212 → ≈ -13 dBov. - let frame: Vec = (0..960) - .map(|i| { - let t = i as f32 / 48_000.0; - 0.3 * (2.0 * std::f32::consts::PI * 1_000.0 * t).sin() - }) - .collect(); - let db = audio_level_dbov(&frame); - assert!( - (-20..=-8).contains(&db), - "expected -20..=-8 dBov for 0.3-amp sine, got {db}", - ); - } - - /// Reserved flag bits MUST be ignored on parse (no error, just preserved - /// for forward-compat). The contract is that *receivers* don't inspect - /// them; the parse path keeps them so introspection still works. - #[test] - fn reserved_flag_bits_are_preserved_for_inspection() { - let mut bytes = FrameHeader { - seq: 0, - ts_48k: 0, - level_dbov: 0, - flags: 0, - } - .encode(); - bytes[7] = 0b1010_1010; // FLAG_DTX bit clear, reserved bits set - let (parsed, _) = FrameHeader::parse(&bytes).expect("parse"); - assert_eq!(parsed.flags, 0b1010_1010); - assert!(!parsed.is_dtx()); - // Caller can still check which reserved bits were set if needed. - assert_eq!(parsed.flags & RESERVED_FLAG_MASK, 0b1010_1010); - } -} +//! Shared Huddle wire implementation. +pub use buzz_audio_client::wire::*; diff --git a/desktop/src-tauri/src/managed_agents/discovery/bounded_command.rs b/desktop/src-tauri/src/managed_agents/discovery/bounded_command.rs index 18286d62b1..16eb01231b 100644 --- a/desktop/src-tauri/src/managed_agents/discovery/bounded_command.rs +++ b/desktop/src-tauri/src/managed_agents/discovery/bounded_command.rs @@ -692,7 +692,10 @@ mod tests { ); let mut cmd = Command::new("/bin/sh"); cmd.args(["-c", &script]); - let result = run_watchdogged(cmd, Duration::from_millis(300), Duration::from_secs(5)); + // Allow cold Perl/POSIX startup under parallel test load. Startup speed is + // not the contract: an escaped writer holds the pipe for 300 seconds, while + // the independent five-second watchdog still bounds drain completion. + let result = run_watchdogged(cmd, Duration::from_secs(2), Duration::from_secs(5)); assert!( result.is_none(), "a timed-out probe must fail closed even when an escaped writer holds the pipe"