jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Extract shared Huddle audio client for desktop consumers (#8129)

Browse Source 
Huddle audio was embedded in the v0 desktop client, forcing other
clients to duplicate its protocol and codec behavior. This extracts
`buzz-audio-client` so v1 can depend on a pinned library revision
without building the v0 application.

The crate owns the bounded v2 audio handshake, roster parsing, wire
format, Opus sender, and per-peer NetEQ jitter buffer. The v0 human and
agent senders use it, and the desktop retains room admission,
identity/signing, TLS policy, device routing, mute, and teardown.
Receive decoding preserves v1's support for 120 ms Opus packets. No
relay protocol changes, speech models, or model downloads are included.
Docker builder stages include CMake because cargo-chef compiles the
workspace dependency graph.

Dependency security: Opus 0.4 replaces the unmaintained `audiopus_sys`
backend with `opusic-sys`. NetEQ is pinned exactly to 0.8.3 because its
published Rust sources never reference its declared ringbuf dependency.
RUSTSEC-2026-0293 is documented as unreachable for this audited source,
and a wrapper restriction rejects every other consumer of vulnerable
ringbuf versions. Independent review confirmed the scope;
re-audit/remove the exception when changing NetEQ. NetEQ 0.9.1 still
declares ringbuf 0.3, so a version bump alone does not remove it. Local
`cargo deny check` passes with this constrained exception.

Validation:
- All 21 shared-crate tests pass, including real loopback WebSocket
authentication/cancellation, real Opus-to-NetEQ playback, malformed
input, and 120 ms decoding. These tests are included in `just
test-unit`.
- Shared-crate, root workspace, and desktop Tauri Clippy pass; `just
desktop-tauri-check` passes.
- The desktop Tauri test workspace passes; native two-person
listening/device checks remain separate.
- All mandatory push checks pass at `d06bac5dc`, including the complete
Rust unit gate and desktop workspace checks. Existing agent timing and
relay log-capture tests failed under concurrent load, passed separately,
and passed in the full serial test run; no tests or hooks were skipped.
- One test-only fixture change allows cold Perl startup before the
escaped-process test's inner timeout (2 seconds instead of 300 ms). Its
independent 5-second watchdog, 300-second surviving pipe writer, and all
assertions remain intact. All eight process-helper tests pass;
independent review confirmed this still tests bounded drain completion.
- Dependency inspection confirms the library brings in no desktop
application, Tauri, identity store, relay server, or speech-model
crates.
- Independent source review found no correctness blockers.

Draft: a native two-person call, mute/leave/cancellation, and
microphone/speaker switching still need a runtime tryout and human
review. Full `just ci` was stopped during the unrelated mobile toolchain
download because of disk pressure; it is not a passing gate. Consumer
integration is tracked in block/buzz-app#489.

---------

Signed-off-by: kenny lopez <klopez4212@gmail.com>
This commit is contained in:
klopez4212
2026-10-13 16:24:27 -04:00
committed by GitHub
parent d6dc813290
commit 5dc693aeaf
23 changed files with 1356 additions and 885 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
Cargo.lock generated
View File

@@ -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",

1
Cargo.toml
View File

@@ -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",

1
Dockerfile
View File

@@ -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 \

2
Dockerfile.push-gateway
View File

@@ -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"

2
Justfile
View File

@@ -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

122
crates/buzz-acp/src/acp.rs
View File

@@ -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]

39
crates/buzz-agent/tests/fake_llm.rs
View File

@@ -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;
}

25
crates/buzz-audio-client/Cargo.toml Normal file
View File

@@ -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"] }

42
crates/buzz-audio-client/README.md Normal file
View File

@@ -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.

165
crates/buzz-audio-client/src/connection.rs Normal file
View File

@@ -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<MaybeTlsStream<tokio::net::TcpStream>>;
/// 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<Peer>,
}
/// Parse a bounded roster without truncating out-of-range indices or epochs.
pub fn parse_peers(value: &Value) -> Result<Vec<Peer>, 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<Value, String> {
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<F, Fut>(
audio_url: &str,
parent: Option<&str>,
connector: Option<Connector>,
sign: F,
) -> Result<Joined, String>
where
F: FnOnce(String) -> Fut,
Fut: Future<Output = Result<Value, String>>,
{
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")?
}

65
crates/buzz-audio-client/src/encoder.rs Normal file
View File

@@ -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<u8>,
}
impl AudioEncoder {
/// Create the standard 32 kbit/s VoIP encoder with discontinuous transmission.
pub fn new() -> Result<Self, String> {
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<Vec<u8>, 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)
}
}

285
crates/buzz-audio-client/src/jitter.rs Normal file
View File

@@ -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<f32>,
}
impl OpusFrameDecoder {
fn new(sample_rate: u32, channels: u8) -> Result<Self, String> {
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<Vec<f32>, 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<Self, NetEqError> {
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<f32>, 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<u8> {
// 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)",
);
}
}

11
crates/buzz-audio-client/src/lib.rs Normal file
View File

@@ -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;

346
crates/buzz-audio-client/src/wire.rs Normal file
View File

@@ -0,0 +1,346 @@
//! Huddle audio wire protocol — versioning + frame header.
//!
//! ## v1 (legacy)
//!
//! Client → relay: `<opus_bytes>`
//! Relay → client: `<peer_index: u8><opus_bytes>`
//!
//! 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: `<header: [u8; 8]><opus_bytes>`
//! Relay → client: `<peer_index: u8><header: [u8; 8]><opus_bytes>`
//!
//! 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::<f64>()
/ 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<f32> = (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<f32> = (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);
}
}

56
crates/buzz-audio-client/tests/codec.rs Normal file
View File

@@ -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<f32> = (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::<f32>();
}
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
);
}

133
crates/buzz-audio-client/tests/connection.rs Normal file
View File

@@ -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::<Value>(&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());
}

9
deny.toml
View File

@@ -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" },
]

45
desktop/src-tauri/Cargo.lock generated
View File

@@ -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",

3
desktop/src-tauri/Cargo.toml
View File

@@ -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" }

269
desktop/src-tauri/src/huddle/jitter.rs
View File

@@ -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<f32>,
}
impl OpusFrameDecoder {
fn new(sample_rate: u32, channels: u8) -> Result<Self, String> {
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<Vec<f32>, 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<Self, NetEqError> {
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<f32>, 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<u8> {
// 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::*;

214
desktop/src-tauri/src/huddle/relay_api.rs
View File

@@ -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, String> {
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<std::sync::atomic::AtomicU64>,
speaker_generation: Arc<std::sync::atomic::AtomicU64>,
) -> 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::<QueuedTtsFrame>::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.
}
}
}

348
desktop/src-tauri/src/huddle/wire.rs
View File

@@ -1,346 +1,2 @@
//! Huddle audio wire protocol — versioning + frame header.
//!
//! ## v1 (legacy)
//!
//! Client → relay: `<opus_bytes>`
//! Relay → client: `<peer_index: u8><opus_bytes>`
//!
//! 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: `<header: [u8; 8]><opus_bytes>`
//! Relay → client: `<peer_index: u8><header: [u8; 8]><opus_bytes>`
//!
//! 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::<f64>()
/ 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<f32> = (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<f32> = (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::*;

5
desktop/src-tauri/src/managed_agents/discovery/bounded_command.rs
View File

@@ -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"