diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 4a82cf6306d..cc96ed3f748 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -35,6 +35,22 @@ pub(crate) const DEFAULT_MAX_TURN_DURATION_SECS: u64 = 7200; /// deadline (`max_turn_duration + IN_FLIGHT_DEADLINE_BUFFER_SECS`). pub(crate) const MAX_TURN_DURATION_CEILING_SECS: u64 = 604_800; +/// Event kinds that open agent turns unless an operator supplies an override. +/// +/// `mentions` and `all` differ only in whether a `p` tag is required; both +/// modes subscribe to and match this same bounded set by default. +pub(crate) fn default_subscription_kinds() -> Vec { + use buzz_core::kind::{ + KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, + }; + + vec![ + KIND_STREAM_MESSAGE, + KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_STREAM_REMINDER, + ] +} + #[derive(Debug, Error)] pub enum ConfigError { #[error("failed to parse nostr keys: {0}")] @@ -1276,10 +1292,6 @@ pub fn resolve_channel_filters( discovered_channels: &[Uuid], rules: &[SubscriptionRule], ) -> HashMap { - use buzz_core::kind::{ - KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, - }; - let target_channels: Vec = if let Some(ref overrides) = config.channels_override { overrides .iter() @@ -1294,13 +1306,10 @@ pub fn resolve_channel_filters( match config.subscribe_mode { SubscribeMode::Mentions => { - let kinds = config.kinds_override.clone().unwrap_or_else(|| { - vec![ - KIND_STREAM_MESSAGE, - KIND_WORKFLOW_APPROVAL_REQUESTED, - KIND_STREAM_REMINDER, - ] - }); + let kinds = config + .kinds_override + .clone() + .unwrap_or_else(default_subscription_kinds); let require_mention = !config.no_mention_filter; for ch in &target_channels { result.insert( @@ -1313,11 +1322,15 @@ pub fn resolve_channel_filters( } } SubscribeMode::All => { + let kinds = config + .kinds_override + .clone() + .unwrap_or_else(default_subscription_kinds); for ch in &target_channels { result.insert( *ch, ChannelFilter { - kinds: config.kinds_override.clone(), + kinds: Some(kinds.clone()), require_mention: false, }, ); @@ -1378,10 +1391,6 @@ pub fn resolve_dynamic_channel_filter( channel_id: Uuid, rules: &[crate::filter::SubscriptionRule], ) -> Option { - use buzz_core::kind::{ - KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, - }; - // In Mentions/All mode, if the operator explicitly constrained channels // with --channels, only allow dynamic subscription to channels in that // allowlist. Config mode ignores --channels (per CLI contract) and uses @@ -1399,17 +1408,21 @@ pub fn resolve_dynamic_channel_filter( match config.subscribe_mode { SubscribeMode::Mentions => Some(ChannelFilter { - kinds: Some(config.kinds_override.clone().unwrap_or_else(|| { - vec![ - KIND_STREAM_MESSAGE, - KIND_WORKFLOW_APPROVAL_REQUESTED, - KIND_STREAM_REMINDER, - ] - })), + kinds: Some( + config + .kinds_override + .clone() + .unwrap_or_else(default_subscription_kinds), + ), require_mention: !config.no_mention_filter, }), SubscribeMode::All => Some(ChannelFilter { - kinds: config.kinds_override.clone(), + kinds: Some( + config + .kinds_override + .clone() + .unwrap_or_else(default_subscription_kinds), + ), require_mention: false, }), SubscribeMode::Config => { @@ -1804,19 +1817,22 @@ mod tests { } #[test] - fn test_all_mode_wildcard() { + fn test_all_mode_defaults_to_message_kinds() { + // SubscribeMode::All should mean "same kinds as Mentions, without + // the mention requirement" — not "all event kinds" (wildcard). + // Without this default, typing indicators (kind 20002) and other + // ephemeral kinds open agent turns and cancel pending wakeups. + // See #4949. let config = test_config(SubscribeMode::All); let channels = vec![Uuid::new_v4(), Uuid::new_v4(), Uuid::new_v4()]; let result = resolve_channel_filters(&config, &channels, &[]); + let expected = default_subscription_kinds(); assert_eq!(result.len(), 3); for ch in &channels { let f = result.get(ch).unwrap(); - assert!( - f.kinds.is_none(), - "all mode with no override = wildcard kinds" - ); - assert!(!f.require_mention); + assert_eq!(f.kinds.as_deref(), Some(expected.as_slice())); + assert!(!f.require_mention, "All mode must not require mention"); } } @@ -1831,6 +1847,20 @@ mod tests { assert_eq!(f.kinds.as_ref().unwrap(), &[9, 7]); } + #[test] + fn test_all_mode_dynamic_filter_defaults_to_message_kinds() { + // resolve_dynamic_channel_filter must also default to the message + // kinds list when BUZZ_ACP_KINDS is not set, not wildcard. + // See #4949. + let config = test_config(SubscribeMode::All); + let ch = Uuid::new_v4(); + let result = resolve_dynamic_channel_filter(&config, ch, &[]); + + let f = result.expect("All mode must return a filter"); + assert_eq!(f.kinds, Some(default_subscription_kinds())); + assert!(!f.require_mention, "All mode must not require mention"); + } + #[test] fn test_channels_override_filters_to_discovered() { let mut config = test_config(SubscribeMode::All); diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 146214197a8..dee4a747ce1 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -22,7 +22,6 @@ use acp::{AcpClient, EnvVar, McpServer}; use anyhow::{ensure, Context, Result}; use buzz_core::kind::{ KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_STREAM_MESSAGE, - KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, }; use buzz_core::observer::{ decrypt_observer_payload, encrypt_observer_payload, OBSERVER_FRAME_TELEMETRY, @@ -2103,13 +2102,10 @@ async fn tokio_main() -> Result<()> { vec![SubscriptionRule { name: "mentions".into(), channels: filter::ChannelScope::All("all".into()), - kinds: config.kinds_override.clone().unwrap_or_else(|| { - vec![ - KIND_STREAM_MESSAGE, - KIND_WORKFLOW_APPROVAL_REQUESTED, - KIND_STREAM_REMINDER, - ] - }), + kinds: config + .kinds_override + .clone() + .unwrap_or_else(config::default_subscription_kinds), require_mention: !config.no_mention_filter, filter: None, compiled_filter: None, @@ -2121,7 +2117,10 @@ async fn tokio_main() -> Result<()> { vec![SubscriptionRule { name: "all".into(), channels: filter::ChannelScope::All("all".into()), - kinds: config.kinds_override.clone().unwrap_or_default(), + kinds: config + .kinds_override + .clone() + .unwrap_or_else(config::default_subscription_kinds), require_mention: false, filter: None, compiled_filter: None, 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]