diff --git a/desktop/src-tauri/src/huddle/playout.rs b/desktop/src-tauri/src/huddle/playout.rs index 5bfce3adad5..4371d8a1313 100644 --- a/desktop/src-tauri/src/huddle/playout.rs +++ b/desktop/src-tauri/src/huddle/playout.rs @@ -33,7 +33,7 @@ use tokio_util::sync::CancellationToken; use super::human_floor::HumanFloor; use super::jitter::{PeerJitterBuffer, SAMPLE_RATE_HZ}; use super::relay_api::{WsStream, REMOTE_SPEECH_THRESHOLD}; -use super::wire::{parse_relay_frame, FLAG_DTX}; +use super::wire::{FrameHeader, FLAG_DTX, V2_HEADER_LEN}; /// Speaker-tick window for emitting `huddle-active-speakers`. Active set is /// cleared each tick — peers that didn't send a frame in the last window are @@ -149,11 +149,18 @@ fn is_agent_peer( }) } -/// Whether `peer_idx` is currently occupied per the authoritative roster. -/// Protocol v2 media carries only the peer index, so roster presence is the -/// strongest routing boundary available until the relay supports v3 epochs. -fn is_current_occupant(peer_idx: u8, index_to_epoch: &std::collections::HashMap) -> bool { - index_to_epoch.contains_key(&peer_idx) +/// Whether `peer_idx` is currently occupied at exactly `epoch`, per the +/// authoritative roster. A frame is deliverable only when both match: an index +/// absent from the roster is stale, and a slot reused by a later occupant has +/// advanced its epoch, so a departed occupant's in-flight frame is fenced +/// rather than mis-attributed to the new occupant. A legacy relay omits the +/// epoch, which degrades to `0` on both sides, making the fence a no-op. +fn is_current_occupant( + peer_idx: u8, + epoch: u8, + index_to_epoch: &std::collections::HashMap, +) -> bool { + index_to_epoch.get(&peer_idx) == Some(&epoch) } fn same_occupancy( @@ -189,7 +196,7 @@ fn f32_samples_to_le_bytes(samples: &[f32]) -> Vec { /// One remote peer's slot: jitter buffer + dedicated rodio Player. /// /// Per-frame seq/timestamp come from the v2 wire header (sender-authored). -/// The relay forwards `peer_index | header | opus_bytes` opaquely; we +/// The relay forwards `peer_index | epoch | header | opus_bytes` opaquely; we /// parse the header here and pass the sender's own monotonic seq + 48 kHz media /// timestamp into NetEq. struct PeerSlot { @@ -415,19 +422,22 @@ pub(crate) async fn run_playout_recv_loop( msg = ws_rx.next() => { match msg { Some(Ok(WsMsg::Binary(data))) => { - // Wire shape (v2): [peer_index: u8][header: 8 bytes][opus payload...] - // The minimum size is 1 (peer index) + 8 (header) + ≥1 Opus byte. - let Some((peer_idx, header, opus_bytes)) = parse_relay_frame(&data) else { - eprintln!( - "buzz-desktop: dropping malformed v2 audio relay frame ({} bytes)", - data.len(), - ); + // Wire shape (v2): [peer_index: u8][epoch: u8][header: 8 bytes][opus payload...] + // The minimum size is 2 (peer_index + epoch) + 8 (header) + ≥1 Opus byte. + if data.len() <= 2 + V2_HEADER_LEN { continue; - }; - // Protocol v2 has no media epoch. Drop frames for slots - // absent from the control roster; delayed frames after - // an index is reassigned cannot be fenced until v3. - if !is_current_occupant(peer_idx, &index_to_epoch) { + } + let peer_idx = data[0]; + let epoch = data[1]; + // Fence the peer-index reuse race: a frame authored by a + // departed occupant that arrives after its index is + // reassigned carries the old epoch. Drop it rather than + // mis-attribute stale audio (and the new occupant's + // human/agent STT policy) to whoever grabbed the index. + // An index absent from the roster is also stale. A slot + // with no known epoch (legacy relay) degrades to 0 on + // both sides, so the fence is a no-op there. + if !is_current_occupant(peer_idx, epoch, &index_to_epoch) { continue; } // Suppress only an agent stream synthesized and @@ -436,6 +446,21 @@ pub(crate) async fn run_playout_recv_loop( if is_locally_synthesized_peer(peer_idx, &local_tts_publishers) { continue; } + let after_idx = &data[2..]; + let Some((header, opus_bytes)) = FrameHeader::parse(after_idx) + else { + // Malformed v2 frame: header parse only fails when + // the slice is too short, which `if data.len() <= ...` + // already guards. Defensive log + drop. + eprintln!( + "buzz-desktop: dropping malformed audio frame from peer {peer_idx} ({} bytes)", + data.len(), + ); + continue; + }; + if opus_bytes.is_empty() { + continue; + } let is_dtx = (header.flags & FLAG_DTX) != 0; // Only count non-DTX arrivals toward the UI's // active-speaker set. DTX/comfort packets are emitted @@ -746,16 +771,34 @@ mod tests { ); } + /// Causal regression for the peer-index reuse race (Jude's blocking + /// finding): a frame authored by a departed occupant that arrives after + /// its slot is reassigned to a new occupant carries the stale epoch and + /// must be fenced, never mis-attributed to the new occupant. #[test] - fn v2_media_is_routed_only_for_current_roster_indices() { + fn stale_epoch_frame_is_fenced_after_its_index_is_reused() { let mut index_to_epoch = std::collections::HashMap::new(); + // Slot 3 first occupied at epoch 0. index_to_epoch.insert(3_u8, 0_u8); assert!( - is_current_occupant(3, &index_to_epoch), + is_current_occupant(3, 0, &index_to_epoch), "current occupant's frame is delivered" ); + + // The occupant departs and a new peer reuses slot 3 at epoch 1. + index_to_epoch.insert(3, 1); + assert!( + !is_current_occupant(3, 0, &index_to_epoch), + "in-flight frame from the departed occupant (epoch 0) is fenced" + ); + assert!( + is_current_occupant(3, 1, &index_to_epoch), + "the new occupant's frame (epoch 1) is delivered" + ); + + // A frame for an index absent from the roster is stale. assert!( - !is_current_occupant(9, &index_to_epoch), + !is_current_occupant(9, 0, &index_to_epoch), "frame for an unoccupied index is dropped" ); } diff --git a/desktop/src-tauri/src/huddle/relay_api.rs b/desktop/src-tauri/src/huddle/relay_api.rs index 190397aa054..20a2be57652 100644 --- a/desktop/src-tauri/src/huddle/relay_api.rs +++ b/desktop/src-tauri/src/huddle/relay_api.rs @@ -114,9 +114,6 @@ async fn connect_authenticated_audio_socket( "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 diff --git a/desktop/src-tauri/src/huddle/wire.rs b/desktop/src-tauri/src/huddle/wire.rs index bcf9c007c2d..518377a60b0 100644 --- a/desktop/src-tauri/src/huddle/wire.rs +++ b/desktop/src-tauri/src/huddle/wire.rs @@ -7,18 +7,25 @@ //! //! 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. +//! v1-pinned rooms — but new clients always speak v3. //! -//! ## v2 (compatibility contract) +//! ## v2 (released) //! //! 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. +//! ## v3 (this commit) +//! +//! Client → relay: `` +//! Relay → client: `` +//! +//! The relay prefixes each forwarded frame with the sender's stable +//! `peer_index` and the current occupancy `epoch` of that index. The epoch +//! advances each time a slot is reused by a new occupant, so a client can +//! fence a frame authored by a departed occupant that arrives after its index +//! is reassigned — it carries the stale epoch and is dropped rather than +//! mis-attributed. The client's own send path is unaffected: it emits only +//! `
` and the relay stamps the prefix. //! //! Header layout (8 bytes, network byte order, big-endian): //! @@ -37,13 +44,13 @@ //! * `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 +//! * Negotiation lives in the WS auth message (`protocol_version: 3`), 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; +pub const PROTOCOL_VERSION: u8 = 3; /// Length of the v2 per-frame header in bytes. pub const V2_HEADER_LEN: usize = 8; @@ -128,19 +135,6 @@ impl FrameHeader { } } -/// 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 = @@ -224,41 +218,6 @@ mod tests { 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] diff --git a/desktop/src-tauri/src/managed_agents/discovery.rs b/desktop/src-tauri/src/managed_agents/discovery.rs index 78592357c9b..89489950bb7 100644 --- a/desktop/src-tauri/src/managed_agents/discovery.rs +++ b/desktop/src-tauri/src/managed_agents/discovery.rs @@ -125,7 +125,7 @@ const KNOWN_ACP_RUNTIMES: &[KnownAcpRuntime] = &[ commands: &["claude-agent-acp", "claude-code-acp"], aliases: &["claude-code", "claudecode"], avatar_url: CLAUDE_CODE_AVATAR_URL, - mcp_command: None, + mcp_command: Some("buzz-dev-mcp"), mcp_hooks: false, underlying_cli: Some("claude"), cli_install_commands: &["curl -fsSL https://claude.ai/install.sh | bash"], diff --git a/desktop/src-tauri/src/managed_agents/runtime/tests.rs b/desktop/src-tauri/src/managed_agents/runtime/tests.rs index 8bedfe53207..7a2f3a5913f 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/tests.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/tests.rs @@ -106,6 +106,18 @@ fn codex_has_mcp_command() { assert_eq!(p.mcp_command, Some("buzz-dev-mcp")); } +#[test] +fn claude_has_mcp_command() { + let p = known_acp_runtime("claude-agent-acp").expect("should resolve"); + assert!(!p.mcp_hooks, "claude does not handle MCP_HOOK_SERVERS"); + assert_eq!( + p.mcp_command, + Some("buzz-dev-mcp"), + "Claude Code needs the MCP server wired through session/new \ + because it does not support native config" + ); +} + #[test] fn goose_has_no_mcp_hooks() { let p = known_acp_runtime("goose").expect("should resolve");